Add processor for concurrent execution of function

pyproject.toml

@@ -21,8 +21,8 @@ dev = [
     "docutils>=0.21.2",
     "pydoctor>=24.11.2",
     "pyright>=1.1.396",
-    "pytest>=8.3.5",
     "pytest-cov>=6.1.1",
+    "pytest>=8.3.5",
     "ruff>=0.11.0",
     "tabulate>=0.9.0",
     "types-requests>=2.32.0.20250306",

resonate/bridge.py

@@ -4,7 +4,7 @@
 import threading
 import time
 from concurrent.futures import Future
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING
 
 from resonate.conventions import Base
 from resonate.delay_q import DelayQ
@@ -34,9 +34,10 @@
     Return,
 )
 from resonate.models.durable_promise import DurablePromise
-from resonate.models.result import Ko, Ok, Result
+from resonate.models.result import Ko, Ok
 from resonate.models.task import Task
 from resonate.options import Options
+from resonate.processor import Processor
 from resonate.scheduler import Done, Info, More, Scheduler
 from resonate.utils import exit_on_exception
 
@@ -84,6 +85,7 @@ def __init__(
             self._unicast,
             self._anycast,
         )
+        self._processor = Processor()
 
         self._bridge_thread = threading.Thread(target=self._process_cq, name="bridge", daemon=True)
         self._message_source_thread = threading.Thread(target=self._process_msgs, name="message-source", daemon=True)
@@ -166,6 +168,8 @@ def get(self, id: str, opts: Options, future: Future) -> DurablePromise:
         return promise
 
     def start(self) -> None:
+        self._processor.start()
+
         if not self._message_source_thread.is_alive():
             self._message_source.start()
             self._message_source_thread.start()
@@ -191,6 +195,7 @@ def stop(self) -> None:
 
     def _stop_no_join(self) -> None:
         """Stop internal components and threads. Does not join the threads, to be able to call it from the bridge itself."""
+        self._processor.stop()
         self._message_source.stop()
         self._cq.put_nowait(None)
         self._heartbeat_active.clear()
@@ -218,7 +223,7 @@ def _process_cq(self) -> None:
                                     return
 
                             case Function(id, cid, func):
-                                self._cq.put_nowait(Return(id, cid, self._handle_function(func)))
+                                self._processor.enqueue(func, lambda r: self._cq.put_nowait(Return(id, cid, r)))
                             case Delayed() as item:
                                 self._handle_delay(item)
 
@@ -341,7 +346,7 @@ def _process_delayed_events(self) -> None:
                 for item in events:
                     match item:
                         case Function(id, cid, func):
-                            self._cq.put_nowait(Return(id, cid, self._handle_function(func)))
+                            self._processor.enqueue(func, lambda r: self._cq.put_nowait(Return(id, cid, r)))
                         case retry:
                             self._cq.put_nowait(retry)
 
@@ -434,10 +439,3 @@ def _handle_network_request(self, cmd_id: str, cid: str, req: CreatePromiseReq |
 
             case _:
                 raise NotImplementedError
-
-    def _handle_function(self, func: Callable[[], Any]) -> Result:
-        try:
-            r = func()
-            return Ok(r)
-        except Exception as e:
-            return Ko(e)

resonate/processor.py

@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+import os
+import queue
+from collections.abc import Callable
+from threading import Thread
+from typing import Any
+
+from resonate.models.result import Ko, Ok, Result
+from resonate.utils import exit_on_exception
+
+
+class Processor:
+    def __init__(self, workers: int | None = None) -> None:
+        self.threads = set[Thread]()
+        for _ in range(min(32, workers or (os.cpu_count() or 1))):
+            self.threads.add(Thread(target=self._run, daemon=True))
+
+        self.sq = queue.Queue[tuple[Callable[[], Any], Callable[[Result[Any]], None]] | None]()
+
+    @exit_on_exception
+    def _run(self) -> None:
+        while sqe := self.sq.get():
+            func, callback = sqe
+
+            try:
+                r = Ok(func())
+            except Exception as e:
+                r = Ko(e)
+
+            callback(r)
+            self.sq.task_done()
+
+        self.sq.task_done()
+
+    def enqueue(self, func: Callable[[], Any], callback: Callable[[Result[Any]], None]) -> None:
+        self.sq.put((func, callback))
+
+    def start(self) -> None:
+        for t in self.threads:
+            if not t.is_alive():
+                t.start()
+
+    def stop(self) -> None:
+        for _ in self.threads:
+            self.sq.put(None)
+
+        self.sq.join()
+
+        for t in self.threads:
+            t.join()

ruff.toml

@@ -9,6 +9,7 @@ ignore = [
     "ANN401",
     "ARG001",
     "ARG002",
+    "B023",
     "BLE001",
     "C901",
     "COM812",

tests/test_processor.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+from queue import Queue
+
+import pytest
+
+from resonate.models.result import Ok, Result
+from resonate.processor import Processor
+
+
+def greet(name: str) -> str:
+    return f"Hi {name}"
+
+
+def callback(q: Queue[tuple[str, str]], expected: str, result: Result[str]) -> None:
+    assert isinstance(result, Ok)
+    q.put((result.value, expected))
+
+
+@pytest.mark.parametrize("workers", [1, 2, 3])
+def test_processor(workers: int) -> None:
+    q = Queue[tuple[str, str]]()
+    p = Processor(workers)
+    assert len(p.threads) == workers
+
+    names = ["A", "B"]
+    expected_greet = [greet("A"), greet("B")]
+    p.start()
+    for name, expected in zip(names, expected_greet, strict=False):
+        p.enqueue(lambda: greet(name), lambda r: callback(q, expected, r))
+
+    p.stop()
+    assert q.qsize() == len(names)
+
+    for _ in range(q.qsize()):
+        actual, expected = q.get()
+        assert actual == expected
+        q.task_done()
+
+    q.join()
+    assert q.empty()