Ensure restart clears taskgroups et al (dask#6944)

Author: fjetter

distributed/scheduler.py

@@ -1431,6 +1431,21 @@ def new_task(
 
         return ts
 
+    def _clear_task_state(self):
+
+        logger.debug("Clear task state")
+        for collection in [
+            self.unrunnable,
+            self.erred_tasks,
+            self.computations,
+            self.task_prefixes,
+            self.task_groups,
+            self.task_metadata,
+            self.unknown_durations,
+            self.replicated_tasks,
+        ]:
+            collection.clear()
+
     #####################
     # State Transitions #
     #####################
@@ -3098,8 +3113,6 @@ def __init__(
         resources = {}
         aliases = {}
 
-        self._task_state_collections = [unrunnable]
-
         self._worker_collections = [
             workers,
             host_info,
@@ -3400,7 +3413,7 @@ async def start_unsafe(self):
 
         enable_gc_diagnosis()
 
-        self.clear_task_state()
+        self._clear_task_state()
 
         for addr in self._start_address:
             await self.listen(
@@ -5183,13 +5196,6 @@ async def gather(self, keys, serializers=None):
         self.log_event("all", {"action": "gather", "count": len(keys)})
         return result
 
-    def clear_task_state(self):
-        # XXX what about nested state such as ClientState.wants_what
-        # (see also fire-and-forget...)
-        logger.info("Clear task state")
-        for collection in self._task_state_collections:
-            collection.clear()
-
     @log_errors
     async def restart(self, client=None, timeout=30, wait_for_workers=True):
         """
@@ -5229,9 +5235,8 @@ async def restart(self, client=None, timeout=30, wait_for_workers=True):
                 stimulus_id=stimulus_id,
             )
 
-        self.clear_task_state()
-        self.erred_tasks.clear()
-        self.computations.clear()
+        self._clear_task_state()
+        assert not self.tasks
         self.report({"op": "restart"})
 
         for plugin in list(self.plugins.values()):

distributed/tests/test_scheduler.py

@@ -615,11 +615,16 @@ async def test_ready_remove_worker(s, a, b):
 
 @gen_cluster(client=True, Worker=Nanny, timeout=60)
 async def test_restart(c, s, a, b):
+
     futures = c.map(inc, range(20))
     await wait(futures)
 
     await s.restart()
 
+    assert not s.computations
+    assert not s.task_prefixes
+    assert not s.task_groups
+
     assert len(s.workers) == 2
 
     for ws in s.workers.values():