Repository: dask/distributed

Issue:
Eventually shut down scheduler when no workers return
Currently, a scheduler is not considered idle if it has tasks waiting to be processed but no workers that could process them: https://github.com/hendrikmakait/distributed/blob/4b10aa7068ff8a67cc756ef0c91f5553429963f3/distributed/scheduler.py#L7966-L7972

This makes sense in adaptive scaling scenarios or when using ephemeral workers to give new workers time to spin up. However, we should eventually shut down the scheduler if no workers return instead of leaving it running indefinitely.

We could do this by adjusting the definition of the idle timeout to include time spent without workers or adding a dedicated timeout mechanism.

Discussion:
GPUtester: Can one of the admins verify this patch?

Admins can comment `ok to test` to allow this one PR to run or `add to allowlist` to allow all future PRs from the same author to run.
github-actions[bot]: ## Unit Test Results
_See [test report](https://dask.github.io/distributed/test_report.html) for an extended history of previous test failures. This is useful for diagnosing flaky tests._

       27 files  ±0         27 suites  ±0   11h 53m 36s [:stopwatch:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v2.11.0/README.md#the-symbols "duration of all tests") + 7m 7s
  3 934 tests ±0    3 819 [:heavy_check_mark:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v2.11.0/README.md#the-symbols "passed tests")  - 1     110 [:zzz:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v2.11.0/README.md#the-symbols "skipped / disabled tests") ±0  5 [:x:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v2.11.0/README.md#the-symbols "failed tests") +1 
49 488 runs  ±0  47 193 [:heavy_check_mark:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v2.11.0/README.md#the-symbols "passed tests")  - 2  2 290 [:zzz:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v2.11.0/README.md#the-symbols "skipped / disabled tests") +1  5 [:x:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v2.11.0/README.md#the-symbols "failed tests") +1 

For more details on these failures, see [this check](https://github.com/dask/distributed/runs/19034852177).

Results for commit 28f3f92f. ± Comparison against base commit dc06ce49.

[test-results]:data:application/gzip;base64,H4sIAC6hY2UC/02MSw7DIAwFrxKx7iI2BHAvU1ESS6j5VARWVe9eaJvPcubZ8xIcxmEV1wbNpRFrDumgPkeXwjIXVmhBF1XWVHdJUm14W7P31Vmgwz3CsziAdlfswlhUt4shxiUWU09inmtXkbL2j1tXGSC5u18Xkbavc/fL56xfpimkAgItSybkHlhLMt6zIy3v0pnWdI46ao0BZCfeH4cEsvEUAQAA

hendrikmakait: @fjetter: Do you have time for another review on this PR?
fjetter: Good looks fine but there are a couple of test failures that appear to be related

There is a test failure in `distributed.tests.test_config` that raises a config schema error. That's likely connected to the new option.

`test_idle_timeout_no_workers` also failed once. This is likely a little flaky but I'd be surprised if this isn't connected. I recommend running the test a couple of times concurrently. This typically triggers those flaky things reliably, e.g. `pytest  -x distributed/tests/test_scheduler.py::test_idle_timeout_no_workers -nauto --count=100` (requires `pytest-xdist` and `pytest-repeat`)
crusaderky: @FTang21 do you have time to look into the oustanding issues?
crusaderky: I can envision how users may get confused by the cluster shutting down unexpectedly. Could you please add logging that explain which timeout tripped the timeout?
crusaderky: @hendrikmakait all comments have been addressed

Patch:
diff --git a/distributed/distributed-schema.yaml b/distributed/distributed-schema.yaml
index fffb9b37..71d3cd18 100644
--- a/distributed/distributed-schema.yaml
+++ b/distributed/distributed-schema.yaml
@@ -76,7 +76,20 @@ properties:
             description: |
               Shut down the scheduler after this duration if no activity has occurred
 
-              This can be helpful to reduce costs and stop zombie processes from roaming the earth.
+          no-workers-timeout:
+            type:
+            - string
+            - "null"
+            description: |
+              Shut down the scheduler after this duration if there are pending tasks,
+              but no workers that can process them. This can either mean that there are
+              no workers running at all, or that there are idle workers but they've been
+              excluded through worker or resource restrictions.
+
+              In adaptive clusters, this timeout must be set to be safely higher than
+              the time it takes for workers to spin up.
+
+              Works in conjunction with idle-timeout.
 
           work-stealing:
             type: boolean
diff --git a/distributed/distributed.yaml b/distributed/distributed.yaml
index 8f30c693..1582cdc2 100644
--- a/distributed/distributed.yaml
+++ b/distributed/distributed.yaml
@@ -17,15 +17,16 @@ distributed:
     # Number of seconds to wait until workers or clients are removed from the events log
     # after they have been removed from the scheduler
     events-cleanup-delay: 1h
-    idle-timeout: null      # Shut down after this duration, like "1h" or "30 minutes"
+    idle-timeout: null       # Shut down after this duration, like "1h" or "30 minutes"
+    no-workers-timeout: null # Shut down if there are tasks but no workers to process them
     work-stealing: True     # workers should steal tasks from each other
     work-stealing-interval: 100ms  # Callback time for work stealing
-    worker-saturation: 1.1 # Send this fraction of nthreads root tasks to workers
+    worker-saturation: 1.1  # Send this fraction of nthreads root tasks to workers
     worker-ttl: "5 minutes" # like '60s'. Time to live for workers.  They must heartbeat faster than this
     preload: []             # Run custom modules with Scheduler
     preload-argv: []        # See https://docs.dask.org/en/latest/how-to/customize-initialization.html
     unknown-task-duration: 500ms  # Default duration for all tasks with unknown durations ("15m", "2h")
-    default-task-durations:  # How long we expect function names to run ("1h", "1s") (helps for long tasks)
+    default-task-durations: # How long we expect function names to run ("1h", "1s") (helps for long tasks)
       rechunk-split: 1us
       split-shuffle: 1us
     validate: False         # Check scheduler state at every step for debugging
diff --git a/distributed/scheduler.py b/distributed/scheduler.py
index 6bc30289..2295ab36 100644
--- a/distributed/scheduler.py
+++ b/distributed/scheduler.py
@@ -3517,6 +3517,12 @@ class Scheduler(SchedulerState, ServerNode):
     default_port = 8786
     _instances: ClassVar[weakref.WeakSet[Scheduler]] = weakref.WeakSet()
 
+    worker_ttl: float | None
+    idle_since: float | None
+    idle_timeout: float | None
+    _no_workers_since: float | None  # Note: not None iff there are pending tasks
+    no_workers_timeout: float | None
+
     def __init__(
         self,
         loop=None,
@@ -3578,16 +3584,19 @@ class Scheduler(SchedulerState, ServerNode):
         self.service_kwargs = service_kwargs or {}
         self.services = {}
         self.scheduler_file = scheduler_file
-        worker_ttl = worker_ttl or dask.config.get("distributed.scheduler.worker-ttl")
-        self.worker_ttl = parse_timedelta(worker_ttl) if worker_ttl else None
-        idle_timeout = idle_timeout or dask.config.get(
-            "distributed.scheduler.idle-timeout"
+
+        self.worker_ttl = parse_timedelta(
+            worker_ttl or dask.config.get("distributed.scheduler.worker-ttl")
+        )
+        self.idle_timeout = parse_timedelta(
+            idle_timeout or dask.config.get("distributed.scheduler.idle-timeout")
         )
-        if idle_timeout:
-            self.idle_timeout = parse_timedelta(idle_timeout)
-        else:
-            self.idle_timeout = None
         self.idle_since = time()
+        self.no_workers_timeout = parse_timedelta(
+            dask.config.get("distributed.scheduler.no-workers-timeout")
+        )
+        self._no_workers_since = None
+
         self.time_started = self.idle_since  # compatibility for dask-gateway
         self._replica_lock = RLock()
         self.bandwidth_workers = defaultdict(float)
@@ -3860,9 +3869,12 @@ class Scheduler(SchedulerState, ServerNode):
             pc = PeriodicCallback(self.check_worker_ttl, self.worker_ttl * 1000)
             self.periodic_callbacks["worker-ttl"] = pc
 
-        pc = PeriodicCallback(self.check_idle, (self.idle_timeout or 1) * 1000 / 4)
+        pc = PeriodicCallback(self.check_idle, 250)
         self.periodic_callbacks["idle-timeout"] = pc
 
+        pc = PeriodicCallback(self._check_no_workers, 250)
+        self.periodic_callbacks["no-workers-timeout"] = pc
+
         if extensions is None:
             extensions = DEFAULT_EXTENSIONS.copy()
             if not dask.config.get("distributed.scheduler.work-stealing"):
@@ -8141,7 +8153,7 @@ class Scheduler(SchedulerState, ServerNode):
 
     def check_idle(self) -> float | None:
         if self.status in (Status.closing, Status.closed):
-            return None
+            return None  # pragma: nocover
 
         if self.transition_counter != self._idle_transition_counter:
             self._idle_transition_counter = self.transition_counter
@@ -8178,6 +8190,41 @@ class Scheduler(SchedulerState, ServerNode):
                 self._ongoing_background_tasks.call_soon(self.close)
         return self.idle_since
 
+    def _check_no_workers(self) -> None:
+        """Shut down the scheduler if there have been tasks ready to run which have
+        nowhere to run for `distributed.scheduler.no-workers-timeout`, and there
+        aren't other tasks running.
+        """
+        if self.status in (Status.closing, Status.closed):
+            return  # pragma: nocover
+
+        if (
+            (not self.queued and not self.unrunnable)
+            or (self.queued and self.workers)
+            or any(ws.processing for ws in self.workers.values())
+        ):
+            self._no_workers_since = None
+            return
+
+        # 1. There are queued or unrunnable tasks and no workers at all
+        # 2. There are unrunnable tasks and no workers satisfy their restrictions
+        # (Only rootish tasks can be queued, and rootish tasks can't have restrictions)
+
+        if not self._no_workers_since:
+            self._no_workers_since = time()
+            return
+
+        if (
+            self.no_workers_timeout
+            and time() > self._no_workers_since + self.no_workers_timeout
+        ):
+            logger.info(
+                "Tasks have been without any workers to run them for %s; "
+                "shutting scheduler down",
+                format_time(self.no_workers_timeout),
+            )
+            self._ongoing_background_tasks.call_soon(self.close)
+
     def adaptive_target(self, target_duration=None):
         """Desired number of workers based on the current workload
 


Test patch:
diff --git a/distributed/tests/test_scheduler.py b/distributed/tests/test_scheduler.py
index 16d5f842..67595c38 100644
--- a/distributed/tests/test_scheduler.py
+++ b/distributed/tests/test_scheduler.py
@@ -2405,11 +2405,11 @@ async def test_idle_timeout(c, s, a, b):
     pc.stop()
 
 
-@gen_cluster(
-    client=True,
-    nthreads=[],
-)
+@gen_cluster(client=True, nthreads=[])
 async def test_idle_timeout_no_workers(c, s):
+    """Test that idle-timeout is not triggered if there are no workers available
+    but there are tasks queued
+    """
     # Cancel the idle check periodic timeout so we can step through manually
     s.periodic_callbacks["idle-timeout"].stop()
 
@@ -2440,6 +2440,108 @@ async def test_idle_timeout_no_workers(c, s):
     assert s.check_idle()
 
 
+@gen_cluster(
+    client=True,
+    nthreads=[],
+    config={"distributed.scheduler.no-workers-timeout": None},
+)
+async def test_no_workers_timeout_disabled(c, s, a, b):
+    """no-workers-timeout has been disabled"""
+    future = c.submit(inc, 1, key="x")
+    await wait_for_state("x", ("queued", "no-worker"), s)
+
+    s._check_no_workers()
+    await asyncio.sleep(0.2)
+    s._check_no_workers()
+    await asyncio.sleep(0.2)
+
+    assert s.status == Status.running
+
+
+@pytest.mark.slow
+@gen_cluster(
+    client=True,
+    nthreads=[],
+    config={"distributed.scheduler.no-workers-timeout": "100ms"},
+)
+async def test_no_workers_timeout_without_workers(c, s):
+    """Trip no-workers-timeout when there are no workers available"""
+    # Don't trip scheduler shutdown when there are no tasks
+    s._check_no_workers()
+    await asyncio.sleep(0.2)
+    s._check_no_workers()
+    await asyncio.sleep(0.2)
+
+    assert s.status == Status.running
+
+    future = c.submit(inc, 1)
+    while s.status != Status.closed:
+        await asyncio.sleep(0.01)
+
+
+@pytest.mark.slow
+@gen_cluster(
+    client=True,
+    config={"distributed.scheduler.no-workers-timeout": "100ms"},
+)
+async def test_no_workers_timeout_bad_restrictions(c, s, a, b):
+    """Trip no-workers-timeout when there are workers available but none satisfies
+    task restrictions
+    """
+    future = c.submit(inc, 1, key="x", workers=["127.0.0.2:1234"])
+    while s.status != Status.closed:
+        await asyncio.sleep(0.01)
+
+
+@gen_cluster(
+    client=True,
+    nthreads=[("", 1)],
+    config={"distributed.scheduler.no-workers-timeout": "100ms"},
+)
+async def test_no_workers_timeout_queued(c, s, a):
+    """Don't trip no-workers-timeout when there are queued tasks AND processing tasks"""
+    ev = Event()
+    futures = [c.submit(lambda ev: ev.wait(), ev, pure=False) for _ in range(3)]
+    while not a.state.tasks:
+        await asyncio.sleep(0.01)
+    assert s.queued or math.isinf(s.WORKER_SATURATION)
+
+    s._check_no_workers()
+    await asyncio.sleep(0.2)
+    s._check_no_workers()
+    await asyncio.sleep(0.2)
+
+    assert s.status == Status.running
+    await ev.set()
+
+
+@pytest.mark.slow
+@gen_cluster(
+    client=True,
+    config={"distributed.scheduler.no-workers-timeout": "100ms"},
+)
+async def test_no_workers_timeout_processing(c, s, a, b):
+    """Don't trip no-workers-timeout when there are tasks processing"""
+    ev = Event()
+    x = c.submit(lambda ev: ev.wait(), ev, key="x")
+    y = c.submit(inc, 1, key="y", workers=["127.0.0.2:1234"])
+    await wait_for_state("x", "processing", s)
+    await wait_for_state("y", "no-worker", s)
+
+    # Scheduler won't shut down for as long as f1 is running
+    s._check_no_workers()
+    await asyncio.sleep(0.2)
+    s._check_no_workers()
+    await asyncio.sleep(0.2)
+    assert s.status == Status.running
+
+    await ev.set()
+    await x
+
+    while s.status != Status.closed:
+        await asyncio.sleep(0.01)
+
+
 @gen_cluster(client=True, config={"distributed.scheduler.bandwidth": "100 GB"})
 async def test_bandwidth(c, s, a, b):
     start = s.bandwidth