pml_ob1_recvfrag.c

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

         *   or if we need to traverse both sets at the same time.
         */
        if (opal_list_get_size(&proc->specific_receives) == 0 ){
            /*
             * There are only wild irecvs, so specialize the algorithm.
             */
            MCA_PML_OB1_CHECK_WILD_RECEIVES_FOR_MATCH(hdr, comm, proc, match);
    
        } else if (opal_list_get_size(&comm->wild_receives) == 0 ) {
            /*
             * There are only specific irecvs, so specialize the algorithm.
             */
            MCA_PML_OB1_CHECK_SPECIFIC_RECEIVES_FOR_MATCH(hdr, comm, proc, match);
        } else {
            /*
             * There are some of each.
             */
            MCA_PML_OB1_CHECK_SPECIFIC_AND_WILD_RECEIVES_FOR_MATCH(hdr, comm, proc, match);
        }

        /* if match found, process data */
        if(match) {
            match->req_recv.req_base.req_proc = proc->ompi_proc;

            /*
             * update delivered sequence number information, if needed.
             */
            if( (match->req_recv.req_base.req_type == MCA_PML_REQUEST_PROBE) ) {

                /* complete the probe */
                mca_pml_ob1_recv_request_matched_probe(match,btl,segments,num_segments);

                /* attempt to match actual request */
                match = NULL;
                goto rematch;
            }
        } else {

            /* if no match found, place on unexpected queue */
            mca_pml_ob1_recv_frag_t* frag;
            MCA_PML_OB1_RECV_FRAG_ALLOC(frag, rc);
            if(OMPI_SUCCESS != rc) {
                OPAL_THREAD_UNLOCK(&comm->matching_lock);
                /**
                 * As we return from the match function, we should generate the expected event.
                 */
                PERUSE_TRACE_MSG_EVENT( PERUSE_COMM_SEARCH_POSTED_Q_END, comm_ptr,
                                        hdr->hdr_src, hdr->hdr_tag, PERUSE_RECV);

                return rc;
            }
            MCA_PML_OB1_RECV_FRAG_INIT(frag,hdr,segments,num_segments,btl);
            opal_list_append( &proc->unexpected_frags, (opal_list_item_t *)frag );
        }

        /**
         * The match is over. We generate the SEARCH_POSTED_Q_END here, before going
         * into the mca_pml_ob1_check_cantmatch_for_match so we can make a difference
         * for the searching time for all messages.
         */
        PERUSE_TRACE_MSG_EVENT( PERUSE_COMM_SEARCH_POSTED_Q_END, comm_ptr,
                                hdr->hdr_src, hdr->hdr_tag, PERUSE_RECV);

        /* 
         * Now that new message has arrived, check to see if
         *   any fragments on the c_c_frags_cant_match list
         *   may now be used to form new matchs
         */
        if (0 < opal_list_get_size(&proc->frags_cant_match)) {
            additional_match = mca_pml_ob1_check_cantmatch_for_match(&additional_matches,comm,proc);
        }

    } else {

        /*
         * This message comes after the next expected, so it
         * is ahead of sequence.  Save it for later.
         */
        mca_pml_ob1_recv_frag_t* frag;
        MCA_PML_OB1_RECV_FRAG_ALLOC(frag, rc);
        if(OMPI_SUCCESS != rc) {
            OPAL_THREAD_UNLOCK(&comm->matching_lock);
            return rc;
        }
        MCA_PML_OB1_RECV_FRAG_INIT(frag,hdr,segments,num_segments,btl);
        opal_list_append(&proc->frags_cant_match, (opal_list_item_t *)frag);

    }
    /* release matching lock before processing fragment */
    OPAL_THREAD_UNLOCK(&comm->matching_lock);

    if(match != NULL) {
        mca_pml_ob1_recv_request_progress(match,btl,segments,num_segments);
    } else {
        PERUSE_TRACE_MSG_EVENT( PERUSE_COMM_MSG_INSERT_IN_UNEX_Q, comm_ptr,
                                hdr->hdr_src, hdr->hdr_tag, PERUSE_RECV);
    } 
    if(additional_match) {
        opal_list_item_t* item;
        while(NULL != (item = opal_list_remove_first(&additional_matches))) {
            mca_pml_ob1_recv_frag_t* frag = (mca_pml_ob1_recv_frag_t*)item;
            mca_pml_ob1_recv_request_progress( frag->request, frag->btl, frag->segments,
                                               frag->num_segments );
            MCA_PML_OB1_RECV_FRAG_RETURN(frag);
        }
    }
    return OMPI_SUCCESS;
}


/**
 * Scan the list of frags that came in ahead of time to see if any
 * can be processed at this time.  If they can, try and match the
 * frags.
 *
 * @param additional_matches List to hold new matches with fragments
 * from the c_frags_cant_match list. (IN/OUT)
 *
 * @param pml_comm Pointer to the communicator structure used for
 * matching purposes. (IN)
 *
 * This routine assumes that the appropriate matching locks are
 * set by the upper level routine.
 */

static bool mca_pml_ob1_check_cantmatch_for_match( opal_list_t *additional_matches,
                                                   mca_pml_ob1_comm_t* comm,
                                                   mca_pml_ob1_comm_proc_t *proc )
{
    /* local parameters */
    int match_found;
    uint16_t next_msg_seq_expected, frag_seq;
    mca_pml_ob1_recv_frag_t *frag;
    bool match_made = false;

    /*
     * Loop over all the out of sequence messages.  No ordering is assumed
     * in the c_frags_cant_match list.
     */

    match_found = 1;
    while ((0 < opal_list_get_size(&proc->frags_cant_match)) && match_found) {

        /* initialize match flag for this search */
        match_found = 0;

        /* get sequence number of next message that can be processed */
        next_msg_seq_expected = proc->expected_sequence;

        /* search the list for a fragment from the send with sequence
         * number next_msg_seq_expected
         */
        for(frag = (mca_pml_ob1_recv_frag_t *) 
            opal_list_get_first(&proc->frags_cant_match);
            frag != (mca_pml_ob1_recv_frag_t *)
            opal_list_get_end(&proc->frags_cant_match);
            frag = (mca_pml_ob1_recv_frag_t *) 
            opal_list_get_next(frag))
        {
            /*
             * If the message has the next expected seq from that proc...
             */
            frag_seq=frag->hdr.hdr_match.hdr_seq;
            if (frag_seq == next_msg_seq_expected) {
                mca_pml_ob1_match_hdr_t* hdr = &frag->hdr.hdr_match;
                mca_pml_ob1_recv_request_t *match = NULL;

                /* We're now expecting the next sequence number. */
                (proc->expected_sequence)++;

                /* signal that match was made */
                match_found = 1;

                /*
                 * remove frag from list
                 */
                opal_list_remove_item(&proc->frags_cant_match,
                        (opal_list_item_t *)frag);

rematch:
                /*
                 * figure out what sort of matching logic to use, if need to
                 *   look only at "specific" receives, or "wild" receives,
                 *   or if we need to traverse both sets at the same time.
                 */
                proc = comm->procs + hdr->hdr_src;
                if (opal_list_get_size(&proc->specific_receives) == 0 ) {
                    /*
                     * There are only wild irecvs, so specialize the algorithm.
                     */
                    MCA_PML_OB1_CHECK_WILD_RECEIVES_FOR_MATCH(hdr, comm, proc, match);
                } else if (opal_list_get_size(&comm->wild_receives) == 0 ) {
                    /*
                     * There are only specific irecvs, so specialize the algorithm.
                     */
                    MCA_PML_OB1_CHECK_SPECIFIC_RECEIVES_FOR_MATCH(hdr, comm, proc, match);
                } else {
                    /*
                     * There are some of each.
                     */
                    MCA_PML_OB1_CHECK_SPECIFIC_AND_WILD_RECEIVES_FOR_MATCH(hdr, comm, proc, match);

                }

                /* if match found, process data */
                if(match) {
                    match->req_recv.req_base.req_proc = proc->ompi_proc;

                    /*
                     * If this was a probe need to queue fragment on unexpected list
                     */
                    if( (match->req_recv.req_base.req_type == MCA_PML_REQUEST_PROBE) ) {

                        /* complete the probe */
                        mca_pml_ob1_recv_request_matched_probe(match,frag->btl,frag->segments,frag->num_segments);

                        /* retry the match */
                        match = NULL;
                        goto rematch;

                    } else {

                        /* associate the receive descriptor with the fragment
                         * descriptor */
                        frag->request=match;

                        /* add this fragment descriptor to the list of
                         * descriptors to be processed later
                         */
                        if(match_made == false) {
                            match_made = true;
                            OBJ_CONSTRUCT(additional_matches, opal_list_t);
                        }
                        opal_list_append(additional_matches, (opal_list_item_t *)frag);
                    }

                } else {
    
                    /* if no match found, place on unexpected queue */
                    opal_list_append( &proc->unexpected_frags, (opal_list_item_t *)frag);
                }

                /* c_frags_cant_match is not an ordered list, so exit loop
                 * and re-start search for next sequence number */
                break;

            } /* end if (frag_seq == next_msg_seq_expected) */
            
        } /* end for (frag) loop */
        
    } /* end while loop */

    return match_made;
}