smpd_ipmi.c

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

	    pmi_err_printf("unable to initialize the local dbs engine.\n");
	    return PMI_FAIL;
	}
	result = smpd_dbs_create(pmi_process.kvs_name);
	if (result != SMPD_SUCCESS)
	{
	    pmi_err_printf("unable to create the process group kvs\n");
	    return PMI_FAIL;
	}
	pmi_process.init_finalized = PMI_INITIALIZED;
	return PMI_SUCCESS;
    }
#endif

    p = getenv("PMI_CLIQUE");
    if (p != NULL)
    {
	parse_clique(p);
    }

    /*
    printf("PMI_ROOT_HOST=%s PMI_ROOT_PORT=%s PMI_RANK=%s PMI_SIZE=%s PMI_KVS=%s PMI_CLIQUE=%s\n",
	getenv("PMI_ROOT_HOST"), getenv("PMI_ROOT_PORT"), getenv("PMI_RANK"), getenv("PMI_SIZE"),
	getenv("PMI_KVS"), getenv("PMI_CLIQUE"));
    fflush(stdout);
    */

    if (pmi_process.iproc == 0)
    {
	p = getenv("PMI_ROOT_LOCAL");
	if (p && strcmp(p, "1") == 0)
	{
#ifdef HAVE_WINDOWS_H
	    pmi_process.hRootThreadReadyEvent = CreateEvent(NULL, TRUE, FALSE, NULL);
	    if (pmi_process.hRootThreadReadyEvent == NULL)
	    {
		pmi_err_printf("unable to create the root listener synchronization event, error: %d\n", GetLastError());
		return PMI_FAIL;
	    }
	    pmi_process.hRootThread = CreateThread(NULL, 0, (LPTHREAD_START_ROUTINE)root_smpd, NULL, 0, NULL);
	    if (pmi_process.hRootThread == NULL)
	    {
		pmi_err_printf("unable to create the root listener thread: error %d\n", GetLastError());
		return PMI_FAIL;
	    }
	    if (WaitForSingleObject(pmi_process.hRootThreadReadyEvent, 60000) != WAIT_OBJECT_0)
	    {
		pmi_err_printf("the root process thread failed to initialize.\n");
		return PMI_FAIL;
	    }
#else
	    result = fork();
	    if (result == -1)
	    {
		pmi_err_printf("unable to fork the root listener, errno %d\n", errno);
		return PMI_FAIL;
	    }
	    if (result == 0)
	    {
		root_smpd(NULL);
		exit(0);
	    }
	    pmi_process.root_pid = result;
#endif
	}
    }

    /* connect to the root */

    result = MPIDU_Sock_create_set(&pmi_process.set);
    if (result != MPI_SUCCESS)
    {
	pmi_err_printf("PMI_Init failed: unable to create a sock set, error: %d\n", result);
	return PMI_FAIL;
    }

    result = uPMI_ConnectToHost(pmi_process.root_host, pmi_process.root_port, SMPD_CONNECTING_RPMI);
    if (result != SMPD_SUCCESS)
    {
	pmi_err_printf("PMI_Init failed.\n");
	return PMI_FAIL;
    }

    pmi_process.init_finalized = PMI_INITIALIZED;

    sprintf(rank_str, "%d", pmi_process.iproc);
    sprintf(size_str, "%d", pmi_process.nproc);
    result = pmi_create_post_command("init", pmi_process.kvs_name, rank_str, size_str);
    if (result != PMI_SUCCESS)
    {
	pmi_err_printf("PMI_Init failed: unable to create an init command.\n");
	return PMI_FAIL;
    }

    /* parse the result of the command */
    if (MPIU_Str_get_string_arg(pmi_process.context->read_cmd.cmd, "result", str, 1024) != MPIU_STR_SUCCESS)
    {
	pmi_err_printf("PMI_Init failed: no result string in the result command.\n");
	return PMI_FAIL;
    }
    if (strcmp(str, SMPD_SUCCESS_STR))
    {
	pmi_err_printf("PMI_Init failed: %s\n", str);
	return PMI_FAIL;
    }

    return PMI_SUCCESS;
}

static int rPMI_Finalize()
{
    int result;
    char rank_str[100];
    char str[1024];
#ifndef HAVE_WINDOWS_H
    int status;
#endif

    if (pmi_process.init_finalized == PMI_FINALIZED)
	return PMI_SUCCESS;

    if (pmi_process.local_kvs)
    {
	smpd_dbs_finalize();
	pmi_process.init_finalized = PMI_FINALIZED;
	return PMI_SUCCESS;
    }

    sprintf(rank_str, "%d", pmi_process.iproc);
    result = pmi_create_post_command("finalize", pmi_process.kvs_name, rank_str, NULL);
    if (result != PMI_SUCCESS)
    {
	pmi_err_printf("PMI_Finalize failed: unable to create an finalize command.\n");
	return PMI_FAIL;
    }

    /* parse the result of the command */
    if (MPIU_Str_get_string_arg(pmi_process.context->read_cmd.cmd, "result", str, 1024) != MPIU_STR_SUCCESS)
    {
	pmi_err_printf("PMI_Finalize failed: no result string in the result command.\n");
	return PMI_FAIL;
    }
    if (strcmp(str, SMPD_SUCCESS_STR))
    {
	pmi_err_printf("PMI_Finalize failed: %s\n", str);
	return PMI_FAIL;
    }

    if (pmi_process.iproc == 0)
    {
	/* the root process tells the root to exit when all the pmi contexts have exited */
	result = pmi_create_post_command("exit_on_done", NULL, NULL, NULL);
	if (result != PMI_SUCCESS)
	{
	    pmi_err_printf("exit_on_done command failed.\n");
	    return PMI_FAIL;
	}
	/*printf("exit_on_done command returned successfully.\n");fflush(stdout);*/
    }

    /*printf("entering finalize pmi_barrier.\n");fflush(stdout);*/
    PMI_Barrier();
    /*printf("after finalize pmi_barrier, posting done command.\n");fflush(stdout);*/

    /* post a done command to close the pmi context */
    result = pmi_create_post_command("done", NULL, NULL, NULL);
    if (result != PMI_SUCCESS)
    {
	pmi_err_printf("failed.\n");
	return PMI_FAIL;
    }

    if (pmi_process.iproc == 0)
    {
#ifdef HAVE_WINDOWS_H
	WaitForSingleObject(pmi_process.hRootThread, INFINITE);
#else
	waitpid(pmi_process.root_pid, &status, WUNTRACED);
#endif
    }

    if (pmi_process.sock != MPIDU_SOCK_INVALID_SOCK)
    {
	result = MPIDU_Sock_finalize();
	if (result != MPI_SUCCESS)
	{
	    pmi_err_printf("MPIDU_Sock_finalize failed, error: %d\n", result);
	}
    }

    pmi_process.init_finalized = PMI_FINALIZED;

    return PMI_SUCCESS;
}

int iPMI_Init(int *spawned)
{
    char *p;
    int result;
    char rank_str[100], size_str[100];
    char str[1024];

    if (spawned == NULL)
	return PMI_ERR_INVALID_ARG;

    /* don't allow pmi_init to be called more than once */
    if (pmi_process.init_finalized == PMI_INITIALIZED)
	return PMI_SUCCESS;

    /* initialize to defaults */

    result = MPIDU_Sock_init();
    if (result != MPI_SUCCESS)
    {
	pmi_err_printf("MPIDU_Sock_init failed,\nsock error: %s\n", get_sock_error_string(result));
	return PMI_FAIL;
    }

    result = smpd_init_process();
    if (result != SMPD_SUCCESS)
    {
	pmi_err_printf("unable to initialize the smpd global process structure.\n");
	return PMI_FAIL;
    }

    p = getenv("PMI_ROOT_HOST");
    if (p != NULL)
    {
	return rPMI_Init(spawned);
    }

    pmi_process.iproc = 0;
    pmi_process.nproc = 1;

    p = getenv("PMI_SPAWN");
    if (p)
    {
	*spawned = atoi(p);
    }
    else
    {
	*spawned = 0;
    }

    p = getenv("PMI_APPNUM");
    if (p)
    {
	pmi_process.appnum = atoi(p);
    }
    else
    {
	pmi_process.appnum = 0;
    }

    p = getenv("PMI_KVS");
    if (p != NULL)
    {
	strncpy(pmi_process.kvs_name, p, PMI_MAX_KVS_NAME_LENGTH);
    }
    else
    {
	pmi_process.local_kvs = PMI_TRUE;
	result = smpd_dbs_init();
	if (result != SMPD_SUCCESS)
	{
	    pmi_err_printf("unable to initialize the local dbs engine.\n");
	    return PMI_FAIL;
	}
	result = smpd_dbs_create(pmi_process.kvs_name);
	if (result != SMPD_SUCCESS)
	{
	    pmi_err_printf("unable to create the process group kvs\n");
	    return PMI_FAIL;
	}
	strncpy(pmi_process.domain_name, smpd_process.domain_name, PMI_MAX_KVS_NAME_LENGTH);
	pmi_process.init_finalized = PMI_INITIALIZED;
	return PMI_SUCCESS;
    }

    p = getenv("PMI_DOMAIN");
    if (p != NULL)
    {
	strncpy(pmi_process.domain_name, p, PMI_MAX_KVS_NAME_LENGTH);
    }
    else
    {
	strncpy(pmi_process.domain_name, "mpich2", PMI_MAX_KVS_NAME_LENGTH);
    }

    p = getenv("PMI_RANK");
    if (p != NULL)
    {
	pmi_process.iproc = atoi(p);
	if (pmi_process.iproc < 0)
	{
	    pmi_err_printf("invalid rank %d, setting to 0\n", pmi_process.iproc);
	    pmi_process.iproc = 0;
	}
    }

    p = getenv("PMI_SIZE");
    if (p != NULL)
    {
	pmi_process.nproc = atoi(p);
	if (pmi_process.nproc < 1)
	{
	    pmi_err_printf("invalid size %d, setting to 1\n", pmi_process.nproc);
	    pmi_process.nproc = 1;
	}
    }

    p = getenv("PMI_SMPD_ID");
    if (p != NULL)
    {
	pmi_process.smpd_id = atoi(p);
	smpd_process.id = pmi_process.smpd_id;
    }

    p = getenv("PMI_SMPD_KEY");
    if (p != NULL)
    {
	pmi_process.smpd_key = atoi(p);
    }

    p = getenv("PMI_SMPD_FD");
    if (p != NULL)
    {
	result = MPIDU_Sock_create_set(&pmi_process.set);
	if (result != MPI_SUCCESS)
	{
	    pmi_err_printf("PMI_Init failed: unable to create a sock set, error:\n%s\n",
		get_sock_error_string(result));
	    return PMI_FAIL;
	}

#ifdef HAVE_WINDOWS_H
	pmi_process.smpd_fd = smpd_decode_handle(p);
#else
	pmi_process.smpd_fd = (MPIDU_SOCK_NATIVE_FD)atoi(p);
#endif
	result = MPIDU_Sock_native_to_sock(pmi_process.set, pmi_process.smpd_fd, NULL, &pmi_process.sock);
	if (result != MPI_SUCCESS)
	{
	    pmi_err_printf("MPIDU_Sock_native_to_sock failed, error %s\n", get_sock_error_string(result));
	    return PMI_FAIL;
	}
	result = smpd_create_context(SMPD_CONTEXT_PMI, pmi_process.set, pmi_process.sock, pmi_process.smpd_id, &pmi_process.context);
	if (result != SMPD_SUCCESS)
	{
	    pmi_err_printf("unable to create a pmi context.\n");
	    return PMI_FAIL;
	}
    }
    else
    {
	p = getenv("PMI_HOST");
	if (p != NULL)
	{
	    strncpy(pmi_process.host, p, 100);
	    p = getenv("PMI_PORT");
	    if (p != NULL)
	    {
		pmi_process.port = atoi(p);

		result = MPIDU_Sock_create_set(&pmi_process.set);
		if (result != MPI_SUCCESS)
		{
		    pmi_err_printf("PMI_Init failed: unable to create a sock set, error: %d\n", result);
		    return PMI_FAIL;
		}

		result = uPMI_ConnectToHost(pmi_process.host, pmi_process.port, SMPD_CONNECTING_PMI);
		if (result != SMPD_SUCCESS)
		{
		    pmi_err_printf("PMI_Init failed.\n");
		    return PMI_FAIL;
		}
	    }
	    else
	    {
		pmi_err_printf("No mechanism specified for connecting to the process manager - host %s but no port provided.\n", pmi_process.host);
		return PMI_FAIL;
	    }
	}
	else
	{
	    pmi_err_printf("No mechanism specified for connecting to the process manager.\n");
	    return PMI_FAIL;
	}
    }

    p = getenv("PMI_CLIQUE");
    if (p != NULL)
    {
	parse_clique(p);
    }

    /*
    printf("PMI_RANK=%s PMI_SIZE=%s PMI_KVS=%s PMI_SMPD_ID=%s PMI_SMPD_FD=%s PMI_SMPD_KEY=%s\n PMI_SPAWN=%s",
	getenv("PMI_RANK"), getenv("PMI_SIZE"), getenv("PMI_KVS"), getenv("PMI_SMPD_ID"),
	getenv("PMI_SMPD_FD"), getenv("PMI_SMPD_KEY"), getenv("PMI_SPAWN"));
    fflush(stdout);
    */

    pmi_process.init_finalized = PMI_INITIALIZED;

    sprintf(rank_str, "%d", pmi_process.iproc);
    sprintf(size_str, "%d", pmi_process.nproc);
    result = pmi_create_post_command("init", pmi_process.kvs_name, rank_str, size_str);
    if (result != PMI_SUCCESS)
    {
	pmi_err_printf("PMI_Init failed: unable to create an init command.\n");
	return PMI_FAIL;
    }

    /* parse the result of the command */
    if (MPIU_Str_get_string_arg(pmi_process.context->read_cmd.cmd, "result", str, 1024) != MPIU_STR_SUCCESS)
    {
	pmi_err_printf("PMI_Init failed: no result string in the result command.\n");
	return PMI_FAIL;
    }
    if (strcmp(str, SMPD_SUCCESS_STR))
    {
	pmi_err_printf("PMI_Init failed: %s\n", str);
	return PMI_FAIL;
    }

    /*
    if (*spawned && pmi_process.iproc == 0)
    {
	char key[1024], val[8192];
	key[0] = '\0';
	result = PMI_KVS_Iter_first(pmi_process.kvs_name, key, 1024, val, 8192);
	if (result != PMI_SUCCESS || key[0] == '\0')
	{
	    printf("No preput values in %s\n", pmi_process.kvs_name);
	}
	while (result == PMI_SUCCESS && key[0] != '\0')
	{
	    printf("PREPUT key=%s, val=%s\n", key, val);
	    result = PMI_KVS_Iter_next(pmi_process.kvs_name, key, 1024, val, 8192);
	}
	fflush(stdout);
    }
    iPMI_Barrier();
    */

    /*printf("iPMI_Init returning success.\n");fflush(stdout);*/
    return PMI_SUCCESS;
}

int iPMI_Finalize()
{
    int result;
    char rank_str[100];
    char str[1024];

    if (pmi_process.init_finalized == PMI_FINALIZED)
	return PMI_SUCCESS;

    /*
    printf("PMI_Finalize called.\n");
    fflush(stdout);
    */

    if (pmi_process.rpmi)
    {
	return rPMI_Finalize();
    }

    if (pmi_process.local_kvs)
    {
	smpd_dbs_finalize();
	pmi_process.init_finalized = PMI_FINALIZED;
	return PMI_SUCCESS;
    }

    sprintf(rank_str, "%d", pmi_process.iproc);
    result = pmi_create_post_command("finalize", pmi_process.kvs_name, rank_str, NULL);
    if (result != PMI_SUCCESS)
    {
	pmi_err_printf("PMI_Finalize failed: unable to create an finalize command.\n");
	goto fn_fail;
    }

    /* parse the result of the command */
    if (MPIU_Str_get_string_arg(pmi_process.context->read_cmd.cmd, "result", str, 1024) != MPIU_STR_SUCCESS)
    {
	pmi_err_printf("PMI_Finalize failed: no result string in the result command.\n");
	goto fn_fail;
    }
    if (strcmp(str, SMPD_SUCCESS_STR))
    {
	pmi_err_printf("PMI_Finalize failed: %s\n", str);
	goto fn_fail;
    }

    PMI_Barrier();