simple_pmi.c

fortran并行计算包

/* -*- Mode: C; c-basic-offset:4 ; -*- */
/*  
 *  (C) 2001 by Argonne National Laboratory.
 *      See COPYRIGHT in top-level directory.
 */

/*********************** PMI implementation ********************************/
/*
 * This file implements the client-side of the PMI interface.
 *
 * Note that the PMI client code must not print error messages (except 
 * when an abort is required) because MPI error handling is based on
 * reporting error codes to which messages are attached.  
 *
 * In v2, we should require a PMI client interface to use MPI error codes
 * to provide better integration with MPICH2.  
 */
/***************************************************************************/

#include "pmiconf.h" 

#define PMI_VERSION    1
#define PMI_SUBVERSION 1

#include <stdio.h>
#ifdef HAVE_UNISTD_H
#include <unistd.h>
#endif
#ifdef HAVE_STDLIB_H
#include <stdlib.h>
#endif
#ifdef HAVE_STRING_H
#include <string.h>
#endif
#ifdef HAVE_STRINGS_H
#include <strings.h>
#endif
#ifdef USE_PMI_PORT
#ifndef MAXHOSTNAME
#define MAXHOSTNAME 256
#endif
#endif
/* This should be moved to pmiu for shutdown */
#if defined(HAVE_SYS_SOCKET_H)
#include <sys/socket.h>
#endif
/* mpimem includes the definitions for MPIU_Snprintf, MPIU_Malloc, and 
   MPIU_Free */
#include "mpimem.h"

/* Temporary debug definitions */
#if 0
#define DBG_PRINTF(args) printf args ; fflush(stdout)
#else
#define DBG_PRINTF(args)
#endif

#include "pmi.h"
#include "simple_pmiutil.h"
#include "mpi.h"		/* to get MPI_MAX_PORT_NAME */

/* 
   These are global variable used *ONLY* in this file, and are hence
   declared static.
 */


static int PMI_fd = -1;
static int PMI_size = 1;
static int PMI_rank = 0;

/* Set PMI_initialized to 1 for singleton init but no process manager
   to help.  Initialized to 2 for normal initialization.  Initialized
   to values higher than 2 when singleton_init by a process manager.
   All values higher than 1 invlove a PM in some way.
*/
typedef enum { PMI_UNINITIALIZED = 0, 
               SINGLETON_INIT_BUT_NO_PM = 1,
	       NORMAL_INIT_WITH_PM,
	       SINGLETON_INIT_WITH_PM } PMIState;
static PMIState PMI_initialized = PMI_UNINITIALIZED;

/* ALL GLOBAL VARIABLES MUST BE INITIALIZED TO AVOID POLLUTING THE 
   LIBRARY WITH COMMON SYMBOLS */
static int PMI_kvsname_max = 0;
static int PMI_keylen_max = 0;
static int PMI_vallen_max = 0;

static int PMI_iter_next_idx = 0;
static int PMI_debug = 0;
static int PMI_debug_init = 0;    /* Set this to true to debug the init
				     handshakes */
static int PMI_spawned = 0;

/* Function prototypes for internal routines */
static int PMII_getmaxes( int *kvsname_max, int *keylen_max, int *vallen_max );
static int PMII_iter( const char *kvsname, const int idx, int *nextidx, 
		      char *key, int key_len, char *val, int val_len );
static int PMII_Set_from_port( int, int );
static int PMII_Connect_to_pm( char *, int );

static int GetResponse( const char [], const char [], int );
static int getPMIFD( int * );

#ifdef USE_PMI_PORT
static int PMII_singinit(void);
static int PMI_totalview = 0;
#endif
static int PMIi_InitIfSingleton(void);
static int accept_one_connection(int);
static char cached_singinit_key[PMIU_MAXLINE];
static char cached_singinit_val[PMIU_MAXLINE];
static char singinit_kvsname[256];

/******************************** Group functions *************************/

int PMI_Init( int *spawned )
{
    char *p;
    int notset = 1;
    int rc;

    /* FIXME: Why is setvbuf commented out? */
    /* FIXME: What if the output should be fully buffered (directed to file)?
       unbuffered (user explicitly set?) */
    /* setvbuf(stdout,0,_IONBF,0); */
    setbuf(stdout,NULL);
    /* PMIU_printf( 1, "PMI_INIT\n" ); */

    /* Get the value of PMI_DEBUG from the environment if possible, since
       we may have set it to help debug the setup process */
    p = getenv( "PMI_DEBUG" );
    if (p) PMI_debug = atoi( p );

    /* Get the fd for PMI commands; if none, we're a singleton */
    rc = getPMIFD(&notset);
    if (rc) {
	return rc;
    }

    if ( PMI_fd == -1 ) {
	/* Singleton init: Process not started with mpiexec, 
	   so set size to 1, rank to 0 */
	PMI_size = 1;
	PMI_rank = 0;
	*spawned = 0;
	
	PMI_initialized = SINGLETON_INIT_BUT_NO_PM;
	/* 256 is picked as the minimum allowed length by the PMI servers */
	PMI_kvsname_max = 256;
	PMI_keylen_max  = 256;
	PMI_vallen_max  = 256;
	
	return( 0 );
    }

    /* If size, rank, and debug are not set from a communication port,
       use the environment */
    if (notset) {
	if ( ( p = getenv( "PMI_SIZE" ) ) )
	    PMI_size = atoi( p );
	else 
	    PMI_size = 1;
	
	if ( ( p = getenv( "PMI_RANK" ) ) ) {
	    PMI_rank = atoi( p );
	    /* Let the util routine know the rank of this process for 
	       any messages (usually debugging or error) */
	    PMIU_Set_rank( PMI_rank );
	}
	else 
	    PMI_rank = 0;
	
	if ( ( p = getenv( "PMI_DEBUG" ) ) )
	    PMI_debug = atoi( p );
	else 
	    PMI_debug = 0;

	/* Leave unchanged otherwise, which indicates that no value
	   was set */
    }

/* FIXME: Why does this depend on their being a port??? */
/* FIXME: What is this for? */
#ifdef USE_PMI_PORT
    if ( ( p = getenv( "PMI_TOTALVIEW" ) ) )
	PMI_totalview = atoi( p );
    if ( PMI_totalview ) {
	char buf[PMIU_MAXLINE], cmd[PMIU_MAXLINE];
	/* FIXME: This should use a cmd/response rather than a expecting the
	   server to set a value in this and only this case */
	/* FIXME: And it most ceratainly should not happen *before* the
	   initialization handshake */
	PMIU_readline( PMI_fd, buf, PMIU_MAXLINE );
	PMIU_parse_keyvals( buf );
	PMIU_getval( "cmd", cmd, PMIU_MAXLINE );
	if ( strncmp( cmd, "tv_ready", PMIU_MAXLINE ) != 0 ) {
	    PMIU_printf( 1, "expecting cmd=tv_ready, got %s\n", buf );
	    return( PMI_FAIL );
	}
    }
#endif

    PMII_getmaxes( &PMI_kvsname_max, &PMI_keylen_max, &PMI_vallen_max );

    /* FIXME: This is something that the PM should tell the process,
       rather than deliver it through the environment */
    if ( ( p = getenv( "PMI_SPAWNED" ) ) )
	PMI_spawned = atoi( p );
    else
	PMI_spawned = 0;
    if (PMI_spawned)
	*spawned = 1;
    else
	*spawned = 0;

    if ( ! PMI_initialized )
	PMI_initialized = NORMAL_INIT_WITH_PM;

    return( 0 );
}

int PMI_Initialized( PMI_BOOL *initialized )
{
    /* Turn this into a logical value (1 or 0) .  This allows us
       to use PMI_initialized to distinguish between initialized with
       an PMI service (e.g., via mpiexec) and the singleton init, 
       which has no PMI service */
    *initialized = PMI_initialized != 0 ? PMI_TRUE : PMI_FALSE;
    return PMI_SUCCESS;
}

int PMI_Get_size( int *size )
{
    if ( PMI_initialized )
	*size = PMI_size;
    else
	*size = 1;
    return( 0 );
}

int PMI_Get_rank( int *rank )
{
    if ( PMI_initialized )
	*rank = PMI_rank;
    else
	*rank = 0;
    return( 0 );
}

/* 
 * Get_universe_size is one of the routines that needs to communicate
 * with the process manager.  If we started as a singleton init, then
 * we first need to connect to the process manager and acquire the 
 * needed information.
 */
int PMI_Get_universe_size( int *size)
{
    int  err;
    char size_c[PMIU_MAXLINE];

    /* Connect to the PM if we haven't already */
    if (PMIi_InitIfSingleton() != 0) return -1;

    if ( PMI_initialized > SINGLETON_INIT_BUT_NO_PM)  {
	err = GetResponse( "cmd=get_universe_size\n", "universe_size", 0 );
	if (err == PMI_SUCCESS) {
	    PMIU_getval( "size", size_c, PMIU_MAXLINE );
	    *size = atoi(size_c);
	    return( PMI_SUCCESS );
	}
	else return err;
    }
    else
	*size = 1;
    return( PMI_SUCCESS );
}

int PMI_Get_appnum( int *appnum )
{
    int  err;
    char appnum_c[PMIU_MAXLINE];

    if ( PMI_initialized > SINGLETON_INIT_BUT_NO_PM) {
	err = GetResponse( "cmd=get_appnum\n", "appnum", 0 );
	if (err == PMI_SUCCESS) {
	    PMIU_getval( "appnum", appnum_c, PMIU_MAXLINE );
	    *appnum = atoi(appnum_c);
	    return( PMI_SUCCESS );
	}
	else return err;
	
    }
    else
	*appnum = -1;

    return( PMI_SUCCESS );
}

int PMI_Barrier( )
{
    int err = PMI_SUCCESS;

    if ( PMI_initialized > SINGLETON_INIT_BUT_NO_PM) {
	err = GetResponse( "cmd=barrier_in\n", "barrier_out", 0 );
    }

    return err;
}

/* */
static int clique_size=-2, *clique_ranks =0;

/* pmiPrivateLocalRanks_<r> gets the local ranks for this process */
int PMI_Get_clique_size( int *size )
{
#if 1
    char buf[PMIU_MAXLINE];
    char pmi_kvsname[1024];
    int i, rc, err;

    /* As the server for the information on the */
    if (clique_size == -2 && PMI_initialized > SINGLETON_INIT_BUT_NO_PM)  {
	PMI_KVS_Get_my_name( pmi_kvsname, sizeof(pmi_kvsname) );
	rc = MPIU_Snprintf( buf, PMIU_MAXLINE, 
			    "cmd=get kvsname=%s key=pmiPrivateLocalRanks_%d\n", 
			    pmi_kvsname, PMI_rank );
	if (rc < 0) return PMI_FAIL;

	err = GetResponse( buf, "get_result", 0 );
	if (err == PMI_SUCCESS) {
	    PMIU_getval( "rc", buf, PMIU_MAXLINE );
	    rc = atoi( buf );
	    if ( rc == 0 ) {
		char *p = buf, *p0;
		/* Allocate clique_ranks and fill it in */
		PMIU_getval( "value", buf, PMIU_MAXLINE );
		/* Count the number of ranks and allocate the space for them */
		clique_size = 1;
		while (*p) {
		    if (*p++ == ',') clique_size++;
		}
		clique_ranks = (int *)MPIU_Malloc( clique_size * sizeof(int) );
		DBG_PRINTF( ("Clique_size = %d\n", clique_size) );
		p0 = p = buf;
		i  = 0;
		while (*p) {
		    while (*p && *p != ',') p++;
		    if (*p == ',') *p++ = 0;
		    clique_ranks[i++] = atoi(p0);
		    p0 = p;
		}
	    }
	    else {
		/* Default case (PM did not understand request) */
		clique_size = 1;
	    }
	}
    }
    if (clique_size < 0) *size = 1;
    else                 *size = clique_size;
#else
    *size = 1;
#endif
    return PMI_SUCCESS;
}

int PMI_Get_clique_ranks( int ranks[], int length )
{
#if 1
    int i;
    if (length < 1) 
	return PMI_ERR_INVALID_ARG;

    if (clique_size > 0 && clique_ranks) {
	for (i=0; i<length && i<clique_size; i++) 
	    ranks[i] = clique_ranks[i];
    }
    else 
	ranks[0] = PMI_rank;
    return PMI_SUCCESS;
#else
    if ( length < 1 )
	return PMI_ERR_INVALID_ARG;
    else
	return PMI_Get_rank( &ranks[0] );
#endif
}

/* Inform the process manager that we're in finalize */
int PMI_Finalize( )
{
    int err = PMI_SUCCESS;

    if ( PMI_initialized > SINGLETON_INIT_BUT_NO_PM) {
	err = GetResponse( "cmd=finalize\n", "finalize_ack", 0 );
	shutdown( PMI_fd, SHUT_RDWR );
	close( PMI_fd );
    }
    /* Free any memory that we've allocated */
    if (clique_ranks) MPIU_Free( clique_ranks );

    return err;
}

int PMI_Abort(int exit_code, const char error_msg[])
{
    PMIU_printf(1, "aborting job:\n%s\n", error_msg);
    MPIU_Exit(exit_code);
    return -1;
}

/************************************* Keymap functions **********************/

/*FIXME: need to return an error if the value of the kvs name returned is 
  truncated because it is larger than length */
/* FIXME: My name should be cached rather than re-acquired, as it is
   unchanging (after singleton init) */
int PMI_KVS_Get_my_name( char kvsname[], int length )
{
    int err;

    if (PMI_initialized == SINGLETON_INIT_BUT_NO_PM) {
	/* Return a dummy name */
	/* FIXME: We need to support a distinct kvsname for each 
	   process group */
	/* FIXME: Should the length be length (from the arg list) 
	   instead of PMIU_MAXLINE? */
	MPIU_Snprintf( kvsname, PMIU_MAXLINE, "singinit_kvs_%d_0", 
		       (int)getpid() );
	return 0;
    }
    err = GetResponse( "cmd=get_my_kvsname\n", "my_kvsname", 0 );
    if (err == PMI_SUCCESS) {
	PMIU_getval( "kvsname", kvsname, length );
    }
    return err;
}

int PMI_KVS_Get_name_length_max( int *maxlen )
{
    if (maxlen == NULL)
	return PMI_ERR_INVALID_ARG;
    *maxlen = PMI_kvsname_max;
    return PMI_SUCCESS;
}

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 )
{
    int rc = PMI_KVS_Get_my_name( id_str, length );
    return rc;
}

/* FIXME: What is this function?  How is it defined and used? */
int PMI_Get_kvs_domain_id( char id_str[], int length )
{
    return PMI_KVS_Get_my_name( id_str, length );
}

/*FIXME: need to return an error if the value of the kvs name returned is 
  truncated  because it is larger than length */
int PMI_KVS_Create( char kvsname[], int length )
{
    int err = PMI_SUCCESS;
    
    if (PMI_initialized == SINGLETON_INIT_BUT_NO_PM) {
	/* It is ok to pretend to *create* a kvs space */
	return 0;
    }

    err = GetResponse( "cmd=create_kvs\n", "newkvs", 0 );
    if (err == PMI_SUCCESS) {
	PMIU_getval( "kvsname", kvsname, length );
    }
    return err;
}

int PMI_KVS_Destroy( const char kvsname[] )
{
    char buf[PMIU_MAXLINE];
    int  err = PMI_SUCCESS;

    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 );
    err = GetResponse( buf, "kvs_destroyed", 1 );
    return err;

}

int PMI_KVS_Put( const char kvsname[], const char key[], const char value[] )