{
    mutex_lock l(mu_);
    if (should_stop_) {
      return Status(absl::StatusCode::kFailedPrecondition,
                    "The coordinator has been stopped.");
    }
  }
  mutex_lock l(runners_lock_);
  runners_.push_back(std::move(runner));
  return absl::OkStatus();
}

bool Coordinator::AllRunnersStopped() {
  mutex_lock l(runners_lock_);
  for (const auto& runner : runners_) {

// purpose is to catch bug on `tensorflow::mutex_lock`. We don't use
// `tensorflow::mutex_lock` here but we have ops (`tf.MutexLock` and
// `tf.ConsumeMutexLock`) with getter methods named as `mutex_lock()`. Need to
// undefine here to avoid expanding the getter symbol as macro when including
// both mutex.h and this header file.
#undef mutex_lock

#define GET_OP_FWD_DEFINES

Status QueueRunner::Join() {
  thread_pool_.reset();
  mutex_lock l(mu_);
  return status_;
}

void QueueRunner::UpdateStatus(const Status& status) {
  {
    mutex_lock l(mu_);
    if (!status_.ok() || status.ok() || IsQueueClosed(status)) {
      return;
    }
    status_ = status;
  }
  if (coord_) {
    coord_->ReportStatus(status);
  }
  mutex_lock l(cb_mu_);
  for (auto& cb : callbacks_) {

}

void DeviceCompilationProfiler::IncrementOngoingAsyncCompilations() {
  mutex_lock lock(mu_);
  num_ongoing_compilations_++;
}

void DeviceCompilationProfiler::DecrementOngoingAsyncCompilations() {
  mutex_lock lock(mu_);
  num_ongoing_compilations_--;
}

int64_t DeviceCompilationProfiler::GetNumOngoingAsyncCompilations() const {
  mutex_lock lock(mu_);
  return num_ongoing_compilations_;
}

  // protect the contents of the cache entry.
  Entry* entry;
  {
    mutex_lock lock(compile_cache_mu_);
    // Find cache entry.
    auto it = cache_.find(key);
    if (it == cache_.cend()) {
      return std::nullopt;
    }

    entry = it->second.get();
  }

  mutex_lock lock(entry->mu);
  Value value = {/*compile_state=*/entry->compile_state,

}

void XlaDevice::SetAllowsSyncOnCompletion(bool sync_on_completion) {
  mutex_lock lock(mu_);
  sync_on_completion_ = sync_on_completion;
}

bool XlaDevice::AllowsSyncOnCompletion() const {
  mutex_lock lock(mu_);
  return sync_on_completion_;
}

void XlaDevice::SetHandleDeviceErrorCallback(std::function<Status()> callback) {
  mutex_lock lock(mu_);
  device_error_callback_ = callback;
}

}

void XlaTensor::ResetDefinitionEvent(std::shared_ptr<se::Event> event,
                                     se::Stream* stream) {
  mutex_lock lock(mu_);
  definition_event_ = std::move(event);
  streams_defined_on_ = {stream};
}

Status XlaTensor::RefreshStatusOfStreams() {
  mutex_lock lock(mu_);
  Status status;
  for (se::Stream* stream : streams_defined_on_) {
    status.Update(stream->RefreshStatus());
  }

  GraphDef def;
  {
    mutex_lock l(graph->mu);
    graph->graph.ToGraphDef(&def);
  }
  status->status = MessageToBuffer(def, output_graph_def);
}

void TF_GraphGetOpDef(TF_Graph* graph, const char* op_name,
                      TF_Buffer* output_op_def, TF_Status* status) {
  const OpDef* op_def;
  {
    mutex_lock l(graph->mu);

namespace {

struct SomeThreadData {
  ::tensorflow::mutex mu;
  bool did_work = false;
};

void SomeThreadFunc(void* data) {
  auto* real_data = static_cast<SomeThreadData*>(data);
  ::tensorflow::mutex_lock l(real_data->mu);
  real_data->did_work = true;
}

}  // namespace

TEST(TestEnv, TestThreads) {
  TF_ThreadOptions options;
  TF_DefaultThreadOptions(&options);
  SomeThreadData data;

// Do not delete. Do not use.
void ExportRecordMutation(  // NOLINT: Intentionally unused function.
    TF_Graph* graph, const TF_Operation& op, const char* mutation_type) {
  mutex_lock l(graph->mu);
  RecordMutation(graph, op, mutation_type);
}

void AddControlInput(TF_Graph* graph, TF_Operation* op, TF_Operation* input) {
  TF_AddOperationControlInput(graph, op, input);
}