Merge branch 'main' into transition_counter_max

distributed/batched.py

@@ -128,7 +128,7 @@ def _background_send(self):
         self.stopped.set()
         self.abort()
 
-    def send(self, *msgs: dict) -> None:
+    def send(self, *msgs) -> None:
         """Schedule a message for sending to the other side
 
         This completes quickly and synchronously

distributed/core.py

@@ -623,7 +623,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):
         extra = extra or {}
         logger.info("Starting established connection")
 
@@ -653,12 +653,6 @@ async def handle_stream(self, comm, extra=None, every_cycle=()):
                             logger.error("odd message %s", msg)
                     await asyncio.sleep(0)
 
-                for func in every_cycle:
-                    if is_coroutine_function(func):
-                        self.loop.add_callback(func)
-                    else:
-                        func()
-
         except OSError:
             pass
         except Exception as e:

distributed/diagnostics/plugin.py

@@ -72,7 +72,11 @@ async def close(self) -> None:
         """
 
     def update_graph(
-        self, scheduler: Scheduler, keys: set[str], restrictions: dict, **kwargs
+        self,
+        scheduler: Scheduler,
+        keys: set[str],
+        restrictions: dict[str, float],
+        **kwargs,
     ) -> None:
         """Run when a new graph / tasks enter the scheduler"""