Method GetRemoteAddressSync

tensorflow/contrib/verbs/grpc_verbs_service.cc:105–154

synchronous method

// synchronous method
Status GrpcVerbsService::GetRemoteAddressSync(
    const GetRemoteAddressRequest* request,
    GetRemoteAddressResponse* response) {
  // analyzing request
  // the channel setting part is redundant.
  const string remote_host_name = request->host_name();
  RdmaChannel* rc = rdma_mgr_->FindChannel(remote_host_name);
  CHECK(rc);
  RdmaAddress ra;
  ra.lid = request->channel().lid();
  ra.qpn = request->channel().qpn();
  ra.psn = request->channel().psn();
  ra.snp = request->channel().snp();
  ra.iid = request->channel().iid();
  rc->SetRemoteAddress(ra, false);
  rc->Connect();
  int i = 0;
  int idx[] = {1, 0};
  std::vector<RdmaMessageBuffer*> mb(rc->message_buffers());
  CHECK_EQ(request->mr_size(), RdmaChannel::kNumMessageBuffers);
  for (const auto& mr : request->mr()) {
    // the connections are crossed, i.e.
    // local tx_message_buffer <---> remote rx_message_buffer_
    // local rx_message_buffer <---> remote tx_message_buffer_
    // hence idx[] = {1, 0}.
    RdmaMessageBuffer* rb = mb[idx[i]];
    RemoteMR rmr;
    rmr.remote_addr = mr.remote_addr();
    rmr.rkey = mr.rkey();
    rb->SetRemoteMR(rmr, false);
    i++;
  }
  CHECK(i == RdmaChannel::kNumMessageBuffers);

  // setting up response
  response->set_host_name(
      worker_env_->session_mgr->LegacySession()->worker_name);
  Channel* channel_info = response->mutable_channel();
  channel_info->set_lid(rc->self().lid);
  channel_info->set_qpn(rc->self().qpn);
  channel_info->set_psn(rc->self().psn);
  channel_info->set_snp(rc->self().snp);
  channel_info->set_iid(rc->self().iid);
  for (int i = 0; i < RdmaChannel::kNumMessageBuffers; i++) {
    MemoryRegion* mr = response->add_mr();
    mr->set_remote_addr(reinterpret_cast<uint64>(mb[i]->buffer()));
    mr->set_rkey(mb[i]->self()->rkey);
  }
  return Status::OK();
}

// Create a GrpcVerbsService, then assign it to a given handle.
void SetNewVerbsService(GrpcVerbsService** handle, const WorkerEnv* worker_env,