btl_mvapi_endpoint.c

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

/*
 * Copyright (c) 2004-2005 The Trustees of Indiana University and Indiana
 *                         University Research and Technology
 *                         Corporation.  All rights reserved.
 * Copyright (c) 2004-2005 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 (c) 2007      Cisco, Inc.  All rights reserved.
 * $COPYRIGHT$
 * 
 * Additional copyrights may follow
 * 
 * $HEADER$
 */


#include "ompi_config.h"
#include <sys/time.h>
#include <time.h>
#include "ompi/types.h"
#include "ompi/mca/pml/base/pml_base_sendreq.h"
#include "orte/mca/rml/rml.h"
#include "orte/mca/errmgr/errmgr.h"
#include "btl_mvapi.h"
#include "btl_mvapi_endpoint.h" 
#include "btl_mvapi_proc.h"
#include "btl_mvapi_frag.h"
#include "ompi/class/ompi_free_list.h" 

static void mca_btl_mvapi_endpoint_construct(mca_btl_base_endpoint_t* endpoint);
static void mca_btl_mvapi_endpoint_destruct(mca_btl_base_endpoint_t* endpoint);

int mca_btl_mvapi_endpoint_create_qp(
    mca_btl_mvapi_module_t* mvapi_btl, 
    VAPI_hca_hndl_t nic,
    VAPI_pd_hndl_t ptag, 
    VAPI_cq_hndl_t cq_hndl, 
#ifdef VAPI_FEATURE_SRQ
    VAPI_srq_hndl_t srq_hndl, 
#endif
    VAPI_qp_hndl_t* qp_hndl, 
    VAPI_qp_prop_t* qp_prop, 
    int transport_type
); 

int mca_btl_mvapi_endpoint_qp_init_query(
    mca_btl_mvapi_module_t* mvapi_btl, 
    VAPI_hca_hndl_t nic, 
    VAPI_qp_hndl_t qp_hndl, 
    VAPI_qp_num_t remote_qp_num, 
    IB_lid_t remote_lid, 
    IB_port_t port_id
); 

/*
 * Note this routine must be called w/ endpoint lock held.
 */
                   
static inline int mca_btl_mvapi_endpoint_post_send(
    mca_btl_mvapi_module_t* mvapi_btl, 
    mca_btl_mvapi_endpoint_t * endpoint, 
    mca_btl_mvapi_frag_t * frag)
{
    int do_rdma = 0; 
    VAPI_qp_hndl_t qp_hndl; 
    int ret;

    if(frag->base.des_flags & MCA_BTL_DES_FLAGS_PRIORITY &&
            frag->size <= mvapi_btl->super.btl_eager_limit){ 

        /* check for a send wqe */
        if (OPAL_THREAD_ADD32(&endpoint->sd_wqe_hp,-1) < 0) {
            OPAL_THREAD_ADD32(&endpoint->sd_wqe_hp,1);
            opal_list_append(&endpoint->pending_frags_hp, (opal_list_item_t *)frag);
            return OMPI_SUCCESS;
        }
        /* check for rdma tocken */
        if (OPAL_THREAD_ADD32(&endpoint->eager_rdma_remote.tokens,-1) < 0) {
            OPAL_THREAD_ADD32(&endpoint->eager_rdma_remote.tokens,1);
            /* check for a token */
            if(!mca_btl_mvapi_component.use_srq &&
                    OPAL_THREAD_ADD32(&endpoint->sd_tokens_hp,-1) < 0) {
                OPAL_THREAD_ADD32(&endpoint->sd_wqe_hp,1);
                OPAL_THREAD_ADD32(&endpoint->sd_tokens_hp,1);
                opal_list_append(&endpoint->pending_frags_hp,
                        (opal_list_item_t *)frag);
                return OMPI_SUCCESS;

            } else if( mca_btl_mvapi_component.use_srq &&
                   OPAL_THREAD_ADD32(&mvapi_btl->sd_tokens_hp,-1) < 0) {
                OPAL_THREAD_ADD32(&endpoint->sd_wqe_hp,1);
                OPAL_THREAD_ADD32(&mvapi_btl->sd_tokens_hp,1);
                OPAL_THREAD_LOCK(&mvapi_btl->ib_lock);
                opal_list_append(&mvapi_btl->pending_frags_hp, (opal_list_item_t *)frag);
                OPAL_THREAD_UNLOCK(&mvapi_btl->ib_lock);
                return OMPI_SUCCESS;
            }
        } else {
            do_rdma = 1;
        }
        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);
        qp_hndl = endpoint->lcl_qp_hndl_hp;
    } else {
        /* check for a send wqe */
        if (OPAL_THREAD_ADD32(&endpoint->sd_wqe_lp,-1) < 0) {

            OPAL_THREAD_ADD32(&endpoint->sd_wqe_lp,1);
            opal_list_append(&endpoint->pending_frags_lp, (opal_list_item_t *)frag);
            return OMPI_SUCCESS;

        /* check for a token */
        } else if(!mca_btl_mvapi_component.use_srq &&
            OPAL_THREAD_ADD32(&endpoint->sd_tokens_lp,-1) < 0 ) {

            OPAL_THREAD_ADD32(&endpoint->sd_wqe_lp,1);
            OPAL_THREAD_ADD32(&endpoint->sd_tokens_lp,1);
            opal_list_append(&endpoint->pending_frags_lp, (opal_list_item_t *)frag);
            return OMPI_SUCCESS;

        } else if(mca_btl_mvapi_component.use_srq &&
            OPAL_THREAD_ADD32(&mvapi_btl->sd_tokens_lp,-1) < 0) {

            OPAL_THREAD_ADD32(&endpoint->sd_wqe_lp,1);
            OPAL_THREAD_ADD32(&mvapi_btl->sd_tokens_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;

        /* queue the request */
        } else {
            frag->hdr->credits = (endpoint->rd_credits_lp > 0) ? endpoint->rd_credits_lp : 0;
            OPAL_THREAD_ADD32(&endpoint->rd_credits_lp, -frag->hdr->credits);
            qp_hndl = endpoint->lcl_qp_hndl_lp;
        }
    } 
    
    frag->desc.sr_desc.remote_qkey = 0; 
    frag->sg_entry.addr = (VAPI_virt_addr_t) (MT_virt_addr_t) frag->hdr; 
    frag->sg_entry.len =
        frag->segment.seg_len + sizeof(mca_btl_mvapi_header_t) + 
        (do_rdma ? sizeof(mca_btl_mvapi_footer_t) : 0);

    if(do_rdma) {
        mca_btl_mvapi_footer_t* ftr =
            (mca_btl_mvapi_footer_t*)(((char*)frag->segment.seg_addr.pval) +
                                       frag->segment.seg_len);
        frag->desc.sr_desc.opcode = VAPI_RDMA_WRITE;
        MCA_BTL_MVAPI_RDMA_FRAG_SET_SIZE(ftr, frag->sg_entry.len);
        MCA_BTL_MVAPI_RDMA_MAKE_LOCAL(ftr);
#ifdef OMPI_ENABLE_DEBUG
        ftr->seq = endpoint->eager_rdma_remote.seq++;
#endif
        frag->desc.sr_desc.r_key = (VAPI_rkey_t)endpoint->eager_rdma_remote.rkey;
        frag->desc.sr_desc.remote_addr = (VAPI_virt_addr_t)
            endpoint->eager_rdma_remote.base.lval +
            endpoint->eager_rdma_remote.head *
            mvapi_btl->eager_rdma_frag_size +
            sizeof(mca_btl_mvapi_frag_t) +
            sizeof(mca_btl_mvapi_header_t) +
            frag->size +
            sizeof(mca_btl_mvapi_footer_t);
        frag->desc.sr_desc.remote_addr -= frag->sg_entry.len;
        MCA_BTL_MVAPI_RDMA_NEXT_INDEX (endpoint->eager_rdma_remote.head);
    } else {
        frag->desc.sr_desc.opcode = VAPI_SEND;
    }


    if(frag->sg_entry.len <= mvapi_btl->ib_inline_max) { 
        ret = EVAPI_post_inline_sr(mvapi_btl->nic, qp_hndl, &frag->desc.sr_desc); 
    } else { 
        ret = VAPI_post_sr(mvapi_btl->nic, qp_hndl, &frag->desc.sr_desc); 
    }

    if(VAPI_OK != ret) {
        BTL_ERROR(("VAPI_post_sr: %s\n", VAPI_strerror(ret)));
        return OMPI_ERROR; 
    }
#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 OMPI_SUCCESS; 
}



OBJ_CLASS_INSTANCE(mca_btl_mvapi_endpoint_t, 
                   opal_list_item_t, mca_btl_mvapi_endpoint_construct, 
                   mca_btl_mvapi_endpoint_destruct);

/*
 * Initialize state of the endpoint instance.
 *
 */

static void mca_btl_mvapi_endpoint_construct(mca_btl_base_endpoint_t* endpoint)
{
    endpoint->endpoint_btl = 0;
    endpoint->endpoint_proc = 0;
    endpoint->endpoint_tstamp = 0.0;
    endpoint->endpoint_state = MCA_BTL_IB_CLOSED;
    endpoint->endpoint_retries = 0;
    OBJ_CONSTRUCT(&endpoint->endpoint_lock, opal_mutex_t);
    OBJ_CONSTRUCT(&endpoint->pending_send_frags, opal_list_t);
    OBJ_CONSTRUCT(&endpoint->pending_frags_hp, opal_list_t);
    OBJ_CONSTRUCT(&endpoint->pending_frags_lp, opal_list_t);
    
    endpoint->rd_posted_hp = 0;
    endpoint->rd_posted_lp = 0;

    /* number of available send wqes */
    endpoint->sd_wqe_hp = mca_btl_mvapi_component.rd_num;
    endpoint->sd_wqe_lp = mca_btl_mvapi_component.rd_num;

    /* zero these out w/ initial posting, so that we start out w/
     * zero credits to return to peer
     */
    endpoint->rd_credits_hp = -(mca_btl_mvapi_component.rd_num + mca_btl_mvapi_component.rd_rsv);
    endpoint->rd_credits_lp = -(mca_btl_mvapi_component.rd_num + mca_btl_mvapi_component.rd_rsv);
    endpoint->sd_credits_hp = 0;
    endpoint->sd_credits_lp = 0;

    /* initialize the high and low priority tokens */ 
    endpoint->sd_tokens_hp = mca_btl_mvapi_component.rd_num; 
    endpoint->sd_tokens_lp = mca_btl_mvapi_component.rd_num; 
    endpoint->get_tokens = mca_btl_mvapi_component.ib_qp_ous_rd_atom;

    /* initialize RDMA eager related parts */
    endpoint->eager_recv_count = 0;
    memset(&endpoint->eager_rdma_remote, 0,
           sizeof(mca_btl_mvapi_eager_rdma_remote_t));
    memset (&endpoint->eager_rdma_local, 0,
           sizeof(mca_btl_mvapi_eager_rdma_local_t));
    OBJ_CONSTRUCT(&endpoint->eager_rdma_local.lock, opal_mutex_t);

    endpoint->rem_info.rem_qp_num_hp = 0; 
    endpoint->rem_info.rem_qp_num_lp = 0; 
    endpoint->rem_info.rem_lid = 0; 
    endpoint->rem_info.rem_subnet = 0; 
}

/*
 * Destroy a endpoint
 *
 */

static void mca_btl_mvapi_endpoint_destruct(mca_btl_base_endpoint_t* endpoint)
{
    OBJ_DESTRUCT(&endpoint->endpoint_lock);
    OBJ_DESTRUCT(&endpoint->pending_send_frags);
    OBJ_DESTRUCT(&endpoint->pending_frags_hp);
    OBJ_DESTRUCT(&endpoint->pending_frags_lp);
}

/*
 * Send connection information to remote endpoint using OOB
 *
 */

static void mca_btl_mvapi_endpoint_send_cb(
    int status,
    orte_process_name_t* endpoint, 
    orte_buffer_t* buffer,
    orte_rml_tag_t tag, 
    void* cbdata)
{
    OBJ_RELEASE(buffer);
}


static int mca_btl_mvapi_endpoint_send_connect_data(mca_btl_base_endpoint_t* endpoint)
{
    orte_buffer_t* buffer = OBJ_NEW(orte_buffer_t);
    int rc;
    if(NULL == buffer) {
         ORTE_ERROR_LOG(ORTE_ERR_OUT_OF_RESOURCE);
         return ORTE_ERR_OUT_OF_RESOURCE;
    }

    /* pack the info in the send buffer */

    rc = orte_dss.pack(buffer, &endpoint->lcl_qp_prop_hp.qp_num, 1, ORTE_UINT32);
    if(rc != ORTE_SUCCESS) {
        ORTE_ERROR_LOG(rc);
        return rc;
    }
    
    rc = orte_dss.pack(buffer, &endpoint->lcl_qp_prop_lp.qp_num, 1, ORTE_UINT32);
    if(rc != ORTE_SUCCESS) {
        ORTE_ERROR_LOG(rc);
        return rc;
    }
    rc = orte_dss.pack(buffer, &endpoint->endpoint_btl->port.lid, 1, ORTE_UINT32);
    if(rc != ORTE_SUCCESS) {
        ORTE_ERROR_LOG(rc);
        return rc;
    }
    
    rc = orte_dss.pack(buffer, &((mca_btl_mvapi_endpoint_t*)endpoint)->subnet, 1, ORTE_UINT32);
    if(rc != ORTE_SUCCESS) {
        ORTE_ERROR_LOG(rc);