Use "put_nowait" for event queue

backend/src/events.py

@@ -121,7 +121,7 @@ class NodeFinishEvent(TypedDict):
 
 class EventConsumer(ABC):
     @abstractmethod
-    async def put(self, event: Event) -> None: ...
+    def put(self, event: Event) -> None: ...
 
     @staticmethod
     def filter(queue: EventConsumer, allowed: set[str]) -> EventConsumer:
@@ -133,9 +133,9 @@ class _FilteredEventConsumer(EventConsumer):
     queue: EventConsumer
     allowed: set[str]
 
-    async def put(self, event: Event) -> None:
+    def put(self, event: Event) -> None:
         if event["event"] in self.allowed:
-            await self.queue.put(event)
+            self.queue.put(event)
 
 
 class EventQueue(EventConsumer):
@@ -145,8 +145,8 @@ def __init__(self):
     async def get(self) -> Event:
         return await self.queue.get()
 
-    async def put(self, event: Event) -> None:
-        await self.queue.put(event)
+    def put(self, event: Event) -> None:
+        self.queue.put_nowait(event)
 
     async def wait_until_empty(self, timeout: float) -> None:
         while timeout > 0:

backend/src/process.py

@@ -575,7 +575,7 @@ async def __process(self, node: Node) -> NodeOutput | CollectorOutput:
         context = self.__get_node_context(node)
 
         await self.progress.suspend()
-        await self.__send_node_start(node)
+        self.__send_node_start(node)
         await self.progress.suspend()
 
         output, execution_time = await self.loop.run_in_executor(
@@ -588,7 +588,7 @@ async def __process(self, node: Node) -> NodeOutput | CollectorOutput:
 
         if isinstance(output, RegularOutput):
             await self.__send_node_broadcast(node, output.output)
-            await self.__send_node_finish(node, execution_time)
+            self.__send_node_finish(node, execution_time)
         elif isinstance(output, IteratorOutput):
             await self.__send_node_broadcast(node, output.partial_output)
             # TODO: execution time
@@ -702,15 +702,15 @@ def fill_partial_output(values: object) -> RegularOutput:
         async def update_progress():
             iter_times.add()
             iterations = iter_times.iterations
-            await self.__send_node_progress(
+            self.__send_node_progress(
                 node,
                 iter_times.times,
                 iterations,
                 max(expected_length, iterations),
             )
 
         # iterate
-        await self.__send_node_progress(node, [], 0, expected_length)
+        self.__send_node_progress(node, [], 0, expected_length)
 
         deferred_errors: list[str] = []
         for values in iterator_output.iterator.iter_supplier():
@@ -765,8 +765,8 @@ async def update_progress():
         await self.__send_node_broadcast(node, iterator_output.partial_output)
 
         # finish iterator
-        await self.__send_node_progress_done(node, iter_times.iterations)
-        await self.__send_node_finish(node, iter_times.get_time_since_start())
+        self.__send_node_progress_done(node, iter_times.iterations)
+        self.__send_node_finish(node, iter_times.get_time_since_start())
 
         # finalize collectors
         for collector, timer, collector_node in collectors:
@@ -778,7 +778,7 @@ async def update_progress():
 
             await self.__send_node_broadcast(collector_node, collector_output.output)
             # TODO: execution time
-            await self.__send_node_finish(collector_node, timer.duration)
+            self.__send_node_finish(collector_node, timer.duration)
 
             self.cache.set(
                 collector_node.id,
@@ -791,7 +791,7 @@ async def update_progress():
             raise Exception(f"Errors occurred during iteration:\n{error_string}")
 
     async def __process_nodes(self):
-        await self.__send_chain_start()
+        self.__send_chain_start()
 
         # we first need to run iterator nodes in topological order
         for node_id in self.chain.topological_order():
@@ -850,12 +850,12 @@ def kill(self):
 
     # events
 
-    async def __send_chain_start(self):
+    def __send_chain_start(self):
         # all nodes except the cached ones
         nodes = set(self.chain.nodes.keys())
         nodes.difference_update(self.cache.keys())
 
-        await self.queue.put(
+        self.queue.put(
             {
                 "event": "chain-start",
                 "data": {
@@ -864,8 +864,8 @@ async def __send_chain_start(self):
             }
         )
 
-    async def __send_node_start(self, node: Node):
-        await self.queue.put(
+    def __send_node_start(self, node: Node):
+        self.queue.put(
             {
                 "event": "node-start",
                 "data": {
@@ -874,7 +874,7 @@ async def __send_node_start(self, node: Node):
             }
         )
 
-    async def __send_node_progress(
+    def __send_node_progress(
         self, node: Node, times: Sequence[float], index: int, length: int
     ):
         def get_eta(times: Sequence[float]) -> float:
@@ -890,7 +890,7 @@ def get_eta(times: Sequence[float]) -> float:
             remaining = max(0, length - index)
             return avg_time * remaining
 
-        await self.queue.put(
+        self.queue.put(
             {
                 "event": "node-progress",
                 "data": {
@@ -903,8 +903,8 @@ def get_eta(times: Sequence[float]) -> float:
             }
         )
 
-    async def __send_node_progress_done(self, node: Node, length: int):
-        await self.queue.put(
+    def __send_node_progress_done(self, node: Node, length: int):
+        self.queue.put(
             {
                 "event": "node-progress",
                 "data": {
@@ -935,7 +935,7 @@ async def send_broadcast():
                 return
 
             data, types = result
-            await self.queue.put(
+            self.queue.put(
                 {
                     "event": "node-broadcast",
                     "data": {
@@ -951,12 +951,12 @@ async def send_broadcast():
             # broadcasts are done is parallel, so don't wait
             self.__broadcast_tasks.append(self.loop.create_task(send_broadcast()))
 
-    async def __send_node_finish(
+    def __send_node_finish(
         self,
         node: Node,
         execution_time: float,
     ):
-        await self.queue.put(
+        self.queue.put(
             {
                 "event": "node-finish",
                 "data": {

backend/src/server.py

@@ -258,7 +258,7 @@ async def run(request: Request):
                 "inputs": exception.inputs,
             }
 
-        await ctx.queue.put({"event": "execution-error", "data": error})
+        ctx.queue.put({"event": "execution-error", "data": error})
         return json(error_response("Error running nodes!", exception), status=500)
     finally:
         if ctx.config.trace and tracer is not None: