btl_mvapi_component.c

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

                OPAL_THREAD_UNLOCK(&endpoint->eager_rdma_local.lock);
        }
    }
    if (count) return count;

    for(i = 0; i < mca_btl_mvapi_component.ib_num_btls; i++) {
        VAPI_ret_t ret; 
        VAPI_wc_desc_t comp; 
        mca_btl_mvapi_module_t* mvapi_btl = &mca_btl_mvapi_component.mvapi_btls[i];
        /* we have two completion queues, one for "high" priority and one for "low". 
         *   we will check the high priority and process them until there are none left. 
         *   note that low priority messages are only processed one per progress call. 
         */ 
        ret = VAPI_poll_cq(mvapi_btl->nic, mvapi_btl->cq_hndl_hp, &comp); 
        if(VAPI_OK == ret) { 
            if(comp.status != VAPI_SUCCESS) { 
                ompi_proc_t* remote_proc = NULL;
                frag = (mca_btl_mvapi_frag_t*) (unsigned long) comp.id;
                if(frag) { 
                    endpoint = (mca_btl_mvapi_endpoint_t*) frag->endpoint; 
                    if(endpoint && 
                       endpoint->endpoint_proc && 
                       endpoint->endpoint_proc->proc_ompi) { 
                        remote_proc = endpoint->endpoint_proc->proc_ompi; 
                    }
                }
                BTL_PEER_ERROR(remote_proc, ("error polling HP CQ with status %s status number %d for Frag : %p", 
                                             VAPI_wc_status_sym(comp.status), 
                                             comp.status, comp.id));  
                if(comp.status == VAPI_RETRY_EXC_ERR) { 
                    opal_show_help("help-mpi-btl-mvapi.txt", "btl_mvapi:retry-exceeded", true);
                }
                return OMPI_ERROR; 
            }
            
            /* Handle work completions */
            switch(comp.opcode) {
            case VAPI_CQE_RQ_RDMA_WITH_IMM: 
                BTL_ERROR(("Got an RDMA with Immediate data!, not supported!")); 
                return OMPI_ERROR; 
           
            case VAPI_CQE_SQ_RDMA_WRITE:
            case VAPI_CQE_SQ_SEND_DATA :

                /* Process a completed send */
                frag = (mca_btl_mvapi_frag_t*) (unsigned long) comp.id; 
                endpoint = (mca_btl_mvapi_endpoint_t*) frag->endpoint; 

                /* Process a completed send */
                frag->base.des_cbfunc(&mvapi_btl->super, endpoint, &frag->base, OMPI_SUCCESS); 

                /* return send wqe */
                OPAL_THREAD_ADD32(&endpoint->sd_wqe_hp, 1);

                /* check to see if we need to progress any pending descriptors */
                while (!opal_list_is_empty(&endpoint->pending_frags_hp) &&
                        endpoint->sd_wqe_hp > 0 && (endpoint->sd_tokens_hp > 0 || endpoint->eager_rdma_remote.tokens > 0)) {
                    opal_list_item_t *frag_item;
                    OPAL_THREAD_LOCK(&endpoint->endpoint_lock);
                    frag_item = opal_list_remove_first(&(endpoint->pending_frags_hp));
                    OPAL_THREAD_UNLOCK(&endpoint->endpoint_lock);
                    if(NULL == (frag = (mca_btl_mvapi_frag_t *) frag_item))
                        break;
                    if(OMPI_SUCCESS !=  mca_btl_mvapi_endpoint_send(frag->endpoint, frag)) {
                        BTL_ERROR(("error in posting pending send\n"));
                        break;
                    }
                }

                if(!mca_btl_mvapi_component.use_srq) {
                    /* check to see if we need to return credits */
                    if((endpoint->rd_credits_hp >= mca_btl_mvapi_component.rd_win || endpoint->eager_rdma_local.credits >= mca_btl_mvapi_component.rd_win) &&
                        OPAL_THREAD_ADD32(&endpoint->sd_credits_hp, 1) == 1) {
                        mca_btl_mvapi_endpoint_send_credits_hp(endpoint);
                    }
                } else if(OPAL_THREAD_ADD32(&mvapi_btl->sd_tokens_hp, 1)  > 0
                          && !opal_list_is_empty(&mvapi_btl->pending_frags_hp)) {
                    /* dequeue resources due to global flow control */
                    opal_list_item_t *frag_item;
                    OPAL_THREAD_LOCK(&mvapi_btl->ib_lock);
                    frag_item = opal_list_remove_first(&mvapi_btl->pending_frags_hp);
                    OPAL_THREAD_UNLOCK(&mvapi_btl->ib_lock);
                    if(NULL != (frag = (mca_btl_mvapi_frag_t *) frag_item) &&
                       OMPI_SUCCESS !=  mca_btl_mvapi_endpoint_send(frag->endpoint, frag)) {
                        BTL_ERROR(("error in posting pending send\n"));
                    }
                }
                count++;
                break;

            case VAPI_CQE_RQ_SEND_DATA:
                /* process a RECV */
                frag = (mca_btl_mvapi_frag_t*) (unsigned long) comp.id;
                hret = mca_btl_mvapi_handle_incoming_hp(mvapi_btl,
                        frag->endpoint, frag, comp.byte_len);
                if (hret != OMPI_SUCCESS)
                    return hret;
                count++;
                break;

            case VAPI_CQE_SQ_RDMA_READ:
            default:
                BTL_ERROR(("Unhandled work completion opcode is %d", comp.opcode));
                break;
            }
        }
        
        ret = VAPI_poll_cq(mvapi_btl->nic, mvapi_btl->cq_hndl_lp, &comp); 
        if(VAPI_OK == ret) { 
            if(comp.status != VAPI_SUCCESS) { 
                BTL_ERROR(("Got error : %s, Vendor code : %d Frag : %p", 
                            VAPI_wc_status_sym(comp.status), 
                            comp.vendor_err_syndrome, comp.id));  
                return OMPI_ERROR; 
            }
            
            /* Handle n/w completions */
            switch(comp.opcode) {
            case VAPI_CQE_RQ_RDMA_WITH_IMM: 
                BTL_ERROR(("Got an RDMA with Immediate data!, not supported!")); 
                return OMPI_ERROR; 
            
            case VAPI_CQE_SQ_SEND_DATA :
                
                frag = (mca_btl_mvapi_frag_t*) (unsigned long) comp.id; 
                endpoint = frag->endpoint;

                /* Process a completed send - receiver must return tokens */
                frag->base.des_cbfunc(&mvapi_btl->super, frag->endpoint, &frag->base, OMPI_SUCCESS);

                /* return send wqe */
                OPAL_THREAD_ADD32(&endpoint->sd_wqe_lp, 1);

                /* check to see if we need to progress any pending desciptors */
                while (!opal_list_is_empty(&endpoint->pending_frags_lp) &&
                       endpoint->sd_wqe_lp > 0 && endpoint->sd_tokens_lp > 0) {
                   opal_list_item_t *frag_item;
                   OPAL_THREAD_LOCK(&endpoint->endpoint_lock);
                   frag_item = opal_list_remove_first(&(endpoint->pending_frags_lp));
                   OPAL_THREAD_UNLOCK(&endpoint->endpoint_lock);
                   if(NULL == (frag = (mca_btl_mvapi_frag_t *) frag_item))
                       break;
                    MCA_BTL_IB_FRAG_PROGRESS(frag);
                }

                if( !mca_btl_mvapi_component.use_srq) {
                    /* check to see if we need to return credits */
                    if( endpoint->rd_credits_lp >= mca_btl_mvapi_component.rd_win &&
                        OPAL_THREAD_ADD32(&endpoint->sd_credits_lp, 1) == 1) {
                        mca_btl_mvapi_endpoint_send_credits_lp(endpoint);
                    }

                /* SRQ case */
                } else if(OPAL_THREAD_ADD32(&mvapi_btl->sd_tokens_lp, 1)  > 0
                          && !opal_list_is_empty(&mvapi_btl->pending_frags_lp)) {
                    opal_list_item_t *frag_item;
                    OPAL_THREAD_LOCK(&mvapi_btl->ib_lock);
                    frag_item = opal_list_remove_first(&mvapi_btl->pending_frags_lp);
                    OPAL_THREAD_UNLOCK(&mvapi_btl->ib_lock);
                    if(NULL != (frag = (mca_btl_mvapi_frag_t *) frag_item)) {
                        MCA_BTL_IB_FRAG_PROGRESS(frag);
                    }
                }
                count++;
                break;

            case VAPI_CQE_SQ_RDMA_READ:

                frag = (mca_btl_mvapi_frag_t*) (unsigned long) comp.id; 
                OPAL_THREAD_ADD32(&frag->endpoint->get_tokens, 1);
                /* fall through */

            case VAPI_CQE_SQ_RDMA_WRITE:

                frag = (mca_btl_mvapi_frag_t*) (unsigned long) comp.id; 
                endpoint = frag->endpoint;

                /* process a completed write */
                frag->base.des_cbfunc(&mvapi_btl->super, frag->endpoint, &frag->base, OMPI_SUCCESS);

                /* return send wqe */
                OPAL_THREAD_ADD32(&endpoint->sd_wqe_lp, 1);

                /* check for pending frags */
                if(!opal_list_is_empty(&endpoint->pending_frags_lp)) {
                    opal_list_item_t *frag_item;
                    OPAL_THREAD_LOCK(&endpoint->endpoint_lock);
                    frag_item = opal_list_remove_first(&endpoint->pending_frags_lp);
                    OPAL_THREAD_UNLOCK(&endpoint->endpoint_lock);
                    if(NULL != (frag = (mca_btl_mvapi_frag_t *) frag_item)) {
                        MCA_BTL_IB_FRAG_PROGRESS(frag);
                    }
                }
                if (mca_btl_mvapi_component.use_srq &&
                    endpoint->sd_wqe_lp > 0 &&
                   !opal_list_is_empty(&mvapi_btl->pending_frags_lp)) {
                    opal_list_item_t *frag_item;
                    OPAL_THREAD_LOCK(&mvapi_btl->ib_lock);
                    frag_item = opal_list_remove_first(&mvapi_btl->pending_frags_lp);
                    OPAL_THREAD_UNLOCK(&mvapi_btl->ib_lock);
                    if(NULL != (frag = (mca_btl_mvapi_frag_t *) frag_item)) {
                        MCA_BTL_IB_FRAG_PROGRESS(frag);
                    }
                }
                count++;
                break;
                
            case VAPI_CQE_RQ_SEND_DATA:
                
                /* Process a RECV */
                frag = (mca_btl_mvapi_frag_t*) (unsigned long) comp.id;
                endpoint = (mca_btl_mvapi_endpoint_t*) frag->endpoint;
                credits = frag->hdr->credits;
                                                                                                                         
                /* advance the segment address past the header and subtract from the length..*/
                frag->segment.seg_len =  comp.byte_len-
                    ((unsigned char*) frag->segment.seg_addr.pval  - (unsigned char*) frag->hdr);
                                                                                                                         
                /* call registered callback */
                mvapi_btl->ib_reg[frag->hdr->tag].cbfunc(&mvapi_btl->super,
                                                          frag->hdr->tag,
                                                          &frag->base,
                                                          mvapi_btl->ib_reg[frag->hdr->tag].cbdata);
                OMPI_FREE_LIST_RETURN(&(mvapi_btl->recv_free_max), (ompi_free_list_item_t*) frag);
                                                                                                                         
#ifdef VAPI_FEATURE_SRQ
                if(mca_btl_mvapi_component.use_srq) {
                    /* repost receive descriptors */
                    OPAL_THREAD_ADD32((int32_t*) &mvapi_btl->srd_posted_lp, -1);
                    MCA_BTL_MVAPI_POST_SRR_LOW(mvapi_btl, 0);
                } else {
#endif
                    /* repost receive descriptors */
                    OPAL_THREAD_ADD32((int32_t*) &endpoint->rd_posted_lp, -1);
                    MCA_BTL_MVAPI_ENDPOINT_POST_RR_LOW(endpoint, 0);
                                                                                                                         
                    /* check to see if we need to progress any pending desciptors */
                    if( OPAL_THREAD_ADD32(&endpoint->sd_tokens_lp, credits) > 0) {
                                                                                                                         
                        while(!opal_list_is_empty(&endpoint->pending_frags_lp) &&
                              endpoint->sd_wqe_lp > 0 && endpoint->sd_tokens_lp > 0) {
                            opal_list_item_t *frag_item;
                            OPAL_THREAD_LOCK(&endpoint->endpoint_lock);
                            frag_item = opal_list_remove_first(&(endpoint->pending_frags_lp));
                            OPAL_THREAD_UNLOCK(&endpoint->endpoint_lock);
                            if(NULL == (frag = (mca_btl_mvapi_frag_t *) frag_item))
                                break;
                            MCA_BTL_IB_FRAG_PROGRESS(frag);
                        }
                    }
                                                                                                                         
                    /* check to see if we need to return credits */
                    if( endpoint->rd_credits_lp >= mca_btl_mvapi_component.rd_win &&
                        OPAL_THREAD_ADD32(&endpoint->sd_credits_lp, 1) == 1) {
                        mca_btl_mvapi_endpoint_send_credits_lp(endpoint);
                    }
                                                                                                                         
#ifdef VAPI_FEATURE_SRQ
                }
#endif
                count++;
                break;

            default:
                BTL_ERROR(("Errorneous network completion"));
                break;
            }
        }
        
    }
    return count;
}