simple_pmi.c

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

}

int PMI_KVS_Get_key_length_max( int *maxlen )
{
    if (maxlen == NULL)
	return PMI_ERR_INVALID_ARG;
    *maxlen = PMI_keylen_max;
    return PMI_SUCCESS;
}

int PMI_KVS_Get_value_length_max( int *maxlen )
{
    if (maxlen == NULL)
	return PMI_ERR_INVALID_ARG;
    *maxlen = PMI_vallen_max;
    return PMI_SUCCESS;
}

/* We will use the default kvsname for both the kvs_domain_id and for the id */
/* Hence the implementation of the following three functions */

int PMI_Get_id_length_max( int *length )
{
    if (length == NULL)
	return PMI_ERR_INVALID_ARG;
    *length = PMI_kvsname_max;
    return PMI_SUCCESS;
}

int PMI_Get_id( char id_str[], int length )
{
    return PMI_KVS_Get_my_name( id_str, length );
}

int PMI_Get_kvs_domain_id( char id_str[], int length )
{
    return PMI_KVS_Get_my_name( id_str, length );
}

int PMI_KVS_Create( char kvsname[], int length )
{
    char buf[PMIU_MAXLINE], cmd[PMIU_MAXLINE];
    
    if (PMI_initialized == SINGLETON_INIT_BUT_NO_PM) {
	/* It is ok to pretend to *create* a kvs space */
	return 0;
    }

    PMIU_writeline( PMI_fd, "cmd=create_kvs\n" );
    PMIU_readline( PMI_fd, buf, PMIU_MAXLINE );
    PMIU_parse_keyvals( buf );
    PMIU_getval( "cmd", cmd, PMIU_MAXLINE );
    if ( strncmp( cmd, "newkvs", PMIU_MAXLINE ) != 0 ) {
	PMIU_printf( 1, "got unexpected response to create_kvs :%s:\n", buf );
	return( -1 );
    }
    else {
	PMIU_getval( "kvsname", kvsname, PMI_kvsname_max );
	return( 0 );
    }
}

int PMI_KVS_Destroy( const char kvsname[] )
{
    char buf[PMIU_MAXLINE], cmd[PMIU_MAXLINE];
    int rc;

    if (PMI_initialized == SINGLETON_INIT_BUT_NO_PM) {
	return 0;
    }

    /* FIXME: Check for tempbuf too short */
    MPIU_Snprintf( buf, PMIU_MAXLINE, "cmd=destroy_kvs kvsname=%s\n", kvsname );
    PMIU_writeline( PMI_fd, buf );
    PMIU_readline( PMI_fd, buf, PMIU_MAXLINE );
    PMIU_parse_keyvals( buf );
    PMIU_getval( "cmd", cmd, PMIU_MAXLINE );
    if ( strncmp( cmd, "kvs_destroyed", PMIU_MAXLINE ) != 0 ) {
	PMIU_printf( 1, "got unexpected response to destroy_kvs :%s:\n", buf );
	return( -1 );
    }
    else {
	PMIU_getval( "rc", buf, PMIU_MAXLINE );
	rc = atoi( buf );
	if ( rc != 0 ) {
	    PMIU_getval( "msg", buf, PMIU_MAXLINE );
	    PMIU_printf( 1, "KVS not destroyed, reason='%s'\n", buf );
	    return( -1 );
	}
	else {
	    return( 0 );
	}
    }
}

int PMI_KVS_Put( const char kvsname[], const char key[], const char value[] )
{
    char buf[PMIU_MAXLINE], cmd[PMIU_MAXLINE], message[PMIU_MAXLINE];
    int  rc;

    if (PMI_initialized == SINGLETON_INIT_BUT_NO_PM) {
	MPIU_Strncpy(cached_singinit_key,key,PMI_keylen_max);
	MPIU_Strncpy(cached_singinit_val,value,PMI_vallen_max);
	return 0;
    }
    
    /* FIXME: Check for tempbuf too short */
    MPIU_Snprintf( buf, PMIU_MAXLINE, "cmd=put kvsname=%s key=%s value=%s\n",
	      kvsname, key, value);
    PMIU_writeline( PMI_fd, buf );
    PMIU_readline( PMI_fd, buf, PMIU_MAXLINE );
    PMIU_parse_keyvals( buf );
    PMIU_getval( "cmd", cmd, PMIU_MAXLINE );
    if ( strncmp( cmd, "put_result", PMIU_MAXLINE ) != 0 ) {
	PMIU_printf( 1, "got unexpected response to put :%s:\n", buf );
	return( -1 );
    }
    else {
	PMIU_getval( "rc", buf, PMIU_MAXLINE );
	rc = atoi( buf );
	if ( rc < 0 ) {
	    PMIU_getval( "msg", message, PMIU_MAXLINE );
	    PMIU_printf( 1, "put failed; reason = %s\n", message );
	    return( -1 );
	}
    }
    return( 0 );
}

int PMI_KVS_Commit( const char kvsname[] )
{
    /* no-op in this implementation */
    return( 0 );
}

int PMI_KVS_Get( const char kvsname[], const char key[], char value[], int length)
{
    char buf[PMIU_MAXLINE], cmd[PMIU_MAXLINE];
    int  rc;

    /* FIXME: Check for tempbuf too short */
    MPIU_Snprintf( buf, PMIU_MAXLINE, "cmd=get kvsname=%s key=%s\n", kvsname, key );
    PMIU_writeline( PMI_fd, buf );
    PMIU_readline( PMI_fd, buf, PMIU_MAXLINE );
    PMIU_parse_keyvals( buf ); 
    PMIU_getval( "cmd", cmd, PMIU_MAXLINE );
    if ( strncmp( cmd, "get_result", PMIU_MAXLINE ) != 0 ) {
	PMIU_printf( 1, "got unexpected response to get :%s:\n", buf );
	return( -1 );
    }
    else {
	PMIU_getval( "rc", buf, PMIU_MAXLINE );
	rc = atoi( buf );
	if ( rc == 0 ) {
	    PMIU_getval( "value", value, PMI_vallen_max );
	    return( 0 );
	}
	else {
	    return( -1 );
	}
    }
}

int PMI_KVS_Iter_first(const char kvsname[], char key[], int key_len, char val[], int val_len)
{
    int rc;

    rc = PMII_iter( kvsname, 0, &PMI_iter_next_idx, key, key_len, val, val_len );
    return( rc );
}

int PMI_KVS_Iter_next(const char kvsname[], char key[], int key_len, char val[], int val_len)
{
    int rc;

    rc = PMII_iter( kvsname, PMI_iter_next_idx, &PMI_iter_next_idx, key, key_len, val, val_len );
    if ( rc == -2 )
	PMI_iter_next_idx = 0;
    return( rc );
}

/******************************** Name Publishing functions *************************/

int PMI_Publish_name( const char service_name[], const char port[] )
{
    char buf[PMIU_MAXLINE], cmd[PMIU_MAXLINE];

    if ( PMI_initialized > 1)  /* Ignore SINGLETON_INIT_BUT_NO_PM */
    {
        MPIU_Snprintf( cmd, PMIU_MAXLINE, "cmd=publish_name service=%s port=%s\n",
		       service_name, port );
	PMIU_writeline( PMI_fd, cmd );
	PMIU_readline( PMI_fd, buf, PMIU_MAXLINE );
	PMIU_parse_keyvals( buf );
	PMIU_getval( "cmd", cmd, PMIU_MAXLINE );
        if ( strncmp( cmd, "publish_result", PMIU_MAXLINE ) != 0 ) {
	    PMIU_printf( 1, "got unexpected response to publish :%s:\n", buf );
	    return( PMI_FAIL );
        }
        else {
	    PMIU_getval( "info", buf, PMIU_MAXLINE );
	    if ( strcmp(buf,"ok") != 0 ) {
	        PMIU_printf( 1, "publish failed; reason = %s\n", buf );
	        return( PMI_FAIL );
	    }
        }
    }
    else
    {
	PMIU_printf( 1, "PMI_Publish_name called before init\n" );
	return( PMI_FAIL );
    }

    return( PMI_SUCCESS );
}

int PMI_Unpublish_name( const char service_name[] )
{
    char buf[PMIU_MAXLINE], cmd[PMIU_MAXLINE];

    if ( PMI_initialized > 1)  /* Ignore SINGLETON_INIT_BUT_NO_PM */
    {
        MPIU_Snprintf( cmd, PMIU_MAXLINE, "cmd=unpublish_name service=%s\n", service_name );
	PMIU_writeline( PMI_fd, cmd );
	PMIU_readline( PMI_fd, buf, PMIU_MAXLINE );
	PMIU_parse_keyvals( buf );
	PMIU_getval( "cmd", cmd, PMIU_MAXLINE );
        if ( strncmp( cmd, "unpublish_result", PMIU_MAXLINE ) != 0 ) {
	    PMIU_printf( 1, "got unexpected response to unpublish :%s:\n", buf );
	    return( PMI_FAIL );
        }
        else {
	    PMIU_getval( "info", buf, PMIU_MAXLINE );
	    if ( strcmp(buf,"ok") != 0 ) {
	        PMIU_printf( 1, "unpublish failed; reason = %s\n", buf );
	        return( PMI_FAIL );
	    }
        }
    }
    else
    {
	PMIU_printf( 1, "PMI_Unpublish_name called before init\n" );
	return( PMI_FAIL );
    }

    return( PMI_SUCCESS );
}

int PMI_Lookup_name( const char service_name[], char port[] )
{
    char buf[PMIU_MAXLINE], cmd[PMIU_MAXLINE];

    if ( PMI_initialized > 1)  /* Ignore SINGLETON_INIT_BUT_NO_PM */
    {
        MPIU_Snprintf( cmd, PMIU_MAXLINE, "cmd=lookup_name service=%s\n", service_name );
	PMIU_writeline( PMI_fd, cmd );
	PMIU_readline( PMI_fd, buf, PMIU_MAXLINE );
	PMIU_parse_keyvals( buf );
	PMIU_getval( "cmd", cmd, PMIU_MAXLINE );
        if ( strncmp( cmd, "lookup_result", PMIU_MAXLINE ) != 0 ) {
	    PMIU_printf( 1, "got unexpected response to lookup :%s:\n", buf );
	    return( PMI_FAIL );
        }
        else {
	    PMIU_getval( "info", buf, PMIU_MAXLINE );
	    if ( strcmp(buf,"ok") != 0 ) {
		/****
	        PMIU_printf( 1, "lookup failed; reason = %s\n", buf );
		****/
	        return( PMI_FAIL );
	    }
	    PMIU_getval( "port", port, PMIU_MAXLINE );
        }
    }
    else
    {
	PMIU_printf( 1, "PMI_Lookup_name called before init\n" );
	return( PMI_FAIL );
    }

    return( PMI_SUCCESS );
}


/******************************** Process Creation functions *************************/

int PMI_Spawn_multiple(int count,
                       const char * cmds[],
                       const char ** argvs[],
                       const int maxprocs[],
                       const int info_keyval_sizes[],
                       const PMI_keyval_t * info_keyval_vectors[],
                       int preput_keyval_size,
                       const PMI_keyval_t preput_keyval_vector[],
                       int errors[])
{
    int  i,rc,argcnt,spawncnt;
    char buf[PMIU_MAXLINE], tempbuf[PMIU_MAXLINE], cmd[PMIU_MAXLINE];

#ifdef USE_PMI_PORT
    if (PMI_initialized < 2)
    {
	rc = PMII_singinit();
	if (rc < 0)
	    return(-1);
	PMI_initialized = SINGLETON_INIT_WITH_PM;    /* do this right away */
	PMI_size = 1;
	PMI_rank = 0;
	PMI_debug = 0;
	PMI_spawned = 0;
	PMII_getmaxes( &PMI_kvsname_max, &PMI_keylen_max, &PMI_vallen_max );
	PMI_KVS_Put( "singinit_kvs_0", cached_singinit_key, cached_singinit_val );
    }
#endif

    for (spawncnt=0; spawncnt < count; spawncnt++)
    {
	/* FIXME: Check for buf too short */
        MPIU_Snprintf(buf, PMIU_MAXLINE, "mcmd=spawn\nnprocs=%d\nexecname=%s\n",
	          maxprocs[spawncnt], cmds[spawncnt] );

	MPIU_Snprintf(tempbuf, PMIU_MAXLINE,"totspawns=%d\nspawnssofar=%d\n",
		      count, spawncnt+1);
	MPIU_Strnapp(buf,tempbuf,PMIU_MAXLINE);

        argcnt = 0;
        if ((argvs != NULL) && (argvs[spawncnt] != NULL)) {
            for (i=0; argvs[spawncnt][i] != NULL; i++)
            {
		/* FIXME (protocol design flaw): command line arguments
		   may contain both = and <space> (and even tab!).
		   Also, command line args may be quite long, leading to 
		   errors when PMIU_MAXLINE is exceeded */
		/* FIXME: Check for tempbuf too short */
                MPIU_Snprintf(tempbuf,PMIU_MAXLINE,"arg%d=%s\n",i+1,argvs[spawncnt][i]);
		/* FIXME: Check for error (buf too short for line) */
                MPIU_Strnapp(buf,tempbuf,PMIU_MAXLINE);
                argcnt++;
            }
        }
	/* FIXME: Check for tempbuf too short */
        MPIU_Snprintf(tempbuf,PMIU_MAXLINE,"argcnt=%d\n",argcnt);
	/* FIXME: Check for error (buf too short for line) */
        MPIU_Strnapp(buf,tempbuf,PMIU_MAXLINE);
    
/*        snprintf(tempbuf,PMIU_MAXLINE,"preput_num=%d ",preput_num);
        strcat(buf,tempbuf);
        for (i=0; i < preput_num; i++)
        {
	    MPIU_Snprintf(tempbuf,PMIU_MAXLINE,"preput_%d=%s:%s ",i,preput_keys[i],preput_vals[i]);
*/ /* FIXME: Check for error (buf too short for line) *//*
	    MPIU_Strnapp(buf,tempbuf,PMIU_MAXLINE);
        }
*/

	/* FIXME: Check for tempbuf too short */
        MPIU_Snprintf(tempbuf,PMIU_MAXLINE,"preput_num=%d\n", preput_keyval_size);
	/* FIXME: Check for error (buf too short for line) */
        MPIU_Strnapp(buf,tempbuf,PMIU_MAXLINE);
        for (i=0; i < preput_keyval_size; i++)
        { 
	    /* FIXME: Check for tempbuf too short */
	    MPIU_Snprintf(tempbuf,PMIU_MAXLINE,"preput_key_%d=%s\n",i,preput_keyval_vector[i].key);
	    /* FIXME: Check for error (buf too short for line) */
	    MPIU_Strnapp(buf,tempbuf,PMIU_MAXLINE); 
	    /* FIXME: Check for tempbuf too short */
	    MPIU_Snprintf(tempbuf,PMIU_MAXLINE,"preput_val_%d=%s\n",i,preput_keyval_vector[i].val);
	    /* FIXME: Check for error (buf too short for line) */
	    MPIU_Strnapp(buf,tempbuf,PMIU_MAXLINE); 
        } 
	/* FIXME: Check for tempbuf too short */
        MPIU_Snprintf(tempbuf,PMIU_MAXLINE,"info_num=%d\n", info_keyval_sizes[spawncnt]);
	/* FIXME: Check for error (buf too short for line) */
        MPIU_Strnapp(buf,tempbuf,PMIU_MAXLINE);
	for (i=0; i < info_keyval_sizes[spawncnt]; i++)
	{
	    /* FIXME: Check for tempbuf too short */
	    MPIU_Snprintf(tempbuf,PMIU_MAXLINE,"info_key_%d=%s\n",i,info_keyval_vectors[spawncnt][i].key);
	    /* FIXME: Check for error (buf too short for line) */
	    MPIU_Strnapp(buf,tempbuf,PMIU_MAXLINE); 
	    /* FIXME: Check for tempbuf too short */
	    MPIU_Snprintf(tempbuf,PMIU_MAXLINE,"info_val_%d=%s\n",i,info_keyval_vectors[spawncnt][i].val);
	    /* FIXME: Check for error (buf too short for line) */
	    MPIU_Strnapp(buf,tempbuf,PMIU_MAXLINE); 
	}

	/* FIXME: Check for error (buf too short for line) */
        MPIU_Strnapp(buf, "endcmd\n", PMIU_MAXLINE);
	DBG_PRINTF( ( "About to writeline %s\n", buf ) );
        PMIU_writeline( PMI_fd, buf );
    }
    DBG_PRINTF( ( "About to readline\n" ) );
    PMIU_readline( PMI_fd, buf, PMIU_MAXLINE );
    DBG_PRINTF( ( "About to parse\n" ) );
    PMIU_parse_keyvals( buf ); 
    DBG_PRINTF( ( "About to getval\n" ) );
    PMIU_getval( "cmd", cmd, PMIU_MAXLINE );
    if ( strncmp( cmd, "spawn_result", PMIU_MAXLINE ) != 0 ) {
	PMIU_printf( 1, "got unexpected response to spawn :%s:\n", buf );
	return( -1 );
    }
    else {
	PMIU_getval( "rc", buf, PMIU_MAXLINE );
	rc = atoi( buf );
	if ( rc != 0 ) {
	    /****
	    PMIU_getval( "status", tempbuf, PMIU_MAXLINE );
	    PMIU_printf( 1, "pmi_spawn_mult failed; status: %s\n",tempbuf);
	    ****/
	    return( -1 );
	}
    }
    return( 0 );
}