smpd_handle_command.c

fortran并行计算包

		    if (ret_val == SMPD_SUCCESS)
		    {
			ret_val = smpd_post_write_command(context, cmd_ptr);
			if (ret_val == SMPD_SUCCESS)
			{
			    ret_val = SMPD_CLOSE;
			}
			else
			{
			    smpd_err_printf("unable to post a write of a done command.\n");
			}
		    }
		    else
		    {
			smpd_err_printf("unable to create a done command.\n");
		    }
		}
		else if (strcmp(iter->cmd_str, "remove_job") == 0)
		{
		    /* print the result of the remove_job command */
		    printf("%s\n", str);
		    /* close the session */
		    ret_val = smpd_create_command("done", smpd_process.id, context->id, SMPD_FALSE, &cmd_ptr);
		    if (ret_val == SMPD_SUCCESS)
		    {
			ret_val = smpd_post_write_command(context, cmd_ptr);
			if (ret_val == SMPD_SUCCESS)
			{
			    ret_val = SMPD_CLOSE;
			}
			else
			{
			    smpd_err_printf("unable to post a write of a done command.\n");
			}
		    }
		    else
		    {
			smpd_err_printf("unable to create a done command.\n");
		    }
		}
		else if (strcmp(iter->cmd_str, "associate_job") == 0)
		{
		    /* print the result of the associate_job command */
		    printf("%s\n", str);
		    /* close the session */
		    ret_val = smpd_create_command("done", smpd_process.id, context->id, SMPD_FALSE, &cmd_ptr);
		    if (ret_val == SMPD_SUCCESS)
		    {
			ret_val = smpd_post_write_command(context, cmd_ptr);
			if (ret_val == SMPD_SUCCESS)
			{
			    ret_val = SMPD_CLOSE;
			}
			else
			{
			    smpd_err_printf("unable to post a write of a done command.\n");
			}
		    }
		    else
		    {
			smpd_err_printf("unable to create a done command.\n");
		    }
		}
		else
		{
		    smpd_err_printf("result returned for unhandled '%s' command:\n command: '%s'\n result: '%s'\n", iter->cmd_str, iter->cmd, str);
		}
	    }
	    else
	    {
		smpd_err_printf("no result string in the result command.\n");
	    }
	    if (trailer == iter)
	    {
		context->wait_list = context->wait_list->next;
	    }
	    else
	    {
		trailer->next = iter->next;
	    }
	    result = smpd_free_command(iter);
	    if (result != SMPD_SUCCESS)
	    {
		smpd_err_printf("unable to free command in the wait_list\n");
	    }
	    smpd_exit_fn(FCNAME);
	    return ret_val;
	}
	else
	{
	    smpd_dbg_printf("tag %d != match_tag %d\n", iter->tag, match_tag);
	}
	if (trailer != iter)
	    trailer = trailer->next;
	iter = iter->next;
    }
    if (context->wait_list == NULL)
    {
	smpd_err_printf("result command received but the wait_list is empty.\n");
    }
    else
    {
	smpd_err_printf("result command did not match any commands in the wait_list.\n");
    }
    smpd_exit_fn(FCNAME);
    return SMPD_FAIL;
}

#undef FCNAME
#define FCNAME "get_name_key_value"
static int get_name_key_value(char *str, char *name, char *key, char *value)
{
    smpd_enter_fn(FCNAME);

    if (str == NULL)
    {
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }

    if (name != NULL)
    {
	if (MPIU_Str_get_string_arg(str, "name", name, SMPD_MAX_DBS_NAME_LEN) != MPIU_STR_SUCCESS)
	{
	    smpd_exit_fn(FCNAME);
	    return SMPD_FAIL;
	}
    }
    if (key != NULL)
    {
	if (MPIU_Str_get_string_arg(str, "key", key, SMPD_MAX_DBS_KEY_LEN) != MPIU_STR_SUCCESS)
	{
	    smpd_exit_fn(FCNAME);
	    return SMPD_FAIL;
	}
    }
    if (value != NULL)
    {
	if (MPIU_Str_get_string_arg(str, "value", value, SMPD_MAX_DBS_VALUE_LEN) != MPIU_STR_SUCCESS)
	{
	    smpd_exit_fn(FCNAME);
	    return SMPD_FAIL;
	}
    }
    smpd_exit_fn(FCNAME);
    return SMPD_SUCCESS;
}

#undef FCNAME
#define FCNAME "smpd_handle_dbs_command"
int smpd_handle_dbs_command(smpd_context_t *context)
{
    int result;
    smpd_command_t *cmd, *temp_cmd;
    char name[SMPD_MAX_DBS_NAME_LEN+1] = "";
    char key[SMPD_MAX_DBS_KEY_LEN+1] = "";
    char value[SMPD_MAX_DBS_VALUE_LEN+1] = "";
    char ctx_key[100];
    char *result_str;

    smpd_enter_fn(FCNAME);

    cmd = &context->read_cmd;

    /*
    printf("handling dbs command on %s context, sock %d.\n", smpd_get_context_str(context), MPIDU_Sock_get_sock_id(context->sock));
    fflush(stdout);
    */

    /* prepare the result command */
    result = smpd_create_command("result", smpd_process.id, cmd->src, SMPD_FALSE, &temp_cmd);
    if (result != SMPD_SUCCESS)
    {
	smpd_err_printf("unable to create a result command for the dbs command '%s'.\n", cmd->cmd);
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }
    /* add the command tag for result matching */
    result = smpd_add_command_int_arg(temp_cmd, "cmd_tag", cmd->tag);
    if (result != SMPD_SUCCESS)
    {
	smpd_err_printf("unable to add the tag to the result command for dbs command '%s'.\n", cmd->cmd);
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }
    result = smpd_add_command_arg(temp_cmd, "cmd_orig", cmd->cmd_str);
    if (result != SMPD_SUCCESS)
    {
	smpd_err_printf("unable to add cmd_orig to the result command for a %s command\n", cmd->cmd_str);
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }
    /* copy the ctx_key for pmi control channel lookup */
    if (MPIU_Str_get_string_arg(cmd->cmd, "ctx_key", ctx_key, 100) != MPIU_STR_SUCCESS)
    {
	smpd_err_printf("no ctx_key in the db command: '%s'\n", cmd->cmd);
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }
    result = smpd_add_command_arg(temp_cmd, "ctx_key", ctx_key);
    if (result != SMPD_SUCCESS)
    {
	smpd_err_printf("unable to add the ctx_key to the result command for dbs command '%s'.\n", cmd->cmd);
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }

    /* check to make sure this node is running the dbs */
    if (!smpd_process.have_dbs)
    {
	/*
	printf("havd_dbs is false for this process %s context.\n", smpd_get_context_str(context));
	fflush(stdout);
	*/

	/* create a failure reply because this node does not have an initialized database */
	smpd_dbg_printf("sending a failure reply because this node does not have an initialized database.\n");
	result = smpd_add_command_arg(temp_cmd, "result", SMPD_FAIL_STR" - smpd does not have an initialized database.");
	if (result != SMPD_SUCCESS)
	{
	    smpd_err_printf("unable to add the result string to the result command for dbs command '%s'.\n", cmd->cmd);
	    smpd_exit_fn(FCNAME);
	    return SMPD_FAIL;
	}
	smpd_dbg_printf("sending result command to %s context: \"%s\"\n", smpd_get_context_str(context), temp_cmd->cmd);
	result = smpd_post_write_command(context, temp_cmd);
	if (result != SMPD_SUCCESS)
	{
	    smpd_err_printf("unable to post a write of the result command to the context: cmd '%s', dbs cmd '%s'", temp_cmd->cmd, cmd->cmd);
	    smpd_exit_fn(FCNAME);
	    return SMPD_FAIL;
	}
	smpd_exit_fn(FCNAME);
	return SMPD_SUCCESS;
    }

    /* do the dbs request */
    if (strcmp(cmd->cmd_str, "dbput") == 0)
    {
	if (get_name_key_value(cmd->cmd, name, key, value) != SMPD_SUCCESS)
	    goto invalid_dbs_command;
	if (smpd_dbs_put(name, key, value) == SMPD_SUCCESS)
	    result_str = DBS_SUCCESS_STR;
	else
	    result_str = DBS_FAIL_STR;
    }
    else if (strcmp(cmd->cmd_str, "dbget") == 0)
    {
	if (get_name_key_value(cmd->cmd, name, key, NULL) != SMPD_SUCCESS)
	    goto invalid_dbs_command;
	if (smpd_dbs_get(name, key, value) == SMPD_SUCCESS)
	{
	    result = smpd_add_command_arg(temp_cmd, "value", value);
	    if (result != SMPD_SUCCESS)
	    {
		smpd_err_printf("unable to add the get value('%s') to the result command.\n", value);
		smpd_exit_fn(FCNAME);
		return SMPD_FAIL;
	    }
	    result_str = DBS_SUCCESS_STR;
	}
	else
	    result_str = DBS_FAIL_STR;
    }
    else if (strcmp(cmd->cmd_str, "dbcreate") == 0)
    {
	if (MPIU_Str_get_string_arg(cmd->cmd, "name", name, SMPD_MAX_DBS_NAME_LEN) == MPIU_STR_SUCCESS)
	{
	    result = smpd_dbs_create_name_in(name);
	}
	else
	{
	    result = smpd_dbs_create(name);
	}
	if (result == SMPD_SUCCESS)
	{
	    result = smpd_add_command_arg(temp_cmd, "name", name);
	    if (result != SMPD_SUCCESS)
	    {
		smpd_err_printf("unable to add the dbcreate name('%s') to the result command.\n", name);
		smpd_exit_fn(FCNAME);
		return SMPD_FAIL;
	    }
	    result_str = DBS_SUCCESS_STR;
	}
	else
	    result_str = DBS_FAIL_STR;
    }
    else if (strcmp(cmd->cmd_str, "dbdestroy") == 0)
    {
	if (get_name_key_value(cmd->cmd, name, NULL, NULL) != SMPD_SUCCESS)
	    goto invalid_dbs_command;
	if (smpd_dbs_destroy(name) == SMPD_SUCCESS)
	    result_str = DBS_SUCCESS_STR;
	else
	    result_str = DBS_FAIL_STR;
    }
    else if (strcmp(cmd->cmd_str, "dbfirst") == 0)
    {
	if (get_name_key_value(cmd->cmd, name, NULL, NULL) != SMPD_SUCCESS)
	    goto invalid_dbs_command;
	if (smpd_dbs_first(name, key, value) == SMPD_SUCCESS)
	{
	    if (*key == '\0')
	    {
		/* this can be changed to a special end_key if we don't want DBS_END_STR to be a reserved key */
		result = smpd_add_command_arg(temp_cmd, "key", DBS_END_STR);
		if (result != SMPD_SUCCESS)
		{
		    smpd_err_printf("unable to add the dbfirst key('%s') to the result command.\n", key);
		    smpd_exit_fn(FCNAME);
		    return SMPD_FAIL;
		}
	    }
	    else
	    {
		result = smpd_add_command_arg(temp_cmd, "key", key);
		if (result != SMPD_SUCCESS)
		{
		    smpd_err_printf("unable to add the dbfirst key('%s') to the result command.\n", key);
		    smpd_exit_fn(FCNAME);
		    return SMPD_FAIL;
		}
		result = smpd_add_command_arg(temp_cmd, "value", value);
		if (result != SMPD_SUCCESS)
		{
		    smpd_err_printf("unable to add the dbfirst value('%s') to the result command.\n", value);
		    smpd_exit_fn(FCNAME);
		    return SMPD_FAIL;
		}
	    }
	    result_str = DBS_SUCCESS_STR;
	}
	else
	{
	    result_str = DBS_FAIL_STR;
	}
    }
    else if (strcmp(cmd->cmd_str, "dbnext") == 0)
    {
	if (get_name_key_value(cmd->cmd, name, NULL, NULL) != SMPD_SUCCESS)
	    goto invalid_dbs_command;
	if (smpd_dbs_next(name, key, value) == SMPD_SUCCESS)
	{
	    if (*key == '\0')
	    {
		/* this can be changed to a special end_key if we don't want DBS_END_STR to be a reserved key */
		result = smpd_add_command_arg(temp_cmd, "key", DBS_END_STR);
		if (result != SMPD_SUCCESS)
		{
		    smpd_err_printf("unable to add the dbndext key('%s') to the result command.\n", key);
		    smpd_exit_fn(FCNAME);
		    return SMPD_FAIL;
		}
	    }
	    else
	    {
		result = smpd_add_command_arg(temp_cmd, "key", key);
		if (result != SMPD_SUCCESS)
		{
		    smpd_err_printf("unable to add the dbnext key('%s') to the result command.\n", key);
		    smpd_exit_fn(FCNAME);
		    return SMPD_FAIL;
		}
		result = smpd_add_command_arg(temp_cmd, "value", value);
		if (result != SMPD_SUCCESS)
		{
		    smpd_err_printf("unable to add the dbnext value('%s') to the result command.\n", value);
		    smpd_exit_fn(FCNAME);
		    return SMPD_FAIL;
		}
	    }
	    result_str = DBS_SUCCESS_STR;
	}
	else
	{
	    result_str = DBS_FAIL_STR;
	}
    }
    else if (strcmp(cmd->cmd_str, "dbfirstdb") == 0)
    {
	if (smpd_dbs_firstdb(name) == SMPD_SUCCESS)
	{
	    /* this can be changed to a special end_key if we don't want DBS_END_STR to be a reserved key */
	    if (*name == '\0')
		result = smpd_add_command_arg(temp_cmd, "name", DBS_END_STR);
	    else
		result = smpd_add_command_arg(temp_cmd, "name", name);
	    if (result != SMPD_SUCCESS)
	    {
		smpd_err_printf("unable to add the dbfirstdb name('%s') to the result command.\n", name);
		smpd_exit_fn(FCNAME);
		return SMPD_FAIL;
	    }
	    result_str = DBS_SUCCESS_STR;
	}
	else
	{
	    result_str = DBS_FAIL_STR;
	}
    }
    else if (strcmp(cmd->cmd_str, "dbnextdb") =