btl_mx_component.c

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

                                                  bool enable_progress_threads,
                                                  bool enable_mpi_threads)
{
    mca_btl_base_module_t** btls;
    mx_return_t status;
    uint32_t size, count;
    int32_t i;
    uint64_t *nic_addrs;
    mca_btl_mx_addr_t *mx_addrs;

    *num_btl_modules = 0;

    if (enable_progress_threads) { 
        opal_output( 0, "mca_btl_mx_component_init: progress threads requested but not supported");
        mca_pml_base_modex_send(&mca_btl_mx_component.super.btl_version, 
                                NULL, 0);
        return NULL;
    }

    /* set the MX error handle to always return. This function is the only MX function
     * allowed to be called before mx_init in order to make sure that if the MX is not
     * up and running the MX library does not exit the application.
     */
    mx_set_error_handler(MX_ERRORS_RETURN);
    /* Until this BTL reach a stable state let MX library generate assert for the errors */
    /*mx_set_error_handler(MX_ERRORS_ARE_FATAL);*/

    /* First check if MX is available ... */
    if(OMPI_SUCCESS!=ompi_common_mx_initialize()) { 
        mca_pml_base_modex_send(&mca_btl_mx_component.super.btl_version, 
                                NULL, 0);
        return NULL;
    }
        
    /* initialize objects */
    OBJ_CONSTRUCT(&mca_btl_mx_component.mx_send_eager_frags, ompi_free_list_t);
    OBJ_CONSTRUCT(&mca_btl_mx_component.mx_send_user_frags, ompi_free_list_t);
    OBJ_CONSTRUCT(&mca_btl_mx_component.mx_recv_frags, ompi_free_list_t);
    OBJ_CONSTRUCT(&mca_btl_mx_component.mx_procs, opal_list_t);
    OBJ_CONSTRUCT(&mca_btl_mx_component.mx_pending_acks, opal_list_t);
    OBJ_CONSTRUCT(&mca_btl_mx_component.mx_lock, opal_mutex_t);

    ompi_free_list_init( &mca_btl_mx_component.mx_send_eager_frags,
                         sizeof(mca_btl_mx_frag_t) + mca_btl_mx_module.super.btl_eager_limit,
                         OBJ_CLASS(mca_btl_mx_frag_t),
                         mca_btl_mx_component.mx_free_list_num,
                         mca_btl_mx_component.mx_free_list_max,
                         mca_btl_mx_component.mx_free_list_inc,
                         NULL ); /* use default allocator */

    ompi_free_list_init( &mca_btl_mx_component.mx_send_user_frags,
                         sizeof(mca_btl_mx_frag_t),
                         OBJ_CLASS(mca_btl_mx_frag_t),
                         mca_btl_mx_component.mx_free_list_num,
                         mca_btl_mx_component.mx_free_list_max,
                         mca_btl_mx_component.mx_free_list_inc,
                         NULL ); /* use default allocator */

    ompi_free_list_init( &mca_btl_mx_component.mx_recv_frags,
                         sizeof(mca_btl_mx_frag_t),
                         OBJ_CLASS(mca_btl_mx_frag_t),
                         mca_btl_mx_component.mx_free_list_num,
                         mca_btl_mx_component.mx_free_list_max,
                         mca_btl_mx_component.mx_free_list_inc,
                         NULL ); /* use default allocator */

    /* intialize process hash table */
    OBJ_CONSTRUCT( &mca_btl_mx_component.mx_procs, opal_list_t );

    /* get the number of card available on the system */
    if( (status = mx_get_info( NULL, MX_NIC_COUNT, NULL, 0,
                               &mca_btl_mx_component.mx_num_btls, sizeof(uint32_t))) != MX_SUCCESS ) {
        opal_output(0, "mca_btl_mx_component_init: mx_get_info(MX_NIC_COUNT) failed with status=%d\n", status);
        mca_pml_base_modex_send(&mca_btl_mx_component.super.btl_version, 
                                NULL, 0);
        return NULL;
    }

    if (0 == mca_btl_mx_component.mx_num_btls) {
        mca_btl_base_error_no_nics("Myrinet/MX", "NIC");
        mca_pml_base_modex_send(&mca_btl_mx_component.super.btl_version, 
                                NULL, 0);
        return NULL;
    }

    /* check for limit on number of btls */
    if(mca_btl_mx_component.mx_num_btls > mca_btl_mx_component.mx_max_btls)
        mca_btl_mx_component.mx_num_btls = mca_btl_mx_component.mx_max_btls;

    /* Now we know how many NIC are available on the system. We will create a BTL for each one
     * and then give a pointer to the BTL to the upper level.
     */
    mca_btl_mx_component.mx_btls = malloc( mca_btl_mx_component.mx_num_btls * sizeof(mca_btl_base_module_t*) );
    if( NULL == mca_btl_mx_component.mx_btls )
        return NULL;

    /* determine the NIC ids */
    size = sizeof(uint64_t) * (mca_btl_mx_component.mx_num_btls + 1);
    if( NULL == (nic_addrs = (uint64_t*)malloc(size)) )
        return NULL;
    if( (status = mx_get_info( NULL, MX_NIC_IDS, NULL, 0,
                               nic_addrs, size)) != MX_SUCCESS) {
        free(nic_addrs);
        return NULL;
    }

    size = sizeof(mca_btl_mx_addr_t) * mca_btl_mx_component.mx_num_btls;
    mx_addrs = (mca_btl_mx_addr_t*)malloc( size );
    if( NULL == mx_addrs ) {
        free( nic_addrs );
        return NULL;
    }

    /* create a btl for each NIC */
    for( i = count = 0; i < mca_btl_mx_component.mx_num_btls; i++ ) {
        mca_btl_mx_module_t* btl = mca_btl_mx_create(nic_addrs[i]);
        if( NULL == btl ) {
            continue;
        }
        status = mx_decompose_endpoint_addr( btl->mx_endpoint_addr, &(mx_addrs[i].nic_id),
                                             &(mx_addrs[i].endpoint_id) );
        if( MX_SUCCESS != status ) {
            OBJ_RELEASE( btl );
            continue;
        }

#if OMPI_ENABLE_HETEROGENEOUS_SUPPORT
        BTL_MX_ADDR_HTON(mx_addrs[i]);
#endif
        mca_btl_mx_component.mx_btls[count++] = btl;
    }
    size = sizeof(mca_btl_mx_addr_t) * count;
    if( 0 == count ) {
        /* No active BTL module */
    }
    mca_btl_mx_component.mx_num_btls = count;

    /* publish the MX addresses via the MCA framework */
    mca_pml_base_modex_send( &mca_btl_mx_component.super.btl_version, mx_addrs, size );

    free( nic_addrs );
    free( mx_addrs );

    btls = malloc( mca_btl_mx_component.mx_num_btls * sizeof(mca_btl_base_module_t*) );

    if( NULL == btls ) {
        free( mca_btl_mx_component.mx_btls );
        mca_btl_mx_component.mx_num_btls = 0;  /* no active BTL modules */
        return NULL;
    }
    memcpy( btls,  mca_btl_mx_component.mx_btls,
            mca_btl_mx_component.mx_num_btls*sizeof(mca_btl_mx_module_t*) );
    *num_btl_modules = mca_btl_mx_component.mx_num_btls;
    return btls;
}

/*
 *  MX component progress.
 */
int mca_btl_mx_component_progress()
{
    int32_t num_progressed = 0, i;
    mx_status_t mx_status;
    mx_return_t mx_return;
    mx_segment_t mx_segment;
    mx_request_t mx_request;
    mca_btl_mx_frag_t* frag;

    for( i = 0; i < mca_btl_mx_component.mx_num_btls; i++ ) {
        mca_btl_mx_module_t* mx_btl = mca_btl_mx_component.mx_btls[i];
        uint32_t mx_result = 0;
        
        /* pre-post receive */
#if 0
        if( mx_btl->mx_recvs_posted == 0 ) {
            OPAL_THREAD_ADD32( &mx_btl->mx_recvs_posted, 1 );
            MCA_BTL_MX_POST( mx_btl, frag );
        }
#endif

        /*if( mx_btl->mx_posted_request ) { */
        mx_return = mx_ipeek( mx_btl->mx_endpoint, &mx_request, &mx_result );
        if( mx_return != MX_SUCCESS ) {
            opal_output(0, "mca_btl_mx_component_progress: mx_ipeek() failed with status %d\n",
                        mx_return);
            continue;
        }
        if( mx_result == 0 ) {
            continue;
        }
        
        mx_return = mx_test( mx_btl->mx_endpoint, &mx_request, &mx_status, &mx_result);
        if( mx_return != MX_SUCCESS ) {
            opal_output(0, "mca_btl_mx_progress: mx_test() failed with status=%dn",
                        mx_return);
            continue;
        }

        frag = mx_status.context;
        if( 0 == frag->base.des_dst_cnt ) {  /* it's a send */
            /* call the completion callback */
            frag->base.des_cbfunc( &(mx_btl->super), frag->endpoint, &(frag->base), OMPI_SUCCESS);
        } else { /* and this one is a receive */
            mca_btl_base_recv_reg_t* reg;

            reg = &(mx_btl->mx_reg[frag->tag]);
            frag->base.des_dst->seg_len = mx_status.msg_length;
            reg->cbfunc( &(mx_btl->super), frag->tag, &(frag->base), reg->cbdata );
            /*
             * The upper level extract the data from the fragment. Now we can register the fragment
             * again with the MX BTL.
             */
            mx_segment.segment_ptr = frag->base.des_dst->seg_addr.pval;
            mx_segment.segment_length = mca_btl_mx_module.super.btl_eager_limit;
            mx_return = mx_irecv( mx_btl->mx_endpoint, &mx_segment, 1, (uint64_t)frag->tag, 
                                  (uint64_t)0xffffffffffffffffULL,
                                  frag, &(frag->mx_request) );
            if( MX_SUCCESS != mx_return ) {
                opal_output( 0, "Fail to re-register a fragment with the MX NIC ...\n" );
            }
        }

        /*MCA_BTL_MX_PROGRESS(mx_btl, mx_status);*/
        /*
         * on the mx_status we have now the pointer attached to the request. This pointer indicate
         * which fragment we are working on. On the status we have the status of the operation, so
         * we know what we are supposed to do next.
         */
        num_progressed++;
    }
    return num_progressed;
}