Method StartParallelExecutors

tensorflow/core/distributed_runtime/graph_mgr.cc:543–643

}

void GraphMgr::StartParallelExecutors(
    const string& handle, int64 step_id, Item* item, Rendezvous* rendezvous,
    Rendezvous* global_rendezvous,
    CollectiveExecutor::Handle* ce_handle, StepStatsCollector* collector,
    CostGraphDef* cost_graph, CancellationManager* cancellation_manager,
    WorkerSession* session, const ExecutorOpts& opts, StatusCallback done) {
  const int num_units = item->units.size();
  CHECK_GE(num_units, 1);
  ScopedStepContainer* step_container = new ScopedStepContainer(
      step_id,
      [this](const string& name) { device_mgr_->ClearContainers({name}); });
#ifdef TENSORFLOW_USE_STAR
  int32 cpu_id = 0;
  if (worker_env_->lockless) {
    cpu_id = seastar::engine().cpu_id();
  }
#endif // TENSORFLOW_USE_STAR
  // NOTE: Transfer one ref of rendezvous and item.
  ExecutorBarrier* barrier =
      new ExecutorBarrier(num_units, rendezvous,
                          [this, item, collector, cost_graph, step_container,
#ifdef TENSORFLOW_USE_STAR
                           done, cpu_id](const Status& s) {
                            BuildCostModel(item, collector, cost_graph);
                            if (worker_env_->lockless) {
                              seastar::alien::submit_to(cpu_id, [s, done] {
                                done(s);
                                return seastar::make_ready_future();
                              });
                            } else  {
                              done(s);
                            }
#else
                           done](const Status& s) {
                            BuildCostModel(item, collector, cost_graph);
                            done(s);
#endif // TENSORFLOW_USE_STAR
                            delete step_container;
                          });
  Executor::Args args;
  args.step_id = step_id;
  args.round_step_id = step_id;
  args.rendezvous = rendezvous;
  args.global_rendezvous = global_rendezvous;
  args.collective_executor = ce_handle ? ce_handle->get() : nullptr;
  args.cancellation_manager = cancellation_manager;
  args.stats_collector = collector;
  args.step_container = step_container;
  args.sync_on_finish = sync_on_finish_;
  if (LogMemory::IsEnabled()) {
    LogMemory::RecordStep(args.step_id, handle);
  }
  args.executor_policy = opts.executor_policy();
  thread::ThreadPool* pool = worker_env_->compute_pool;
  using std::placeholders::_1;
  using std::placeholders::_2;
  // Line below is equivalent to this code, but does one less indirect call:
  //  args.runner = [pool](std::function<void()> fn) { pool->Schedule(fn); };