Function to process incoming messages There are two types of messages: 1. IBV_WC_RECV_RDMA_WITH_IMM (receive) 2. IBV_WC_RDMA_WRITE (send))
| 435 | // 1. IBV_WC_RECV_RDMA_WITH_IMM (receive) |
| 436 | // 2. IBV_WC_RDMA_WRITE (send)) |
| 437 | void RdmaAdapter::Process_CQ() { |
| 438 | while (true) { |
| 439 | ibv_cq* cq; |
| 440 | void* cq_context; |
| 441 | CHECK(!ibv_get_cq_event(event_channel_, &cq, &cq_context)); |
| 442 | CHECK(cq == cq_); |
| 443 | ibv_ack_cq_events(cq, 1); |
| 444 | CHECK(!ibv_req_notify_cq(cq_, 0)); |
| 445 | |
| 446 | int ne = |
| 447 | ibv_poll_cq(cq_, MAX_CONCURRENT_WRITES * 2, static_cast<ibv_wc*>(wc_)); |
| 448 | CHECK_GE(ne, 0); |
| 449 | for (int i = 0; i < ne; ++i) { |
| 450 | CHECK(wc_[i].status == IBV_WC_SUCCESS) |
| 451 | << "Failed status \n" |
| 452 | << ibv_wc_status_str(wc_[i].status) << " " << wc_[i].status << " " |
| 453 | << static_cast<int>(wc_[i].wr_id) << " " << wc_[i].vendor_err; |
| 454 | if (wc_[i].opcode == IBV_WC_RECV_RDMA_WITH_IMM) { |
| 455 | RdmaChannel* rc = reinterpret_cast<RdmaChannel*>(wc_[i].wr_id); |
| 456 | // put back a recv wr. |
| 457 | rc->Recv(); |
| 458 | // imm_data is the index of RX buffer in the buffer table. |
| 459 | uint32_t imm_data = wc_[i].imm_data; |
| 460 | RdmaMessageBuffer* rb; |
| 461 | RdmaMessage rm; |
| 462 | |
| 463 | if (imm_data == RDMA_IMM_DATA_ACK) { |
| 464 | // receive an ack to a message |
| 465 | rb = rc->tx_message_buffer_; |
| 466 | rb->SetBufferStatus(remote, idle); |
| 467 | rb->SendNextItem(); |
| 468 | continue; |
| 469 | } |
| 470 | |
| 471 | if (imm_data <= RDMA_IMM_MAX_REQUEST_ID) { |
| 472 | // receive a tensor RDMA write |
| 473 | uint32_t request_index = imm_data; |
| 474 | RdmaTensorRequest* request = rc->GetTensorRequest(request_index); |
| 475 | request->RecvTensorContent(); |
| 476 | continue; |
| 477 | } |
| 478 | |
| 479 | // receive a control message |
| 480 | rb = rc->rx_message_buffer_; |
| 481 | RdmaMessage::ParseMessage(rm, rb->buffer_); |
| 482 | RdmaMessageBuffer::SendAck(rc); |
| 483 | RDMA_LOG(1) << "Step 0x" << std::hex << rm.step_id_ << std::dec |
| 484 | << ": Received " << MessageTypeToString(rm.type_) << " " |
| 485 | << "#" << rm.request_index_ << ": " << rm.name_; |
| 486 | |
| 487 | if (rm.type_ == RDMA_MESSAGE_TENSOR_REQUEST) { |
| 488 | RdmaTensorResponse* response = rc->AddTensorResponse(rm); |
| 489 | response->Start(); |
| 490 | } else if (rm.type_ == RDMA_MESSAGE_META_DATA_UPDATE) { |
| 491 | RdmaTensorRequest* request = rc->GetTensorRequest(rm.request_index_); |
| 492 | request->RecvTensorMetaData(rm.data_type_, rm.tensor_shape_, |
| 493 | rm.is_dead_, rm.tensor_bytes_); |
| 494 | #ifdef RDMA_DATA_VALIDATION |
nothing calls this directly
no test coverage detected