Extract selector-based I/O loop helpers into a shared module#67175
Closed
jason810496 wants to merge 1 commit into
Closed
Extract selector-based I/O loop helpers into a shared module#67175jason810496 wants to merge 1 commit into
jason810496 wants to merge 1 commit into
Conversation
Pull the buffered socket reader and the selector dispatch loop out of ``WatchedSubprocess`` into ``airflow.sdk.execution_time.selector_loop``, together with a new ``make_raw_forwarder`` helper. ``WatchedSubprocess`` now delegates its event loop to ``service_selector`` and imports ``make_buffered_socket_reader`` from the shared module, so the same primitives can back other selector-driven bridges without copy/paste. Behavior is unchanged: the loop still clamps the select timeout to 0.01 s, treats EOF / ``BrokenPipeError`` / ``ConnectionResetError`` as "no more reads", invokes the per-socket ``on_close`` callback, and closes the socket. Re-exporting ``make_buffered_socket_reader`` from ``supervisor`` keeps the existing ``triggerer_job_runner`` import path working.
Contributor
There was a problem hiding this comment.
Pull request overview
This PR refactors selector-based I/O bridging utilities out of WatchedSubprocess into a reusable airflow.sdk.execution_time.selector_loop module, keeping supervisor.py’s behavior the same while improving reusability and typing.
Changes:
- Introduces a shared
selector_loop.pymodule withservice_selector,make_buffered_socket_reader, and a newmake_raw_forwarderhelper. - Updates
WatchedSubprocessto delegate its selector servicing to the sharedservice_selector. - Adds a dedicated test module covering the new selector-loop helpers (unit-style + small integration tests).
Reviewed changes
Copilot reviewed 3 out of 3 changed files in this pull request and generated 4 comments.
| File | Description |
|---|---|
| task-sdk/src/airflow/sdk/execution_time/selector_loop.py | Adds shared selector-loop helpers for buffered line reading, raw forwarding, and one-iteration selector dispatch. |
| task-sdk/src/airflow/sdk/execution_time/supervisor.py | Switches subprocess selector servicing to service_selector and imports buffered reader from the new shared module. |
| task-sdk/tests/task_sdk/execution_time/test_selector_loop.py | Adds tests for buffered reader, raw forwarder, and selector servicing behavior. |
Comment on lines
+403
to
+408
| sender.sendall(b"first line\nsecond line\n") | ||
|
|
||
| service_selector(sel, timeout=1.0) | ||
|
|
||
| assert b"first line\n" in received | ||
| assert b"second line\n" in received |
Comment on lines
+437
to
+442
| service_selector(sel, timeout=1.0) | ||
|
|
||
| dst_read.setblocking(False) | ||
| forwarded = dst_read.recv(4096) | ||
|
|
||
| assert forwarded == b"raw data payload" |
Comment on lines
+468
to
+472
| sender.close() | ||
| sender = None | ||
| service_selector(sel, timeout=0.5) | ||
|
|
||
| # on_close should have been called, and socket closed by service_selector |
Comment on lines
+49
to
+53
| # Sockets, even the `.makefile()` function don't correctly do line buffering on reading. If a chunk is read | ||
| # and it doesn't contain a new line character, `.readline()` will just return the chunk as is. | ||
| # | ||
| # This returns a callback suitable for attaching to a `selector` that reads in to a buffer, and yields lines | ||
| # to a (sync) generator |
Member
|
Why do we need this? |
Member
Author
I thought even with new |
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.
Why
Extract the
selector_loopmodularization from #65958 PR for further reusability and enhance the type annotation.This improvement should not be blocked by the AIP-108 vote.
What
Pull the buffered socket reader and the selector dispatch loop out of
WatchedSubprocessintoairflow.sdk.execution_time.selector_loop, together with a newmake_raw_forwarderhelper.WatchedSubprocessnow delegates its event loop toservice_selectorand importsmake_buffered_socket_readerfrom the shared module, so the same primitives can back other selector-driven bridges without copy/paste.The behavior should be unchanged after this refactor.
Was generative AI tooling used to co-author this PR?