smpd_handle_command.c

mpi并行计算的c++代码 可用vc或gcc编译通过 可以用来搭建并行计算试验环境

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") == 0)
    {
	if (smpd_dbs_nextdb(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 dbnextdb 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, "dbdelete") == 0)
    {
	if (get_name_key_value(cmd->cmd, name, key, NULL) != SMPD_SUCCESS)
	    goto invalid_dbs_command;
	if (smpd_dbs_delete(name, key) == SMPD_SUCCESS)
	    result_str = DBS_SUCCESS_STR;
	else
	    result_str = DBS_FAIL_STR;
    }
    else
    {
	smpd_err_printf("unknown dbs command '%s', replying with failure.\n", cmd->cmd_str);
	goto invalid_dbs_command;
    }

    /* send the reply */
    smpd_dbg_printf("sending reply to dbs command '%s'.\n", cmd->cmd);
    result = smpd_add_command_arg(temp_cmd, "result", result_str);
    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;

invalid_dbs_command:

    result = smpd_add_command_arg(temp_cmd, "result", SMPD_FAIL_STR" - invalid dbs command.");
    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;
}

#undef FCNAME
#define FCNAME "smpd_handle_launch_command"
int smpd_handle_launch_command(smpd_context_t *context)
{
    int result;
    char err_msg[256];
    int iproc;
    int i, nmaps;
    char drive_arg_str[20];
    smpd_command_t *cmd, *temp_cmd;
    smpd_process_t *process;
    char share[SMPD_MAX_EXE_LENGTH];
    int priority_class = SMPD_DEFAULT_PRIORITY_CLASS, priority_thread = SMPD_DEFAULT_PRIORITY;

    smpd_enter_fn(FCNAME);

    cmd = &context->read_cmd;

    /* parse the command */
    if (MPIU_Str_get_int_arg(cmd->cmd, "i", &iproc) != MPIU_STR_SUCCESS)
	iproc = 0;
    result = smpd_create_process_struct(iproc, &process);
    if (result != SMPD_SUCCESS)
    {
	smpd_err_printf("unable to create a process structure.\n");
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }

    if (MPIU_Str_get_string_arg(cmd->cmd, "c", process->exe, SMPD_MAX_EXE_LENGTH) != MPIU_STR_SUCCESS)
    {
	smpd_err_printf("launch command received with no executable: '%s'\n", cmd->cmd);
	smpd_exit_fn(FCNAME);
	return SMPD_FAIL;
    }
    MPIU_Str_get_string_arg(cmd->cmd, "e", process->env, SMPD_MAX_ENV_LENGTH);
    MPIU_Str_get_string_arg(cmd->cmd, "d", process->dir, SMPD_MAX_DIR_LENGTH);
    MPIU_Str_get_string_arg(cmd->cmd, "p", process->path, SMPD_MAX_PATH_LENGTH);
    MPIU_Str_get_string_arg(cmd->cmd, "k", process->kvs_name, SMPD_MAX_DBS_NAME_LEN);
    MPIU_Str_get_string_arg(cmd->cmd, "kd", process->domain_name, SMPD_MAX_DBS_NAME_LEN);
    MPIU_Str_get_string_arg(cmd->cmd, "q", process->