mpid_rma_common.c

fortran并行计算包

 *
 * Searches the group lpid list for a match.
 *
 * \param[in] lpid	World rank of the node in question
 * \param[in] grp	Group to validate against
 * \return TRUE is lpid is in group
 */
int MPIDU_valid_group_rank(int lpid, MPID_Group *grp) {
        int size = grp->size;
        int z;

        for (z = 0; z < size &&
                lpid != grp->lrank_to_lpid[z].lpid; ++z);
        return (z < size);
}

/*
 * Remote (receiver) Callbacks.
 */

/**
 * \brief Receive callback for RMA protocol and operations messages
 *
 * "Small" message callback - the entire message is already here.
 * Process it now and return.
 *
 * \param[in] _mi	Pointer to msginfo
 * \param[in] ct	Number of DCQuad's in msginfo
 * \param[in] or	Rank of origin
 * \param[in] sb	Pointer to send buffer (data received)
 * \param[in] sl	Length (bytes) of data
 * \return nothing
 *
 * \ref msginfo_usage
 */
void recv_sm_cb(void *cd, const DCQuad *_mi, unsigned ct, unsigned or,
                        const char *sb, const unsigned sl) {
        MPID_Win *win;
        char *rb;
        MPIDU_Onesided_ctl_t *mc = (MPIDU_Onesided_ctl_t *)_mi;
        MPIDU_Onesided_info_t *mi = (MPIDU_Onesided_info_t *)_mi;

        switch (_mi[0].w0) {
        /* The following all use msginfo as DCMF_Control_t (DCQuad[1]) */
        case MPID_MSGTYPE_COMPLETE:
                MPID_assert_debug(ct == 1);
                MPID_assert_debug(sl == 0);
                MPID_Win_get_ptr((MPI_Win)mc->mpid_ctl_w1, win);
                MPID_assert_debug(win != NULL);
                win->_dev.coll_info[win->_dev.comm_ptr->rank].rma_sends += mc->mpid_ctl_w3;
                ++win->_dev.my_sync_done;
                break;
        case MPID_MSGTYPE_POST:
                MPID_assert_debug(ct == 1);
                MPID_assert_debug(sl == 0);
                MPID_Win_get_ptr((MPI_Win)mc->mpid_ctl_w1, win);
                MPID_assert_debug(win != NULL);
                ++win->_dev.my_sync_begin;
                break;
        case MPID_MSGTYPE_LOCK:
                MPID_assert_debug(ct == 1);
                lock_cb(mc, or);
                break;
        case MPID_MSGTYPE_UNLOCK:
                MPID_assert_debug(ct == 1);
                unlk_cb(mc, or);
                break;
        case MPID_MSGTYPE_LOCKACK:
        case MPID_MSGTYPE_UNLOCKACK:
                MPID_assert_debug(ct == 1);
                MPID_Win_get_ptr((MPI_Win)mc->mpid_ctl_w1, win);
                MPID_assert_debug(win != NULL);
                ++win->_dev.my_sync_done;
                break;

        /* The following all use msginfo as DCQuad[2] */
        case MPID_MSGTYPE_PUT:
                MPID_assert_debug(ct == 2);
                MPID_assert_debug(sl != 0);
                MPID_Win_get_ptr((MPI_Win)mi->mpid_info_w1, win);
                MPID_assert_debug(win != NULL);
                MPIDU_assert_PUTOK(win);
                if (win->_dev.epoch_assert & MPI_MODE_NOPUT) {
                        /** \todo exact error handling */
                }
                memcpy((char *)mi->mpid_info_w3, sb, sl);
                rma_recvs_cb(win, mi->mpid_info_w2, or);
                break;
        case MPID_MSGTYPE_DT_MAP:
                MPID_assert_debug(ct == 2);
                rb = MPID_Prepare_rem_dt(mi);
                if (rb) {
                        DCQuad xtra;

                        xtra.w0 = mi->mpid_info_w0;
                        xtra.w1 = mi->mpid_info_w1;
                        xtra.w2 = mi->mpid_info_w2;
                        xtra.w3 = mi->mpid_info_w3;
                        memcpy(rb, sb, sl);
                        MPID_Recvdone1_rem_dt(&xtra);
                }
                break;
        case MPID_MSGTYPE_DT_IOV:
#ifdef NOT_USED
                MPID_assert_debug(ct == 2);
                rb = mpid_update_rem_dt(mi->mpid_info_w2, mi->mpid_info_w3, mi->mpid_info_w1);
                if (rb) {
                        memcpy(rb, sb, sl);
                        mpid_recvdone2_rem_dt(_mi);
                }
                break;
#endif /* NOT_USED */
                MPID_abort();
        case MPID_MSGTYPE_ACC:
                MPID_assert_debug(ct == 2);
                MPID_Win_get_ptr((MPI_Win)mi->mpid_info_w1, win);
                MPID_assert_debug(win != NULL);
                MPIDU_assert_PUTOK(win);
                if (win->_dev.epoch_assert & MPI_MODE_NOPUT) {
                        /** \todo exact error handling */
                }
                target_accumulate(mi, sb, or);
                break;

        /* Not supported message types */
        case MPID_MSGTYPE_GET: /* GET can't generate these */
                MPID_abort();
        default:
                /*
                 * Don't know what to do with this... we have some data
                 * (possibly) but don't have a target address to copy it
                 * to (or know what else to do with it).
                 */
                break;
        }
}

/**
 * \brief Callback for DCMF_Control() messages
 *
 * Simple pass-through to recv_sm_cb() with zero-length data.
 *
 * \param[in] ctl	Control message (one quad)
 * \param[in] or	Origin node lpid
 * \return	nothing
 */
void recv_ctl_cb(void *cd, const DCMF_Control_t *ctl, unsigned or) {
        recv_sm_cb(cd, (const DCQuad *)ctl, 1, or, NULL, 0);
}

/**
 * \brief Receive callback for RMA operations messages
 *
 * "Message receive initiated" callback.
 * This one should never get called for protocol messages.
 * Setup buffers, get a request object, and return so receive can begin.
 * In some cases (e.g. MPID_MSGTYPE_ACC) the processing is done in the
 * receive completion callback, otherwise that callback just frees
 * the request and cleans up (updates counters).
 *
 * \param[in] _mi	Pointer to msginfo
 * \param[in] ct	Number of DCQuad's in msginfo
 * \param[in] or	Rank of origin
 * \param[in] sl	Length (bytes) of sent data
 * \param[out] rl	Length (bytes) of data to receive
 * \param[out] rb	receive buffer
 * \param[out] cb	callback to invoke after receive
 * \return Pointer to DCMF request object to use for receive,
 *	or NULL to discard received data
 *
 * \ref msginfo_usage
 */
DCMF_Request_t *recv_cb(void *cd, const DCQuad *_mi, unsigned ct,
                        unsigned or, const unsigned sl, unsigned *rl,
                        char **rb, DCMF_Callback_t *cb)
{
        DCMF_Request_t *req;
        MPID_Win *win;
        MPIDU_Onesided_info_t *mi = (MPIDU_Onesided_info_t *)_mi;

        switch (_mi[0].w0) {
        /* The following all use msginfo as DCQuad[2] */
        case MPID_MSGTYPE_PUT:
                MPID_assert_debug(ct == 2);
                MPID_Win_get_ptr((MPI_Win)mi->mpid_info_w1, win);
                MPID_assert_debug(win != NULL);
                MPIDU_assert_PUTOK(win);
                if (win->_dev.epoch_assert & MPI_MODE_NOPUT) {
                        /** \todo exact error handling */
                }
                MPID_assert_debug(mi->mpid_info_w3 >=
                                        (unsigned)win->base &&
                        mi->mpid_info_w3 + sl <=
                                        (unsigned)win->base + win->size);
                *rl = sl;
                *rb = (char *)mi->mpid_info_w3;
                {	DCQuad xtra;
                        xtra.w0 = mi->mpid_info_w3;
                        xtra.w1 = mi->mpid_info_w1;
                        xtra.w2 = mi->mpid_info_w2;
                        xtra.w3 = or;
                        req = MPIDU_get_req(&xtra, NULL);
                }
                cb->clientdata = req;
                cb->function = rma_rqc_cb;
                return req;
        case MPID_MSGTYPE_DT_MAP:
                MPID_assert_debug(ct == 2);
                *rb = MPID_Prepare_rem_dt(mi);
                if (!*rb) {
                        return NULL;
                }
                *rl = sl;
                {	DCQuad xtra;
                        xtra.w0 = mi->mpid_info_w0;
                        xtra.w1 = mi->mpid_info_w1;
                        xtra.w2 = mi->mpid_info_w2;
                        xtra.w3 = mi->mpid_info_w3;
                        req = MPIDU_get_req(&xtra, NULL);
                }
                cb->clientdata = req;
                cb->function = dtc1_rqc_cb;
                return req;
        case MPID_MSGTYPE_DT_IOV:
#ifdef NOT_USED
                MPID_assert_debug(ct == 2);
                *rb = mpid_update_rem_dt(mi->mpid_info_w2, mi->mpid_info_w3, mi->mpid_info_w1);
                if (!*rb) {
                        return NULL;
                }
                *rl = sl;
                {	DCQuad xtra;
                        xtra.w0 = mi->mpid_info_w0;
                        xtra.w1 = mi->mpid_info_w1;
                        xtra.w2 = mi->mpid_info_w2;
                        xtra.w3 = mi->mpid_info_w3;
                        req = MPIDU_get_req(&xtra, NULL);
                }
                cb->clientdata = req;
                cb->function = dtc2_rqc_cb;
                return req;
#endif /* NOT_USED */
                MPID_abort();
        case MPID_MSGTYPE_ACC:
                MPID_assert_debug(ct == 2);
        { /* block */
                MPIDU_Onesided_info_t *info;
                DCQuad xtra = {0};

                MPID_Win_get_ptr((MPI_Win)mi->mpid_info_w1, win);
                MPID_assert_debug(win != NULL);
                MPIDU_assert_PUTOK(win);
                if (win->_dev.epoch_assert & MPI_MODE_NOPUT) {
                        /** \todo exact error handling */
                }
                /** \note These embedded DCQuads are not directly
                 * used in any communications. */
                MPIDU_MALLOC(info, MPIDU_Onesided_info_t,
                        sizeof(MPIDU_Onesided_info_t) + sl, e, "MPID_MSGTYPE_ACC");
                MPID_assert_debug(info != NULL);
                *rb = (char *)(info + 1);
                *rl = sl;
                memcpy(info, mi, sizeof(MPIDU_Onesided_info_t));
                xtra.w2 = (unsigned)info;
                xtra.w3 = or;
                req = MPIDU_get_req(&xtra, NULL);
                cb->clientdata = req;
                cb->function = accum_cb;
                return req;
        } /* block */

        /* The following all use msginfo as DCMF_Control_t (DCQuad[1]) */
        case MPID_MSGTYPE_POST:
        case MPID_MSGTYPE_COMPLETE:
                /* Win_post/Win_complete messages are always small. */
        case MPID_MSGTYPE_LOCK:
        case MPID_MSGTYPE_UNLOCK:
        case MPID_MSGTYPE_LOCKACK:
        case MPID_MSGTYPE_UNLOCKACK:
                MPID_abort();
        case MPID_MSGTYPE_GET: /* GET can't generate these */
                MPID_abort();
        default:
                break;
        }
        return NULL;
}

/*
 * End of remote callbacks.
 */

#ifdef NOT_USED
/**
 * \brief Send local datatype to target node
 *
 * Routine to send target datatype to target node.
 * These sends are handled by recv callbacks above...
 *
 * \param[in] dt	datatype handle to send
 * \param[in] o_lpid	Origin lpid
 * \param[in] t_lpid	Target lpid
 * \param[out] pending	Pointer to send done counter
 * \param[in,out] consistency	Pointer for consistency used for sends (out)
 * \return MPI_SUCCESS, or error returned by DCMF_Send.
 *
 * \ref msginfo_usage\n
 * \ref dtcache_design
 */
int mpid_queue_datatype(MPI_Datatype dt,
                        int o_lpid, int t_lpid, volatile unsigned *pending,
                        DCMF_Consistency *consistency) {
        MPIDU_Onesided_info_t *info;
        DCQuad xtra = {0};
        DCMF_Callback_t cb_send;
        DCMF_Request_t *reqp;
        int mpi_errno = MPI_SUCCESS;
        mpid_dt_info dti;

        if (MPIDU_check_dt(t_lpid, dt, &dti)) {
                /* we've previously sent this datatype to that target */
                return mpi_errno;
        }
        /** \todo need to ensure we don't LOWER consistency... */
        *consistency = DCMF_WEAK_CONSISTENCY;
        xtra.w0 = (unsigned)pending;

        reqp = MPIDU_get_req(&xtra, &info);
        info->mpid_info_w0 = MPID_MSGTYPE_DT_MAP;
        info->mpid_info_w1 = dti.map_len;
        info->mpid_info_w2 = o_lpid;
        info->mpid_info_w3 = dt;
        info->mpid_info_w4 = dti.dtp->extent;
        info->mpid_info_w5 = dti.dtp->eltype;
        info->mpid_info_w6 = dti.dtp->element_size;
        ++(*pending);
        cb_send.function = done_rqc_cb;
        cb_send.clientdata = reqp;
        mpi_errno = DCMF_Send(&bg1s_sn_proto, reqp, cb_send,
                        *consistency, t_lpid,
                        dti.map_len * sizeof(*dti.map), (char *)dti.map,
                        info->info, 2);
        if (mpi_errno) { return(mpi_errno); }
        reqp = MPIDU_get_req(&xtra, &info);
        info->mpid_info_w0 = MPID_MSGTYPE_DT_IOV;
        info->mpid_info_w1 = dti.iov_len;
        info->mpid_info_w2 = o_lpid;
        info->mpid_info_w3 = dt;
        info->mpid_info_w4 = dti.dtp->extent;
        info->mpid_info_w5 = dti.dtp->eltype;
        info->mpid_info_w6 = dti.dtp->element_size;
        ++(*pending);
        cb_send.function = done_rqc_cb;
        cb_send.clientdata = reqp;
        mpi_errno = DCMF_Send(&bg1s_sn_proto, reqp, cb_send,
                        *consistency, t_lpid,
                        dti.iov_len * sizeof(*dti.iov), (char *)dti.iov,
                        info->info, 2);
        return mpi_errno;
}
#endif /* NOT_USED */