btl_mvapi.c

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

        /* we didn't get a memory registration passed in, so we have to register the region
         * ourselves 
         */ 
        rc = btl->btl_mpool->mpool_register(btl->btl_mpool,
            frag->segment.seg_addr.pval, *size, 0, &registration);
        if(OMPI_SUCCESS != rc || NULL == registration) {
            BTL_ERROR(("mpool_register(%p,%lu) failed: base %p lb %lu offset %lu",
                frag->segment.seg_addr.pval, *size, convertor->pBaseBuf, lb, convertor->bConverted));
            MCA_BTL_IB_FRAG_RETURN(btl, frag);
            return NULL;
        }
        frag->registration = (mca_btl_mvapi_reg_t*)registration;
    }
    mvapi_reg = (mca_btl_mvapi_reg_t*)registration;
    
    frag->sg_entry.len = *size;
    frag->sg_entry.lkey = mvapi_reg->l_key;
    frag->sg_entry.addr = (VAPI_virt_addr_t) (MT_virt_addr_t) frag->segment.seg_addr.pval; 
    
    frag->segment.seg_key.key32[0] =mvapi_reg->r_key;

    frag->base.des_dst = &frag->segment; 
    frag->base.des_dst_cnt = 1; 
    frag->base.des_src = NULL; 
    frag->base.des_src_cnt = 0; 
    
    return &frag->base; 
    
}

int mca_btl_mvapi_finalize(struct mca_btl_base_module_t* btl)
{
    mca_btl_mvapi_module_t* mvapi_btl; 
    mvapi_btl = (mca_btl_mvapi_module_t*) btl; 
    return OMPI_SUCCESS;
}

/*
 *  Initiate a send. 
 */

int mca_btl_mvapi_send( 
    struct mca_btl_base_module_t* btl,
    struct mca_btl_base_endpoint_t* endpoint,
    struct mca_btl_base_descriptor_t* descriptor, 
    mca_btl_base_tag_t tag)
   
{
    mca_btl_mvapi_frag_t* frag = (mca_btl_mvapi_frag_t*)descriptor; 
    frag->endpoint = endpoint; 
    frag->hdr->tag = tag; 
    frag->desc.sr_desc.opcode = VAPI_SEND;
    return mca_btl_mvapi_endpoint_send(endpoint, frag);
}

/*
 * RDMA local buffer to remote buffer address.
 */

int mca_btl_mvapi_put( mca_btl_base_module_t* btl,
                       mca_btl_base_endpoint_t* endpoint,
                       mca_btl_base_descriptor_t* descriptor)
{
    int rc; 
    mca_btl_mvapi_module_t* mvapi_btl = (mca_btl_mvapi_module_t*) btl; 
    mca_btl_mvapi_frag_t* frag = (mca_btl_mvapi_frag_t*) descriptor; 

    /* setup for queued requests */
    frag->endpoint = endpoint;
    frag->desc.sr_desc.opcode = VAPI_RDMA_WRITE; 

    /* check for a send wqe */
    if (OPAL_THREAD_ADD32(&endpoint->sd_wqe_lp,-1) < 0) {
        OPAL_THREAD_ADD32(&endpoint->sd_wqe_lp,1);
        OPAL_THREAD_LOCK(&endpoint->endpoint_lock);
        opal_list_append(&endpoint->pending_frags_lp, (opal_list_item_t *)frag);
        OPAL_THREAD_UNLOCK(&endpoint->endpoint_lock);
        return OMPI_SUCCESS;
                                                                                                                        
    /* post descriptor */
    } else {
        
        frag->desc.sr_desc.remote_qp = endpoint->rem_info.rem_qp_num_lp; 
        frag->desc.sr_desc.remote_addr = (VAPI_virt_addr_t) frag->base.des_dst->seg_addr.lval; 
        frag->desc.sr_desc.r_key = frag->base.des_dst->seg_key.key32[0]; 
        frag->sg_entry.addr = (VAPI_virt_addr_t) (MT_virt_addr_t) frag->base.des_src->seg_addr.pval; 
        frag->sg_entry.len  = frag->base.des_src->seg_len; 
        if(VAPI_OK != VAPI_post_sr(mvapi_btl->nic, endpoint->lcl_qp_hndl_lp, &frag->desc.sr_desc)) {
            rc =  OMPI_ERROR; 
        } else { 
            rc = OMPI_SUCCESS; 
        }
#ifdef VAPI_FEATURE_SRQ
        if(mca_btl_mvapi_component.use_srq) { 
            MCA_BTL_MVAPI_POST_SRR_HIGH(mvapi_btl, 1); 
            MCA_BTL_MVAPI_POST_SRR_LOW(mvapi_btl, 1); 
        } else 
#endif
        { 
            MCA_BTL_MVAPI_ENDPOINT_POST_RR_HIGH(endpoint, 1); 
            MCA_BTL_MVAPI_ENDPOINT_POST_RR_LOW(endpoint, 1); 
        }
    }
    return rc; 
}

/* 
 * RDMA read remote buffer to local buffer address. 
 */ 

int mca_btl_mvapi_get( mca_btl_base_module_t* btl, 
                       mca_btl_base_endpoint_t* endpoint,
                       mca_btl_base_descriptor_t* descriptor)
{
    int rc; 
    mca_btl_mvapi_module_t* mvapi_btl = (mca_btl_mvapi_module_t*) btl; 
    mca_btl_mvapi_frag_t* frag = (mca_btl_mvapi_frag_t*) descriptor; 
    
    frag->endpoint = endpoint;
    frag->desc.sr_desc.opcode = VAPI_RDMA_READ; 

   /* check for a send wqe */
    if (OPAL_THREAD_ADD32(&endpoint->sd_wqe_lp,-1) < 0) {
                                                                                                                        
        OPAL_THREAD_ADD32(&endpoint->sd_wqe_lp,1);
        OPAL_THREAD_LOCK(&mvapi_btl->ib_lock);
        opal_list_append(&mvapi_btl->pending_frags_lp, (opal_list_item_t *)frag);
        OPAL_THREAD_UNLOCK(&mvapi_btl->ib_lock);
        return OMPI_SUCCESS;
                                                                                                                        
    /* check for a get token */
    } else if(OPAL_THREAD_ADD32(&endpoint->get_tokens,-1) < 0) {
                                                                                                                        
        OPAL_THREAD_ADD32(&endpoint->sd_wqe_lp,1);
        OPAL_THREAD_ADD32(&endpoint->get_tokens,1);
        OPAL_THREAD_LOCK(&endpoint->endpoint_lock);
        opal_list_append(&endpoint->pending_frags_lp, (opal_list_item_t*)frag);
        OPAL_THREAD_UNLOCK(&endpoint->endpoint_lock);
        return OMPI_SUCCESS;
                                                                                                                        

    } else {
        
        frag->desc.sr_desc.remote_qp = endpoint->rem_info.rem_qp_num_lp; 
        frag->desc.sr_desc.remote_addr = (VAPI_virt_addr_t) frag->base.des_src->seg_addr.lval; 
        frag->desc.sr_desc.r_key = frag->base.des_src->seg_key.key32[0]; 
        frag->sg_entry.addr = (VAPI_virt_addr_t) (MT_virt_addr_t) frag->base.des_dst->seg_addr.pval; 
        frag->sg_entry.len  = frag->base.des_dst->seg_len; 

        if(VAPI_OK != VAPI_post_sr(mvapi_btl->nic, endpoint->lcl_qp_hndl_lp, &frag->desc.sr_desc)) {
            rc =  OMPI_ERROR; 
        } else { 
            rc = OMPI_SUCCESS; 
        }
#ifdef VAPI_FEATURE_SRQ
        if(mca_btl_mvapi_component.use_srq) { 
            MCA_BTL_MVAPI_POST_SRR_HIGH(mvapi_btl, 1); 
            MCA_BTL_MVAPI_POST_SRR_LOW(mvapi_btl, 1); 
        } else 
#endif
        { 
            MCA_BTL_MVAPI_ENDPOINT_POST_RR_HIGH(endpoint, 1); 
            MCA_BTL_MVAPI_ENDPOINT_POST_RR_LOW(endpoint, 1); 
        }
    }
    return rc; 
    
}

                       

/*
 * Asynchronous event handler to detect unforseen
 * events. Usually, such events are catastrophic.
 * Should have a robust mechanism to handle these
 * events and abort the OMPI application if necessary.
 *
 */
static void async_event_handler(VAPI_hca_hndl_t hca_hndl,
        VAPI_event_record_t * event_p,
        void *priv_data)
{
    switch (event_p->type) {
    case VAPI_QP_PATH_MIGRATED:
    case VAPI_EEC_PATH_MIGRATED:
    case VAPI_QP_COMM_ESTABLISHED:
    case VAPI_EEC_COMM_ESTABLISHED:
    case VAPI_SEND_QUEUE_DRAINED:
    case VAPI_PORT_ACTIVE:
        {
            BTL_VERBOSE(("Got an asynchronous event: %s\n", VAPI_event_record_sym(event_p->type)));
            break;
        }
    case VAPI_CQ_ERROR:
    case VAPI_LOCAL_WQ_INV_REQUEST_ERROR:
    case VAPI_LOCAL_WQ_ACCESS_VIOL_ERROR:
    case VAPI_LOCAL_WQ_CATASTROPHIC_ERROR:
    case VAPI_PATH_MIG_REQ_ERROR:
    case VAPI_LOCAL_EEC_CATASTROPHIC_ERROR:
    case VAPI_LOCAL_CATASTROPHIC_ERROR:
    case VAPI_PORT_ERROR:
        {
            BTL_ERROR(("Got an asynchronous event: %s (%s)",
                       VAPI_event_record_sym(event_p->type),
                       VAPI_event_syndrome_sym(event_p->syndrome)));
            break;
        }
#ifdef VAPI_FEATURE_SRQ
    case VAPI_SRQ_LIMIT_REACHED: 
        { 
            size_t i;
            BTL_ERROR(("SRQ limit is reached, posting more buffers %s\n", VAPI_event_record_sym(event_p->type))); 
            for(i = 0; i < mca_btl_mvapi_component.ib_num_btls; i++) { 
                mca_btl_mvapi_module_t* mvapi_btl = &mca_btl_mvapi_component.mvapi_btls[i]; 
            
                MCA_BTL_MVAPI_POST_SRR_HIGH(mvapi_btl, 1); 
                MCA_BTL_MVAPI_POST_SRR_LOW(mvapi_btl, 1);
            }
        }
#endif 

/* BWB - is this right? */       
#ifdef VAPI_FEATURE_SRQ
    case VAPI_RECEIVE_QUEUE_DRAINED: { 
        fprintf(stderr, "VAPI_RECEIVE_QUEUE_DRAINEDD\n");
        
    }
#endif

    default:
        BTL_ERROR(("Warning!! Got an undefined "
                   "asynchronous event %s", VAPI_event_record_sym(event_p->type)));
    }

}


/* 
 * Initialize the btl module by allocating a protection domain 
 *  and creating both the high and low priority completion queues 
 */ 
int mca_btl_mvapi_module_init(mca_btl_mvapi_module_t *mvapi_btl)
{

    /* Allocate Protection Domain */ 
    VAPI_ret_t ret;
    uint32_t cqe_cnt = 0;
#ifdef VAPI_FEATURE_SRQ
    VAPI_srq_attr_t srq_attr, srq_attr_out, srq_attr_mod; 
    VAPI_srq_attr_mask_t srq_attr_mask;
    uint32_t max_outs_wr; 
#endif
    ret = VAPI_alloc_pd(mvapi_btl->nic, &mvapi_btl->ptag);
    
    if(ret != VAPI_OK) {
        BTL_ERROR(("error in VAPI_alloc_pd: %s", VAPI_strerror(ret)));
        return OMPI_ERROR;
    }

#ifdef VAPI_FEATURE_SRQ    
    if(mca_btl_mvapi_component.use_srq) { 
        mvapi_btl->srd_posted_hp = 0; 
        mvapi_btl->srd_posted_lp = 0; 
        srq_attr.pd_hndl = mvapi_btl->ptag; 
        srq_attr.max_outs_wr = mca_btl_mvapi_component.srq_rd_max;
        srq_attr.max_sentries = mca_btl_mvapi_component.ib_sg_list_size; 
        
        srq_attr_mod.srq_limit = mvapi_btl->rd_num * 0.9;
        ret = VAPI_create_srq(mvapi_btl->nic, 
                              &srq_attr, 
                              &mvapi_btl->srq_hndl_hp, 
                              &srq_attr_out); 
        if(ret != VAPI_OK) {
            BTL_ERROR(("error in VAPI_create_srq: %s", VAPI_strerror(ret)));
            return OMPI_ERROR;
        }
        
        srq_attr_mask = 0; 
        srq_attr_mask |= VAPI_SRQ_ATTR_LIMIT; 
                                  
        ret = VAPI_modify_srq 
            ( 
             mvapi_btl->nic, 
             mvapi_btl->srq_hndl_hp, 
             &srq_attr_mod, 
             srq_attr_mask, 
             &max_outs_wr 
             );

        if(ret != VAPI_OK) {
       /*      BTL_ERROR(("error in VAPI_modify_srq: %s", VAPI_strerror(ret))); */
/*             return OMPI_ERROR; */
        }
        
        ret = VAPI_create_srq(mvapi_btl->nic, 
                              &srq_attr, 
                              &mvapi_btl->srq_hndl_lp, 
                              &srq_attr_out); 
        if(ret != VAPI_OK) {
            BTL_ERROR(("error in VAPI_create_srq: %s", VAPI_strerror(ret)));
            return OMPI_ERROR;
        }
        
        srq_attr_mask = 0; 
        srq_attr_mask |= VAPI_SRQ_ATTR_LIMIT; 
                                  
        ret = VAPI_modify_srq 
            ( 
             mvapi_btl->nic, 
             mvapi_btl->srq_hndl_lp, 
             &srq_attr_mod, 
             srq_attr_mask, 
             &max_outs_wr 
             );

        if(ret != VAPI_OK) {
         /*    BTL_ERROR(("error in VAPI_modify_srq: %s", VAPI_strerror(ret))); */
/*             return OMPI_ERROR; */
        }
        

    } else {
        mvapi_btl->srq_hndl_hp = VAPI_INVAL_SRQ_HNDL; 
        mvapi_btl->srq_hndl_lp = VAPI_INVAL_SRQ_HNDL; 
    } 
#endif /* VAPI_FEATURE_SRQ */
    ret = VAPI_create_cq(mvapi_btl->nic, mca_btl_mvapi_component.ib_cq_size,
                         &mvapi_btl->cq_hndl_lp, &cqe_cnt);

    
    if( VAPI_OK != ret) {  
        BTL_ERROR(("error in VAPI_create_cq: %s", VAPI_strerror(ret)));
        return OMPI_ERROR;
    }
    
    ret = VAPI_create_cq(mvapi_btl->nic, mca_btl_mvapi_component.ib_cq_size,
                         &mvapi_btl->cq_hndl_hp, &cqe_cnt);

    
    if( VAPI_OK != ret) {  
        BTL_ERROR(("error in VAPI_create_cq: %s", VAPI_strerror(ret)));
        return OMPI_ERROR;
    }

    
    if(cqe_cnt <= 0) { 
        BTL_ERROR(("error creating completion queue ")); 
        return OMPI_ERROR; 
    } 

    ret = EVAPI_set_async_event_handler(mvapi_btl->nic,
            async_event_handler, 0, &mvapi_btl->async_handler);

    if(VAPI_OK != ret) {
        BTL_ERROR(("error in EVAPI_set_async_event_handler: %s", VAPI_strerror(ret)));
        return OMPI_ERROR;
    }
    
    
    return OMPI_SUCCESS;
}


/*
 * Dump state of btl/queues
 */
/*#include "orte/mca/ns/ns_types.h"*/
void mca_btl_mvapi_dump(
    struct mca_btl_base_module_t* btl,
    struct mca_btl_base_endpoint_t* endpoint,
    int verbose)
{
    mca_btl_mvapi_module_t* mvapi_btl = (mca_btl_mvapi_module_t*)btl;

    if( NULL == endpoint ) {
        opal_output( 0, "No endpoint for this peer\n" );
        return;
    }
    opal_output( 0, "endpoint with processor (%lu.%lu.%lu)\n", 
                 ORTE_NAME_ARGS( &(endpoint->endpoint_proc->proc_ompi->proc_name) ) );
    opal_output( 0, "endpoint state: %s\n",
                 (endpoint->endpoint_state == MCA_BTL_IB_CONNECTING ? "connecting" :
                  (endpoint->endpoint_state == MCA_BTL_IB_CONNECT_ACK ? "waiting ack" :
                   (endpoint->endpoint_state == MCA_BTL_IB_WAITING_ACK ? "waiting final ack" :
                    (endpoint->endpoint_state == MCA_BTL_IB_CONNECTED ? "connected" :
                     (endpoint->endpoint_state == MCA_BTL_IB_CLOSED ? "closed" :
                      (endpoint->endpoint_state == MCA_BTL_IB_FAILED ? "failed" : "unknown")))))));

    opal_output( 0, "pending send frags: %d\n", opal_list_get_size(&endpoint->pending_send_frags) );
    opal_output( 0, "pending frags hp : %d\n", opal_list_get_size(&endpoint->pending_frags_hp) );
    opal_output( 0, "pending frags lp : %d\n", opal_list_get_size(&endpoint->pending_frags_lp) );
#ifdef VAPI_FEATURE_SRQ
    if( mca_btl_mvapi_component.use_srq ) {
        opal_output( 0, "mvapi_btl->srd_posted_hp %d\n", mvapi_btl->srd_posted_hp );
        opal_output( 0, "mvapi_btl->srd_posted_lp %d\n", mvapi_btl->srd_posted_lp );
        opal_output( 0, "mvapi_btl->sd_tokens_hp %d\n", mvapi_btl->sd_tokens_hp );
        opal_output( 0, "mvapi_btl->sd_tokens_lp %d\n", mvapi_btl->sd_tokens_lp );
    } else {
#endif  /* VAPI_FEATURE_SRQ */
        opal_output( 0, "sd_tokens_hp %d\n", endpoint->sd_tokens_hp );
        opal_output( 0, "sd_tokens_lp %d\n", endpoint->sd_tokens_lp );
        opal_output( 0, "get_tokens %d\n", endpoint->get_tokens );
        opal_output( 0, "rd_posted_hp %d\n", endpoint->rd_posted_hp );
        opal_output( 0, "rd_posted_lp %d\n", endpoint->rd_posted_lp );
        opal_output( 0, "rd_credits_hp %d\n", endpoint->rd_credits_hp );
        opal_output( 0, "rd_credits_lp %d\n", endpoint->rd_credits_lp );
        opal_output( 0, "sd_credits_hp %d\n", endpoint->sd_credits_hp );
        opal_output( 0, "sd_credits_lp %d\n", endpoint->sd_credits_lp );
#ifdef VAPI_FEATURE_SRQ
    }
#endif  /* VAPI_FEATURE_SRQ */
    opal_output( 0, "sd_wqe_hp %d\n", endpoint->sd_wqe_hp );
    opal_output( 0, "sd_wqe_lp %d\n", endpoint->sd_wqe_lp );
}