Method Initialize

tensorflow/core/distributed_runtime/base_rendezvous_mgr.cc:221–264

}

Status BaseRemoteRendezvous::Initialize(WorkerSession* session) {
  CHECK_NE(session, nullptr) << "session must not be null!";
  std::vector<DeferredCall> deferred_calls;
  {
    mutex_lock l(mu_);
    if (session_ != nullptr) {
      if (worker_name_ != session->worker_name) {
        Status s = errors::Internal(
            "Double init! Worker names would have changed from: ",
            worker_name_, " -> ", session->worker_name);
        LOG(WARNING) << s;
        return s;
      }
    }
    session_ = session;
    worker_name_ = session->worker_name;
    std::swap(deferred_calls, deferred_calls_);
  }
  for (auto& call : deferred_calls) {
    RecvLocalAsyncInternal(call.parsed, std::move(call.done));
  }

  std::vector<DeferredFuseCall> deferred_fuse_calls;
  {
    mutex_lock l(mu_);
    std::swap(deferred_fuse_calls, deferred_fuse_calls_);
  }
  for (auto& fuse_call : deferred_fuse_calls) {
    FuseRecvLocalAsyncInternal(fuse_call.parsed_keys,
                               std::move(fuse_call.done));
  }

  std::vector<DeferredFlowControlCall> deferred_flow_control_calls;
  {
    mutex_lock l(mu_);
    std::swap(deferred_flow_control_calls, deferred_flow_control_calls_);
  }
  for (auto& fc_call : deferred_flow_control_calls) {
    FlowControlRecvLocalAsyncInternal(fc_call.tag, fc_call.parsed,
                                      std::move(fc_call.done));
  }

  return Status::OK();
}

WorkerSession* BaseRemoteRendezvous::session() {
  tf_shared_lock l(mu_);