pmiserv.c

fortran并行计算包

   spawnssofar=%d
   argcnt=%d
   arg%d=%s
   preput_num=%d
   preput_key_%d=%s
   preput_val_%d=%s
   info_num=%d
   info_key_%d=%s
   info_val_%d=%s
   endcmd

   After all of the data is collected, a ProcessWorld structure is
   created and passed to the user's spawner command.  That command
   is responsible for adding the ProcessWorld to the pUniv (ProcessUniverse)
   structure (in case there is an error and the spawn fails, the user
   can decide whether to included the failed world in the pUniv structure).


   Note that this routine must create the new "current" PMI group so that
   there will be a place for the associated KV Space.
  
*/

/* Maximum number of arguments */
#define PMI_MAX_ARGS 256
static int fPMI_Handle_spawn( PMIProcess *pentry )
{
    char          inbuf[PMIU_MAXLINE];
    char          *(args[PMI_MAX_ARGS]);
    char          key[MAXKEYLEN];
    char          outbuf[PMIU_MAXLINE];
    ProcessWorld *pWorld;
    ProcessApp   *app = 0;
    int           preputNum = 0, rc;
    int           i;
    int           totspawns=0, spawnnum=0;
    PMIKVSpace    *kvs = 0;
    /* Variables for info */
    char curInfoKey[PMI_MAX_INFO_KEY], curInfoVal[PMI_MAX_INFO_VAL];
    int  curInfoIdx = -1;

    DBG_PRINTFCOND(pmidebug,( "Entering fPMI_Handle_spawn\n" ));

    if (!pentry->spawnWorld) {
	pWorld = (ProcessWorld *)MPIU_Malloc( sizeof(ProcessWorld) );
	if (!pWorld) return 1;
	
	pentry->spawnWorld = pWorld;
	pWorld->apps       = 0;
	pWorld->nProcess   = 0;
	pWorld->nextWorld  = 0;
	pWorld->nApps      = 0;
	pWorld->worldNum   = pUniv.nWorlds++;
	/* FIXME: What should be the defaults for the spawned env? 
	   Should the default be the env ov the spawner? */
	pWorld->genv       = 0;
	pentry->spawnKVS   = fPMIKVSAllocate();
    }
    else {
	pWorld = pentry->spawnWorld;
    }
    kvs    = pentry->spawnKVS;

    /* Note that each mcmd=spawn creates an app.  When all apps
       are present, then then can be linked to a world.  A 
       spawnmultiple command makes use of multiple mcmd=spawn PMI
       commands */ 

    /* Create a new app */
    app = (ProcessApp *)MPIU_Malloc( sizeof(ProcessApp) );
    if (!app) return 1;
    app->myAppNum  = 0;
    app->exename   = 0;
    app->arch      = 0;
    app->path      = 0;
    app->wdir      = 0;
    app->hostname  = 0;
    app->args      = 0;
    app->nArgs     = 0;
    app->soft.nelm = 0;
    app->nProcess  = 0;
    app->pState    = 0;
    app->nextApp   = 0;
    app->env       = 0;
    app->pWorld    = pWorld;

    /* Add to the pentry spawn structure */
    if (pentry->spawnAppTail) {
	pentry->spawnAppTail->nextApp = app;
    }
    else {
	pentry->spawnApp = app;
	pWorld->apps     = app;
    }
    pentry->spawnAppTail = app;

    for (i=0; i<PMI_MAX_ARGS; i++) args[i] = 0;

    /* Get lines until we find either cmd or mcmd (an error) or endcmd 
       (expected end) */
    while ((rc = PMIUBufferedReadLine( pentry, inbuf, sizeof(inbuf) )) > 0) {
	char *cmdPtr, *valPtr, *p;

	/* Find the command = format */
	p = inbuf;
	/* Find first nonblank */
	while (*p && isascii(*p) && isspace(*p)) p++;
	if (!*p) {
	    /* Empty string.  Ignore */
	    continue;
	}
	cmdPtr = p++;
	/* Find '=' */
	while (*p && *p != '=') p++;
	if (!*p) {
	    /* No =.  Check for endcmd */
	    p--;
	    /* Trim spaces */
	    while (isascii(*p) && isspace(*p)) p--;
	    /* Add null to end */
	    *++p = 0;
	    if (strcmp( "endcmd", cmdPtr ) == 0) { break; }
	    /* FIXME: Otherwise, we have a problem */
	    MPIU_Error_printf( "Malformed PMI command (no endcmd seen\n" );
	    return 1;
	}
	else {
	    *p = 0;
	}
	
	/* Found an = .  value is the rest of the line */
	valPtr = ++p; 
	while (*p && *p != '\n') p++;
	if (*p) *p = 0;     /* Remove the newline */

	/* Now, process the cmd and value */
	if (strcmp( "nprocs", cmdPtr ) == 0) {
	    app->nProcess     = atoi(valPtr);
	    pWorld->nProcess += app->nProcess;
	}
	else if (strcmp( "execname", cmdPtr ) == 0) {
	    app->exename = MPIU_Strdup( valPtr );
	}
	else if (strcmp( "totspawns", cmdPtr ) == 0) {
	    /* This tells us how many separate spawn commands
	       we expect to see (e.g., for spawn multiple).
	       Each spawn command is a separate "app" */
	    totspawns = atoi(valPtr);
	}
	else if (strcmp( "spawnssofar", cmdPtr ) == 0) {
	    /* This tells us which app we are (starting from 1) */
	    spawnnum      = atoi(valPtr);
	    app->myAppNum = spawnnum - 1;
	}
	else if (strcmp( "argcnt", cmdPtr ) == 0) {
	    /* argcnt may not be set before the args */
	    app->nArgs = atoi(valPtr);
	}
	else if (strncmp( "arg", cmdPtr, 3 ) == 0) {
	    int argnum;
	    /* argcnt may not be set before the args */
	    /* Handle arg%d.  Values are 1 - origin */
	    argnum = atoi( cmdPtr + 3 ) - 1;
	    if (argnum < 0 || argnum >= PMI_MAX_ARGS) {
		MPIU_Error_printf( "Malformed PMI Spawn command; the index of an argument in the command is %d but must be between 0 and %d\n",
				   argnum, PMI_MAX_ARGS );
		return 1;
	    }
	    args[argnum] = MPIU_Strdup( valPtr );
	}
	else if (strcmp( "preput_num", cmdPtr ) == 0) {
	    preputNum = atoi(valPtr);
	}
	else if (strncmp( "preput_key_", cmdPtr, 11 ) == 0) {
	    /* Save the key */
	    MPIU_Strncpy( key, valPtr, sizeof(key) );
	}
	else if (strncmp( "preput_val_", cmdPtr, 11 ) == 0) {
	    /* Place the key,val into the space associate with the current 
	       PMI group */
	    fPMIKVSAddPair( kvs, key, valPtr );
	}
	/* Info is on a per-app basis (it is an array of info items in
	   spawn multiple).  We can ignore most info values.
	   The ones that are handled are processed by a 
	   separate routine (not yet implemented).
	   simple_pmi.c sends (key,value), so we can keep just the
	   last key and pass the key/value to the registered info
	   handler, along with tha app structure.  Alternately,
	   we could save all info items and let the user's 
	   spawner handle it */
	else if (strcmp( "info_num", cmdPtr ) == 0) {
	    /* Number of info values */
	    ;
	}
	else if (strncmp( "info_key_", cmdPtr, 9 ) == 0) {
	    /* The actual name has a digit, which indicates *which* info 
	       key this is */
	    curInfoIdx = atoi( cmdPtr + 9 );
	    MPIU_Strncpy( curInfoKey, valPtr, sizeof(curInfoKey) );
	}
	else if (strncmp( "info_val_", cmdPtr, 9 ) == 0) {
	    /* The actual name has a digit, which indicates *which* info 
	       value this is */
	    int idx = atoi( cmdPtr + 9 );
	    if (idx != curInfoIdx) {
		MPIU_Error_printf( "Malformed PMI command: info keys and values not ordered as expected (expected value %d but got %d)\n", curInfoIdx, idx );
		return 1;
	    }
	    else {
		MPIU_Strncpy( curInfoVal, valPtr, sizeof(curInfoVal) );
		/* Apply this info item */
		fPMIInfoKey( app, curInfoKey, curInfoVal );
		/* printf( "Got info %s+%s\n", curInfoKey, curInfoVal ); */
	    }
	}
	else {
	    MPIU_Error_printf( "Unrecognized PMI subcommand on spawnmult: %s\n",
			       cmdPtr );
	    return 1;
	}
    }	

    if (app->nArgs > 0) {
	app->args  = (const char **)MPIU_Malloc( app->nArgs * sizeof(char *) );
	for (i=0; i<app->nArgs; i++) {
	    app->args[i] = args[i];
	    args[i]      = 0;
	}
    }

    pWorld->nApps ++;

    /* Now that we've read the commands, invoke the user's spawn command */
    if (totspawns == spawnnum) {
	PMISetupNewGroup( pWorld->nProcess, kvs );
	
	if (userSpawner) {
	    rc = (*userSpawner)( pWorld, userSpawnerData );
	}
	else {
	    MPIU_Error_printf( "Unable to spawn %s\n", app->exename );
	    rc = 1;
	    MPIE_PrintProcessWorld( stdout, pWorld );
	}
	
	MPIU_Snprintf( outbuf, PMIU_MAXLINE, "cmd=spawn_result rc=%d\n", rc );
	PMIWriteLine( pentry->fd, outbuf );
	DBG_PRINTFCOND(pmidebug,( "%s", outbuf ));

	/* Clear for the next spawn */
	pentry->spawnApp     = 0;
	pentry->spawnAppTail = 0;
	pentry->spawnKVS     = 0;
	pentry->spawnWorld   = 0;
    }
    
    /* If totspawnnum != spawnnum, then we are expecting a 
       spawnmult with additional items */
    return 0;
}

/* ------------------------------------------------------------------------- */
/*  
 * FIXME:
 * Question: What does this need to do?
 * 1.  Is nproces in range?
 * 2.  Is the program executable?
 * 3.  Create the KVS group
 * 4.  Invoke startup -> the process startup procedure must have
 *     been registered by the calling program (e.g., usually by
 *     mpiexec, but possibly a separate pmiserver process)
 * 5. return kvsname; return code
 *    How do we handle soft (no specific return size required).
 *    Also, is part fo the group associated with these processes or
 *    another group (the spawner?) of processes?
 *
 * This should be called after receiving the cmd=initack from the client.
 */
void PMI_Init_remote_proc( int fd, PMIProcess *pentry )
{
    /* Everything else should be setup by PMISetupNewProcess */
    fPMI_Handle_init_port( pentry );
}

/* 
 * This is a special routine.  It accepts the first input from the
 * remote process, and returns the PMI_ID value.  -1 is returned on error 
 */
int PMI_Init_port_connection( int fd )
{
    char message[PMIU_MAXLINE], cmd[MAXPMICMD];
    int pmiid = -1;

    DBG_PRINTFCOND(pmidebug,( "Beginning initial handshake read\n" ));
    PMIReadLine( fd, message, PMIU_MAXLINE );
    DBG_PRINTFCOND(pmidebug,( "received message %s\n", message ));

    PMIU_parse_keyvals( message );
    PMIU_getval( "cmd", cmd, MAXPMICMD );
    if (strcmp(cmd,"initack")) {
	PMIU_printf( 1, "Unexpected cmd %s, expected initack\n", cmd );
	return -1;
    }
    PMIU_getval( "pmiid", cmd, MAXPMICMD );
    pmiid = atoi(cmd);

    return pmiid;
}

/* Implement the singleton init handshake.  See the discussion in 
   simplepmi.c for the protocol */
int PMI_InitSingletonConnection( int fd, PMIProcess *pmiprocess )
{
    char buf[PMIU_MAXLINE], cmd[PMIU_MAXLINE];
    int  rc;
    char version[PMIU_MAXLINE], subversion[PMIU_MAXLINE];
    
    /* We start with the singinit command, wait for the singinit from
       the client, and then send the singinit_info */
    MPIU_Snprintf( buf, PMIU_MAXLINE, 
   "cmd=singinit pmi_version=%d pmi_subversion=%d stdio=no authtype=none\n",
	   PMI_VERSION, PMI_SUBVERSION );
    PMIWriteLine( fd, buf );
    PMIReadLine( fd, buf, PMIU_MAXLINE );
    PMIU_parse_keyvals( buf );
    PMIU_getval( "cmd", cmd, MAXPMICMD );
    if (strcmp(cmd,"singinit")) {
	PMIU_printf( 1, "Unexpected cmd %s\n", cmd );
	return -1;
    }
    /* Could look at authtype */
    /* 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;
    
    MPIU_Snprintf( buf, PMIU_MAXLINE,
		   "cmd=singinit_info versionok=%s stdio=no kvsname=%s\n",
		   (rc == 0) ? "yes" : "no",  
		   (char *)(pmiprocess->group->kvs->kvsname) );
    PMIWriteLine( fd, buf );

    return 0;
}

/* Handle the default info values */
static int fPMIInfoKey( ProcessApp *app, const char key[], const char val[] )
{
    if (strcmp( key, "host" ) == 0) {
	app->hostname = MPIU_Strdup( val );
    }
    else if (strcmp( key, "arch" ) == 0) {
	app->arch     = MPIU_Strdup( val );
    }
    else if (strcmp( key, "wdir" ) == 0) {
	app->wdir     = MPIU_Strdup( val );
    }
    else if (strcmp( key, "path" ) == 0) {
	app->path     = MPIU_Strdup( val );
    }
    else if (strcmp( key, "soft" ) == 0) {
	MPIE_ParseSoftspec( val, &app->soft );
    }
    else {
	/* FIXME: call user-specified info handler, if any.
	   Unspecified info keys are ignored */
    }
    return 0;
}

/* ------------------------------------------------------------------------- */
#ifndef PMIWriteLine
int PMIWriteLine( int fd, const char *buf )
{
    int rc;

    DBG_PRINTFCOND(pmidebug,( "Writing to fd %d the line :%s:\n", fd, buf ));
    rc = PMIU_writeline( fd, (char*)buf );
    DBG_PRINTFCOND(pmidebug&&rc<0,( "Write on fd %d returned rc %d\n", fd, rc ));

    return rc;
}
int PMIReadLine( int fd, char *buf, int maxlen )
{
    int rc;

    /* If no data is read, PMIU_readline doesn't place a null in the
       buffer.  We make sure that buf[0] is always valid */
    buf[0] = 0;
    rc = PMIU_readline( fd, buf, maxlen );
    if (pmidebug) {
	if (rc < 0) {
	    DBG_PRINTFCOND(1,( "Read on fd %d returned rc %d\n", fd, rc ) );
	}
	else {
	    DBG_PRINTFCOND(1,( "Read from fd %d the line :%s:\n", fd, buf ) );
	}
	DBG_COND(1,fflush(stdout));
    }

    return rc;
}
#endif
/* ------------------------------------------------------------------------- */
/* 
 * To handle multiple incoming command streams, we need to have a separate
 * states buffered input (replacing PMIU_readline from the client-side
 * code)
 */
/* ------------------------------------------------------------------------- */
/* 
 * Return the next newline-terminated string of maximum length maxlen.
 * This is a buffered version, and reads from fd as necessary.  A
 */
static int PMIUBufferedReadLine( PMIProcess *pentry, char *buf, int maxlen )
{
    int curlen, n;
    char *p, ch;
    int  fd = pentry->fd;
    char *readbuf  = pentry->readBuf;
    char *nextChar = pentry->nextChar;
    char *endChar  = pentry->endChar;

    p      = buf;
    curlen = 1;    /* Make room for the null */
    while (curlen < maxlen) {
	if (nextChar == endChar) {
	    do {
		n = read( fd, readbuf, sizeof(readbuf)-1 );
	    } while (n == -1 && errno == EINTR);
	    if (n == 0) {
		/* EOF */
		break;
	    }
	    else if (n < 0) {
		/* Error.  Return a negative value if there is no
		   data.  Save the errno in case we need to return it
		   later. */
		if (curlen == 1) {
		    curlen = 0;
		}
		break;
	    }
	    nextChar = readbuf;
	    endChar  = readbuf + n;
	    pentry->endChar  = endChar;
	    /* Add a null at the end just to make it easier to print
	       the read buffer */
	    readbuf[n] = 0;
	    /* FIXME: Make this an optional output */
	    /* printf( "Readline %s\n", readbuf ); */
	}
	
	ch   = *nextChar++;
	*p++ = ch;
	curlen++;
	if (ch == '\n') break;
    }

    /* We null terminate the string for convenience in printing */
    *p = 0;

    /* Save the state of the buffer */
    pentry->nextChar = nextChar;

    /* Return the number of characters, not counting the null */
    return curlen-1;
}

pmix_preput( const char *key, const char *val )
{
    fPMIKVSAddPair( curPMIGroup->kvs, key, val );
}