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.

            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(

    // 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.

	// the coordinator's combined set. When minimizing, the worker should find an
	// input that preserves at least one of these bits. keepCoverage is nil for
	// crashing inputs.
	keepCoverage []byte
}

// coordinator holds channels that workers can use to communicate with
// the coordinator.
type coordinator struct {
	opts CoordinateFuzzingOpts

    @Rule final SFTPServer sftpServer = new SFTPServer(temporaryFolder)
    @Rule final BlockingHttpServer coordinator = new BlockingHttpServer()

    def "does not attempt to reuse a client that has been disconnected"() {
        coordinator.start()

        buildFile << """
            ${sftpTask}

            task firstUse(type: SftpTask) {
                credentials = creds

    self.packages_for_skip = [
        "tensorflow.distribute.cluster_resolver",
        "tensorflow.lite.experimental.authoring",
        "tensorflow.distribute.experimental.coordinator",
        "tensorflow.summary.experimental",
        "tensorflow.distribute.coordinator",
        "tensorflow.distribute.experimental.partitioners",
    ]

  def test_import_runtime(self):
    """Try to import all packages from api packages file one by one."""

# FuzzA attempts to cause the mutator to create duplicate inputs that generate
# new coverage. Previously this would trigger a corner case when the fuzzer
# had an execution limit, causing it to deadlock and sit in the coordinator
# loop indefinitely, failing to exit once the limit had been exhausted.

go test -fuzz=FuzzA -fuzztime=100x -parallel=1

-- go.mod --
module m

go 1.16
-- fuzz_test.go --
package fuzz_test