Migrate ensure_executing to WorkerState event mechanism (dask#6003)

distributed/core.py

@@ -596,7 +596,7 @@ async def handle_comm(self, comm):
                         "Failed while closing connection to %r: %s", address, e
                     )
 
-    async def handle_stream(self, comm, extra=None, every_cycle=[]):
+    async def handle_stream(self, comm, extra=None, every_cycle=()):
         extra = extra or {}
         logger.info("Starting established connection")
 

distributed/tests/test_worker_state_machine.py

@@ -1,9 +1,13 @@
+from itertools import chain
+
 import pytest
 
 from distributed.utils import recursive_to_dict
 from distributed.worker_state_machine import (
+    Instruction,
     ReleaseWorkerDataMsg,
     SendMessageToScheduler,
+    StateMachineEvent,
     TaskState,
     UniqueTaskHeap,
 )
@@ -82,13 +86,32 @@ def test_unique_task_heap():
     assert repr(heap) == "<UniqueTaskHeap: 0 items>"
 
 
-@pytest.mark.parametrize("cls", SendMessageToScheduler.__subclasses__())
-def test_sendmsg_slots(cls):
-    smsg = cls(**dict.fromkeys(cls.__annotations__))
-    assert not hasattr(smsg, "__dict__")
+@pytest.mark.parametrize(
+    "cls",
+    chain(
+        [UniqueTaskHeap],
+        Instruction.__subclasses__(),
+        SendMessageToScheduler.__subclasses__(),
+        StateMachineEvent.__subclasses__(),
+    ),
+)
+def test_slots(cls):
+    params = [
+        k
+        for k in dir(cls)
+        if not k.startswith("_") and k != "op" and not callable(getattr(cls, k))
+    ]
+    inst = cls(**dict.fromkeys(params))
+    assert not hasattr(inst, "__dict__")
 
 
 def test_sendmsg_to_dict():
     # Arbitrary sample class
     smsg = ReleaseWorkerDataMsg(key="x")
     assert smsg.to_dict() == {"op": "release-worker-data", "key": "x"}
+
+
+@pytest.mark.parametrize("cls", StateMachineEvent.__subclasses__())
+def test_event_slots(cls):
+    smsg = cls(**dict.fromkeys(cls.__annotations__), stimulus_id="test")
+    assert not hasattr(smsg, "__dict__")