Allow a loop cycle after every message in the batch

dask · Feb 15, 2022 · 9da4c46 · 9da4c46
1 parent f25f1fc
commit 9da4c46
Showing 1 changed file with 1 addition and 1 deletion.
diff --git a/distributed/core.py b/distributed/core.py
@@ -586,7 +586,7 @@ async def handle_stream(self, comm, extra=None, every_cycle=[]):
                                 handler(**merge(extra, msg))
                         else:
                             logger.error("odd message %s", msg)
-                    await asyncio.sleep(0)
+                        await asyncio.sleep(0)
 
                 for func in every_cycle:
                     if is_coroutine_function(func):