btl_portals.c

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

            PtlMDUnlink(frag->md_h);
            frag->md_h = PTL_INVALID_HANDLE;
        }
        OMPI_BTL_PORTALS_FRAG_RETURN_MAX(&mca_btl_portals_module.super, frag); 

    } else if (BTL_PORTALS_FRAG_TYPE_USER == frag->type) {
        if (frag->md_h != PTL_INVALID_HANDLE) {
            PtlMDUnlink(frag->md_h);
            frag->md_h = PTL_INVALID_HANDLE;
        }
        OPAL_THREAD_ADD32(&mca_btl_portals_module.portals_outstanding_ops, -1);
        OMPI_BTL_PORTALS_FRAG_RETURN_USER(&mca_btl_portals_module.super, frag); 
    } else {
        return OMPI_ERR_BAD_PARAM;
    }

    return OMPI_SUCCESS; 
}


mca_btl_base_descriptor_t* 
mca_btl_portals_prepare_src(struct mca_btl_base_module_t* btl_base,
                            struct mca_btl_base_endpoint_t* peer,
                            mca_mpool_base_registration_t* registration, 
                            struct ompi_convertor_t* convertor,
                            size_t reserve,
                            size_t* size)
{
    mca_btl_portals_frag_t* frag;
    size_t max_data = *size;
    struct iovec iov;
    uint32_t iov_count = 1;
    int ret;

    assert(&mca_btl_portals_module == (mca_btl_portals_module_t*) btl_base);

    if (0 != reserve || 0 != ompi_convertor_need_buffers(convertor)) {
        frag = (mca_btl_portals_frag_t*) 
            mca_btl_portals_alloc(btl_base, max_data + reserve);
        if (NULL == frag)  {
            return NULL;
        }

        if (max_data + reserve > frag->size) {
            max_data = frag->size - reserve;
        }
        
        iov.iov_len = max_data;
        iov.iov_base = (unsigned char*) frag->segments[0].seg_addr.pval + reserve;
        ret = ompi_convertor_pack(convertor, &iov, &iov_count, 
                                  &max_data );
        *size  = max_data;
        if ( ret < 0 ) {
            return NULL;
        }

        frag->segments[0].seg_len = max_data + reserve;
        frag->base.des_src_cnt = 1;

    } else {
        /* no need to pack - rdma operation out of user's buffer */
        ptl_md_t md;
        ptl_handle_me_t me_h;

        /* reserve space in the event queue for rdma operations immediately */
        while (OPAL_THREAD_ADD32(&mca_btl_portals_module.portals_outstanding_ops, 1) >
               mca_btl_portals_module.portals_max_outstanding_ops) {
            OPAL_THREAD_ADD32(&mca_btl_portals_module.portals_outstanding_ops, -1);
            mca_btl_portals_component_progress();
        }

        OMPI_BTL_PORTALS_FRAG_ALLOC_USER(&mca_btl_portals_module.super, frag, ret);
        if(NULL == frag){
            OPAL_THREAD_ADD32(&mca_btl_portals_module.portals_outstanding_ops, -1);
            return NULL;
        }
        iov.iov_len = max_data;
        iov.iov_base = NULL;

        ompi_convertor_pack(convertor, &iov, &iov_count, &max_data );

        frag->segments[0].seg_len = max_data;
        frag->segments[0].seg_addr.pval = iov.iov_base;
        frag->segments[0].seg_key.key64 = 
            OPAL_THREAD_ADD64(&(mca_btl_portals_module.portals_rdma_key), 1);
        frag->base.des_src_cnt = 1;

        /* either a put or get.  figure out which later */
        OPAL_OUTPUT_VERBOSE((90, mca_btl_portals_component.portals_output,
                             "rdma src posted for frag 0x%x, callback 0x%x, bits %lld",
                             frag, frag->base.des_cbfunc, frag->segments[0].seg_key.key64));

        /* create a match entry */
        ret = PtlMEAttach(mca_btl_portals_module.portals_ni_h,
                          OMPI_BTL_PORTALS_RDMA_TABLE_ID,
                          *((mca_btl_base_endpoint_t*) peer),
                          frag->segments[0].seg_key.key64, /* match */
                          0, /* ignore */
                          PTL_UNLINK,
                          PTL_INS_AFTER,
                          &me_h);
        if (PTL_OK != ret) {
            opal_output(mca_btl_portals_component.portals_output,
                        "Error creating rdma src ME: %d", ret);
            OMPI_BTL_PORTALS_FRAG_RETURN_USER(&mca_btl_portals_module.super, frag);
            OPAL_THREAD_ADD32(&mca_btl_portals_module.portals_outstanding_ops, -1);
            return NULL;
        }

        /* setup the memory descriptor */
        md.start = frag->segments[0].seg_addr.pval;
        md.length = frag->segments[0].seg_len;
        md.threshold = PTL_MD_THRESH_INF;
        md.max_size = 0;
        md.options = PTL_MD_OP_PUT | PTL_MD_OP_GET | PTL_MD_EVENT_START_DISABLE;
        md.user_ptr = frag; /* keep a pointer to ourselves */
        md.eq_handle = mca_btl_portals_module.portals_eq_handles[OMPI_BTL_PORTALS_EQ_SEND];

        ret = PtlMDAttach(me_h, 
                          md,
                          PTL_UNLINK,
                          &(frag->md_h));
        if (PTL_OK != ret) {
            opal_output(mca_btl_portals_component.portals_output,
                        "Error creating rdma src MD: %d", ret);
            PtlMEUnlink(me_h);
            OMPI_BTL_PORTALS_FRAG_RETURN_USER(&mca_btl_portals_module.super, frag);
            OPAL_THREAD_ADD32(&mca_btl_portals_module.portals_outstanding_ops, -1);
            return NULL;
        }
    }

    frag->base.des_src = frag->segments;
    frag->base.des_dst = NULL;
    frag->base.des_dst_cnt = 0;
    frag->base.des_flags = 0;

    return &frag->base;
}


mca_btl_base_descriptor_t* 
mca_btl_portals_prepare_dst(struct mca_btl_base_module_t* btl_base, 
                            struct mca_btl_base_endpoint_t* peer,
                            mca_mpool_base_registration_t* registration, 
                            struct ompi_convertor_t* convertor,
                            size_t reserve,
                            size_t* size)
{
    mca_btl_portals_frag_t* frag;
    ptl_md_t md;
    ptl_handle_me_t me_h;
    int ret;
    ptrdiff_t lb;

    assert(&mca_btl_portals_module == (mca_btl_portals_module_t*) btl_base);

    /* reserve space in the event queue for rdma operations immediately */
    while (OPAL_THREAD_ADD32(&mca_btl_portals_module.portals_outstanding_ops, 1) >
           mca_btl_portals_module.portals_max_outstanding_ops) {
        OPAL_THREAD_ADD32(&mca_btl_portals_module.portals_outstanding_ops, -1);
        mca_btl_portals_component_progress();
    }

    OMPI_BTL_PORTALS_FRAG_ALLOC_USER(&mca_btl_portals_module.super, frag, ret);
    if(NULL == frag) {
        OPAL_THREAD_ADD32(&mca_btl_portals_module.portals_outstanding_ops, -1);
        return NULL;
    }

    ompi_ddt_type_lb(convertor->pDesc, &lb);
    frag->segments[0].seg_len = *size;
    frag->segments[0].seg_addr.pval = convertor->pBaseBuf + lb + convertor->bConverted;
    frag->segments[0].seg_key.key64 = 
        OPAL_THREAD_ADD64(&(mca_btl_portals_module.portals_rdma_key), 1);
    frag->base.des_src = NULL;
    frag->base.des_src_cnt = 0;
    frag->base.des_dst = frag->segments;
    frag->base.des_dst_cnt = 1;
    frag->base.des_flags = 0;

    OPAL_OUTPUT_VERBOSE((90, mca_btl_portals_component.portals_output,
                         "rdma dest posted for frag 0x%x, callback 0x%x, bits %lld",
                         frag, frag->base.des_cbfunc, frag->segments[0].seg_key.key64));

    /* create a match entry */
    ret = PtlMEAttach(mca_btl_portals_module.portals_ni_h,
                      OMPI_BTL_PORTALS_RDMA_TABLE_ID,
                      *((mca_btl_base_endpoint_t*) peer),
                      frag->segments[0].seg_key.key64, /* match */
                      0, /* ignore */
                      PTL_UNLINK,
                      PTL_INS_AFTER,
                      &me_h);
    if (PTL_OK != ret) {
        opal_output(mca_btl_portals_component.portals_output,
                    "Error creating rdma dest ME: %d", ret);
        OPAL_THREAD_ADD32(&mca_btl_portals_module.portals_outstanding_ops, -1);
        OMPI_BTL_PORTALS_FRAG_RETURN_USER(&mca_btl_portals_module.super, frag);
        return NULL;
    }

    /* setup the memory descriptor. */
    md.start = frag->segments[0].seg_addr.pval;
    md.length = frag->segments[0].seg_len;
    md.threshold = PTL_MD_THRESH_INF;
    md.max_size = 0;
    md.options = PTL_MD_OP_PUT | PTL_MD_OP_GET | PTL_MD_EVENT_START_DISABLE;
    md.user_ptr = frag; /* keep a pointer to ourselves */
    md.eq_handle = mca_btl_portals_module.portals_eq_handles[OMPI_BTL_PORTALS_EQ_SEND];

    ret = PtlMDAttach(me_h, 
                      md,
                      PTL_UNLINK,
                      &(frag->md_h));
    if (PTL_OK != ret) {
        opal_output(mca_btl_portals_component.portals_output,
                    "Error creating rdma dest MD: %d", ret);
        PtlMEUnlink(me_h);
        OPAL_THREAD_ADD32(&mca_btl_portals_module.portals_outstanding_ops, -1);
        OMPI_BTL_PORTALS_FRAG_RETURN_USER(&mca_btl_portals_module.super, frag);
        return NULL;
    }

    return &frag->base;
}


int
mca_btl_portals_finalize(struct mca_btl_base_module_t *btl_base)
{
    int ret;

    assert(&mca_btl_portals_module == (mca_btl_portals_module_t*) btl_base);

    /* finalize all communication */
    while (mca_btl_portals_module.portals_outstanding_ops > 0) {
        mca_btl_portals_component_progress();
    }

    if (mca_btl_portals_module.portals_num_procs != 0) {
        int i;

        ret = mca_btl_portals_recv_disable(&mca_btl_portals_module);

        /* destroy eqs */
        for (i = 0 ; i < OMPI_BTL_PORTALS_EQ_SIZE ; ++i) {
            int ptl_ret = PtlEQFree(mca_btl_portals_module.portals_eq_handles[i]);
            if (PTL_OK != ptl_ret) {
#if OMPI_PORTALS_CRAYXT3
                if (i != OMPI_BTL_PORTALS_EQ_SEND && PTL_EQ_IN_USE != ptl_ret) {
                    /* The PML isn't great about cleaning up after itself.
                       Ignore related errors. */
#endif
                    opal_output(mca_btl_portals_component.portals_output,
                            "Error freeing EQ %d: %d", i, ptl_ret);
#if OMPI_PORTALS_CRAYXT3
                }
#endif
            }
        }

    } 

    OBJ_DESTRUCT(&mca_btl_portals_module.portals_recv_blocks);
    OBJ_DESTRUCT(&mca_btl_portals_module.portals_recv_frag);
    OBJ_DESTRUCT(&mca_btl_portals_module.portals_frag_eager);
    OBJ_DESTRUCT(&mca_btl_portals_module.portals_frag_max);
    OBJ_DESTRUCT(&mca_btl_portals_module.portals_frag_user);

    ompi_common_portals_ni_finalize();
    ompi_common_portals_finalize();

    opal_output_verbose(20, mca_btl_portals_component.portals_output,
                        "successfully finalized module");

    return OMPI_SUCCESS;
}