From c7417f1b67bc214d42a20636dbb23ac5aef6c2b2 Mon Sep 17 00:00:00 2001 From: Atishay Kasliwal Date: Sun, 8 Nov 2026 17:29:54 -0500 Subject: [PATCH] fix: update asyncio_reduce example to the current reducer API Both scripts used AsyncReducer and aiorun, which are gone. Move them to ReduceAsyncServer, make the executor handler a Reducer that uses its own pool, and stop reading the event loop before one is running. The Dockerfile ran a missing example.py, and requests was not a dependency. Add the example to the image build matrix, as the TODO there asked. Fixes #338 Signed-off-by: Atishay Kasliwal --- .github/workflows/build-push.yaml | 4 +- .../examples/reduce/asyncio_reduce/Dockerfile | 2 +- .../examples/reduce/asyncio_reduce/README.md | 3 +- .../reduce/asyncio_reduce/async_exec.py | 7 ++- .../reduce/asyncio_reduce/executorpool.py | 43 +++++---------- .../reduce/asyncio_reduce/pyproject.toml | 4 +- .../examples/reduce/asyncio_reduce/uv.lock | 54 +++++++++---------- 7 files changed, 49 insertions(+), 68 deletions(-) diff --git a/.github/workflows/build-push.yaml b/.github/workflows/build-push.yaml index 0d54ed6e..ecfa9e7a 100644 --- a/.github/workflows/build-push.yaml +++ b/.github/workflows/build-push.yaml @@ -18,12 +18,10 @@ jobs: strategy: matrix: - # TODO: rewrite asyncio_reduce example using latest SDK version, as it is currently using old methods - # then add to example_directories matrix example_directories: [ "examples/map/even_odd", "examples/map/flatmap", "examples/map/forward_message", "examples/map/multiproc_map", "examples/mapstream/flatmap_stream", "examples/reduce/counter", - "examples/reducestream/counter", "examples/reducestream/sum", + "examples/reduce/asyncio_reduce", "examples/reducestream/counter", "examples/reducestream/sum", "examples/sideinput/simple_sideinput", "examples/sideinput/simple_sideinput/udf", "examples/sink/async_log", "examples/sink/log", "examples/sink/all_sinks", "examples/source/simple_source", "examples/sourcetransform/event_time_filter", diff --git a/packages/pynumaflow/examples/reduce/asyncio_reduce/Dockerfile b/packages/pynumaflow/examples/reduce/asyncio_reduce/Dockerfile index b6e1789b..8c6ca9cf 100644 --- a/packages/pynumaflow/examples/reduce/asyncio_reduce/Dockerfile +++ b/packages/pynumaflow/examples/reduce/asyncio_reduce/Dockerfile @@ -33,6 +33,6 @@ COPY --from=builder $EXAMPLE_PATH/ $EXAMPLE_PATH/ # is not present (by design, to keep the image small), so uv run fails. # Instead, we activate the pre-built .venv via PATH and run python directly. ENV PATH="$EXAMPLE_PATH/.venv/bin:$PATH" -CMD ["python", "example.py"] +CMD ["python", "async_exec.py"] EXPOSE 5000 diff --git a/packages/pynumaflow/examples/reduce/asyncio_reduce/README.md b/packages/pynumaflow/examples/reduce/asyncio_reduce/README.md index 56193baa..698d9b32 100644 --- a/packages/pynumaflow/examples/reduce/asyncio_reduce/README.md +++ b/packages/pynumaflow/examples/reduce/asyncio_reduce/README.md @@ -85,7 +85,8 @@ Default executor pool would work just as well in the use case. ```python if __name__ == "__main__": ... - threadPool = ExecutorPool(exec_type=e_type, max_workers=max_workers) + exec_pool = ExecutorPool(exec_type=e_type, max_workers=mx_workers) + grpc_server = ReduceAsyncServer(ReduceHandler, init_args=(exec_pool,)) ... ``` diff --git a/packages/pynumaflow/examples/reduce/asyncio_reduce/async_exec.py b/packages/pynumaflow/examples/reduce/asyncio_reduce/async_exec.py index 9e04fa41..cc30737d 100644 --- a/packages/pynumaflow/examples/reduce/asyncio_reduce/async_exec.py +++ b/packages/pynumaflow/examples/reduce/asyncio_reduce/async_exec.py @@ -1,5 +1,4 @@ import aiohttp -import aiorun import asyncio import json import time @@ -11,7 +10,7 @@ Message, Datum, Metadata, - AsyncReducer, + ReduceAsyncServer, ) _LOGGER = setup_logging(__name__) @@ -47,5 +46,5 @@ async def reduce_handler(keys: list[str], datums: AsyncIterable[Datum], md: Meta if __name__ == "__main__": - grpc_server = AsyncReducer(handler=reduce_handler) - aiorun.run(grpc_server.start()) + grpc_server = ReduceAsyncServer(reduce_handler) + grpc_server.start() diff --git a/packages/pynumaflow/examples/reduce/asyncio_reduce/executorpool.py b/packages/pynumaflow/examples/reduce/asyncio_reduce/executorpool.py index 50d392cd..874f5163 100644 --- a/packages/pynumaflow/examples/reduce/asyncio_reduce/executorpool.py +++ b/packages/pynumaflow/examples/reduce/asyncio_reduce/executorpool.py @@ -1,4 +1,3 @@ -import aiorun import asyncio import concurrent.futures import os @@ -12,19 +11,18 @@ Message, Datum, Metadata, - AsyncReducer, + ReduceAsyncServer, + Reducer, ) _LOGGER = setup_logging(__name__) -class ReduceHandler: - def __init__(self, exec_pool=None): +class ReduceHandler(Reducer): + def __init__(self, exec_pool): self.exec_pool = exec_pool - async def reduce_handler( - self, key: list[str], datums: AsyncIterable[Datum], md: Metadata - ) -> Messages: + async def handler(self, key: list[str], datums: AsyncIterable[Datum], md: Metadata) -> Messages: """ handler function for executing operations on the messages received by the reduce vertex Here the function is used to execute certain blocking operations to demonstrate the use of @@ -32,13 +30,12 @@ async def reduce_handler( """ interval_window = md.interval_window tasks = [] - self.exec_pool.set_loop(asyncio.get_event_loop()) start_time = time.time() async for _ in datums: url = "http://host.docker.internal:9888/ping" - fut = threadPool.submit(blocking_call, url) + fut = self.exec_pool.submit(blocking_call, url) tasks.append(fut) - await threadPool.gather(tasks) + await self.exec_pool.gather(tasks) end_time = time.time() msg = ( f"batch_time:{end_time - start_time} interval_window_start:{interval_window.start} " @@ -60,9 +57,6 @@ class ExecutorPool: thread - Used to create a ThreadPool Executor process - Used to create a ProcessPool Executor - loop : asyncio event loop - the event loop on which the tasks should be submitted - max_workers : Number of workers in the executor pool the max number of threads/process to be created in the pool @@ -74,22 +68,18 @@ class ExecutorPool: async gather(tasks, return_exceptions, return_when) Wait for the coroutines to execute and gather the results - set_loop(event_loop) - Update the event loop associated with the executor - close(wait) Shutdown the executor """ - def __init__(self, exec_type="thread", max_workers=None, loop=None): - self.loop = loop or asyncio.get_event_loop() + def __init__(self, exec_type="thread", max_workers=None): if exec_type == "thread": self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) elif exec_type == "process": self.executor = concurrent.futures.ProcessPoolExecutor(max_workers=max_workers) def submit(self, func, *args): - return self.loop.run_in_executor(self.executor, func, args) + return asyncio.get_running_loop().run_in_executor(self.executor, func, args) async def gather(self, tasks, return_exceptions=False, return_when="ALL_COMPLETED"): completed, pending = await asyncio.wait(tasks, return_when=return_when) @@ -99,9 +89,6 @@ async def gather(self, tasks, return_exceptions=False, return_when="ALL_COMPLETE # self.tasks = [] return results - def set_loop(self, event_loop): - self.loop = event_loop - def close(self, wait=True): self.executor.shutdown(wait) @@ -126,15 +113,11 @@ def blocking_call(url): EXEC_TYPE is environment var to decide the type of executor to be used by the UDF """ if __name__ == "__main__": - e_type = os.getenv("EXEC_TYPE") - if e_type: - e_type = e_type.lower() + e_type = os.getenv("EXEC_TYPE", "thread").lower() mx_workers = os.getenv("MAX_WORKERS") if mx_workers: mx_workers = int(mx_workers) - threadPool = ExecutorPool(exec_type=e_type, max_workers=mx_workers) - handler = ReduceHandler(exec_pool=threadPool) - - grpc_server = AsyncReducer(handler=handler.reduce_handler) - aiorun.run(grpc_server.start()) + exec_pool = ExecutorPool(exec_type=e_type, max_workers=mx_workers) + grpc_server = ReduceAsyncServer(ReduceHandler, init_args=(exec_pool,)) + grpc_server.start() diff --git a/packages/pynumaflow/examples/reduce/asyncio_reduce/pyproject.toml b/packages/pynumaflow/examples/reduce/asyncio_reduce/pyproject.toml index fc4f8095..dd0e7d1c 100644 --- a/packages/pynumaflow/examples/reduce/asyncio_reduce/pyproject.toml +++ b/packages/pynumaflow/examples/reduce/asyncio_reduce/pyproject.toml @@ -1,12 +1,12 @@ [project] -name = "reduce-counter" +name = "asyncio-reduce" version = "0.2.4" description = "" requires-python = ">=3.13" dependencies = [ "pynumaflow", - "aiorun>=2025.1.1", "aiohttp>=3.8.4", + "requests>=2.32.0", ] [tool.uv.sources] diff --git a/packages/pynumaflow/examples/reduce/asyncio_reduce/uv.lock b/packages/pynumaflow/examples/reduce/asyncio_reduce/uv.lock index 74e0b07b..ae74a923 100644 --- a/packages/pynumaflow/examples/reduce/asyncio_reduce/uv.lock +++ b/packages/pynumaflow/examples/reduce/asyncio_reduce/uv.lock @@ -1,5 +1,5 @@ version = 1 -revision = 3 +revision = 5 requires-python = ">=3.13" [[package]] @@ -100,6 +100,23 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/fb/76/641ae371508676492379f16e2fa48f4e2c11741bd63c48be4b12a6b09cba/aiosignal-1.4.0-py3-none-any.whl", hash = "sha256:053243f8b92b990551949e63930a839ff0cf0b0ebbe0597b0f3fb19e1a0fe82e", size = 7490, upload-time = "2025-07-03T22:54:42.156Z" }, ] +[[package]] +name = "asyncio-reduce" +version = "0.2.4" +source = { editable = "." } +dependencies = [ + { name = "aiohttp" }, + { name = "pynumaflow" }, + { name = "requests" }, +] + +[package.metadata] +requires-dist = [ + { name = "aiohttp", specifier = ">=3.8.4" }, + { name = "pynumaflow", directory = "../../../" }, + { name = "requests", specifier = ">=2.32.0" }, +] + [[package]] name = "attrs" version = "25.4.0" @@ -711,7 +728,7 @@ wheels = [ [[package]] name = "pynumaflow" -version = "0.12.1" +version = "0.14.0" source = { directory = "../../../" } dependencies = [ { name = "aiorun" }, @@ -738,16 +755,16 @@ requires-dist = [ [package.metadata.requires-dev] dev = [ - { name = "black", specifier = ">=23.1" }, + { name = "black", specifier = ">=26.10.1" }, { name = "grpc-stubs", specifier = ">=1.53.0.6" }, { name = "grpcio-testing", specifier = ">=1.48.1" }, - { name = "mypy", specifier = ">=1.18.2" }, - { name = "pre-commit", specifier = ">=3.3.1" }, - { name = "pytest", specifier = ">=7.2.1" }, - { name = "pytest-cov", specifier = ">=3.0" }, - { name = "ruff", specifier = ">=0.0.264" }, - { name = "types-protobuf", specifier = ">=6.32.1.20250918" }, - { name = "types-psutil", specifier = ">=7.0.0.20251001" }, + { name = "mypy", specifier = ">=2.4.0" }, + { name = "pre-commit", specifier = ">=4.6.2" }, + { name = "pytest", specifier = ">=9.1.1" }, + { name = "pytest-cov", specifier = ">=7.1.0" }, + { name = "ruff", specifier = ">=0.17.0" }, + { name = "types-protobuf", specifier = ">=6.32.1.20260221,<7" }, + { name = "types-psutil", specifier = ">=7.2.2.20260906" }, ] docs = [ { name = "mike", specifier = ">=2.1" }, @@ -757,23 +774,6 @@ docs = [ { name = "mkdocstrings", extras = ["python"], specifier = ">=0.27" }, ] -[[package]] -name = "reduce-counter" -version = "0.2.4" -source = { editable = "." } -dependencies = [ - { name = "aiohttp" }, - { name = "aiorun" }, - { name = "pynumaflow" }, -] - -[package.metadata] -requires-dist = [ - { name = "aiohttp", specifier = ">=3.8.4" }, - { name = "aiorun", specifier = ">=2025.1.1" }, - { name = "pynumaflow", directory = "../../../" }, -] - [[package]] name = "requests" version = "2.32.5"