pmiserv.c

fortran并行计算包

}
/* Create a kvs and generate its name; return that name as the argument */
static int fPMIKVSGetNewSpace( char kvsname[], int maxlen )
{
    PMIKVSpace *kvs;

    kvs = fPMIKVSAllocate( );
    MPIU_Strncpy( kvsname, kvs->kvsname, maxlen );
    return 0;
}
static int fPMIKVSFindKey( PMIKVSpace *kvs, 
			   const char key[], char val[], int maxval )
{
    PMIKVPair *p;
    int       rc;

    p = kvs->pairs;
    while (p) {
	rc = strcmp( p->key, key );
	if (rc == 0) {
	    /* Found it.  Get the value and return success */
	    MPIU_Strncpy( val, p->val, maxval );
	    return 0;
	}
	if (rc > 0) {
	    /* We're past the point in the sorted list where the
	       key could be found */
	    return 1;
	}
	p = p->nextPair;
    }
    return 1;
}

static int fPMIKVSAddPair( PMIKVSpace *kvs, 
			   const char key[], const char val[] )
{
    PMIKVPair *pair, *p, **pprev;
    int       rc;

    /* Find the location in which to insert the pair (if the
       same key already exists, that is an error) */
    p = kvs->pairs;
    pprev = &(kvs->pairs);
    while (p) {
	rc = strcmp( p->key, key );
	if (rc == 0) {
	    /* Duplicate.  Indicate an error */
	    return 1;
	}
	if (rc > 0) {
	    /* We've found the location (after pprev, before p) */
	    break;
	}
	pprev = &(p->nextPair);
	p = p->nextPair;
    }
    pair = (PMIKVPair *)MPIU_Malloc( sizeof(PMIKVPair) );
    if (!pair) {
	return -1;
    }
    MPIU_Strncpy( pair->key, key, sizeof(pair->key) );
    MPIU_Strncpy( pair->val, val, sizeof(pair->val) );

    /* Insert into the list */
    pair->nextPair = p;
    *pprev         = pair;

    /* Since the list has been modified, clear the index helpers */
    kvs->lastByIdx = 0;
    kvs->lastIdx   = -1;

    return 0;
}

static PMIKVSpace *fPMIKVSFindSpace( const char kvsname[] )
{
    PMIKVSpace *kvs = pmimaster.kvSpaces;
    int rc;

    /* We require the kvs spaces to be stored in a sorted order */
    while (kvs) {
	rc = strcmp( kvs->kvsname, kvsname );
	if (rc == 0) {
	    /* Found */
	    return kvs;
	}
	if (rc > 0) {
	    /* Did not find (kvsname is sorted after kvs->kvsname) */
	    return 0;
	}
	kvs = kvs->nextKVS;
    }
    /* Did not find */
    return 0;
}
static int PMIKVSFree( PMIKVSpace *kvs )
{
    PMIKVPair *p, *pNext;
    PMIKVSpace **kPrev, *k;
    int        rc;
    
    /* Recover the pairs */
    p = kvs->pairs;
    while (p) {
	pNext = p->nextPair;
	MPIU_Free( p );
	p = pNext;
    }

    /* Recover the KVS space, and remove it from the master's list */
    kPrev = &pmimaster.kvSpaces;
    k     = pmimaster.kvSpaces;
    rc    = 1;
    while (k) {
	rc = strcmp( k->kvsname, kvs->kvsname );
	if (rc == 0) {
	    *kPrev = k->nextKVS;
	    MPIU_Free( k );
	    break;
	}
	kPrev = &(k->nextKVS);
	k = k->nextKVS;
    }
    
    /* Note that if we did not find the kvs, we have an internal 
       error, since all kv spaces are maintained within the pmimaster list */
    if (rc != 0) {
	MPIU_Internal_error_printf( "Could not find KV Space %s\n", 
				    kvs->kvsname );
	return 1;
    }
    return 0;
}
/* 
 * Handle an incoming "create_kvs" command
 */
static int fPMI_Handle_create_kvs( PMIProcess *pentry )
{
    char kvsname[MAXKVSNAME], outbuf[PMIU_MAXLINE];
    int rc;

    rc = fPMIKVSGetNewSpace( kvsname, sizeof(kvsname) );
    if (rc) {
	/* PANIC - allocation failed */
	return 1;
    }
    MPIU_Snprintf( outbuf, PMIU_MAXLINE, "cmd=newkvs kvsname=%s\n", kvsname );
    PMIWriteLine( pentry->fd, outbuf );
    DBG_PRINTFCOND(pmidebug, ("Handle_create_kvs new name %s\n", kvsname ) );
    return 0;
}

/* 
 * Handle an incoming "destroy_kvs" command 
 */
static int fPMI_Handle_destroy_kvs( PMIProcess *pentry )
{
    int         rc=0;
    PMIKVSpace *kvs;
    char        kvsname[MAXKVSNAME];
    char        message[PMIU_MAXLINE], outbuf[PMIU_MAXLINE];
    
    PMIU_getval( "kvsname", kvsname, MAXKVSNAME );
    kvs = fPMIKVSFindSpace( kvsname );
    if (kvs) {
	PMIKVSFree( kvs );
	MPIU_Snprintf( message, PMIU_MAXLINE,
		       "KVS_%s_successfully_destroyed", kvsname );
    }
    else {
	MPIU_Snprintf( message, PMIU_MAXLINE, "KVS %s not found", kvsname );
	rc = -1;
    }
    MPIU_Snprintf( outbuf, PMIU_MAXLINE, "cmd=kvs_destroyed rc=%d msg=%s\n",
	      rc, message );
    PMIWriteLine( pentry->fd, outbuf );
    return 0;
}

/* 
 * Handle an incoming "put" command
 */
static int fPMI_Handle_put( PMIProcess *pentry )
{
    int         rc=0;
    PMIKVSpace *kvs;
    char        kvsname[MAXKVSNAME];
    char        message[PMIU_MAXLINE], outbuf[PMIU_MAXLINE];
    char        key[MAXKEYLEN], val[MAXVALLEN];

    PMIU_getval( "kvsname", kvsname, MAXKVSNAME );
    DBG_PRINTFCOND(pmidebug,( "Put: Finding kvs %s\n", kvsname ) );

    kvs = fPMIKVSFindSpace( kvsname );
    if (kvs) {
	/* should check here for duplicate key and raise error */
	PMIU_getval( "key", key, MAXKEYLEN );
	PMIU_getval( "value", val, MAXVALLEN );
	rc = fPMIKVSAddPair( kvs, key, val );
	if (rc == 1) {
	    rc = -1;          /* no duplicate keys allowed */
	    MPIU_Snprintf( message, PMIU_MAXLINE, "duplicate_key %s", key );
	}
	else if (rc == -1) {
	    rc = -1;
	    MPIU_Snprintf( message, PMIU_MAXLINE, "no_room_in_kvs_%s",
			   kvsname );
	}
	else {
	    rc = 0;
	    MPIU_Strncpy( message, "success", PMIU_MAXLINE );
	}
    }
    else {
	rc = -1;
	MPIU_Snprintf( message, PMIU_MAXLINE, "kvs_%s_not_found", kvsname );
    }
    MPIU_Snprintf( outbuf, PMIU_MAXLINE, "cmd=put_result rc=%d msg=%s\n",
	      rc, message );
    PMIWriteLine( pentry->fd, outbuf );
    return 0;
}

/*
 * Handle incoming "get" command
 */
static int fPMI_Handle_get( PMIProcess *pentry )
{
    PMIKVSpace *kvs;
    int         rc=0;
    char        kvsname[MAXKVSNAME];
    char        message[PMIU_MAXLINE], key[PMIU_MAXLINE], value[PMIU_MAXLINE];
    char        outbuf[PMIU_MAXLINE];
    
    PMIU_getval( "kvsname", kvsname, MAXKVSNAME );
    DBG_PRINTFCOND(pmidebug,( "Get: Finding kvs %s\n", kvsname ) );

    kvs = fPMIKVSFindSpace( kvsname );
    if (kvs) {
	PMIU_getval( "key", key, PMIU_MAXLINE );
	/* Here we could intercept internal keys, e.g., 
	   pmiPrivate keys. */
	rc = fPMIKVSFindKey( kvs, key, value, sizeof(value) );
	if (rc == 0) {
	    rc = 0;
	    MPIU_Strncpy( message, "success", PMIU_MAXLINE );
	}
	else if (rc) {
	    rc = -1;
	    MPIU_Strncpy( value, "unknown", PMIU_MAXLINE );
	    MPIU_Snprintf( message, PMIU_MAXLINE, "key_%s_not_found", 
			   kvsname );
	}
    }
    else { 
	rc = -1;
	MPIU_Strncpy( value, "unknown", PMIU_MAXLINE );
	MPIU_Snprintf( message, PMIU_MAXLINE, "kvs_%s_not_found", kvsname );
    }
    MPIU_Snprintf( outbuf, PMIU_MAXLINE, 
		   "cmd=get_result rc=%d msg=%s value=%s\n",
		   rc, message, value );
    PMIWriteLine( pentry->fd, outbuf );
    DBG_PRINTFCOND( pmidebug, ("%s", outbuf ));
    return rc;
}

/* Handle an incoming get_my_kvsname command */
static int fPMI_Handle_get_my_kvsname( PMIProcess *pentry )
{
    char outbuf[PMIU_MAXLINE];
    PMIKVSpace *kvs;

    kvs = pentry->group->kvs;
    if (kvs && kvs->kvsname) {
	MPIU_Snprintf( outbuf, PMIU_MAXLINE, "cmd=my_kvsname kvsname=%s\n",
		       kvs->kvsname );
    }
    else {
	MPIU_Internal_error_printf( "Group has no associated KVS" );
	return -1;
    }
    PMIWriteLine( pentry->fd, outbuf );
    DBG_PRINTFCOND(pmidebug,( "%s", outbuf ));
    return 0;
}

/* Handle an incoming get_universe_size command */
static int fPMI_Handle_get_universe_size( PMIProcess *pentry )
{
    char outbuf[PMIU_MAXLINE];
    /* Import the universe size from the process structures */
    MPIU_Snprintf( outbuf, PMIU_MAXLINE, "cmd=universe_size size=%d\n",
		   pUniv.size );
    PMIWriteLine( pentry->fd, outbuf );
    DBG_PRINTFCOND(pmidebug,( "%s", outbuf ));
    return 0;
}

/* Handle an incoming get_appnum command */
static int fPMI_Handle_get_appnum( PMIProcess *pentry )
{
    ProcessApp *app = pentry->pState->app;
    char outbuf[PMIU_MAXLINE];
    MPIU_Snprintf( outbuf, PMIU_MAXLINE, "cmd=appnum appnum=%d\n",
		   app->myAppNum );		
    PMIWriteLine( pentry->fd, outbuf );
    DBG_PRINTFCOND(pmidebug,( "%s", outbuf ));
    return 0;
}

/* Handle an incoming "init" command */
static int fPMI_Handle_init( PMIProcess *pentry )
{
    char version[PMIU_MAXLINE];
    char subversion[PMIU_MAXLINE];
    char outbuf[PMIU_MAXLINE];
    int rc;

    /* check version compatibility with PMI client library */
    PMIU_getval( "pmi_version", version, PMIU_MAXLINE );
    PMIU_getval( "pmi_subversion", subversion, PMIU_MAXLINE );
    if (PMI_VERSION == atoi(version) && PMI_SUBVERSION >= atoi(subversion))
	rc = 0;
    else
	rc = -1;

    pentry->pState->status = PROCESS_COMMUNICATING;

    MPIU_Snprintf( outbuf, PMIU_MAXLINE,
	   "cmd=response_to_init pmi_version=%d pmi_subversion=%d rc=%d\n",
		   PMI_VERSION, PMI_SUBVERSION, rc);
    PMIWriteLine( pentry->fd, outbuf );
    DBG_PRINTFCOND(pmidebug,( "%s", outbuf ));
    return 0;
}

/* Handle an incoming "get_maxes" command */
static int fPMI_Handle_get_maxes( PMIProcess *pentry )
{
    char outbuf[PMIU_MAXLINE];
    MPIU_Snprintf( outbuf, PMIU_MAXLINE,
	      "cmd=maxes kvsname_max=%d keylen_max=%d vallen_max=%d\n",
	      MAXKVSNAME, MAXKEYLEN, MAXVALLEN );
    PMIWriteLine( pentry->fd, outbuf );
    DBG_PRINTFCOND(pmidebug,( "%s", outbuf ));
    return 0;
}

/*
 * Handle incoming "getbyidx" command
 */
static int fPMI_Handle_getbyidx( PMIProcess *pentry )
{
    int j, jNext, rc=0;
    PMIKVSpace *kvs;
    char kvsname[MAXKVSNAME], j_char[8], outbuf[PMIU_MAXLINE];
    PMIKVPair *p;

    PMIU_getval( "kvsname", kvsname, MAXKVSNAME );
    kvs = fPMIKVSFindSpace( kvsname );
    if (kvs) {
	PMIU_getval( "idx", j_char, sizeof(j_char) );
	j = atoi( j_char );
	jNext = j+1;
	if (kvs->lastIdx >= 0 && j >= kvs->lastIdx) {
	    for (p = kvs->lastByIdx, j-= kvs->lastIdx; j-- > 0 && p; 
		 p = p->nextPair );
	}
	else {
	    for (p = kvs->pairs; j-- > 0 && p; p = p->nextPair) ;
	}
	if (p) {
	    MPIU_Snprintf( outbuf, PMIU_MAXLINE, "cmd=getbyidx_results "
			   "rc=0 nextidx=%d key=%s val=%s\n",
			   jNext, p->key, p->val );
	    kvs->lastIdx   = jNext-1;
	    kvs->lastByIdx = p;
	}
	else {
	    MPIU_Snprintf( outbuf, PMIU_MAXLINE, "cmd=getbyidx_results rc=-1 "
			   "reason=no_more_keyvals\n" );
	    kvs->lastIdx   = -1;
	    kvs->lastByIdx = 0;
	}
    }
    else {
	rc = -1;
	MPIU_Snprintf( outbuf, PMIU_MAXLINE, "cmd=getbyidx_results rc=-1 "
		  "reason=kvs_%s_not_found\n", kvsname );
    }

    PMIWriteLine( pentry->fd, outbuf );
    DBG_PRINTFCOND(pmidebug,( "%s", outbuf ));
    return rc;
}

/* ------------------------------------------------------------------------- */
/* Using a host:port instead of a pre-existing FD to establish communication.
 * Steps:
 * The server sets up a listener with 
 *      PMIServSetupPort - also returns the host:port name (in that form)
 * When the server creates the process, the routine
 *      PMISetupSockets( 1, &pmiinfo )
 * will check that there is a host/port name, and initialize the fd for
 * communication to the PMI server to -1 (to indicate that no connection is
 * set up)
 * After the fork, the child will call
 *      PMISetupInClient( 1, &pmiinfo )
 * This adds the PMI_PORT and PMI_ID values to the enviroment
 * The parent also calls
 *      PMISetupFinishInServer( 1, &pmiinfo, pState )
 * ? What should this do, since there is no connection yet?
 * ? Remember about the process in pmiinfo (curPMIGroup, pState?)
 *
 * The connection process is as follows:
 *    The client (the created process), if using simple_pmi, does the
 *    following:
 *    PMI_Init checks for the environment variables PMI_FD and PMI_PORT,
 *      If PMI_FD is not set but PMI_PORT is set, then call 
 *        PMII_Connect_to_pm( hostname, portnum )
 *          This returns an open FD that will be used for further
 *          communication.  (the server side handles this through
 *          the listener)
 *      Next, use the value in the environment variable PMI_ID and
 *      call
 *        PMII_Set_from_port
 *           First, write cmd=initack pmiid=<value>
 *           Read line, require cmd=initack
 *           (This completes the handshake)
 *           Next, read a sequence of 3 commands in this order
 *           cmd=set size=<value>
 *           cmd=set rank=<value>
 *           cmd=set debug=<value>
 *        At this point, the connection is in the same state (with respect
 *        to the PMI protocol) as a new if PMI_FD was set.  In particular,
 *        the next command will be cmd=init.  Think of the port setup as
 *        a prefix on the wire protocol.
 *
 * To handle this in the server
 *       
 *
 */
/* ------------------------------------------------------------------------- */
/*
 * These routines are called when communication is established through 
 * a port instead of an fd, and no information is communicated
 * through environment variables.
 */
static int fPMI_Handle_init_port( PMIProcess *pentry )
{
    char outbuf[PMIU_MAXLINE];

    DBG_PRINTFCOND(pmidebug,
		   ( "Entering fPMI_Handle_init_port to start connection\n" ));

    /* simple_pmi wants to see cmd=initack after the initack request before
       the other data */
    PMIWriteLine( pentry->fd, "cmd=initack\n" );
    MPIU_Snprintf( outbuf, PMIU_MAXLINE, "cmd=set size=%d\n", 
		   pentry->group->nProcess );
    PMIWriteLine( pentry->fd, outbuf );
    MPIU_Snprintf( outbuf, PMIU_MAXLINE, "cmd=set rank=%d\n", 
		   pentry->pState->wRank );
    PMIWriteLine( pentry->fd, outbuf );
    MPIU_Snprintf( outbuf, PMIU_MAXLINE, "cmd=set debug=%d\n", pmidebug );
    PMIWriteLine( pentry->fd, outbuf );
    return 0;
}
/* ------------------------------------------------------------------------- */

/* Handle a spawn command.  This makes use of the spawner routine that
   was set during PMIServInit
   
   Spawn command in the simple PMI is handled as multiple lines with the
   subcommands:
   nprocs=%d
   execname=%s
   totspawns=%d