Notification* start = nullptr) {
    thread_pool_->Schedule(
        std::bind(&MockQueueRunner::CountThread, this, counter, until, start));
  }

  void StartSettingStatus(const Status& status, BlockingCounter* counter,
                          Notification* start) {
    thread_pool_->Schedule(std::bind(&MockQueueRunner::SetStatusThread, this,
                                     status, counter, start));
  }

  SetRunArgumentsAndCostGraph(run_options);
  return Start(sess, 0);
}

Status QueueRunner::Start(Session* sess, int wait_for) {
  counter_.reset(new BlockingCounter(runs_));
  for (const string& enqueue_op : enqueue_op_names_) {
    thread_pool_->Schedule(
        std::bind(&QueueRunner::Run, this, sess, enqueue_op));
  }
  if (coord_) {

  std::unique_ptr<thread::ThreadPool> thread_pool_;
  mutex mu_;
  int runs_ = 0;
  Status status_ TF_GUARDED_BY(mu_);
  Status enqueue_status_ TF_GUARDED_BY(mu_);
  std::unique_ptr<BlockingCounter> counter_;

  Coordinator* coord_;

  std::atomic<bool> stopped_;

  mutex cb_mu_;
  std::vector<std::function<void(Status)>> callbacks_;

  mutable std::unique_ptr<mutex> cg_mu_;

  // concurrently (at which point it will respond with success to all callers),
  // or 10 seconds have elapsed (at which point it will respond with an error).
  const int callers = 4;
  BlockingCounter counter(callers);
  auto fetcher = [&counter](const string& filename, size_t offset, size_t n,
                            char* buffer, TF_Status* status) -> int64_t {
    counter.DecrementCount();