btl_mvapi_endpoint.c

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

    qp_attr.qp_state = VAPI_RTR;
    QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_QP_STATE);
    qp_attr.qp_ous_rd_atom = mca_btl_mvapi_component.ib_qp_ous_rd_atom;
    QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_QP_OUS_RD_ATOM);
    qp_attr.path_mtu = mca_btl_mvapi_component.ib_mtu;
    QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_PATH_MTU);
    qp_attr.rq_psn = mca_btl_mvapi_component.ib_psn;
    QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_RQ_PSN);
    qp_attr.pkey_ix = mca_btl_mvapi_component.ib_pkey_ix;
    QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_PKEY_IX);
    qp_attr.min_rnr_timer = mca_btl_mvapi_component.ib_min_rnr_timer;
    QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_MIN_RNR_TIMER);

    qp_attr.av.sl = mca_btl_mvapi_component.ib_service_level;
    qp_attr.av.grh_flag = FALSE;
    qp_attr.av.static_rate = mca_btl_mvapi_component.ib_static_rate;
    qp_attr.av.src_path_bits = mca_btl_mvapi_component.ib_src_path_bits;

    qp_attr.dest_qp_num = remote_qp_num;
    QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_DEST_QP_NUM);
    qp_attr.av.dlid = remote_lid;
    QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_AV);

    ret = VAPI_modify_qp(nic, qp_hndl,
            &qp_attr, &qp_attr_mask, &qp_cap);

    if(VAPI_OK != ret) {
        BTL_ERROR(("Error modifying the queue pair: %s", VAPI_strerror(ret)));
        return OMPI_ERROR;
    }
    
    BTL_VERBOSE(("Modified to RTR..Qp %d", qp_hndl));

    /************** RTS *******************/
    QP_ATTR_MASK_CLR_ALL(qp_attr_mask);
    qp_attr.qp_state = VAPI_RTS;
    QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_QP_STATE);
    qp_attr.sq_psn = mca_btl_mvapi_component.ib_psn;
    QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_SQ_PSN);
    qp_attr.timeout = mca_btl_mvapi_component.ib_timeout;
    QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_TIMEOUT);
    qp_attr.retry_count = mca_btl_mvapi_component.ib_retry_count;
    QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_RETRY_COUNT);
    qp_attr.rnr_retry = mca_btl_mvapi_component.ib_rnr_retry;
    QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_RNR_RETRY);
    qp_attr.ous_dst_rd_atom = mca_btl_mvapi_component.ib_max_rdma_dst_ops;
    QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_OUS_DST_RD_ATOM);

    ret = VAPI_modify_qp(nic, qp_hndl,
            &qp_attr, &qp_attr_mask, &qp_cap);

    if(VAPI_OK != ret) {
        return OMPI_ERROR;
    }
    BTL_VERBOSE(("Modified to RTS..Qp %d", qp_hndl));
    
    ret = VAPI_query_qp(nic, qp_hndl, &qp_attr, &qp_attr_mask, &qp_init_attr );          
    if (ret != VAPI_OK) {                                                   
        BTL_ERROR(("Error modifying the queue pair: %s", VAPI_strerror(ret)));
        return OMPI_ERROR; 
    }                      
    
    mvapi_btl->ib_inline_max = qp_init_attr.cap.max_inline_data_sq;  
    
    return OMPI_SUCCESS;
}

                                                                                                                       
/**
 * Return control fragment.
 */

static void mca_btl_mvapi_endpoint_credits_lp(
    mca_btl_base_module_t* btl,
    struct mca_btl_base_endpoint_t* endpoint,
    struct mca_btl_base_descriptor_t* descriptor,
    int status)
{
    int32_t credits;

    /* we don't acquire a wqe or token for credit message - so decrement */
    OPAL_THREAD_ADD32(&endpoint->sd_wqe_lp,-1);

    /* check to see if there are addditional credits to return */
    if ((credits = OPAL_THREAD_ADD32(&endpoint->sd_credits_lp,-1)) > 0) {
        OPAL_THREAD_ADD32(&endpoint->sd_credits_lp,-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);
        }
    }
    MCA_BTL_IB_FRAG_RETURN(((mca_btl_mvapi_module_t*)btl), 
                           ((mca_btl_mvapi_frag_t*)descriptor));
}

/**
 * Return credits to peer
 */

void mca_btl_mvapi_endpoint_send_credits_lp(
    mca_btl_mvapi_endpoint_t* endpoint)
{
    mca_btl_mvapi_module_t* mvapi_btl = endpoint->endpoint_btl;
    mca_btl_mvapi_frag_t* frag;
    int ret;

    MCA_BTL_IB_FRAG_ALLOC_EAGER(mvapi_btl, frag, ret);
    if(NULL == frag) {
        BTL_ERROR(("error allocating fragment"));
        return;
    }

    frag->base.des_cbfunc = mca_btl_mvapi_endpoint_credits_lp;
    frag->base.des_cbdata = NULL;
    frag->endpoint = endpoint;

    frag->hdr->tag = MCA_BTL_TAG_BTL;
    frag->hdr->credits = endpoint->rd_credits_lp;
    OPAL_THREAD_ADD32(&endpoint->rd_credits_lp, -frag->hdr->credits);
    ((mca_btl_mvapi_control_header_t *)frag->segment.seg_addr.pval)->type = MCA_BTL_MVAPI_CONTROL_NOOP;

    frag->desc.sr_desc.opcode = VAPI_SEND; 
    frag->sg_entry.addr = (VAPI_virt_addr_t) (MT_virt_addr_t) frag->hdr; 
    frag->sg_entry.len = sizeof(mca_btl_mvapi_header_t) +
        sizeof(mca_btl_mvapi_control_header_t);

    if(sizeof(mca_btl_mvapi_header_t) <= mvapi_btl->ib_inline_max) {
        ret = EVAPI_post_inline_sr(mvapi_btl->nic, endpoint->lcl_qp_hndl_lp, &frag->desc.sr_desc);
    } else {
        ret = VAPI_post_sr(mvapi_btl->nic, endpoint->lcl_qp_hndl_lp, &frag->desc.sr_desc);
    }
    if(ret != VAPI_SUCCESS) {
        OPAL_THREAD_ADD32(&endpoint->sd_credits_lp, -1);
        OPAL_THREAD_ADD32(&endpoint->rd_credits_lp, frag->hdr->credits);
        MCA_BTL_IB_FRAG_RETURN(mvapi_btl, frag);
        BTL_ERROR(("error posting send request errno %d says %s", strerror(errno)));
        return;
    }
}

/**
 * Return control fragment.
 */

static void mca_btl_mvapi_endpoint_credits_hp(
    mca_btl_base_module_t* btl,
    struct mca_btl_base_endpoint_t* endpoint,
    struct mca_btl_base_descriptor_t* descriptor,
    int status)
{
    int32_t credits;

    /* we don't acquire a wqe or token for credit message - so decrement */
    OPAL_THREAD_ADD32(&endpoint->sd_wqe_hp,-1);

    /* check to see if there are addditional credits to return */
    if ((credits = OPAL_THREAD_ADD32(&endpoint->sd_credits_hp,-1)) > 0) {
        OPAL_THREAD_ADD32(&endpoint->sd_credits_hp,-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);
        }
    }
    MCA_BTL_IB_FRAG_RETURN(((mca_btl_mvapi_module_t*)btl), 
                           ((mca_btl_mvapi_frag_t*)descriptor));
}

/**
 * Return credits to peer
 */

void mca_btl_mvapi_endpoint_send_credits_hp(
    mca_btl_mvapi_endpoint_t* endpoint)
{
    mca_btl_mvapi_module_t* mvapi_btl = endpoint->endpoint_btl;
    mca_btl_mvapi_frag_t* frag;
    int ret;

    MCA_BTL_IB_FRAG_ALLOC_EAGER(mvapi_btl, frag, ret);
    if(NULL == frag) {
        BTL_ERROR(("error allocating fragment"));
        return;
    }

    frag->base.des_cbfunc = mca_btl_mvapi_endpoint_credits_hp;
    frag->base.des_cbdata = NULL;
    frag->endpoint = endpoint;

    frag->hdr->tag = MCA_BTL_TAG_BTL;
    frag->hdr->credits =
        (endpoint->rd_credits_hp > 0) ? endpoint->rd_credits_hp: 0;
    OPAL_THREAD_ADD32(&endpoint->rd_credits_hp, -frag->hdr->credits);
    frag->hdr->rdma_credits = endpoint->eager_rdma_local.credits;
    OPAL_THREAD_ADD32(&endpoint->eager_rdma_local.credits,
            -frag->hdr->rdma_credits);
    ((mca_btl_mvapi_control_header_t *)frag->segment.seg_addr.pval)->type = MCA_BTL_MVAPI_CONTROL_NOOP;


    frag->desc.sr_desc.opcode = VAPI_SEND; 
    frag->sg_entry.addr = (VAPI_virt_addr_t) (MT_virt_addr_t) frag->hdr; 
    frag->sg_entry.len = sizeof(mca_btl_mvapi_header_t) +
        sizeof(mca_btl_mvapi_control_header_t);

    if(sizeof(mca_btl_mvapi_header_t) <= mvapi_btl->ib_inline_max) {
        ret = EVAPI_post_inline_sr(mvapi_btl->nic, endpoint->lcl_qp_hndl_hp, &frag->desc.sr_desc);
    } else {
        ret = VAPI_post_sr(mvapi_btl->nic, endpoint->lcl_qp_hndl_hp, &frag->desc.sr_desc);
    }
    if(ret != VAPI_SUCCESS) {
        OPAL_THREAD_ADD32(&endpoint->sd_credits_lp, -1);
        OPAL_THREAD_ADD32(&endpoint->rd_credits_lp, frag->hdr->credits);
        MCA_BTL_IB_FRAG_RETURN(mvapi_btl, frag);
        BTL_ERROR(("error posting send request errno %d says %s", strerror(errno)));
        return;
    }
}

static void mca_btl_mvapi_endpoint_eager_rdma(
    mca_btl_base_module_t* btl,
    struct mca_btl_base_endpoint_t* endpoint,
    struct mca_btl_base_descriptor_t* descriptor,
    int status)
{
    MCA_BTL_IB_FRAG_RETURN(((mca_btl_mvapi_module_t*)btl),
                           ((mca_btl_mvapi_frag_t*)descriptor));
}

static int mca_btl_mvapi_endpoint_send_eager_rdma(
    mca_btl_base_endpoint_t* endpoint)
{
    mca_btl_mvapi_module_t* mvapi_btl = endpoint->endpoint_btl;
    mca_btl_mvapi_eager_rdma_header_t *rdma_hdr;
    mca_btl_mvapi_frag_t* frag;
    int rc;

    MCA_BTL_IB_FRAG_ALLOC_EAGER(mvapi_btl, frag, rc);
    if(NULL == frag) {
        BTL_ERROR(("error allocating fragment"));
        return -1;
    }

    frag->base.des_cbfunc = mca_btl_mvapi_endpoint_eager_rdma;
    frag->base.des_cbdata = NULL;
    frag->endpoint = endpoint;
    frag->base.des_flags |= MCA_BTL_DES_FLAGS_PRIORITY;

    frag->hdr->tag = MCA_BTL_TAG_BTL;
    rdma_hdr = (mca_btl_mvapi_eager_rdma_header_t*)frag->segment.seg_addr.pval;
    rdma_hdr->control.type = MCA_BTL_MVAPI_CONTROL_RDMA;
    rdma_hdr->rkey = endpoint->eager_rdma_local.reg->r_key;
    rdma_hdr->rdma_start.lval = ompi_ptr_ptol(endpoint->eager_rdma_local.base.pval);
    frag->segment.seg_len = sizeof(mca_btl_mvapi_eager_rdma_header_t);
    if (mca_btl_mvapi_endpoint_post_send(mvapi_btl, endpoint, frag) !=
            OMPI_SUCCESS) {
        MCA_BTL_IB_FRAG_RETURN(mvapi_btl, frag);
        BTL_ERROR(("Error sending RDMA buffer", strerror(errno)));
        return -1;
    }
    return 0;
}
/* create RDMA buffer for eager messages */
void mca_btl_mvapi_endpoint_connect_eager_rdma(
        mca_btl_mvapi_endpoint_t* endpoint)
{
    mca_btl_mvapi_module_t* mvapi_btl = endpoint->endpoint_btl;
    char *buf;
    unsigned int i;

    OPAL_THREAD_LOCK(&endpoint->eager_rdma_local.lock);
    if (endpoint->eager_rdma_local.base.pval)
        goto unlock_rdma_local;

    buf = mvapi_btl->super.btl_mpool->mpool_alloc(mvapi_btl->super.btl_mpool,
            mvapi_btl->eager_rdma_frag_size * 
            mca_btl_mvapi_component.eager_rdma_num, 0,
            MCA_MPOOL_FLAGS_CACHE_BYPASS,
            (mca_mpool_base_registration_t**)&endpoint->eager_rdma_local.reg);

    if(!buf)
       goto unlock_rdma_local;

    for(i = 0; i < mca_btl_mvapi_component.eager_rdma_num; i++) {
        ompi_free_list_item_t *item = (ompi_free_list_item_t *)(buf +
                i*mvapi_btl->eager_rdma_frag_size);
        item->user_data = (void*)endpoint->eager_rdma_local.reg;
        OBJ_CONSTRUCT(item, mca_btl_mvapi_recv_frag_eager_t);
        ((mca_btl_mvapi_frag_t*)item)->endpoint = endpoint;
        ((mca_btl_mvapi_frag_t*)item)->type = MCA_BTL_MVAPI_FRAG_EAGER_RDMA;
    }

    OPAL_THREAD_LOCK(&mvapi_btl->eager_rdma_lock);
    if(orte_pointer_array_add (&endpoint->eager_rdma_index,
                mvapi_btl->eager_rdma_buffers, endpoint) < 0)
      goto cleanup;

    endpoint->eager_rdma_local.base.pval = buf;
    mvapi_btl->eager_rdma_buffers_count++;
    if (mca_btl_mvapi_endpoint_send_eager_rdma(endpoint) == 0) {
        OPAL_THREAD_UNLOCK(&mvapi_btl->eager_rdma_lock);
        OPAL_THREAD_UNLOCK(&endpoint->eager_rdma_local.lock);
        return;
    }

    mvapi_btl->eager_rdma_buffers_count--;
    endpoint->eager_rdma_local.base.pval = NULL;
    orte_pointer_array_set_item(mvapi_btl->eager_rdma_buffers,
            endpoint->eager_rdma_index, NULL);

cleanup:
    OPAL_THREAD_UNLOCK(&mvapi_btl->eager_rdma_lock);
    mvapi_btl->super.btl_mpool->mpool_free(mvapi_btl->super.btl_mpool,
            buf, (mca_mpool_base_registration_t*)endpoint->eager_rdma_local.reg);
unlock_rdma_local:
    OPAL_THREAD_UNLOCK(&endpoint->eager_rdma_local.lock);
}