btl_portals_component.c

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

                        NULL);

    /* send frags */
    ompi_free_list_init(&(mca_btl_portals_module.portals_frag_max),
                        sizeof(mca_btl_portals_frag_max_t) + 
                        mca_btl_portals_module.super.btl_max_send_size,
                        OBJ_CLASS(mca_btl_portals_frag_max_t),
                        mca_btl_portals_component.portals_free_list_init_num,
                        mca_btl_portals_component.portals_free_list_max_num,
                        mca_btl_portals_component.portals_free_list_inc_num,
                        NULL);

    /* user frags */
    ompi_free_list_init(&(mca_btl_portals_module.portals_frag_user),
                        sizeof(mca_btl_portals_frag_user_t),
                        OBJ_CLASS(mca_btl_portals_frag_user_t),
                        mca_btl_portals_component.portals_free_list_init_num,
                        mca_btl_portals_component.portals_free_list_max_num,
                        mca_btl_portals_component.portals_free_list_inc_num,
                        NULL);

    /* recv frags */
    OBJ_CONSTRUCT(&(mca_btl_portals_module.portals_recv_frag),
                  mca_btl_portals_frag_recv_t);

    /* receive block list */
    OBJ_CONSTRUCT(&(mca_btl_portals_module.portals_recv_blocks), opal_list_t);

    /* list for send requests that have to be delayed */
    OBJ_CONSTRUCT(&(mca_btl_portals_module.portals_queued_sends),
                  opal_list_t);

    *num_btls = 1;

    opal_output_verbose(20, mca_btl_portals_component.portals_output,
                        "initialized Portals module");

    return btls;
}


int
mca_btl_portals_component_progress(void)
{
    int num_progressed = 0;
    int ret, which;
    static ptl_event_t ev;
    mca_btl_portals_frag_t *frag = NULL;
    mca_btl_portals_recv_block_t *block = NULL;
    mca_btl_base_tag_t tag;

    if (0 == mca_btl_portals_module.portals_num_procs) {
        return 0;
    }

    while (true) {
        ret = PtlEQPoll(mca_btl_portals_module.portals_eq_handles,
                        OMPI_BTL_PORTALS_EQ_SIZE,
                        0,       /* timeout */
                        &ev,     /* event structure to update */
                        &which); /* which queue the event came from - we don't care */
        switch (ret) {
        case PTL_OK:
            frag = ev.md.user_ptr;
            num_progressed++;

            switch (ev.type) {
            case PTL_EVENT_GET_START:
                /* generated on source (target) when a get from memory starts */
                OPAL_OUTPUT_VERBOSE((900, mca_btl_portals_component.portals_output,
                                     "PTL_EVENT_GET_START for 0x%x, %d",
                                     frag, (int) ev.hdr_data));

                break;

            case PTL_EVENT_GET_END:
                /* generated on source (target) when a get from memory ends */
                OPAL_OUTPUT_VERBOSE((900, mca_btl_portals_component.portals_output,
                                     "PTL_EVENT_GET_END for 0x%x, %d",
                                     frag, (int) ev.hdr_data));

                break;

            case PTL_EVENT_PUT_START:
                /* generated on destination (target) when a put into memory starts */
                OPAL_OUTPUT_VERBOSE((900, mca_btl_portals_component.portals_output,
                                     "PTL_EVENT_PUT_START for 0x%x, %d",
                                     frag, (int) ev.hdr_data));

#if OMPI_ENABLE_DEBUG
                if (ev.ni_fail_type != PTL_NI_OK) {
                    opal_output(mca_btl_portals_component.portals_output,
                                "Failure to start event\n");
                    return OMPI_ERROR;
                }
#endif
                /* if it's a pending unexpected receive, do book keeping. */
                if (ev.hdr_data < MCA_BTL_TAG_MAX) {
                    block = ev.md.user_ptr;
                    OPAL_THREAD_ADD32(&(block->pending), 1);
                }

                break;

            case PTL_EVENT_PUT_END: 
                /* generated on destination (target) when a put into memory ends */
                OPAL_OUTPUT_VERBOSE((900, mca_btl_portals_component.portals_output,
                                     "PTL_EVENT_PUT_END for 0x%x, %d",
                                     frag, (int) ev.hdr_data));

#if OMPI_ENABLE_DEBUG
                if (ev.ni_fail_type != PTL_NI_OK) {
                    opal_output(mca_btl_portals_component.portals_output,
                                "Failure to end event\n");
                    mca_btl_portals_return_block_part(&mca_btl_portals_module,
                                                      block);
                    return OMPI_ERROR;
                }
#endif
                /* if it's an unexpected receive, do book keeping and send to PML */
                if (ev.hdr_data < MCA_BTL_TAG_MAX) {
                    block = ev.md.user_ptr;
                    tag = ev.hdr_data;

                    /* if we ever make this thread hot, need to do
                       something with the receive fragments */
                    frag = &mca_btl_portals_module.portals_recv_frag;
                    frag->segments[0].seg_addr.pval = (((char*) ev.md.start) + ev.offset);
                    frag->segments[0].seg_len = ev.mlength;

                    OPAL_OUTPUT_VERBOSE((90, mca_btl_portals_component.portals_output,
                                         "received send fragment %x (thresh: %d, length %d)", 
                                         frag, ev.md.threshold, (int) ev.mlength));

                    if (ev.md.length - (ev.offset + ev.mlength) < ev.md.max_size ||
                        ev.md.threshold == 1) {
                        /* the block is full.  It's deactivated automagically, but we
                           can't start it up again until everyone is done with it.
                           The actual reactivation and all that will happen after the
                           free completes the last operation... */
                        OPAL_OUTPUT_VERBOSE((90, mca_btl_portals_component.portals_output,
                                             "marking block 0x%x as full", block->start));
                        block->full = true;
                    }

                    assert(NULL != mca_btl_portals_module.portals_reg[tag].cbfunc);

                    mca_btl_portals_module.portals_reg[tag].cbfunc(
                                             &mca_btl_portals_module.super,
                                             tag,
                                             &frag->base,
                                             mca_btl_portals_module.portals_reg[tag].cbdata);
                    mca_btl_portals_return_block_part(&mca_btl_portals_module, block);
                }
                break;

            case PTL_EVENT_REPLY_START:
                /* generated on destination (origin) when a get starts
                   returning data */

                OPAL_OUTPUT_VERBOSE((900, mca_btl_portals_component.portals_output,
                                     "PTL_EVENT_REPLY_START for 0x%x, %d",
                                     frag, (int) ev.hdr_data));

                break;

            case PTL_EVENT_REPLY_END:
                /* generated on destination (origin) when a get is
                   done returning data */

                OPAL_OUTPUT_VERBOSE((90, mca_btl_portals_component.portals_output,
                                     "PTL_EVENT_REPLY_END for 0x%x", frag));

                /* let the PML know we're done */
                frag->base.des_cbfunc(&mca_btl_portals_module.super,
                                      frag->endpoint,
                                      &frag->base,
                                      OMPI_SUCCESS);

                break;

            case PTL_EVENT_SEND_START:
                /* generated on source (origin) when put starts sending */

#if OMPI_ENABLE_DEBUG
                OPAL_OUTPUT_VERBOSE((900, mca_btl_portals_component.portals_output,
                                     "PTL_EVENT_SEND_START for 0x%x, %d",
                                     frag, (int) ev.hdr_data));

                if (ev.ni_fail_type != PTL_NI_OK) {
                    opal_output(mca_btl_portals_component.portals_output,
                                "Failure to start send event\n");
                    frag->base.des_cbfunc(&mca_btl_portals_module.super,
                                          frag->endpoint,
                                          &frag->base,
                                          OMPI_ERROR);
                }
#endif
                break;

            case PTL_EVENT_SEND_END:
                /* generated on source (origin) when put stops sending */
#if OMPI_ENABLE_DEBUG
                OPAL_OUTPUT_VERBOSE((90, mca_btl_portals_component.portals_output,
                                     "PTL_EVENT_SEND_END for 0x%x, %d",
                                     frag, (int) ev.hdr_data));

                if (ev.ni_fail_type != PTL_NI_OK) {
                    opal_output(mca_btl_portals_component.portals_output,
                                "Failure to end send event\n");
                    frag->base.des_cbfunc(&mca_btl_portals_module.super,
                                          frag->endpoint,
                                          &frag->base,
                                          OMPI_ERROR);
                }
#endif
                break;

            case PTL_EVENT_ACK:
                /* ack that a put as completed on other side */

                /* ACK for either send or RDMA put.  Either way, we
                   just call the callback function on goodness.
                   Requeue the put on badness */

                OPAL_OUTPUT_VERBOSE((90, mca_btl_portals_component.portals_output,
                                     "PTL_EVENT_ACK for 0x%x", frag));

#if OMPI_ENABLE_DEBUG
                if (ev.ni_fail_type != PTL_NI_OK) {
                    opal_output(mca_btl_portals_component.portals_output,
                                "Failure to ack event\n");
                    frag->base.des_cbfunc(&mca_btl_portals_module.super,
                                          frag->endpoint,
                                          &frag->base,
                                          OMPI_ERROR);
                } else
#endif

                if (0 == ev.mlength) {
                    /* other side received message but truncated to 0.
                       This should only happen for unexpected
                       messages, and only when the other side has no
                       buffer space available for receiving */
                    opal_output_verbose(50,
                                        mca_btl_portals_component.portals_output,
                                        "message was dropped.  Trying again");
                    
                    mca_btl_portals_send(&mca_btl_portals_module.super,
                                         frag->endpoint,
                                         &frag->base,
                                         frag->hdr.tag);
                } else {
                    /* other side received the message.  should have
                       received entire thing */
                    /* let the PML know we're done */
                    frag->base.des_cbfunc(&mca_btl_portals_module.super,
                                          frag->endpoint,
                                          &frag->base,
                                          OMPI_SUCCESS);
                }

                opal_output_verbose(50, mca_btl_portals_component.portals_output, "fuck");

                if (0 != frag->size) {
                    OPAL_THREAD_ADD32(&mca_btl_portals_module.portals_outstanding_ops,
                                      -1);
                    MCA_BTL_PORTALS_PROGRESS_QUEUED_SENDS();
                }

                break;

            default:
                break;
            }
            break;

        case PTL_EQ_EMPTY:
            /* there's nothing in the queue.  This is actually the
               common case, so the easiest way to make the compiler
               emit something that doesn't completely blow here is to
               just to go back to a good old goto */
            goto done;
            break;

        case PTL_EQ_DROPPED:
            /* not sure how we could deal with this more gracefully */
            opal_output(mca_btl_portals_component.portals_output,
                        "WARNING: EQ events dropped.  Too many messages pending.");
            opal_output(mca_btl_portals_component.portals_output,
                        "WARNING: Giving up in dispair");
            abort();
            break;
        
        default:
            opal_output(mca_btl_portals_component.portals_output,
                        "WARNING: Error in PtlEQPoll (%d).  This shouldn't happen",
                        ret);
            abort();
            break;
        }
    }

 done:
    return num_progressed;
}