feat(api): add async output collector and thread-to-loop pump bridge - #89
Merged
Conversation
Introduces the consumer-side mailbox (OutputCollector/CollectorRegistry) and the pump thread that bridges engine-thread events onto the asyncio event loop via call_soon_threadsafe, laying groundwork for moving request handling off FastAPI's worker-thread pool.
There was a problem hiding this comment.
Pull request overview
Adds new asyncio-friendly primitives to collect per-request output events and a background thread “pump” to bridge engine-thread events onto the asyncio event loop via call_soon_threadsafe, plus accompanying unit tests and packaging/test configuration updates.
Changes:
- Introduce
OutputCollector/CollectorRegistryfor per-request buffering + async wake-ups. - Add
OutputPumpbackground thread to drain a sharedQueue[Event]and dispatch onto the loop thread. - Add unit tests under
tests/api/and updatepyproject.tomldependencies + pytest async configuration + package discovery.
Reviewed changes
Copilot reviewed 5 out of 5 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
server/api/collector.py |
Adds per-request output mailbox (OutputCollector) and routing registry (CollectorRegistry). |
server/api/pump.py |
Adds background thread pump to bridge engine-thread queue events to asyncio loop dispatch. |
tests/api/test_collector.py |
Adds unit tests validating collector buffering, ordering, timeouts, and shutdown failure fan-out. |
tests/api/test_pump.py |
Adds unit tests validating pump dispatch across thread/loop boundary and shutdown edge cases. |
pyproject.toml |
Pins transformers<5, adds protobuf + pytest-asyncio, enables pytest asyncio_mode=auto, and configures package discovery. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+60
to
+64
| if self._thread is not None: | ||
| raise RuntimeError("OutputPump.start() called twice") | ||
|
|
||
| self._loop = loop | ||
| self._thread = threading.Thread( |
Comment on lines
+107
to
+109
| collector = OutputCollector() | ||
| self._collectors[request_id] = collector | ||
| return collector |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Introduces the consumer-side mailbox (
OutputCollector/CollectorRegistry) and theOutputPumpbackground thread that bridges engine-thread events onto the asyncio event loop viacall_soon_threadsafe. This lays the groundwork for moving request handling off FastAPI's worker-thread pool so a waiting request becomes a suspended coroutine instead of a parked OS thread.server/api/collector.py:OutputCollector(per-request event buffer +asyncio.Event, coalesces bursts into a single wake-up) andCollectorRegistry(request_id→ collector routing, plusfail_allfor shutdown)server/api/pump.py:OutputPump, a single background thread that drains the engine's shared event queue and schedules dispatch onto the event loop viacall_soon_threadsafe, including a shutdown drain to guarantee delivery of final eventstests/api/pyproject.toml: pintransformers<5, addprotobufdependency, addpytest-asynciodev dependency withasyncio_mode = "auto", and configure package discovery forserver/kernels