smpd_handle_spawn.c

fortran并行计算包

		if (env_str == NULL)
		{
		    goto spawn_failed;
		}
		token = strtok(env_str, " ");
		while (token)
		{
		    char *env_key, *env_val;
		    env_key = MPIU_Strdup(token);
		    if (env_key == NULL)
		    {
			goto spawn_failed;
		    }
		    env_val = env_key;
		    while (*env_val != '\0' && *env_val != '=')
			env_val++;
		    if (*env_val == '=')
		    {
			*env_val = '\0';
			env_val++;
			MPIU_Str_add_string_arg(&cur_env_loc, &env_maxlen, env_key, env_val);
			/* Check for special environment variables */
			if (strcmp(env_val, "MPICH2_CHANNEL") == 0)
			{
			    env_channel_specified = SMPD_TRUE;
			}
			else if (strcmp(env_val, "MPI_DLL_NAME") == 0)
			{
			    env_dll_specified = SMPD_TRUE;
			}
			else if (strcmp(env_val, "MPI_WRAP_DLL_NAME") == 0)
			{
			    env_wrap_dll_specified = SMPD_TRUE;
			}
		    }
		    MPIU_Free(env_key);
		    token = strtok(NULL, " ");
		}
		MPIU_Free(env_str);
	    }
	    /* log */
	    if (strcmp(info[j].key, "log") == 0)
	    {
		if (smpd_is_affirmative(info[j].val) || (strcmp(info[j].val, "1") == 0))
		{
		    MPIU_Str_add_string_arg(&cur_env_loc, &env_maxlen, "MPI_WRAP_DLL_NAME", "mpe");
		    env_wrap_dll_specified = SMPD_TRUE;
		}
	    }
	    /* wdir */
	    if ((strcmp(info[j].key, "wdir") == 0) || (strcmp(info[j].key, "dir") == 0))
	    {
		strcpy(node.dir, info[j].val);
		smpd_dbg_printf("wdir = %s\n", info[j].val);
	    }
	    /* map */
	    if (strcmp(info[j].key, "map") == 0)
	    {
		if (smpd_parse_map_string(info[j].val, &drive_map_list) != SMPD_SUCCESS)
		{
		    goto spawn_failed;
		}
	    }
	    /* localonly */
	    if (strcmp(info[j].key, "localonly") == 0)
	    {
		smpd_get_hostname(node.hostname, SMPD_MAX_HOST_LENGTH);
		if (smpd_get_host_id(node.hostname, &node.host_id) != SMPD_SUCCESS)
		{
		    node.hostname[0] = '\0';
		    node.host_id = -1;
		}
	    }
	    /* machinefile */
	    if (strcmp(info[j].key, "machinefile") == 0)
	    {
		if (smpd_parse_machine_file(info[j].val))
		{
		    /*use_machine_file = SMPD_TRUE;*/
		}
	    }
	    /* configfile */
	    if (strcmp(info[j].key, "configfile") == 0)
	    {
	    }
	    /* file */
	    if (strcmp(info[j].key, "file") == 0)
	    {
	    }
	    /* priority */
	    if (strcmp(info[j].key, "priority") == 0)
	    {
		if (smpd_isnumbers_with_colon(info[j].val))
		{
		    char *str;
		    node.priority_class = atoi(info[j].val); /* This assumes atoi will stop at the colon and return a number */
		    str = strchr(info[j].val, ':');
		    if (str)
		    {
			str++;
			node.priority_thread = atoi(str);
		    }
		    if (node.priority_class < 0 || node.priority_class > 4 || node.priority_thread < 0 || node.priority_thread > 5)
		    {
			smpd_err_printf("Error: priorities must be between 0-4:0-5\n");
			node.priority_class = SMPD_DEFAULT_PRIORITY_CLASS;
			node.priority_thread = SMPD_DEFAULT_PRIORITY;
		    }
		}
	    }
	    /* timeout */
	    if (strcmp(info[j].key, "timeout") == 0)
	    {
#if 0 /* FIXME: create a mechanism to timeout spawned processes and only those spawned processes */
		smpd_process.timeout = atoi(info[j].val);
		if (smpd_process.timeout < 1)
		{
		    smpd_dbg_printf("Warning: invalid timeout specified, ignoring timeout value of '%s'\n", info[j].val);
		    smpd_process.timeout = -1;
		}
#endif
	    }
	    /* exitcodes */
	    if (strcmp(info[j].key, "exitcodes") == 0)
	    {
		/* FIXME: This will turn on exit code printing for all processes.  Implement a new mechanism for only printing codes for an individual process group. */
		if (smpd_is_affirmative(info[j].val) || (strcmp(info[j].val, "1") == 0))
		{
		    smpd_process.output_exit_codes = SMPD_TRUE;
		}
	    }
	    /* nompi */
	    if (strcmp(info[j].key, "nompi") == 0)
	    {
		/* FIXME: Tell MPICH that the spawned processes will not make any MPI calls, including MPI_Init - so don't to a comm_accept or it will hang! */
	    }
	    /* etc */
	}

	/* Add special environment variables */
	if (env_channel_specified == SMPD_FALSE && env_dll_specified == SMPD_FALSE)
	{
	    if (smpd_process.env_dll[0] != '\0')
	    {
		MPIU_Str_add_string_arg(&cur_env_loc, &env_maxlen, "MPI_DLL_NAME", smpd_process.env_dll);
	    }
	    else if (smpd_process.env_channel[0] != '\0')
	    {
		MPIU_Str_add_string_arg(&cur_env_loc, &env_maxlen, "MPICH2_CHANNEL", smpd_process.env_channel);
	    }
	}
	if (env_wrap_dll_specified == SMPD_FALSE)
	{
	    if (smpd_process.env_wrap_dll[0] != '\0')
	    {
		MPIU_Str_add_string_arg(&cur_env_loc, &env_maxlen, "MPI_WRAP_DLL_NAME", smpd_process.env_wrap_dll);
	    }
	}

	/* create launch nodes for the current command */
	for (j=0; j<maxprocs[i]; j++)
	{
	    if (launch_list == NULL)
	    {
		launch_list = (smpd_launch_node_t*)MPIU_Malloc(sizeof(smpd_launch_node_t));
		launch_iter = launch_list;
		if (launch_iter)
		{
		    launch_iter->prev = NULL;
		}
	    }
	    else
	    {
		launch_iter->next = (smpd_launch_node_t*)MPIU_Malloc(sizeof(smpd_launch_node_t));
		if (launch_iter->next)
		{
		    launch_iter->next->prev = launch_iter;
		    launch_iter = launch_iter->next;
		}
		else
		{
		    launch_iter = NULL;
		}
	    }
	    if (launch_iter == NULL)
	    {
		smpd_err_printf("unable to allocate a launch node structure for the %dth command.\n", cur_iproc);
		goto spawn_failed;
	    }
	    launch_iter->appnum = i;
	    launch_iter->iproc = cur_iproc++;
	    launch_iter->args[0] = '\0';
	    launch_iter->clique[0] = '\0';
	    /*launch_iter->dir[0] = '\0';*/
	    strcpy(launch_iter->dir, node.dir);
	    strcpy(launch_iter->env_data, node.env_data);
	    launch_iter->env = launch_iter->env_data;
	    launch_iter->exe[0] = '\0';
	    if (node.host_id != -1)
	    {
		launch_iter->host_id = node.host_id;
		strcpy(launch_iter->hostname, node.hostname);
		strcpy(launch_iter->alt_hostname, node.alt_hostname);
	    }
	    else
	    {
		launch_iter->host_id = -1;
		launch_iter->hostname[0] = '\0';
		launch_iter->alt_hostname[0] = '\0';
	    }
	    launch_iter->map_list = drive_map_list;
	    if (drive_map_list)
	    {
		drive_map_list->ref_count++;
	    }
	    launch_iter->priority_class = node.priority_class;
	    launch_iter->priority_thread = node.priority_thread;
	    launch_iter->next = NULL;

	    strcpy(launch_iter->exe, node.exe);
	    /*strcpy(launch_iter->args, node.args);*/
	    if (strlen(node.args) > 0)
	    {
		strncat(launch_iter->exe, " ", SMPD_MAX_EXE_LENGTH);
		strncat(launch_iter->exe, node.args, SMPD_MAX_EXE_LENGTH);
	    }
	    strcpy(launch_iter->path, node.path);
	}
    }
    if (info != NULL)
    {
	/* free the last round of infos */
	for (j=0; j<nkeyvals[i-1]; j++)
	{
	    MPIU_Free(info[j].key);
	    MPIU_Free(info[j].val);
	}
	MPIU_Free(info);
    }
    info = NULL;

    /* create a spawn context to save parameters, state, etc. */
    if (context->spawn_context != NULL)
    {
	MPIU_Free(context->spawn_context);
    }
    context->spawn_context = (smpd_spawn_context_t*)MPIU_Malloc(sizeof(smpd_spawn_context_t));
    if (context->spawn_context == NULL)
    {
	smpd_err_printf("unable to create a spawn context.\n");
	goto spawn_failed;
    }
    context->spawn_context->context = context;
    context->spawn_context->kvs_name[0] = '\0';
    context->spawn_context->launch_list = NULL;
    context->spawn_context->npreput = -1;
    context->spawn_context->num_outstanding_launch_cmds = -1;
    context->spawn_context->preput[0] = '\0';
    context->spawn_context->result_cmd = NULL;

    /* Get the keyval pairs to be put in the process group keyval space before the processes are launched. */
    if (MPIU_Str_get_int_arg(cmd->cmd, "npreput", &context->spawn_context->npreput) != MPIU_STR_SUCCESS)
    {
	smpd_err_printf("unable to get the npreput parameter from the spawn command '%s'.\n", cmd->cmd);
	goto spawn_failed;
    }
    /*printf("npreput = %d\n", context->spawn_context->npreput);fflush(stdout);*/
    if (context->spawn_context->npreput > 0 && MPIU_Str_get_string_arg(cmd->cmd, "preput", context->spawn_context->preput, SMPD_MAX_CMD_LENGTH) != MPIU_STR_SUCCESS)
    {
	smpd_err_printf("unablet to get the preput parameter from the spawn command '%s'.\n", cmd->cmd);
	goto spawn_failed;
    }
    MPIU_Free(maxprocs);
    MPIU_Free(nkeyvals);


    /* do the spawn stuff */

    /* count the number of processes to spawn */
    nproc = 0;
    launch_iter = launch_list;
    while (launch_iter)
    {
	nproc++;
	launch_iter = launch_iter->next;
    }

    /* create the host list and add nproc to the launch list */
    host_list = NULL;
    launch_iter = launch_list;
    while (launch_iter)
    {
	if (launch_iter->host_id == -1)
	{
	    smpd_get_next_host(&host_list, launch_iter);
	}
	if (launch_iter->alt_hostname[0] != '\0')
	{
	    if (smpd_append_env_option(launch_iter->env_data, SMPD_MAX_ENV_LENGTH, "MPICH_INTERFACE_HOSTNAME", launch_iter->alt_hostname) != SMPD_SUCCESS)
	    {
		smpd_err_printf("unable to add the MPICH_INTERFACE_HOSTNAME to the launch node environment block\n");
		goto spawn_failed;
	    }
	}
	launch_iter->nproc = nproc;
	launch_iter = launch_iter->next;
    }
    smpd_create_cliques(launch_list);

    /* connect up the new smpd hosts */
    context = smpd_process.left_context;
    /* save the launch list to be used after the new hosts are connected */
    context->spawn_context->launch_list = launch_list;
    context->spawn_context->num_outstanding_launch_cmds = 0;/*nproc;*/ /* this assumes all launch commands will be successfully posted. */
    smpd_fix_up_host_tree(smpd_process.host_list);

#if 0 /* old way */
    host_iter = smpd_process.host_list;
    while (host_iter)
    {
	if (!host_iter->connected)
	{
	    context->connect_to = host_iter;

	    /* create a connect command to be sent to the parent */
	    result = smpd_create_command("connect", 0, context->connect_to->parent, SMPD_TRUE, &cmd);
	    if (result != SMPD_SUCCESS)
	    {
		smpd_err_printf("unable to create a connect command.\n");
		goto spawn_failed;
	    }
	    host_iter->connect_cmd_tag = cmd->tag;
	    result = smpd_add_command_arg(cmd, "host", context->connect_to->host);
	    if (result != SMPD_SUCCESS)
	    {
		smpd_err_printf("unable to add the host parameter to the connect command for host %s\n", context->connect_to->host);
		goto spawn_failed;
	    }
	    result = smpd_add_command_int_arg(cmd, "id", context->connect_to->id);
	    if (result != SMPD_SUCCESS)
	    {
		smpd_err_printf("unable to add the id parameter to the connect command for host %s\n", context->connect_to->host);
		goto spawn_failed;
	    }
	    if (smpd_process.plaintext)
	    {
		/* propagate the plaintext option to the manager doing the connect */
		result = smpd_add_command_arg(cmd, "plaintext", "yes");
		if (result != SMPD_SUCCESS)
		{
		    smpd_err_printf("unable to add the plaintext parameter to the connect command for host %s\n", context->connect_to->host);
		    goto spawn_failed;
		}
	    }

	    smpd_dbg_printf("sending first connect command to add new hosts for the spawn command.\n");