caffe-distri/src/main/cpp/util/rdma.cpp
/**
 * Polling for events on a inner thread allows processing of management messages
 * like buffer connection immediately, even if the user is not polling.
 * Otherwise buffer constructors would block indefinitely.
 *
 * Deep learning workloads are about sending small numbers of large messages,
 * in which case this model works great. If the library was to be used to
 * exchange large numbers of short messages, it would be useful to split
 * management and data messages over two different queue pairs. User threads
 * could then wait or poll on the data queue pair directly.
 */
Polling for events on a inner thread allows processing of management messages
like buffer connection immediately, even if the user is not polling.
Otherwise buffer constructors would block indefinitely.

Deep learning workloads are about sending small numbers of large messages,
in which case this model works great. If the library was to be used to
exchange large numbers of short messages, it would be useful to split
management and data messages over two different queue pairs. User threads
could then wait or poll on the data queue pair directly.
Polling for events on a inner thread allows processing of management messages like buffer connection immediately even if the user is not polling . Otherwise buffer constructors would block indefinitely . Deep learning workloads are about sending small numbers of large messages in which case this model works great . If the library was to be used to exchange large numbers of short messages it would be useful to split management and data messages over two different queue pairs . User threads could then wait or poll on the data queue pair directly .
void RDMAAdapter::InternalThreadEntry() {
  while (!must_stop()) {
    ibv_cq* cq;
    void* cq_context;
    CHECK(!ibv_get_cq_event(channel_, &cq, &cq_context));
    CHECK(cq == cq_);
    ibv_ack_cq_events(cq, 1);
    CHECK(!ibv_req_notify_cq(cq_, 0));
    int ne = ibv_poll_cq(cq_, MAX_CONCURRENT_WRITES * 2,
      static_cast<ibv_wc*>(wc_));
    CHECK_GE(ne, 0);
    for (int i = 0; i < ne; ++i) {
      CHECK(wc_[i].status == IBV_WC_SUCCESS) << "Failed status \n"
                                             << ibv_wc_status_str(wc_[i].status)
                                             << " " << wc_[i].status << " "
                                             << static_cast<int>(wc_[i].wr_id)
                                             << " "<< wc_[i].vendor_err;
      if (wc_[i].opcode == IBV_WC_RECV_RDMA_WITH_IMM) {
        RDMAChannel* channel = reinterpret_cast<RDMAChannel*>(wc_[i].wr_id);
        channel->recv();
        int id = wc_[i].imm_data;
        if (id >= CTRL_ID_OFFSET) {
          ctrl_received_.push(channel->buffers_[id - CTRL_ID_OFFSET]);
        } else {
          received_.push(channel->buffers_[id]);
        }
      } else {
        if (wc_[i].opcode & IBV_WC_RECV) {
          RDMAChannel* channel = reinterpret_cast<RDMAChannel*>(wc_[i].wr_id);
          int id = wc_[i].imm_data;
          channel->memory_regions_queue_.push(channel->memory_regions_[id]);
          CHECK(id == channel->memory_regions_received_++);
          CHECK(!ibv_dereg_mr(channel->region_regions_[id]));
        }
      }
    }
  }
}
void RDMAAdapter :: InternalThreadEntry ( ) { while ( ! must_stop ( ) ) { ibv_cq * cq ; void * cq_context ; CHECK ( ! ibv_get_cq_event ( channel_ , & cq , & cq_context ) ) ; CHECK ( cq == cq_ ) ; ibv_ack_cq_events ( cq , 1 ) ; CHECK ( ! ibv_req_notify_cq ( cq_ , 0 ) ) ; int ne = ibv_poll_cq ( cq_ , MAX_CONCURRENT_WRITES * 2 , static_cast < ibv_wc * > ( wc_ ) ) ; CHECK_GE ( ne , 0 ) ; for ( int i = 0 ; i < ne ; ++ i ) { CHECK ( wc_ [ i ] . status == IBV_WC_SUCCESS ) << " \n " << ibv_wc_status_str ( wc_ [ i ] . status ) << " " << wc_ [ i ] . status << " " << static_cast < int > ( wc_ [ i ] . wr_id ) << " " << wc_ [ i ] . vendor_err ; if ( wc_ [ i ] . opcode == IBV_WC_RECV_RDMA_WITH_IMM ) { RDMAChannel * channel = reinterpret_cast < RDMAChannel * > ( wc_ [ i ] . wr_id ) ; channel -> recv ( ) ; int id = wc_ [ i ] . imm_data ; if ( id >= CTRL_ID_OFFSET ) { ctrl_received_ . push ( channel -> buffers_ [ id - CTRL_ID_OFFSET ] ) ; } else { received_ . push ( channel -> buffers_ [ id ] ) ; } } else { if ( wc_ [ i ] . opcode & IBV_WC_RECV ) { RDMAChannel * channel = reinterpret_cast < RDMAChannel * > ( wc_ [ i ] . wr_id ) ; int id = wc_ [ i ] . imm_data ; channel -> memory_regions_queue_ . push ( channel -> memory_regions_ [ id ] ) ; CHECK ( id == channel -> memory_regions_received_ ++ ) ; CHECK ( ! ibv_dereg_mr ( channel -> region_regions_ [ id ] ) ) ; } } } } }
Polling for events on a inner thread allows processing of management messages
 like buffer connection immediately, even if the user is not polling.
Polling for events on a inner thread allows processing of management messages like buffer connection immediately even if the user is not polling .
// Data message, add it to user received queue // ctrl signal // data // Buffer connection message