Repository navigation
Conversation
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #40507 +/- ##
=============================================
- Coverage 79.83% 56.60% -23.24%
- Complexity 0 2288 +2288
=============================================
Files 494 1131 +637
Lines 83575 179665 +96090
Branches 0 1487 +1487
=============================================
+ Hits 66722 101696 +34974
- Misses 16853 75448 +58595
- Partials 0 2521 +2521
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
There was a problem hiding this comment.
📋 Review Summary
This PR fixes a race condition in AsyncWrapper.teardown() where the wrapped DoFn's resources could be torn down while background thread-pool futures or asyncio tasks owned by that wrapper instance were still running or cleaning up after cancellation. The per-instance tracking of futures and asyncio tasks (while keeping the shared executor and event loop alive for other instances) and the accompanying unit and DirectRunner tests are well-structured.
🔍 General Feedback
- Consider guarding
teardown()againstAsyncWrapper._event_loopbeingNone(e.g., if_sync_fn.setup()fails before the event loop is started, or ifreset_state()is called prior toteardown()) and initializing_pending_futures/_pending_async_tasksin__init__as well assetup().
💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.
| def teardown(self): | ||
| """Forwards to the wrapped dofn's teardown method.""" | ||
| """Waits for this instance's work before tearing down the wrapped dofn.""" | ||
| if self._use_asyncio: |
There was a problem hiding this comment.
🟡 Medium: If teardown() is called when AsyncWrapper._event_loop is None (for example, if self._sync_fn.setup() raises an exception before the event loop thread is initialized in setup(), or if AsyncWrapper.reset_state() has already been called), asyncio.run_coroutine_threadsafe(self._wait_for_async_tasks(), AsyncWrapper._event_loop) will raise an AttributeError (and emit an unawaited coroutine warning for _wait_for_async_tasks()), preventing self._sync_fn.teardown() from running.
Additionally, initializing self._pending_futures = set() and self._pending_async_tasks = set() in __init__ as well as setup() (or using getattr) would avoid an AttributeError if teardown() or schedule_if_room() is called on an instance where setup() was not run.
| if self._use_asyncio: | |
| if self._use_asyncio and AsyncWrapper._event_loop is not None: |
There was a problem hiding this comment.
I checked the runner lifecycle behind this case. In the portable runner, setup runs during operation construction. The operation tree must finish construction before it becomes the tree that the bundle processor later shuts down. DirectRunner also calls setup while constructing the evaluator. I have not found a supported runner path that calls this wrapper's teardown after failed setup.
reset_state() has test call sites only in the current tree. It stops the shared event loop, so a bare None guard could proceed to wrapped cleanup without waiting for active work after that reset. I am leaving the implementation unchanged and this thread open for human review, rather than weakening the wait-before-cleanup behavior for that test-only transition.
|
Assigning reviewers: R: @damccorm for label python. This pull request likely touches a core component ("core" label). Please review with scrutiny. Note: If you would like to opt out of this review, comment Available commands:
The PR bot will only process comments in the main thread (not review comments). |
|
@dustin12 would you mind taking a first look at this one? |
|
R: @dustin12 |
|
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment |
On Python 3.10, Future.result raises concurrent.futures.TimeoutError, which is a different class from the builtin TimeoutError until Python 3.11. The teardown tests expected the builtin class, so they failed on Python 3.10, which setup.py still supports. The pull request checks did not show this because they run on Python 3.11 and later. Import the class from concurrent.futures, as model_manager_test.py does.
Master is at 2.79.0.dev and the release-2.78 branch already exists. A merge to master ships in 2.79.0, so the entry belongs in that section.
AsyncWrapper.teardown()can close the wrapped DoFn's resources while background work still uses them. A bounded DirectRunner pipeline reportsDONE, then a delayed write fails withValueError: I/O operation on closed file.This change tracks work owned by each wrapper and waits for it before tearing down the wrapped DoFn. Asyncio tasks stay tracked through cancellation cleanup, and the shared executor and event loop remain available to other wrappers.
The AsyncWrapper tests and DirectRunner reproductions passed locally in both thread-pool and asyncio modes. Teardown has no timeout, so hung processing or cleanup can delay shutdown. Pending-output delivery and timer behavior are unchanged.
Run the Python unit suite with:
addresses #123orfixes #456).CHANGES.mdwith noteworthy changes.See CONTRIBUTING.md for contribution guidelines.