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_) {
    thread_pool_->Schedule(std::bind(&QueueRunner::Stop, this, sess));
  }
  // Wait for up to 'wait_for' milliseconds.
  if (wait_for > 0) {

                     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,

  xla::TransferManager* transfer_manager_;

  XlaShapeLayoutHelpers::ShapeDeterminationFns shape_determination_fns_;

  // Thread pool used for running closures
  thread::ThreadPool* thread_pool_;

  absl::Mutex mu_;
  int next_stream_ TF_GUARDED_BY(mu_) = 0;
};

}  // namespace tensorflow

  string close_op_name_;
  string cancel_op_name_;
  // code::Code casted to int to avoid a hash function.
  std::unordered_set<int> queue_closed_exception_types_;

  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_;

      client_(client),
      transfer_manager_(client->backend().transfer_manager()),
      shape_determination_fns_(std::move(shape_determination_fns)),
      thread_pool_(thread_pool) {
  CHECK(host_to_device_stream_ != nullptr);
  CHECK(stream_ != nullptr);
}

void XlaDeviceContext::CopyTensorInSameDevice(const Tensor* input_tensor,
                                              Device* device,

  VLOG(1) << "XlaDevice options: use_multiple_streams: "
          << options.use_multiple_streams << " use_global_compute_stream: "
          << options.use_global_compute_stream;
  thread_pool_.reset(new thread::ThreadPool(session_options.env, "xla_device",
                                            /*num_threads=*/1));

  // We have multiple device to device streams to allow for some concurrency

  std::unique_ptr<DeviceBase::AcceleratorDeviceInfo> accelerator_device_info_
      TF_GUARDED_BY(mu_);

  // Thread pool used for running closures
  std::unique_ptr<thread::ThreadPool> thread_pool_;

  // True if the device allows XlaDevice::Sync to be called on completion
  // regardless of status.
  bool sync_on_completion_ TF_GUARDED_BY(mu_) = true;

  {
    thread::ThreadPool thread_pool(Env::Default(), "pool", 2);
    thread_pool.Schedule([&session, a]() {
      std::vector<Tensor> outputs;
      TF_EXPECT_OK(session.Run({a}, &outputs));
      test::ExpectTensorEqual<int>(outputs[0],
                                   test::AsTensor<int>({4, 6}, {2}));
    });
    thread_pool.Schedule([&session, b]() {
      std::vector<Tensor> outputs;

    }

    public SuggesterBuilder analyzer(final SuggestAnalyzer analyzer) {
        this.analyzer = analyzer;
        return this;
    }

    public SuggesterBuilder threadPool(final ExecutorService threadPool) {
        this.threadPool = threadPool;
        return this;
    }

    public SuggesterBuilder threadPoolSize(final int threadPoolSize) {
        this.threadPoolSize = threadPoolSize;
        return this;
    }

   * be a SynchronousQueue.
   */
  private ExecutorService threadPool;

  @Override
  public void setUp() {
    threadPool = newCachedThreadPool();
  }

  @Override
  public void tearDown() throws InterruptedException {
    threadPool.shutdown();
    assertTrue("Some worker didn't finish in time", threadPool.awaitTermination(10, SECONDS));
  }

  private static <T> int drain(