limitations under the License.
==============================================================================*/

#include "tensorflow/cc/training/coordinator.h"

namespace tensorflow {

Coordinator::Coordinator() : Coordinator(std::vector<error::Code>()) {}

Coordinator::Coordinator(const std::vector<error::Code>& clean_stop_errors)
    : should_stop_(false) {
  if (clean_stop_errors.empty()) {

  /// its queue.
  virtual bool IsRunning() const = 0;
};

/// Coordinator class manages the termination of a collection of QueueRunners.
/// Without a coordinator, QueueRunners have to be joined in a specific order;
/// otherwise the QueueRunner::Join() could sometimes hang. The
/// Coordinator::RequestStop() plays the key role which notifies all running
/// threads under a coordinator to stop. This function could be called by any
/// thread or any client.

    private final BuildInclusionCoordinator coordinator;
    private final PublicBuildPath publicBuildPath;
    private final Instantiator instantiator;
    private final GradleInternal gradle;
    private final List<BuildDefinition> pluginBuildDefinitions = new ArrayList<>();

# Test that when the coordinator experiences an I/O error communicating
# with a worker, the coordinator stops the worker and reports the error.
# The coordinator should not record a crasher.
#
# We simulate an I/O error in the test by writing garbage to fuzz_out.
# This is unlikely, but possible. It's difficult to simulate interruptions
# due to ^C and EOF errors which are more common. We don't report those.
[short] skip
[!fuzz] skip
env GOCACHE=$WORK/cache

    private final ProjectsPreparer delegate;
    private final BuildInclusionCoordinator coordinator;
    private final BuildSourceBuilder buildSourceBuilder;
    private final BuildLoader buildLoader;

    public BuildTreePreparingProjectsPreparer(ProjectsPreparer delegate, BuildLoader buildLoader, BuildInclusionCoordinator coordinator, BuildSourceBuilder buildSourceBuilder) {
        this.delegate = delegate;

            coordinator.getIdleMillis() == 0L
        }
    }

    def "idle millis is > 0 when daemon is idle"() {
        when:
        coordinator.updateActivityTimestamp()
        sleep(10)

        then:
        idle
        coordinator.getIdleMillis() > 0
    }

    boolean isNotStopped() {
        return coordinator.state != Stopped && coordinator.state != ForceStopped
    }

        "//tensorflow/core:test",
        "//tensorflow/core:test_main",
        "//tensorflow/core:testlib",
    ],
)

cc_library(
    name = "coordinator",
    srcs = ["training/coordinator.cc"],
    hdrs = ["training/coordinator.h"],
    deps = [
        "//tensorflow/core:lib",
        "//tensorflow/core:lib_internal",
        "//tensorflow/core:protos_all_cc",
    ],
)

tf_cc_test(

namespace tensorflow {
namespace {

using error::Code;

void WaitForStopThread(Coordinator* coord, Notification* about_to_wait,
                       Notification* done) {
  about_to_wait->Notify();
  coord->WaitForStop();
  done->Notify();
}

TEST(CoordinatorTest, TestStopAndWaitOnStop) {
  Coordinator coord;
  EXPECT_EQ(coord.ShouldStop(), false);

  Notification about_to_wait;
  Notification done;

    // Queue a task to interrupt the waiting coordinator.
    serialTaskQueue +=
      object : SerialTask {
        override fun start() {
          taskRunner.assertThreadHoldsLock()
          waitingCoordinatorInterrupted = true
          val coordinatorTask = waitingCoordinatorTask ?: error("no coordinator waiting")
          currentTask = coordinatorTask

// If a test process terminates unexpectedly while fuzzing, coordinate will
// attempt to restart and continue unless the termination can be attributed
// to an interruption (from a timer or the user).
//
// While looping, coordinate receives inputs from the coordinator, passes
// those inputs to the worker process, then passes the results back to
// the coordinator.
func (w *worker) coordinate(ctx context.Context) error {
	// Main event loop.