pmiserv.c

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

 * 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
   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 = PMIReadLine( pentry->fd, 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\n", cmd );
	return -1;
    }
    PMIU_getval( "pmiid", cmd, MAXPMICMD );
    pmiid = atoi(cmd);

    return pmiid;
}

/* 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
/* ------------------------------------------------------------------------- */