Method MPIBackgroundThread

tensorflow/contrib/mpi/mpi_rendezvous_mgr.cc:277–318

}

void MPIRendezvousMgr::MPIBackgroundThread() {
  std::list<std::unique_ptr<MPISendTensorCall>> active_sends;

  while (1) {
    MPI_Status status;

    // Check for incoming Tensor requests
    RecvTensorRequest request;
    if (ProbeForData(TAG_REQTENSOR, &status, &request)) {
      this->AddRequest(request, status.MPI_SOURCE);
    }

    // Check for incoming Tensor reply
    MPIRecvTensorResponse mRes;
    if (ProbeForData(TAG_SENDTENSOR, &status, &mRes)) {
      const int64 step_id = mRes.step_id();
      std::string key = mRes.key();

      std::shared_ptr<MPIRequestTensorCall> call;
      GetRecvCall(step_id, key, &call);
      call->recv_call_(mRes);
      RemoveRecvCall(step_id, key);
    }

    // Remove sends that have been completed
    active_sends.remove_if(
        [](std::unique_ptr<MPISendTensorCall>& i) { return i->IsFinished(); });

    // send a Tensor request
    RequestQueueEntry req;
    if (GetRequest(&req)) req.second();

    // Send a Tensor response
    SendQueueEntry send;
    if (GetResponse(&send)) {
      std::unique_ptr<MPISendTensorCall> p(send.second());
      active_sends.push_back(std::move(p));
    }

    //    std::this_thread::sleep_for(std::chrono::microseconds(1));
  }
}

}  // namespace tensorflow
#endif  // TENSORFLOW_USE_MPI