Repository navigation
Conversation
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #40507 +/- ##
============================================
+ Coverage 56.16% 56.46% +0.29%
Complexity 2288 2288
============================================
Files 1124 1127 +3
Lines 178336 178889 +553
Branches 1489 1487 -2
============================================
+ Hits 100170 101012 +842
+ Misses 75645 75356 -289
Partials 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: |
|
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? |
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.