mpidi_pg.c

fortran并行计算包

/* Return the kvsname associated with the MPI_COMM_WORLD of this process. */
int MPIDI_PG_GetConnKVSname( char ** kvsname )
{
    *kvsname = pg_world->connData;
    return MPI_SUCCESS;
}

/* For process groups that are not our MPI_COMM_WORLD, store the connection
   information in an array of strings.  These routines and structure
   implement the access to this information. */
typedef struct {
    int     toStringLen;   /* Length needed to encode this connection info */
    char ** connStrings;   /* pointer to an array, indexed by rank, containing
			      connection information */
} MPIDI_ConnInfo;

static int getConnInfo( int rank, char *buf, int bufsize, MPIDI_PG_t *pg )
{
    MPIDI_ConnInfo *connInfo = (MPIDI_ConnInfo *)pg->connData;

    /* printf( "Entering getConnInfo\n" ); fflush(stdout); */
    if (!connInfo || !connInfo->connStrings || !connInfo->connStrings[rank]) {
	/* FIXME: Turn this into a valid error code create/return */
	printf( "Fatal error in getConnInfo (rank = %d)\n", rank );
	printf( "connInfo = %p\n", connInfo );fflush(stdout);
	if (connInfo) {
	    printf( "connInfo->connStrings = %p\n", connInfo->connStrings );
	}
	/* Fatal error.  Connection information missing */
	fflush(stdout);   
    }

    /* printf( "Copying %s to buf\n", connInfo->connStrings[rank] ); fflush(stdout); */
    
    MPIU_Strncpy( buf, connInfo->connStrings[rank], bufsize );
    return MPI_SUCCESS;
}
static int connToString( char **buf_p, int *slen, MPIDI_PG_t *pg )
{
    char *string = 0, *str, *pg_id;
    int  i, len=0;
    
    MPIDI_ConnInfo *connInfo = (MPIDI_ConnInfo *)pg->connData;

    /* Create this from the string array */
    string = (char *)MPIU_Malloc( connInfo->toStringLen );
    str = string;

    pg_id = pg->id;
    /* FIXME: This is a hack, and it doesn't even work */
    /*    MPIDI_PrintConnStrToFile( stdout, __FILE__, __LINE__, 
	  "connToString: pg id is", (char *)pg_id );*/
    /* This is intended to cause a process to transition from a singleton
       to a non-singleton. */
    if (strstr( pg_id, "singinit_kvs" ) == pg_id) {
	PMI_Get_id( pg->id, 256 );
    }
    while (*pg_id) str[len++] = *pg_id++;
    str[len++] = 0;
    
    MPIU_Snprintf( &str[len], 20, "%d", pg->size);
    /* Skip over the length */
    while (str[len++]);

    /* Copy each connection string */
    for (i=0; i<pg->size; i++) {
	char *p = connInfo->connStrings[i];
	while (*p) { str[len++] = *p++; }
	str[len++] = 0;
    }

    if (len > connInfo->toStringLen) {
	*buf_p = 0;
	*slen  = 0;
	return MPIR_Err_create_code(MPI_SUCCESS,MPIR_ERR_FATAL,"connToString",
			    __LINE__, MPI_ERR_INTERN, "**intern", NULL);
    }

    *buf_p = string;
    *slen = len;

    return MPI_SUCCESS;
}
static int connFromString( const char *buf, MPIDI_PG_t *pg )
{
    MPIDI_ConnInfo *conninfo = 0;
    int i, mpi_errno = MPI_SUCCESS;
    const char *buf0 = buf;   /* save the start of buf */

    /* printf( "Starting with buf = %s\n", buf );fflush(stdout); */

    /* Skip the pg id */
    while (*buf) buf++; buf++;

    /* Determine the size of the pg */
    pg->size = atoi( buf );
    while (*buf) buf++; buf++;

    conninfo = (MPIDI_ConnInfo *)MPIU_Malloc( sizeof(MPIDI_ConnInfo) );
    conninfo->connStrings = (char **)MPIU_Malloc( pg->size * sizeof(char *));

    /* For now, make a copy of each item */
    for (i=0; i<pg->size; i++) {
	/* printf( "Adding conn[%d] = %s\n", i, buf );fflush(stdout); */
	conninfo->connStrings[i] = MPIU_Strdup( buf );
	while (*buf) buf++;
	buf++;
    }
    pg->connData = conninfo;
	
    /* Save the length of the string needed to encode the connection
       information */
    conninfo->toStringLen = (int)(buf - buf0) + 1;

    return mpi_errno;
}
static int connFree( MPIDI_PG_t *pg )
{
    MPIDI_ConnInfo *conninfo = (MPIDI_ConnInfo *)pg->connData;
    int i;

    for (i=0; i<pg->size; i++) {
	MPIU_Free( conninfo->connStrings[i] );
    }
    MPIU_Free( conninfo->connStrings );
    MPIU_Free( conninfo );

    return MPI_SUCCESS;
}

#ifdef USE_DBG_LOGGING
/* This is a temporary routine that is used to print out the pg string.
   A better approach may be to convert it into a single (long) string
   with no nulls. */
int MPIDI_PrintConnStr( const char *file, int line, 
			const char *label, const char *str )
{
    int pg_size, i;

    MPIU_DBG_Outevent( file, line, MPIU_DBG_CH3_CONNECT, 0, label );
    MPIU_DBG_Outevent( file, line, MPIU_DBG_CH3_CONNECT, 0, str );
    
    /* Skip the pg id */
    while (*str) str++; str++;

    /* Determine the size of the pg */
    pg_size = atoi( str );
    while (*str) str++; str++;

    for (i=0; i<pg_size; i++) {
	MPIU_DBG_Outevent( file, line, MPIU_DBG_CH3_CONNECT, 0, str );
	while (*str) str++;
	str++;
    }
    return 0;
}
int MPIDI_PrintConnStrToFile( FILE *fd, const char *file, int line, 
			      const char *label, const char *str )
{
    int pg_size, i;

    fprintf( fd, "ConnStr from %s(%d); %s\n\t%s\n", file, line, label, str );
    
    /* Skip the pg id */
    while (*str) str++; str++;

    fprintf( fd, "\t%s\n", str );
    /* Determine the size of the pg */
    pg_size = atoi( str );
    while (*str) str++; str++;

    for (i=0; i<pg_size; i++) {
	fprintf( fd, "\t%s\n", str ); 
	while (*str) str++;
	str++;
    }
    fflush(stdout);
    return 0;
}
#endif

int MPIDI_PG_InitConnString( MPIDI_PG_t *pg )
{
    int mpi_errno = MPI_SUCCESS;

    pg->connData           = 0;
    pg->getConnInfo        = getConnInfo;
    pg->connInfoToString   = connToString;
    pg->connInfoFromString = connFromString;
    pg->freeConnInfo       = connFree;

    return mpi_errno;
}

/* Temp to get connection value for rank r */
#undef FUNCNAME
#define FUNCNAME MPIDI_PG_GetConnString
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
int MPIDI_PG_GetConnString( MPIDI_PG_t *pg, int rank, char *val, int vallen )
{
    int mpi_errno = MPI_SUCCESS;

    if (pg->getConnInfo) {
	mpi_errno = (*pg->getConnInfo)( rank, val, vallen, pg );
    }
    else {
	MPIU_Internal_error_printf( "Panic: no getConnInfo defined!\n" );
    }

    return mpi_errno;
}


#undef FUNCNAME
#define FUNCNAME MPIDI_PG_Dup_vcr
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
/*@
  MPIDI_PG_Dup_vcr - Duplicate a virtual connection from a process group

  Notes:
  This routine provides a dup of a virtual connection given a process group
  and a rank in that group.  This routine is used only in initializing
  the MPI-1 communicators 'MPI_COMM_WORLD' and 'MPI_COMM_SELF', and in creating
  the initial intercommunicator after an 'MPI_Comm_spawn', 
  'MPI_Comm_spawn_multiple', or 'MPI_Comm_connect/MPI_Comm_accept'.  

  In addition to returning a dup of the virtual connection, it manages the
  reference count of the process group, which is always the number of inuse
  virtual connections.
  @*/
int MPIDI_PG_Dup_vcr( MPIDI_PG_t *pg, int rank, MPIDI_VC_t **vc_p )
{
    MPIDI_VC_t *vc;
    MPIDI_STATE_DECL(MPID_STATE_MPIDI_PG_DUP_VCR);

    MPIDI_FUNC_ENTER(MPID_STATE_MPIDI_PG_DUP_VCR);

    vc = &pg->vct[rank];
    /* Increase the reference count of the vc.  If the reference count 
       increases from 0 to 1, increase the reference count of the 
       process group *and* the reference count of the vc (this
       allows us to distinquish between Comm_free and Comm_disconnect) */
    /* FIXME: This should be a fetch and increment for thread-safety */
    if (vc->ref_count == 0) {
	MPIDI_PG_add_ref(pg);
	MPIDI_VC_add_ref(vc);
    }
    MPIDI_VC_add_ref(vc);
    *vc_p = vc;

    MPIDI_FUNC_EXIT(MPID_STATE_MPIDI_PG_DUP_VCR);
    return MPI_SUCCESS;
}

/* FIXME: This routine should invoke a close method on the connection,
   rather than have all of the code here */
#undef FUNCNAME
#define FUNCNAME MPIDI_PG_Close_VCs
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
/*@
  MPIDI_PG_Close_VCs - Close all virtual connections on all process groups.
  
  Note:
  This routine is used in MPID_Finalize.  It is here to keep the process-group
  related functions together.
  @*/
int MPIDI_PG_Close_VCs( void )
{
    MPIDI_PG_t * pg = MPIDI_PG_list;
    int mpi_errno = MPI_SUCCESS;
    MPIDI_STATE_DECL(MPID_STATE_MPIDI_PG_CLOSE_VCS);

    MPIDI_FUNC_ENTER(MPID_STATE_MPIDI_PG_CLOSE_VCS);

    while (pg) {
	int i, inuse;

	MPIU_DBG_MSG_S(CH3_DISCONNECT,VERBOSE,"Closing vcs for pg %s",
		       (char *)pg->id );


	for (i = 0; i < pg->size; i++)
	{
	    MPIDI_VC_t * vc = &pg->vct[i];
	    /* If the VC is myself then skip the close message */
	    if (pg == MPIDI_Process.my_pg && i == MPIDI_Process.my_pg_rank) {
                if (vc->ref_count != 0) {
                    MPIDI_PG_release_ref(pg, &inuse);
                }
		continue;
	    }

	    if (vc->state == MPIDI_VC_STATE_ACTIVE || 
		vc->state == MPIDI_VC_STATE_REMOTE_CLOSE
#if defined(MPIDI_CH3_USES_SSHM) && 0
		/* FIXME: Remove this IFDEF */
		/* sshm queues are uni-directional.  A VC that is connected 
		 * in the read direction is marked MPIDI_VC_STATE_INACTIVE
		 * so that a connection will be formed on the first write.  
		 * Since the other side is marked MPIDI_VC_STATE_ACTIVE for 
		 * writing 
		 * we need to initiate the close protocol on the read side 
		 * even if the write state is MPIDI_VC_STATE_INACTIVE. */
		|| ((vc->state == MPIDI_VC_STATE_INACTIVE) && 
		    ((MPIDI_CH3I_VC *)(vc->channel_private))->shm_read_connected)
#endif
		)
	    {
		MPIDI_CH3U_VC_SendClose( vc, i );
	    }
	    else
	    {
                if (vc->state == MPIDI_VC_STATE_INACTIVE && vc->ref_count != 0) {
		    /* FIXME: If the reference count for the vc is not 0,
		       something is wrong */
                    MPIDI_PG_release_ref(pg, &inuse);
                }

		MPIU_DBG_MSG_FMT(CH3_DISCONNECT,VERBOSE,(MPIU_DBG_FDEST,
		     "vc=%p: not sending a close to %d, vc in state %s", vc,i,
		     MPIDI_VC_GetStateString(vc->state)));
	    }
	}
	pg = pg->next;
    }
    /* Note that we do not free the process groups within this routine, even
       if the reference counts have gone to zero. That is done once the 
       connections are in fact closed (by the final progress loop that
       handles any close requests that this code generates) */

    MPIDI_FUNC_EXIT(MPID_STATE_MPIDI_PG_CLOSE_VCS);
    return mpi_errno;
}

/*
 * This routine may be called to print the contents (including states and
 * reference counts) for process groups.
 */
int MPIU_PG_Printall( FILE *fp )
{
    MPIDI_PG_t *pg;
    int         i;
		  
    pg = MPIDI_PG_list;

    fprintf( fp, "Process groups:\n" );
    while (pg) {
	fprintf( fp, "size = %d, refcount = %d, id = %s\n", 
		 pg->size, pg->ref_count, (char *)pg->id );
	for (i=0; i<pg->size; i++) {
	    fprintf( fp, "\tVCT rank = %d, refcount = %d, lpid = %d, state = %d \n", 
		     pg->vct[i].pg_rank, pg->vct[i].ref_count,
		     pg->vct[i].lpid, (int)pg->vct[i].state );
	}
	fflush(fp);
	pg = pg->next;
    }

    return 0;
}

int MPIDI_PG_CheckForSingleton( void )
{
    if (strstr((char*)pg_world->id,"singinit_kvs") == (char *)pg_world->id) {
	char buf[256];
	/* Force an enroll */
	PMI_KVS_Get( "foobar", "foobar", buf, sizeof(buf) );
	PMI_Get_id( pg_world->id, 256 );
	PMI_KVS_Get_my_name( pg_world->connData, 256 );
    }
    return MPI_SUCCESS;
}