pml_ob1_recvfrag.c

来自「MPI stands for the Message Passing Inter」· C语言 代码 · 共 772 行 · 第 1/3 页

/*
 * Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
 *                         University Research and Technology
 *                         Corporation.  All rights reserved.
 * Copyright (c) 2004-2006 The University of Tennessee and The University
 *                         of Tennessee Research Foundation.  All rights
 *                         reserved.
 * Copyright (c) 2004-2005 High Performance Computing Center Stuttgart, 
 *                         University of Stuttgart.  All rights reserved.
 * Copyright (c) 2004-2005 The Regents of the University of California.
 *                         All rights reserved.
 * $COPYRIGHT$
 * 
 * Additional copyrights may follow
 * 
 * $HEADER$
 */

/**
 * @file
 */

#include "ompi_config.h"

#include "opal/class/opal_list.h"
#include "opal/threads/mutex.h"
#include "ompi/constants.h"
#include "ompi/communicator/communicator.h"
#include "ompi/mca/pml/pml.h"
#include "pml_ob1.h"
#include "pml_ob1_comm.h"
#include "pml_ob1_recvfrag.h"
#include "pml_ob1_recvreq.h"
#include "pml_ob1_sendreq.h"
#include "pml_ob1_hdr.h"
#include "ompi/datatype/dt_arch.h"
#include "ompi/peruse/peruse-internal.h"

OBJ_CLASS_INSTANCE( mca_pml_ob1_buffer_t,
                    ompi_free_list_item_t,
                    NULL,
                    NULL );

OBJ_CLASS_INSTANCE( mca_pml_ob1_recv_frag_t,
                    opal_list_item_t,
                    NULL,
                    NULL );

/**
 * Static functions.
 */

/**
 * Match incoming recv_frags against posted receives.  
 * Supports out of order delivery.
 * 
 * @param frag_header (IN)          Header of received recv_frag.
 * @param frag_desc (IN)            Received recv_frag descriptor.
 * @param match_made (OUT)          Flag indicating wether a match was made.
 * @param additional_matches (OUT)  List of additional matches 
 * @return                          OMPI_SUCCESS or error status on failure.
 */
static int mca_pml_ob1_recv_frag_match( mca_btl_base_module_t *btl, 
                                        mca_pml_ob1_match_hdr_t *hdr,
                                        mca_btl_base_segment_t* segments,
                                        size_t num_segments );

/**
 *  Callback from BTL on receive.
 */
                                                                                                                      
void mca_pml_ob1_recv_frag_callback( mca_btl_base_module_t* btl, 
                                     mca_btl_base_tag_t tag,
                                     mca_btl_base_descriptor_t* des,
                                     void* cbdata )
{
    mca_btl_base_segment_t* segments = des->des_dst;
    mca_pml_ob1_hdr_t* hdr = (mca_pml_ob1_hdr_t*)segments->seg_addr.pval;

    if(segments->seg_len < sizeof(mca_pml_ob1_common_hdr_t)) {
        return;
    }

    /* hdr_type and hdr_flags are uint8_t, so no endian problems */
    switch(hdr->hdr_common.hdr_type) {
    case MCA_PML_OB1_HDR_TYPE_MATCH:
        {
#if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT
            if (hdr->hdr_common.hdr_flags & MCA_PML_OB1_HDR_FLAGS_NBO) {
                MCA_PML_OB1_MATCH_HDR_NTOH(hdr->hdr_match);
            }
#endif
            mca_pml_ob1_recv_frag_match(btl, &hdr->hdr_match, segments,des->des_dst_cnt);
            break;
        }
    case MCA_PML_OB1_HDR_TYPE_RNDV:
        {
#if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT
            if (hdr->hdr_common.hdr_flags & MCA_PML_OB1_HDR_FLAGS_NBO) {
                MCA_PML_OB1_RNDV_HDR_NTOH(hdr->hdr_rndv);
            }
#endif
            mca_pml_ob1_recv_frag_match(btl, &hdr->hdr_match, segments,des->des_dst_cnt);
            break;
        }
    case MCA_PML_OB1_HDR_TYPE_RGET:
        {
#if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT
            /* RDMA is currently disabled by bml if arch doesn't
               match, so this shouldn't be needed.  here to make sure
               we remember if we ever change the bml. */
            assert(0 == (hdr->hdr_common.hdr_flags & MCA_PML_OB1_HDR_FLAGS_NBO));
#endif
            mca_pml_ob1_recv_frag_match(btl, &hdr->hdr_match, segments,des->des_dst_cnt);
            break;
        }
    case MCA_PML_OB1_HDR_TYPE_ACK:
        {
            mca_pml_ob1_send_request_t* sendreq;
#if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT
            if (hdr->hdr_common.hdr_flags & MCA_PML_OB1_HDR_FLAGS_NBO) {
                MCA_PML_OB1_ACK_HDR_NTOH(hdr->hdr_ack);
            }
#endif
            sendreq = (mca_pml_ob1_send_request_t*)
                hdr->hdr_ack.hdr_src_req.pval;
            sendreq->req_recv = hdr->hdr_ack.hdr_dst_req;
            sendreq->req_rdma_offset = (size_t)hdr->hdr_ack.hdr_rdma_offset;
            if(OPAL_THREAD_ADD32(&sendreq->req_state, 1) == 2 &&
                    sendreq->req_bytes_delivered >=
                    sendreq->req_send.req_bytes_packed) {
                MCA_PML_OB1_SEND_REQUEST_PML_COMPLETE(sendreq);
            } else {
                mca_pml_ob1_send_request_schedule(sendreq);
            }
                
            break;
        }
    case MCA_PML_OB1_HDR_TYPE_FRAG:
        {
            mca_pml_ob1_recv_request_t* recvreq;
#if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT
            if (hdr->hdr_common.hdr_flags & MCA_PML_OB1_HDR_FLAGS_NBO) {
                MCA_PML_OB1_FRAG_HDR_NTOH(hdr->hdr_frag);
            }
#endif
            recvreq = (mca_pml_ob1_recv_request_t*)
                hdr->hdr_frag.hdr_dst_req.pval;
            mca_pml_ob1_recv_request_progress(recvreq,btl,segments,des->des_dst_cnt);
            break;
        }
    case MCA_PML_OB1_HDR_TYPE_PUT:
        {
            mca_pml_ob1_send_request_t* sendreq;
#if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT
            /* RDMA is currently disabled by bml if arch doesn't
               match, so this shouldn't be needed.  here to make sure
               we remember if we ever change the bml. */
            assert(0 == (hdr->hdr_common.hdr_flags & MCA_PML_OB1_HDR_FLAGS_NBO));
#endif
            sendreq = (mca_pml_ob1_send_request_t*)
                hdr->hdr_rdma.hdr_req.pval;
            mca_pml_ob1_send_request_put(sendreq,btl,&hdr->hdr_rdma);
            break;
        }
    case MCA_PML_OB1_HDR_TYPE_FIN:
        {
            mca_btl_base_descriptor_t* rdma;
#if !defined(WORDS_BIGENDIAN) && OMPI_ENABLE_HETEROGENEOUS_SUPPORT
            if (hdr->hdr_common.hdr_flags & MCA_PML_OB1_HDR_FLAGS_NBO) {
                MCA_PML_OB1_FIN_HDR_NTOH(hdr->hdr_fin);
            }
#endif
            rdma = (mca_btl_base_descriptor_t*)
                hdr->hdr_fin.hdr_des.pval;
            rdma->des_cbfunc(btl, NULL, rdma, OMPI_SUCCESS);
            break;
        }
    default:
        break;
    }
}

/**
 * Try and match the incoming message fragment to a generic
 * list of receives
 *
 * @param hdr Matching data from received fragment (IN)
 *
 * @param generic_receives Pointer to the receive list used for
 * matching purposes. (IN)
 *
 * @return Matched receive
 *
 * This routine assumes that the appropriate matching locks are
 * set by the upper level routine.
 */
#define MCA_PML_OB1_MATCH_GENERIC_RECEIVES(hdr,generic_receives,proc,return_match) \
    do {                                                                           \
        /* local variables */                                                      \
        mca_pml_ob1_recv_request_t *generic_recv = (mca_pml_ob1_recv_request_t *)  \
                     opal_list_get_first(generic_receives);                        \
        mca_pml_ob1_recv_request_t *last_recv = (mca_pml_ob1_recv_request_t *)     \
            opal_list_get_end(generic_receives);                                   \
        register int recv_tag, frag_tag = hdr->hdr_tag;                            \
                                                                                   \
        /* Loop over the receives. If the received tag is less than zero  */       \
        /* enter in a special mode, where we match only our internal tags */       \
        /* (such as those used by the collectives.*/                               \
        if( 0 <= frag_tag ) {                                                      \
            for( ; generic_recv != last_recv;                                      \
                 generic_recv = (mca_pml_ob1_recv_request_t *)                     \
                     ((opal_list_item_t *)generic_recv)->opal_list_next) {         \
                /* Check for a match */                                            \
                recv_tag = generic_recv->req_recv.req_base.req_tag;                \
                if ( (frag_tag == recv_tag) || (recv_tag == OMPI_ANY_TAG) ) {      \
                    break;                                                         \
                }                                                                  \
            }                                                                      \
        } else {                                                                   \
            for( ; generic_recv != last_recv;                                      \
                 generic_recv = (mca_pml_ob1_recv_request_t *)                     \
                     ((opal_list_item_t *)generic_recv)->opal_list_next) {         \
                /* Check for a match */                                            \
                recv_tag = generic_recv->req_recv.req_base.req_tag;                \
                if(frag_tag == recv_tag) {                        \
                    break;                                                         \
                }                                                                  \
            }                                                                      \
        }                                                                          \
        if( generic_recv != (mca_pml_ob1_recv_request_t *)                         \
            opal_list_get_end(generic_receives) ) {                                \
                                                                                   \
            /* Match made */                                                       \
            return_match = generic_recv;                                           \
                                                                                   \
            /* remove descriptor from posted specific ireceive list */             \
            opal_list_remove_item(generic_receives,                                \
                                  (opal_list_item_t *)generic_recv);               \
                                                                                   \
        }                                                                          \
    } while(0)

/**
 * Try and match the incoming message fragment to the list of
 * "wild" receives
 *
 * @param hdr Matching data from recived fragment (IN)
 *
 * @param pml_comm Pointer to the communicator structure used for
 * matching purposes. (IN)
 *
 * @return Matched receive
 *
 * This routine assumes that the appropriate matching locks are
 * set by the upper level routine.
 */