btl_mvapi_endpoint.c

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

            case MCA_BTL_IB_CLOSED :
                /* We had this connection closed before.
                 * The endpoint is trying to connect. Move the
                 * status of this connection to CONNECTING,
                 * and then reply with our QP information */

                if(OMPI_SUCCESS != (rc = mca_btl_mvapi_endpoint_reply_start_connect(ib_endpoint, &rem_info))) {
                    BTL_ERROR(("error in endpoint reply start connect")); 
                    break;
                }
                /** As long as we expect a message from the peer (in order to setup the connection)
                 * let the event engine pool the OOB events. Note: we increment it once peer active
                 * connection.
                 */
                opal_progress_event_increment();
                break;
                
            case MCA_BTL_IB_CONNECTING :
                
                mca_btl_mvapi_endpoint_set_remote_info(ib_endpoint, &rem_info);
                if(OMPI_SUCCESS != (rc = mca_btl_mvapi_endpoint_connect(ib_endpoint))) {
                    BTL_ERROR(("endpoint connect error: %d", rc)); 
                    break;
                }

                /* Setup state as awaiting ack from peer */
                mca_btl_mvapi_endpoint_waiting_ack(ib_endpoint);

                /* Send him an ack */
                mca_btl_mvapi_endpoint_send_connect_data(ib_endpoint);
                break;

            case MCA_BTL_IB_WAITING_ACK: 
                mca_btl_mvapi_endpoint_connected(ib_endpoint);
                break; 
                
            case MCA_BTL_IB_CONNECT_ACK:

                mca_btl_mvapi_endpoint_send_connect_data(ib_endpoint);
                mca_btl_mvapi_endpoint_connected(ib_endpoint);
                break;

            case MCA_BTL_IB_CONNECTED :
                break;
            default :
                BTL_ERROR(("Invalid endpoint state %d", endpoint_state));
            }

            OPAL_THREAD_UNLOCK(&ib_endpoint->endpoint_lock);
            break;
        }
    }
}

void mca_btl_mvapi_post_recv()
{
    orte_rml.recv_buffer_nb(
        ORTE_NAME_WILDCARD, 
        ORTE_RML_TAG_DYNAMIC-1, 
        ORTE_RML_PERSISTENT,
        mca_btl_mvapi_endpoint_recv,
        NULL);
}


/*
 * Attempt to send a fragment using a given endpoint. If the endpoint is not
 * connected, queue the fragment and start the connection as required.
 */

int mca_btl_mvapi_endpoint_send(
                             mca_btl_base_endpoint_t* endpoint,
                             mca_btl_mvapi_frag_t* frag
                             )
{
    int rc, call_progress = 0;
    mca_btl_mvapi_module_t *mvapi_btl; 
    
    OPAL_THREAD_LOCK(&endpoint->endpoint_lock);
    switch(endpoint->endpoint_state) {
        case MCA_BTL_IB_CONNECTING:

            BTL_VERBOSE(("Queing because state is connecting"));

            opal_list_append(&endpoint->pending_send_frags,
                    (opal_list_item_t *)frag);

            rc = OMPI_SUCCESS;
            call_progress = 1;
            break;

        case MCA_BTL_IB_WAITING_ACK: 
        case MCA_BTL_IB_CONNECT_ACK:

            BTL_VERBOSE(("Queuing because waiting for ack"));

            opal_list_append(&endpoint->pending_send_frags,
                    (opal_list_item_t *)frag);

            rc = OMPI_SUCCESS;
            call_progress = 1;
            break;

        case MCA_BTL_IB_CLOSED:

            BTL_VERBOSE(("Connection to endpoint closed ... connecting ..."));

            opal_list_append(&endpoint->pending_send_frags,
                    (opal_list_item_t *)frag);

            rc = mca_btl_mvapi_endpoint_start_connect(endpoint);
            /**
             * As long as we expect a message from the peer (in order to setup the connection)
             * let the event engine pool the OOB events. Note: we increment it once peer active
             * connection.
             */
            opal_progress_event_increment();
            call_progress = 1;
            break;

        case MCA_BTL_IB_FAILED:

            rc = OMPI_ERR_UNREACH;
            break;

        case MCA_BTL_IB_CONNECTED:
            {
                mvapi_btl = endpoint->endpoint_btl;
                BTL_VERBOSE(("Send to : %d, len : %d, frag : %p", 
                             endpoint->endpoint_proc->proc_guid.vpid,
                             frag->sg_entry.len,
                             frag));
                rc = mca_btl_mvapi_endpoint_post_send(mvapi_btl, endpoint, frag); 
            }   
            break;
            
    default:
        rc = OMPI_ERR_UNREACH;
    }
    
    OPAL_THREAD_UNLOCK(&endpoint->endpoint_lock);
    if( call_progress ) opal_progress();
    return rc;
}




/*
 * Complete connection to endpoint.
 */

int mca_btl_mvapi_endpoint_connect(
    mca_btl_mvapi_endpoint_t *endpoint)
{
    int rc;
    /* Connection establishment RC */
    rc = mca_btl_mvapi_endpoint_qp_init_query(endpoint->endpoint_btl, 
                                              endpoint->endpoint_btl->nic, 
                                              endpoint->lcl_qp_hndl_hp, 
                                              endpoint->rem_info.rem_qp_num_hp, 
                                              endpoint->rem_info.rem_lid,                                            
                                              endpoint->endpoint_btl->port_id); 
    
    rc = mca_btl_mvapi_endpoint_qp_init_query(endpoint->endpoint_btl, 
                                              endpoint->endpoint_btl->nic, 
                                              endpoint->lcl_qp_hndl_lp, 
                                              endpoint->rem_info.rem_qp_num_lp, 
                                              endpoint->rem_info.rem_lid,                                            
                                              endpoint->endpoint_btl->port_id); 
    
    

    if(rc != OMPI_SUCCESS) {
        return rc;
    }
    
#ifdef VAPI_FEATURE_SRQ
    if(mca_btl_mvapi_component.use_srq) { 
        MCA_BTL_MVAPI_POST_SRR_HIGH(endpoint->endpoint_btl, 0); 
        MCA_BTL_MVAPI_POST_SRR_LOW(endpoint->endpoint_btl, 0);
    } else 
#endif
    {
        MCA_BTL_MVAPI_ENDPOINT_POST_RR_HIGH(endpoint, 0); 
        MCA_BTL_MVAPI_ENDPOINT_POST_RR_LOW(endpoint, 0); 
    }
    return OMPI_SUCCESS;
}

/* 
 * Create the queue pair note that this is just the initial 
 *  queue pair creation and we need to get the remote queue pair 
 *  info from the peer before the qp is usable, 
 */ 

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)
{
    
    VAPI_ret_t ret;
    VAPI_qp_init_attr_t qp_init_attr;
#ifdef VAPI_FEATURE_SRQ
    VAPI_qp_init_attr_ext_t qp_init_attr_ext;
#endif

    /* worst case number of credit messages could be queued */
    switch(transport_type) {

    case VAPI_TS_RC: /* Set up RC qp parameters */
        qp_init_attr.cap.max_oust_wr_sq = mca_btl_mvapi_component.rd_num + 1;
        qp_init_attr.cap.max_oust_wr_rq = mca_btl_mvapi_component.rd_num + mca_btl_mvapi_component.rd_rsv;
        qp_init_attr.cap.max_sg_size_sq = mca_btl_mvapi_component.ib_sg_list_size;
        qp_init_attr.cap.max_sg_size_rq = mca_btl_mvapi_component.ib_sg_list_size;
        qp_init_attr.pd_hndl            = ptag;
        /* We don't have Reliable Datagram Handle right now */
        qp_init_attr.rdd_hndl           = 0;
        
        /* Signal all work requests on this queue pair */
        qp_init_attr.rq_sig_type        = VAPI_SIGNAL_REQ_WR;
        qp_init_attr.sq_sig_type        = VAPI_SIGNAL_REQ_WR;
        
        /* Use Reliable Connected  transport service */
        qp_init_attr.ts_type            = VAPI_TS_RC;
        
            
        
        /* Set Send and Recv completion queues */
        qp_init_attr.rq_cq_hndl         = cq_hndl;
        qp_init_attr.sq_cq_hndl         = cq_hndl;
         
            break;
        case VAPI_TS_UD: /* Set up UD qp parameters */
        default:
            return OMPI_ERR_NOT_IMPLEMENTED;
    }

#ifdef VAPI_FEATURE_SRQ    
    if(mca_btl_mvapi_component.use_srq) { 
        qp_init_attr_ext.srq_hndl = srq_hndl; 
        
        ret = VAPI_create_qp_ext(nic, 
                                 &qp_init_attr, 
                                 &qp_init_attr_ext,  
                                 qp_hndl, 
                                 qp_prop);
    } else 
#endif
    { 
        ret = VAPI_create_qp(nic, 
                             &qp_init_attr, 
                             qp_hndl, 
                             qp_prop); 
    }
    if(VAPI_OK != ret) {
        BTL_ERROR(("error creating the queue pair: %s", VAPI_strerror(ret))); 
        return OMPI_ERROR;
    }
    return OMPI_SUCCESS;
}

/* 
 * The queue pair has been created and we have received the remote 
 *  queue pair information from the peer so we init this queue pair 
 *  and are ready to roll. 
 */ 
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
                                      )
     
     
{
    
    VAPI_ret_t              ret;
    VAPI_qp_attr_t          qp_attr;

    VAPI_qp_attr_mask_t     qp_attr_mask;
    VAPI_qp_init_attr_t     qp_init_attr; 
    VAPI_qp_cap_t           qp_cap;

    /* Modifying  QP to INIT */
    QP_ATTR_MASK_CLR_ALL(qp_attr_mask);
    qp_attr.qp_state = VAPI_INIT;
    QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_QP_STATE);
    qp_attr.pkey_ix = mca_btl_mvapi_component.ib_pkey_ix;
    QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_PKEY_IX);
    qp_attr.port = port_id; 
    QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_PORT);
    qp_attr.remote_atomic_flags = VAPI_EN_REM_WRITE | VAPI_EN_REM_READ;
    QP_ATTR_MASK_SET(qp_attr_mask, QP_ATTR_REMOTE_ATOMIC_FLAGS);

    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 init..Qp %d", qp_hndl));

    /**********************  INIT --> RTR  ************************/
    QP_ATTR_MASK_CLR_ALL(qp_attr_mask);