btl_mvapi_endpoint.c

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

        return rc;
    }

    /* send to endpoint */
    rc = orte_rml.send_buffer_nb(&endpoint->endpoint_proc->proc_guid, buffer, ORTE_RML_TAG_DYNAMIC-1, 0,
         mca_btl_mvapi_endpoint_send_cb, NULL);
    
    
    BTL_VERBOSE(("Sending High Priority QP num = %d, Low Priority QP num = %d, LID = %d",
              endpoint->lcl_qp_prop_hp.qp_num,
              endpoint->lcl_qp_prop_lp.qp_num,
              endpoint->endpoint_btl->port.lid));
    
    if(rc < 0) {
        ORTE_ERROR_LOG(rc);
        return rc;
    }
    return OMPI_SUCCESS;
}

/*
 * Set remote connection info
 *
 * XXX: Currently size is unutilized, this shall change
 * as soon as we add more info to be exchanged at connection
 * setup.
 *
 */
static int mca_btl_mvapi_endpoint_set_remote_info(mca_btl_base_endpoint_t* endpoint, mca_btl_mvapi_rem_info_t* rem_info)
{
    
    memcpy(&((mca_btl_mvapi_endpoint_t*) endpoint)->rem_info, rem_info, sizeof(mca_btl_mvapi_rem_info_t)); 
    
    BTL_VERBOSE(("Setting High Priority QP num = %d, Low Priority QP num %d,  LID = %d",
                 endpoint->rem_info.rem_qp_num_hp,
                 endpoint->rem_info.rem_qp_num_lp, 
                 endpoint->rem_info.rem_lid));

    return ORTE_SUCCESS;
}



/*
 * Start to connect to the endpoint. We send our Queue Pair
 * information over the TCP OOB communication mechanism.

 * On completion of our send, a send completion handler 
 * is called.
 *
 */

static int mca_btl_mvapi_endpoint_start_connect(mca_btl_base_endpoint_t* endpoint)
{
    int rc;
    
    
    /* Create the High Priority Queue Pair */
    if(OMPI_SUCCESS != (rc = mca_btl_mvapi_endpoint_create_qp(endpoint->endpoint_btl, 
                                                              endpoint->endpoint_btl->nic, 
                                                              endpoint->endpoint_btl->ptag, 
                                                              endpoint->endpoint_btl->cq_hndl_hp, 
#ifdef VAPI_FEATURE_SRQ
                                                              endpoint->endpoint_btl->srq_hndl_hp, 
#endif
                                                              &endpoint->lcl_qp_hndl_hp, 
                                                              &endpoint->lcl_qp_prop_hp, 
                                                              VAPI_TS_RC))) {
        BTL_ERROR(("error creating queue pair, error code %d", rc)); 
        return rc;
    }

    
    /* Create the Low Priority Queue Pair */
    if(OMPI_SUCCESS != (rc = mca_btl_mvapi_endpoint_create_qp(endpoint->endpoint_btl, 
                                                              endpoint->endpoint_btl->nic, 
                                                              endpoint->endpoint_btl->ptag, 
                                                              endpoint->endpoint_btl->cq_hndl_lp, 
#ifdef VAPI_FEATURE_SRQ
                                                              endpoint->endpoint_btl->srq_hndl_lp, 
#endif
                                                              &endpoint->lcl_qp_hndl_lp, 
                                                              &endpoint->lcl_qp_prop_lp, 
                                                              VAPI_TS_RC))) {
        
        BTL_ERROR(("error creating queue pair, error code %d", rc)); 
        return rc;
    }

    BTL_VERBOSE(("Initialized High Priority QP num = %d, Low Priority QP num = %d,  LID = %d",
              endpoint->lcl_qp_prop_hp.qp_num,
              endpoint->lcl_qp_prop_lp.qp_num,
              endpoint->endpoint_btl->port.lid));

    /* Send connection info over to remote endpoint */
    endpoint->endpoint_state = MCA_BTL_IB_CONNECTING;
    if(OMPI_SUCCESS != (rc = mca_btl_mvapi_endpoint_send_connect_data(endpoint))) {
        BTL_ERROR(("error sending connect request, error code %d", rc)); 
        return rc;
    }
    return OMPI_SUCCESS;
}

/*
 * Reply to a `start - connect' message
 *
 */
static int mca_btl_mvapi_endpoint_reply_start_connect(mca_btl_mvapi_endpoint_t *endpoint, 
                                                      mca_btl_mvapi_rem_info_t* rem_info)
{
    int rc;

    
    /* Create the High Priority Queue Pair */
    if(OMPI_SUCCESS != (rc = mca_btl_mvapi_endpoint_create_qp(endpoint->endpoint_btl, 
                                                              endpoint->endpoint_btl->nic, 
                                                              endpoint->endpoint_btl->ptag, 
                                                              endpoint->endpoint_btl->cq_hndl_hp, 
#ifdef VAPI_FEATURE_SRQ
                                                              endpoint->endpoint_btl->srq_hndl_hp, 
#endif
                                                              &endpoint->lcl_qp_hndl_hp, 
                                                              &endpoint->lcl_qp_prop_hp, 
                                                              VAPI_TS_RC))) {
        BTL_ERROR(("error creating queue pair, error code %d", rc)); 
        return rc;
    }


    /* Create the Low Priority Queue Pair */
    if(OMPI_SUCCESS != (rc = mca_btl_mvapi_endpoint_create_qp(endpoint->endpoint_btl, 
                                                              endpoint->endpoint_btl->nic, 
                                                              endpoint->endpoint_btl->ptag, 
                                                              endpoint->endpoint_btl->cq_hndl_lp, 
#ifdef VAPI_FEATURE_SRQ
                                                              endpoint->endpoint_btl->srq_hndl_lp, 
#endif
                                                              &endpoint->lcl_qp_hndl_lp, 
                                                              &endpoint->lcl_qp_prop_lp, 
                                                              VAPI_TS_RC))) {
        BTL_ERROR(("error creating queue pair, error code %d", rc)); 
        return rc;
    }

    BTL_VERBOSE(("Initialized High Priority QP num = %d, Low Priority QP num = %d,  LID = %d",
              endpoint->lcl_qp_prop_hp.qp_num,
              endpoint->lcl_qp_prop_lp.qp_num,
              endpoint->endpoint_btl->port.lid));




    /* Set the remote side info */
    mca_btl_mvapi_endpoint_set_remote_info(endpoint, rem_info);
    
    /* Connect to endpoint */
    rc = mca_btl_mvapi_endpoint_connect(endpoint);
    if(rc != OMPI_SUCCESS) {
        BTL_ERROR(("error in endpoint connect error code is %d", rc)); 
        return rc;
    }

    /* Send connection info over to remote endpoint */
    endpoint->endpoint_state = MCA_BTL_IB_CONNECT_ACK;
    if(OMPI_SUCCESS != (rc = mca_btl_mvapi_endpoint_send_connect_data(endpoint))) {
        BTL_ERROR(("error in endpoint send connect request error code is %d", rc)); 
        return rc;
    }
    return OMPI_SUCCESS;
}

static void mca_btl_mvapi_endpoint_waiting_ack(mca_btl_mvapi_endpoint_t *endpoint) { 
    endpoint->endpoint_state = MCA_BTL_IB_WAITING_ACK; 
}

/*
 *
 */

static void mca_btl_mvapi_endpoint_connected(mca_btl_mvapi_endpoint_t *endpoint)
{
    opal_list_item_t *frag_item;
    mca_btl_mvapi_frag_t *frag;
    mca_btl_mvapi_module_t* mvapi_btl; 

    /* While there are frags in the list, process them */
    endpoint->endpoint_state = MCA_BTL_IB_CONNECTED;

    /**
     * The connection is correctly setup. Now we can decrease the event trigger.
     */
    opal_progress_event_decrement();

    while(!opal_list_is_empty(&(endpoint->pending_send_frags))) {
        frag_item = opal_list_remove_first(&(endpoint->pending_send_frags));
        frag = (mca_btl_mvapi_frag_t *) frag_item;
        mvapi_btl = endpoint->endpoint_btl;
        /* We need to post this one */
        
        if(OMPI_SUCCESS !=  mca_btl_mvapi_endpoint_post_send(mvapi_btl, endpoint, frag))
            BTL_ERROR(("error in mca_btl_mvapi_endpoint_send"));
    }
}

/*
 * Non blocking OOB recv callback.
 * Read incoming QP and other info, and if this endpoint
 * is trying to connect, reply with our QP info, 
 * otherwise try to modify QP's and establish
 * reliable connection
 *
 */

static void mca_btl_mvapi_endpoint_recv(
    int status,
    orte_process_name_t* endpoint, 
    orte_buffer_t* buffer,
    orte_rml_tag_t tag, 
    void* cbdata)
{
    mca_btl_mvapi_proc_t *ib_proc;
    mca_btl_mvapi_endpoint_t *ib_endpoint = NULL;
    int endpoint_state;
    int rc; 
    uint32_t i;
    int32_t cnt = 1; 
    mca_btl_mvapi_rem_info_t rem_info; 


    /* start by unpacking data first so we know who is knocking at 
       our door */ 

    rc = orte_dss.unpack(buffer, &rem_info.rem_qp_num_hp, &cnt, ORTE_UINT32);
    if(ORTE_SUCCESS != rc) {
        ORTE_ERROR_LOG(rc);
        return;
    }
    rc = orte_dss.unpack(buffer, &rem_info.rem_qp_num_lp, &cnt, ORTE_UINT32);
    if(ORTE_SUCCESS != rc) {
        ORTE_ERROR_LOG(rc);
        return;
    }
    rc = orte_dss.unpack(buffer, &rem_info.rem_lid, &cnt, ORTE_UINT32);
    if(ORTE_SUCCESS != rc) {
        ORTE_ERROR_LOG(rc);
        return;
    }
    rc = orte_dss.unpack(buffer, &rem_info.rem_subnet, &cnt, ORTE_UINT32);
    if(ORTE_SUCCESS != rc) {
        ORTE_ERROR_LOG(rc);
        return;
    }

    BTL_VERBOSE(("Received High Priority QP num = %d, Low Priority QP num %d,  LID = %d",
                 rem_info.rem_qp_num_hp,
                 rem_info.rem_qp_num_lp, 
                 rem_info.rem_lid));

    for(ib_proc = (mca_btl_mvapi_proc_t*)
            opal_list_get_first(&mca_btl_mvapi_component.ib_procs);
            ib_proc != (mca_btl_mvapi_proc_t*)
            opal_list_get_end(&mca_btl_mvapi_component.ib_procs);
            ib_proc  = (mca_btl_mvapi_proc_t*)opal_list_get_next(ib_proc)) {

        if(orte_ns.compare_fields(ORTE_NS_CMP_ALL, &ib_proc->proc_guid, endpoint) == ORTE_EQUAL) {
            bool found = false;
            
            /* Try to get the endpoint instance of this proc */
            /* first match the endpoint based on lid meaning we've seen */ 
            /* this endpoint before.. */ 
            for(i = 0; i < ib_proc->proc_endpoint_count; i++) { 
                mca_btl_mvapi_port_info_t port_info; 
                port_info = ib_proc->proc_ports[i]; 
                ib_endpoint = ib_proc->proc_endpoints[i]; 
                if(ib_endpoint->rem_info.rem_lid && 
                   ib_endpoint->rem_info.rem_lid  == rem_info.rem_lid) { 
                    /* we've seen them before! */ 
                    found = true; 
                    break;
                }
            }
            /* If we haven't seen this remote lid before then try to match on 
               endpoint */ 
            for(i = 0; !found && i < ib_proc->proc_endpoint_count; i++) { 
                mca_btl_mvapi_port_info_t port_info; 
                port_info = ib_proc->proc_ports[i]; 
                ib_endpoint = ib_proc->proc_endpoints[i]; 
                if(!ib_endpoint->rem_info.rem_lid && 
                   ib_endpoint->subnet  == rem_info.rem_subnet) { 
                    /* found a match based on subnet! */ 
                    found = true; 
                    break;
                }
            }
            /* try finding an open port, even if subnets  
               don't match
            */ 
            for(i = 0; !found && i < ib_proc->proc_endpoint_count; i++) { 
                mca_btl_mvapi_port_info_t port_info; 
                port_info = ib_proc->proc_ports[i]; 
                ib_endpoint = ib_proc->proc_endpoints[i]; 
                if(!ib_endpoint->rem_info.rem_lid) { 
                    /* found an unused end-point */ 
                    found = true; 
                    break;
                }
            }
            
            if(!found) { 
                BTL_ERROR(("can't find suitable endpoint for this peer\n")); 
                return; 
            }

            OPAL_THREAD_LOCK(&ib_endpoint->endpoint_lock);
            endpoint_state = ib_endpoint->endpoint_state;
            
            /* Update status */
            switch(endpoint_state) {