clog_commset.c

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

    /*
       Broadcast local_intracommIDs's GID to everyone in remote_intracomm, i.e.
       Send local_intracommIDs' GID from the root of local intracomms to the
       root of the remote intracomms and save it as remote_intracommIDs' GID.
       Then broadcast the received GID to the rest of intracomm.
    */
    if ( intercommIDs->comm_rank == 0 ) {
       PMPI_Irecv( remote_intracommIDs->global_ID, CLOG_UUID_SIZE, MPI_CHAR,
                   0, 9999, intercomm, &request );            
       PMPI_Send( local_intracommIDs->global_ID, CLOG_UUID_SIZE, MPI_CHAR,
                  0, 9999, intercomm );
       PMPI_Wait( &request, &status );
    }
    PMPI_Bcast( remote_intracommIDs->global_ID, CLOG_UUID_SIZE, MPI_CHAR,
                0, orig_intracommIDs->comm );
    /*
       Since REMOTE intracomm is NOT known or undefinable in LOCAL intracomm,
       (that is why we have intercomm).  Set comm and comm_rank to NULL
    */
    remote_intracommIDs->comm       = MPI_COMM_NULL;
    remote_intracommIDs->comm_rank  = -1;

    /* Set the related CLOG_CommIDs_t* to the local and remote intracommIDs */
    intercommIDs->next              = local_intracommIDs;
    local_intracommIDs->next        = remote_intracommIDs;

    return intercommIDs;
}



/*
    The input argument MPI_Comm is assumed not to be MPI_COMM_NULL
*/
CLOG_CommLID_t CLOG_CommSet_get_LID( CLOG_CommSet_t *commset, MPI_Comm comm )
{
    MPI_Aint  ptrlen_value;
    int       istatus;

    PMPI_Comm_get_attr( comm, commset->LID_key, &ptrlen_value, &istatus );
    if ( !istatus ) {
        fprintf( stderr, __FILE__":CLOG_CommSet_get_LID() - \n"
                         "\t""PMPI_Comm_get_attr() fails!\n" );
        fflush( stderr );
        CLOG_Util_abort( 1 );
    }
    return (CLOG_CommLID_t) ptrlen_value;
}

const CLOG_CommIDs_t* CLOG_CommSet_get_IDs( CLOG_CommSet_t *commset,
                                            MPI_Comm comm )
{
    MPI_Aint  ptrlen_value;
    int       istatus;

    PMPI_Comm_get_attr( comm, commset->LID_key, &ptrlen_value, &istatus );
    if ( !istatus ) {
        fprintf( stderr, __FILE__":CLOG_CommSet_get_IDs() - \n"
                         "\t""PMPI_Comm_get_attr() fails!\n" );
        fflush( stderr );
        CLOG_Util_abort( 1 );
    }
    return &(commset->table[ (CLOG_CommLID_t) ptrlen_value ]);
}

const CLOG_CommIDs_t* CLOG_CommTable_get( const CLOG_CommIDs_t *table,
                                                int             table_count,
                                          const CLOG_CommGID_t  commgid )
{
    int              idx;

    for ( idx = 0; idx < table_count; idx++ ) {
        if ( CLOG_Uuid_compare( table[idx].global_ID, commgid ) == 0 )
            return &(table[idx]);
    }
    return NULL;
}

const CLOG_CommIDs_t *CLOG_CommSet_add_GID( CLOG_CommSet_t *commset,
                                            CLOG_CommGID_t  commgid )
{
    CLOG_CommIDs_t  *commIDs;

    /* Update the next available table entry in CLOG_CommSet_t */
    commIDs              = CLOG_CommSet_get_new_IDs( commset );
    commIDs->kind        = CLOG_COMM_KIND_UNKNOWN;

    /* Set the global Comm ID */
    CLOG_Uuid_copy( commgid, commIDs->global_ID );

    /* Set the Comm field's */
    commIDs->comm        = MPI_COMM_NULL;
    commIDs->comm_rank   = -1;

    return commIDs;
}

void CLOG_CommSet_merge( CLOG_CommSet_t *commset )
{
          int              comm_world_size, comm_world_rank;
          int              rank_sep, rank_quot, rank_rem;
          int              rank_src, rank_dst;
          int              recv_table_count, recv_table_size;
          CLOG_CommIDs_t  *recv_table;
    const CLOG_CommIDs_t  *commIDs;
          MPI_Status       status;
          int              idx;

    comm_world_rank  = commset->world_rank;
    comm_world_size  = commset->world_size;

    /* Collect CLOG_CommIDs_t table through Recursive Doubling algorithm */
    rank_sep   = 1;
    rank_quot  = comm_world_rank >> 1; /* rank_quot  = comm_world_rank / 2; */
    rank_rem   = comm_world_rank &  1; /* rank_rem   = comm_world_rank % 2; */

    while ( rank_sep < comm_world_size ) {
        if ( rank_rem == 0 ) {
            rank_src = comm_world_rank + rank_sep;
            if ( rank_src < comm_world_size ) {
                /* Recv from rank_src */
                PMPI_Recv( &recv_table_count, 1, MPI_INT, rank_src,
                           CLOG_COMM_TAG_START, MPI_COMM_WORLD, &status );
                recv_table_size  = recv_table_count * sizeof(CLOG_CommIDs_t);
                recv_table  = (CLOG_CommIDs_t *) MALLOC( recv_table_size );
                if ( recv_table == NULL ) {
                    fprintf( stderr, __FILE__":CLOG_CommSet_merge() - \n"
                                     "\t""MALLOC(%d) fails!\n",
                                     recv_table_size );
                    fflush( stderr );
                    CLOG_Util_abort( 1 );
                }
                /*
                   For simplicity, receive commset's whole table and uses only
                   CLOG_CommGID_t column from the table.  The other columns
                   are relevant only to the sending process.
                */
                PMPI_Recv( recv_table, recv_table_size, MPI_CHAR, rank_src,
                           CLOG_COMM_TAG_START+1, MPI_COMM_WORLD, &status );

                for ( idx = 0; idx < recv_table_count; idx++ ) {
                    /*
                       If recv_table[ idx ].global_ID is not in commset,
                       add recv_table[ idx ].global_ID to commset
                    */
                    commIDs = CLOG_CommTable_get( commset->table,
                                                  commset->count,
                                                  recv_table[ idx ].global_ID );
                    if ( commIDs == NULL ) 
                        CLOG_CommSet_add_GID( commset,
                                              recv_table[ idx ].global_ID );
                }
                if ( recv_table != NULL ) {
                    FREE( recv_table );
                    recv_table = NULL;
                }
            }
        }
        else /* if ( rank_rem != 0 ) */ {
            /* After sending CLOG_CommIDs_t table, the process does a barrier */
            rank_dst = comm_world_rank - rank_sep;
            if ( rank_dst >= 0 ) {
                recv_table_count  = commset->count;
                /* Send from rank_dst */
                PMPI_Send( &recv_table_count, 1, MPI_INT, rank_dst,
                           CLOG_COMM_TAG_START, MPI_COMM_WORLD );
                /*
                   For simplicity, send commset's whole table including
                   useless things even though only CLOG_CommGID_t column
                   will be used from the table in the receiving process.
                */
                recv_table_size  = recv_table_count * sizeof(CLOG_CommIDs_t);
                PMPI_Send( commset->table, recv_table_size, MPI_CHAR, rank_dst,
                           CLOG_COMM_TAG_START+1, MPI_COMM_WORLD );
                break;  /* get out of the while loop */
            }
        }

        rank_rem    = rank_quot & 1; /* rank_rem   = rank_quot % 2; */
        rank_quot >>= 1;             /* rank_quot /= 2; */
        rank_sep  <<= 1;             /* rank_sep  *= 2; */
    }   /* endof while ( rank_sep < comm_world_size ) */

    /*
       Synchronize everybody in MPI_COMM_WORLD
       before broadcasting result back to everybody.
    */
    PMPI_Barrier( MPI_COMM_WORLD );

    if ( comm_world_rank == 0 )
        recv_table_count  = commset->count;
    else
        recv_table_count  = 0;
    MPI_Bcast( &recv_table_count, 1, MPI_INT, 0, MPI_COMM_WORLD );

    /* Allocate a buffer for root process's commset->table */
    recv_table_size  = recv_table_count * sizeof(CLOG_CommIDs_t);
    recv_table  = (CLOG_CommIDs_t *) MALLOC( recv_table_size );
    if ( recv_table == NULL ) {
        fprintf( stderr, __FILE__":CLOG_CommSet_merge() - \n"
                         "\t""MALLOC(%d) fails!\n", recv_table_size );
        fflush( stderr );
        CLOG_Util_abort( 1 );
    }

    if ( comm_world_rank == 0 )
        memcpy( recv_table, commset->table, recv_table_size );
    MPI_Bcast( recv_table, recv_table_size, MPI_CHAR, 0, MPI_COMM_WORLD );

    /* Update the local_ID in CLOG_CommSet_t's table to globally unique ID */
    for ( idx = 0; idx < commset->count; idx++ ) {
        commIDs  = CLOG_CommTable_get( recv_table, recv_table_count,
                                       commset->table[idx].global_ID ); 
        if ( commIDs == NULL ) {
            char uuid_str[CLOG_UUID_STR_SIZE] = {0};
            CLOG_Uuid_sprint( commset->table[idx].global_ID, uuid_str );
            fprintf( stderr, __FILE__":CLOG_CommSet_merge() - \n"
                             "\t""collected table from root does not contain "
                             "%d-th GID %s/n", idx, uuid_str );
            fflush( stderr );
            CLOG_Util_abort( 1 );
        }
        /* if commIDs != NULL */
        commset->table[idx].local_ID = commIDs->local_ID;
    }
    if ( recv_table != NULL ) {
        FREE( recv_table );
        recv_table  = NULL;
    }

    PMPI_Barrier( MPI_COMM_WORLD );
}

#endif