same_dtype.c

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

            dim_A   = ints[ 0 ] + 1;
            dim_B   = 2 * ints[ 0 ];
            for ( ii=dim_A; ii<=dim_B; ii++ ) {
                tot_cnt += ints[ ii ];
            }
            return tot_cnt;
        case MPI_COMBINER_DARRAY :
            for ( ii=3; ii<=ints[2]+2; ii++ ) {
                tot_cnt += ints[ ii ];
            }
            return tot_cnt;
#endif
    }
    return tot_cnt;
}


void CollChk_dtype_hash(MPI_Datatype type, int cnt, CollChk_hash_t *type_hash)
{
    int             nints, naddrs, ntypes, combiner;
    int             *ints; 
    MPI_Aint        *addrs; 
    MPI_Datatype    *types;
    CollChk_hash_t  curr_hash, next_hash;
    int             type_cnt;
    int             ii;

    /*  Don't know if this makes sense or not */
    if ( cnt <= 0 ) {
        /* (value,count)=(0,0) => skipping of this (type,cnt) in addition */
        type_hash->value = 0;
        type_hash->count = 0;
        return;
    }

    MPI_Type_get_envelope(type, &nints, &naddrs, &ntypes, &combiner);
    if (combiner != MPI_COMBINER_NAMED) {
#if ! defined( HAVE_ALLOCA )
        ints = NULL;
        if ( nints > 0 )
            ints = (int *) malloc(nints * sizeof(int)); 
        addrs = NULL;
        if ( naddrs > 0 )
            addrs = (MPI_Aint *) malloc(naddrs * sizeof(MPI_Aint)); 
        types = NULL;
        if ( ntypes > 0 )
            types = (MPI_Datatype *) malloc(ntypes * sizeof(MPI_Datatype));
#else
        ints = NULL;
        if ( nints > 0 )
            ints = (int *) alloca(nints * sizeof(int)); 
        addrs = NULL;
        if ( naddrs > 0 )
            addrs = (MPI_Aint *) alloca(naddrs * sizeof(MPI_Aint)); 
        types = NULL;
        if ( ntypes > 0 )
            types = (MPI_Datatype *) alloca(ntypes * sizeof(MPI_Datatype));
#endif

        MPI_Type_get_contents(type, nints, naddrs, ntypes, ints, addrs, types);
        type_cnt = CollChk_derived_count(0, ints, combiner);
        CollChk_dtype_hash(types[0], type_cnt, &curr_hash);

        /*
            ntypes > 1 only for MPI_COMBINER_STRUCT(_INTEGER)
        */
        for( ii=1; ii < ntypes; ii++) {
            type_cnt = CollChk_derived_count(ii, ints, combiner); 
            CollChk_dtype_hash(types[ii], type_cnt, &next_hash);
            CollChk_hash_add(&curr_hash, &next_hash, &curr_hash);
        }

#if ! defined( HAVE_ALLOCA )
        if ( ints != NULL )
            free( ints );
        if ( addrs != NULL )
            free( addrs );
        if ( types != NULL )
            free( types );
#endif
    }
    else {
        curr_hash.value = CollChk_basic_value(type);
        curr_hash.count = CollChk_basic_count(type);
    }

    type_hash->value = curr_hash.value;
    type_hash->count = curr_hash.count;
    for ( ii=1; ii < cnt; ii++ ) {
        CollChk_hash_add(type_hash, &curr_hash, type_hash);
    }
}

/*
    A wrapper that calls PMPI_Allreduce() provides different send and receive
    buffers so use of MPI_Allreduce() conforms to MPI-1 standard, section 2.2.
*/
int CollChk_Allreduce_int( int ival, MPI_Op op, MPI_Comm comm );
int CollChk_Allreduce_int( int ival, MPI_Op op, MPI_Comm comm )
{
    int  local_ival;
    PMPI_Allreduce( &ival, &local_ival, 1, MPI_INT, op, comm );
    return local_ival;
}

/*
   Checking if (type,cnt) is the same in all processes within the communicator.
*/
int CollChk_dtype_bcast(MPI_Comm comm, MPI_Datatype type, int cnt, int root,
                        char* call)
{
#if 0
    CollChk_hash_t  local_hash;        /* local hash value */
    CollChk_hash_t  root_hash;         /* root's hash value */
    char            err_str[COLLCHK_STD_STRLEN];
    int             rank, size;        /* rank, size */
    int             are_hashes_equal;  /* go flag, ok flag */

    /* get the rank and size */
    MPI_Comm_rank(comm, &rank);
    MPI_Comm_size(comm, &size);

    /* get the hash values */
    CollChk_dtype_hash(type, cnt, &local_hash);

    if (rank == root) {
        root_hash.value = local_hash.value;
        root_hash.count = local_hash.count;
    }
    /* broadcast root's datatype hash to all other processes */
    PMPI_Bcast(&root_hash, 2, MPI_UNSIGNED, root, comm);

    /* Compare root's datatype hash to the local hash */
    are_hashes_equal = CollChk_hash_equal( &local_hash, &root_hash );
    if ( !are_hashes_equal )
        sprintf(err_str, "Inconsistent datatype signatures detected "
                         "between rank %d and rank %d.\n", rank, root);
    else
        sprintf(err_str, COLLCHK_NO_ERROR_STR);

    /* Find out if there is unequal hashes in the communicator */
    are_hashes_equal = CollChk_Allreduce_int(are_hashes_equal, MPI_LAND, comm);

    if ( !are_hashes_equal )
        return CollChk_err_han(err_str, COLLCHK_ERR_DTYPE, call, comm);

    return MPI_SUCCESS;
#endif
#if defined( DEBUG )
    fprintf( stdout, "CollChk_dtype_bcast()\n" );
#endif
    return CollChk_dtype_scatter(comm, type, cnt, type, cnt, root, 1, call );
}


/*
  The (sendtype,sendcnt) is assumed to be known in root process.
  (recvtype,recvcnt) is known in every process.  The routine checks if
  (recvtype,recvcnt) on each process is the same as (sendtype,sendcnt)
  on process root.
*/
int CollChk_dtype_scatter(MPI_Comm comm,
                          MPI_Datatype sendtype, int sendcnt,
                          MPI_Datatype recvtype, int recvcnt,
                          int root, int are2buffs, char *call)
{
    CollChk_hash_t  root_hash;         /* root's hash value */
    CollChk_hash_t  recv_hash;         /* local hash value */
    char            err_str[COLLCHK_STD_STRLEN];
    int             rank, size;
    int             are_hashes_equal;

#if defined( DEBUG )
    fprintf( stdout, "CollChk_dtype_scatter()\n" );
#endif

    /* get the rank and size */
    MPI_Comm_rank(comm, &rank);
    MPI_Comm_size(comm, &size);

    /*
       Scatter() only cares root's send datatype signature,
       i.e. ignore not-root's send datatype signatyre
    */
    /* Set the root's hash value */
    if (rank == root)
        CollChk_dtype_hash(sendtype, sendcnt, &root_hash);

    /* broadcast root's datatype hash to all other processes */
    PMPI_Bcast(&root_hash, 2, MPI_UNSIGNED, root, comm);

    /* Compare root_hash with the input/local hash */
    if ( are2buffs ) {
        CollChk_dtype_hash( recvtype, recvcnt, &recv_hash );
        are_hashes_equal = CollChk_hash_equal( &root_hash, &recv_hash );
    }
    else
        are_hashes_equal = 1;

    if ( !are_hashes_equal )
        sprintf(err_str, "Inconsistent datatype signatures detected "
                         "between rank %d and rank %d.\n", rank, root);
    else
        sprintf(err_str, COLLCHK_NO_ERROR_STR);

    /* Find out if there is unequal hashes in the communicator */
    are_hashes_equal = CollChk_Allreduce_int(are_hashes_equal, MPI_LAND, comm);

    if ( !are_hashes_equal )
        return CollChk_err_han(err_str, COLLCHK_ERR_DTYPE, call, comm);

    return MPI_SUCCESS;
}

/*
  The vector of (sendtype,sendcnts[]) is assumed to be known in root process.
  (recvtype,recvcnt) is known in every process.  The routine checks if
  (recvtype,recvcnt) on process P is the same as (sendtype,sendcnt[P])
  on process root. 
*/
int CollChk_dtype_scatterv(MPI_Comm comm,
                           MPI_Datatype sendtype, int *sendcnts,
                           MPI_Datatype recvtype, int recvcnt,
                           int root, int are2buffs, char *call)
{
    CollChk_hash_t  *hashes;         /* hash array for (sendtype,sendcnts[]) */
    CollChk_hash_t  root_hash;       /* root's hash value */
    CollChk_hash_t  recv_hash;       /* local hash value */
    char            err_str[COLLCHK_STD_STRLEN];
    int             rank, size, idx;
    int             are_hashes_equal;

#if defined( DEBUG )
    fprintf( stdout, "CollChk_dtype_scatterv()\n" );
#endif

    /* get the rank and size */
    MPI_Comm_rank(comm, &rank);
    MPI_Comm_size(comm, &size);

    /*
       Scatter() only cares root's send datatype signature[],
       i.e. ignore not-root's send datatype signatyre
    */
    hashes = NULL;
    if ( rank == root ) {
        /* Allocate hash buffer memory */
#if ! defined( HAVE_ALLOCA )
        hashes = (CollChk_hash_t *) malloc( size * sizeof(CollChk_hash_t) );
#else
        hashes = (CollChk_hash_t *) alloca( size * sizeof(CollChk_hash_t) );
#endif
        for ( idx = 0; idx < size; idx++ )
            CollChk_dtype_hash( sendtype, sendcnts[idx], &(hashes[idx]) );
    }

    /* Send the root's hash array to update other processes's root_hash */
    PMPI_Scatter(hashes, 2, MPI_UNSIGNED, &root_hash, 2, MPI_UNSIGNED,
                 root, comm);

    /* Compare root_hash with the input/local hash */
    if ( are2buffs ) {
        CollChk_dtype_hash( recvtype, recvcnt, &recv_hash );    
        are_hashes_equal = CollChk_hash_equal( &root_hash, &recv_hash );
    }
    else
        are_hashes_equal = 1;

    if ( !are_hashes_equal )
        sprintf(err_str, "Inconsistent datatype signatures detected "
                         "between rank %d and rank %d.\n", rank, root);
    else
        sprintf(err_str, COLLCHK_NO_ERROR_STR);

    /* Find out if there is unequal hashes in the communicator */
    are_hashes_equal = CollChk_Allreduce_int(are_hashes_equal, MPI_LAND, comm);

#if ! defined( HAVE_ALLOCA )
    if ( hashes != NULL )
        free( hashes );
#endif

    if ( !are_hashes_equal )
        return CollChk_err_han(err_str, COLLCHK_ERR_DTYPE, call, comm);

    return MPI_SUCCESS;
}


/*
   (sendtype,sendcnt) and (recvtype,recvcnt) are known in every process.
   The routine checks if (recvtype,recvcnt) on local process is the same
   as (sendtype,sendcnt) collected from all the other processes.
*/
int CollChk_dtype_allgather(MPI_Comm comm,
                            MPI_Datatype sendtype, int sendcnt,
                            MPI_Datatype recvtype, int recvcnt,
                            int are2buffs, char *call)
{
    CollChk_hash_t  *hashes;      /* hashes from other senders' */
    CollChk_hash_t  send_hash;    /* local sender's hash value */
    CollChk_hash_t  recv_hash;    /* local receiver's hash value */
    char            err_str[COLLCHK_STD_STRLEN];
    char            rank_str[COLLCHK_SM_STRLEN];
    int             *isOK2chks;   /* boolean array, true:sendbuff=\=recvbuff */
    int             *err_ranks;   /* array of ranks that have mismatch hashes */
    int             err_rank_size;
    int             err_str_sz, str_sz;
    int             rank, size, idx;

#if defined( DEBUG )
    fprintf( stdout, "CollChk_dtype_allgather()\n" );
#endif

    /* get the rank and size */
    MPI_Comm_rank(comm, &rank);
    MPI_Comm_size(comm, &size);

    CollChk_dtype_hash( sendtype, sendcnt, &send_hash );
    /* Allocate hash buffer memory */
#if ! defined( HAVE_ALLOCA )
    hashes    = (CollChk_hash_t *) malloc( size * sizeof(CollChk_hash_t) );
    err_ranks = (int *) malloc( size * sizeof(int) );
    isOK2chks = (int *) malloc( size * sizeof(int) );
#else
    hashes    = (CollChk_hash_t *) alloca( size * sizeof(CollChk_hash_t) );
    err_ranks = (int *) alloca( size * sizeof(int) );
    isOK2chks = (int *) alloca( size * sizeof(int) );
#endif

    /* Gather other senders' datatype hashes as local hash arrary */
    PMPI_Allgather(&send_hash, 2, MPI_UNSIGNED, hashes, 2, MPI_UNSIGNED, comm);
    PMPI_Allgather(&are2buffs, 1, MPI_INT, isOK2chks, 1, MPI_INT, comm);

    /* Compute the local datatype hash value */
    CollChk_dtype_hash( recvtype, recvcnt, &recv_hash );

    /* Compare the local datatype hash with other senders' datatype hashes */
    /*
       The checks are more exhaustive and redundant tests on all processes,
       but matches what user expects
    */