Allow a loop cycle after every message in the batch

dask · Feb 4, 2022 · 4c078a3 · 4c078a3
1 parent 3cc145b
commit 4c078a3
Showing 1 changed file with 1 addition and 1 deletion.
diff --git a/distributed/core.py b/distributed/core.py
@@ -591,7 +591,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):