Repository navigation
Free the data of failed S3 requests and clean up async copies at loop shutdown - #1192
Conversation
| # those of the errors chained to it, such as the ClientError | ||
| # that S3Core.call() raises an OSError from. clear_frames() | ||
| # skips this frame, which is still executing. | ||
| chained: list[BaseException | None] = [e] |
There was a problem hiding this comment.
Self-review round 1 (behavior, failure paths, resources, simplicity, tests). Base 7dd8ee410262f37fc1c6edd441625a624caa65a3, initial head b9bf9832, repaired head 39b5300bd882aa050e1c21d6e8f72adbb14bd664. Inputs: my own pass, /code-review high (6 findings), /simplify (4 angles).
Result: FINDINGS, repaired.
- A real S3 error is an OSError raised
froma ClientError, whose traceback frames (retry, tenacity, botocore) still held the part body afterclear_frames()of the top error only. Reproduced with a botocore Stubber AccessDenied: 8.1 MiB kept while the error was held, 0.1 MiB after the repair. Repaired in 78be56f (walk__cause__/__context__), simplified in 39b5300 (one walk frome;clear_frames()skips the executing frame ofrun()). Covered bytest_submit_to_failure_frees_arguments[...-True], which fails before the repair. run()deletes closure cells thatsettle()shares. Tried passing the arguments torun()throughsubmit()(the executor's own frames, reachable throughf_back, kept them: test failed) and bindingsettlewithfunctools.partial(the submission's done callback kept the future: file not freed). Kept the shared cells;settle()reads the future only whenrun()did not run, which the comment states. The# noqa: F821is a ruff false positive for the else branch after adelin the except branch.
| self.commit() | ||
| return not final | ||
|
|
||
| def _submit_upload(self, writer: S3MultipartWriter, part_number: int, data: bytes) -> None: |
There was a problem hiding this comment.
Round 1, finding 3 (/code-review): a single write of several parts kept the last submitted part in the part local of _upload_chunk(), which is in the traceback when a later wait of the same call raises that part's error, so the file stayed in a cycle (reproduced: write(b'a' * 12) with block_size=4, max_workers=1 → file not freed). Repaired first with del part (78be56f), then, after the altitude review, by submitting from _submit_upload() on the file (39b5300), so the structure keeps the future out of the caller's frame. Same for AioS3StreamedFile._submit_upload() (part and waiter). Covered by test_failed_request_freed_by_reference_counting[write] in both classes, which fail before the repair.
| # interrupted. The writer records it before resolving the future. | ||
| self._close_without_commit(creation=creation) | ||
| # Not kept by this frame, which the traceback of its error keeps. | ||
| del creation |
There was a problem hiding this comment.
Round 1, finding 4 (own pass while repairing): a failed CreateMultipartUpload also kept the file in a cycle: the creation local of _initiate_upload() here, and AioS3StreamedFile._creation, which _discard() now drops. Covered by test_failed_request_freed_by_reference_counting[creation] in both classes, which fail before the repair.
| future.result() | ||
| # Raised without waiting for the parts before it, so that the | ||
| # parts queued after it are cancelled. | ||
| raise_failure(done) |
There was a problem hiding this comment.
Round 1, finding 5 (/code-review), deferred: if the abort in discard() also fails, discard() raises before it resets multipart_upload_parts, so the file keeps the failed future and stays in a cycle until the garbage collector runs (reproduced). Not changed: the parts are kept on purpose. A later discard() retries the abort, and a later commit() must still see the parts, or it would upload the buffer with PutObject. This needs a double failure, and the garbage collector still frees it. AioS3StreamedFile._discard() resets unconditionally because it logs the abort failure and has no retry or second commit (flush(force=True) cannot run twice).
| # for the semaphore do not start any more. | ||
| await asyncio.gather( | ||
| *[asyncio.wrap_future(c) for c in part_copies if not c.cancel()], | ||
| *tasks, |
There was a problem hiding this comment.
Round 1, finding 6 (/code-review): on a cancellation, _abort() waited for the part tasks before it cancelled the part copies, so copies queued in the filesystem's pool, which bounds all its copies, ran before the abort. It now cancels the copies that have not started, then waits for the running ones and the tasks in one gather (78be56f, merged in 39b5300). Covered by test_copy_object_with_multipart_upload_cancelled_queued_parts, which fails before the repair.
| future.result() | ||
| # Raised without waiting for the parts before it, so that the | ||
| # parts queued after it are cancelled. | ||
| raise_failure(done) |
There was a problem hiding this comment.
Round 1, /simplify (altitude): _finish_multipart_upload() kept the loop that raise_failure() replaced in the commit, so the codebase had two idioms for one check; converted (39b5300). The copy paths still form cycles through their futures locals, but these hold no written data; the PR body lists them as unchanged. Other /simplify results: reuse found no duplicates; efficiency items (wrapping finished copies in _abort, the two scans in raise_failure, a fixed 0.1 s sleep that sibling tests also use) were skipped as negligible; inlining copied in _upload_part was skipped because mypy needs it to infer the part type.
| kept. A failed part copy aborts the multipart upload. So does an interrupt, or the | ||
| cancellation of an `AioS3FileSystem` copy, unless the upload has already completed. A | ||
| cancellation of an `AioS3FileSystem` copy, including one by the shutdown of the event | ||
| loop at the end of `asyncio.run()`, unless the upload has already completed. A |
There was a problem hiding this comment.
Round 1, finding 7 (own pass): the copy cancellation paragraph did not cover the shutdown of the event loop, which this PR now handles; added (b9bf983). raise_failure() is public, so it is listed in docs/api/filesystem.rst with the other executor symbols, and its summary line now states that it empties the set (39b5300).
| self.shutdown(wait=True) | ||
|
|
||
|
|
||
| def raise_failure(done: set[Future[Any]] | set[asyncio.Future[Any]]) -> None: |
There was a problem hiding this comment.
Self-review round 2 (claims, callers, AWS operation, docs, evidence). Base 7dd8ee410262f37fc1c6edd441625a624caa65a3, head 39b5300bd882aa050e1c21d6e8f72adbb14bd664. Result: FINDINGS in the PR description, corrected; no code change.
Claims checked:
- Memory figures, re-measured at this head on Python 3.15 without
gc.collect(): Failed parts of an S3File multipart write keep their data and do not stop the write #1173 reproduction 16.2 MiB held / 0.2 MiB dropped (master 32.2 / 32.2);open_async()0.2 andAioS3File0.4 MiB after the error is dropped (master 25.2 and 50.5, file kept); botocore Stubber AccessDenied for an 8 MiB part throughsubmit_to(): 0.1 MiB while the error is held (master 8.1 MiB, kept by theClientErrorcause). The description now says the 8.1 MiB is master's. - Release-note scope:
submit_to()has 8 call sites; besides the files,S3FileSystem._copy_object_with_multipart_upload()uses it for the creation and the part copies, so its tracebacks also lose worker-frame locals. The description listed only the files; corrected. - Existing callers:
submit_to()keeps its signature and outcomes; a customS3Executorstill only implementssubmit()/shutdown().raise_failure()keeps the previous priority (a failed future before a cancelled one; the commit used to raise the cancellation from itsresult()list instead, with the same exception type)._submit_upload()does not collide with fsspec names. - Evidence: the
raise_failuretests cover a new function, so they are now listed apart from the regression tests that were checked to fail by reverting the source. Python 3.11 and 3.13 counts re-run at this head: 69 passed each (the description said 68 before the copy stage was added).
| # its request keeps running, so that _abort() can wait for an upload | ||
| # that is created after a cancellation; scheduled right before the | ||
| # try, so that nothing can fail between the two. | ||
| creation = asyncio.wrap_future( |
There was a problem hiding this comment.
Round 2, AWS operator view: when the shutdown of the event loop cancels a copy, the cleanup now waits for the running CreateMultipartUpload, UploadPartCopy (up to the part size each), and CompleteMultipartUpload requests, so asyncio.run() returns only after they end; a cancellation of the awaited task (for example Ctrl-C, which cancels the main task) already waited since #1056. The requests sent decrease: queued part copies are cancelled, and no AbortMultipartUpload races a completion. The description now states the later return of asyncio.run() in the release notes. Not covered: a second interrupt during the shutdown, which stops run_until_complete() and the cleanup, as before.
| kept. A failed part copy aborts the multipart upload. So does an interrupt, or the | ||
| cancellation of an `AioS3FileSystem` copy, unless the upload has already completed. A | ||
| cancellation of an `AioS3FileSystem` copy, including one by the shutdown of the event | ||
| loop at the end of `asyncio.run()`, unless the upload has already completed. A |
There was a problem hiding this comment.
Round 2, documentation reader: docs/filesystem.md (this paragraph) and docs/aio.md (threads section, comparison table, async file handles) still match the behavior: part copies in flight finish first, queued ones do not start, and the copy runs one asyncio task per part. No other page describes garbage collection of failed writes or traceback contents.
| cleanup.add_done_callback(_cleanup_tasks.discard) | ||
| await asyncio.shield(cleanup) | ||
| # the cleanup. | ||
| await self._thread_pool.run_shielded(_abort) |
There was a problem hiding this comment.
Independent review (relayed Codex result). Reviewer: Codex CLI 0.162.1, model gpt-6-astra, codex exec -s read-only, session 01a14035-f02c-7e43-bb77-9d5ebe34f6b0. Static review of a clean detached snapshot of 39b5300bd882aa050e1c21d6e8f72adbb14bd664 against base 7dd8ee410262f37fc1c6edd441625a624caa65a3; prompt without PR text or prior findings; no edits, builds, tests, or network. Covered: the diff, surrounding source, conventions, docs; submit_to()/settle() ordering and reference lifetimes across the three executors; raise_failure(); creation, uploads, append copies, commit, and abort of the three file types; async copy cancellation, repeated cancellation, loop shutdown; test sensitivity and Python declarations. Result: FINDINGS (4 × P2, 1 × P3). Snapshot and PR worktree were unchanged afterwards.
Finding 1 (P2): "Shutdown can cancel cleanup that has already started. Cancel a copy while a request is running, let _abort() start waiting, then return from main(). asyncio.run() cancels the existing cleanup task directly; shield(cleanup) only prevents cancellation propagated through its waiter. Cleanup exits while the request continues, leaving the upload without an abort."
Verified: reproduced (['create'], no abort); master behaves the same. Repaired in 8b69b92 by the maintainer's choice: the cleanup runs in a thread of the filesystem through run_shielded(), waiting for the requests as concurrent futures, so the event loop cannot stop it (_cleanup_tasks removed); the creation and the part copies that have not started are cancelled before the cleanup is scheduled. Reproduction now ['create', 'abort']; an abort that ends after asyncio.run() returns still completes before the interpreter exits. Covered by test_copy_object_with_multipart_upload_loop_shutdown[True-create|part], which fail before the repair.
| future.result() | ||
| # Raised without waiting for the parts before it, so that the | ||
| # parts queued after it are cancelled. | ||
| raise_failure(done) |
There was a problem hiding this comment.
Independent review, finding 2 (P2): "Abort failure preserves the synchronous file's reference cycle. If an upload part fails and AbortMultipartUpload also fails, discard() exits before clearing the future collections ... Retain the upload identity for retry, but release settled futures even when abort fails."
Verified (same as round 1 finding 5) and deferred: releasing the parts while keeping the upload would make a later commit() take the not self.multipart_upload_parts branch and upload the buffer with PutObject, and changing that guard widens the contract of a double-failure path that the garbage collector still frees.
| seen.add(id(error)) | ||
| traceback.clear_frames(error.__traceback__) | ||
| chained += [error.__cause__, error.__context__] | ||
| if isinstance(error, BaseExceptionGroup): |
There was a problem hiding this comment.
Independent review, finding 3 (P2): "Exception-group members retain argument data. A callable can collect an exception from a helper receiving body, leave the except block, then raise ExceptionGroup(\"failed\", errors). The group need not have a cause or context."
Verified and repaired in 01d870a: the walk also follows BaseExceptionGroup.exceptions. Covered by test_submit_to_failure_frees_arguments[*-group], which fail before the repair.
| def fail(*args, **kwargs): | ||
| raise PermissionError("Access Denied") | ||
|
|
||
| def fail_copy(**kwargs): |
There was a problem hiding this comment.
Independent review, finding 4 (P2): "The append reference-counting test has a setup race. For stage=\"copy\", the mocked copy fails immediately. If the worker finishes before the initial write checks _copies_in_flight, that setup write raises outside pytest.raises."
Verified by reading _upload_chunk() and repaired in 01d870a: the copy is held until the first write returns, as in test_append_failed_part_copy.
| self.shutdown(wait=True) | ||
|
|
||
|
|
||
| def raise_failure(done: set[Future[Any]] | set[asyncio.Future[Any]]) -> None: |
There was a problem hiding this comment.
Independent review, finding 5 (P3): "raise_failure() leaves additional asyncio exceptions unretrieved. Given two failed asyncio futures owned only by done, next() retrieves one exception and done.clear() releases the other without retrieving its exception."
Verified: no current caller logs, because the waiters of AioS3StreamedFile retrieve their errors in a done callback, but the public function should not rely on that. Repaired in 01d870a with one pass that retrieves every error (which also fixed an unbound loop variable on an empty set). Covered by test_raise_failure_retrieves_errors, which fails before the repair.
| cleanup.add_done_callback(_cleanup_tasks.discard) | ||
| await asyncio.shield(cleanup) | ||
| # the cleanup. | ||
| await self._thread_pool.run_shielded(_abort) |
There was a problem hiding this comment.
Self-review of the independent-review repair (rounds 1 and 2 perspectives), git range-diff 7dd8ee41..39b5300b 7dd8ee41..8b69b92034cd1acaa2dfa49913d87f9737c9dcc1: commits 1-4 unchanged, new 01d870a and 8b69b92; base unchanged (7dd8ee410262f37fc1c6edd441625a624caa65a3). Result: CLEAN after the repairs above.
Behavior (round 1):
- Thread cleanup cannot deadlock the filesystem pool:
S3AioThreadPoolstarts calls in submission order, and the cleanup waits only for its copy's requests, which were submitted before it; the queued ones are cancelled on the event loop before the cleanup is submitted, and no part task submits afterfailed = True, which is set in the same step without an await. - Errors stay retrieved: the shield's inner wrappers of the creation and completion are retrieved by
shield, the part tasks by their own done callbacks (previously by the cleanup'sgather), and_abort_multipart_upload()logs instead of raising. - A creation that fails, a part that fails, a completion that fails, and the plain and repeated cancellations keep their previous outcomes: the existing copy tests pass (
-k copy_object_with_multipart_upload, 40 passed), and the full filesystem suite passes (1295 passed, 9 skipped). - The creation that has not started is now cancelled, as
S3FileSystem._copy_object_with_multipart_upload()does withcreation.cancel().
Claims (round 2):
- The description now states that a copy cancelled before the shutdown lets
asyncio.run()return while the abort runs; checked that the abort still completes before the interpreter exits (concurrent.futuresjoins the pool threads at exit). docs/filesystem.md("the part copies in flight finish first") and the method docstring were checked against the new behavior; the docstring now says the cleanup runs in a thread and cancels the requests that have not started.- Python 3.11 and 3.13: 109 related tests passed each; the new tests 5 times on 3.15.
| multipart_upload = await asyncio.shield(creation) | ||
| # A cancellation does not cancel the creation, so that _abort() | ||
| # can wait for an upload that it creates. | ||
| multipart_upload = await self._thread_pool.wait_shielded(creation) |
There was a problem hiding this comment.
Independent follow-up review (relayed Codex result). Reviewer: Codex CLI 0.162.1, model gpt-6-astra, codex exec -s read-only, session 01a141e7-6db0-7d02-a814-848a94f42b25. Static review of a clean detached snapshot of 8b69b92034cd1acaa2dfa49913d87f9737c9dcc1: the new commits 39b5300b..8b69b920, with the range-diff showing the earlier commits unchanged (base 7dd8ee410262f37fc1c6edd441625a624caa65a3). Covered: exception-group clearing, raise_failure() and its callers, shared-pool ordering ("No pool dependency cycle found: cleanup waits only for earlier submissions, and failed prevents further part submissions before cleanup starts"), cancellation, creation/part/completion/annotation failures, loop shutdown/stop/close, pool shutdown and fork, Python 3.11-3.15 APIs, both docs pages, and the changed tests. Result: FINDINGS (1 × P2). Limits stated by the reviewer: runtime behavior, fork/interpreter exit, and scheduling under load unverified; the cleanup survives loop stop/close once submitted, not if the loop stops before the cancellation handler runs.
Finding (P2): "Creation and completion errors can become unretrieved asyncio exceptions after cancellation. Cancel a copy while creation or completion is running, let cancellation callbacks execute, then have that request fail while the loop remains running. Cancellation detaches shield() from the pending asyncio wrapper. Cleanup now calls .exception() on the underlying concurrent future; that does not retrieve the exception subsequently stored in the asyncio wrapper."
Verified: reproduced on 3.11/3.13 ("Future exception was never retrieved") and 3.15 ("OSError exception in shielded future"); master logs nothing, so this was a regression of 8b69b92. Repaired in a5f4d5e/ef7bfd65 with the public S3AioThreadPool.wait_shielded(future), which waits with asyncio.wait() and retrieves the outcome; shield() on Python 3.14+ logs such a failure even when it is retrieved, so it cannot be used. The copy and run_shielded() use it. Covered by test_wait_shielded_cancelled and new log assertions in test_copy_object_with_multipart_upload_cancelled_creation|completion[True], which fail before the repair on 3.11 and 3.15.
| # Unlike shield(), which on Python 3.14 and later also logs the | ||
| # failure of a future that ends after a cancellation, wait() | ||
| # leaves the future alone when the waiting task is cancelled. | ||
| await asyncio.wait([waiter]) |
There was a problem hiding this comment.
Self-review of the follow-up repair (both perspectives), 8b69b920..ef7bfd65, base unchanged. Result: CLEAN after one correction.
- Behavior:
wait_shielded()raisesCancelledErroron a cancellation of the waiting task asshield()did, leaves the call running, and returns or raises the call's outcome otherwise; it deletesfutureandwaiterinfinally, because the traceback keeps its frame (the reference-counting test of theAioS3StreamedFilecommit failed untilfuturewas deleted too).run_shielded()keeps its contract. - Claims: the comment said
shield()logs "since Python 3.15"; the 3.14.5 source has the same_log_on_exception, so it now says 3.14 and later (ef7bfd6), and the description says so. Related tests: 110 passed on Python 3.11, 3.13, and 3.14 each. Full filesystem suite on 3.15: 1296 passed, 9 skipped in three consecutive runs; one earlier run at the same code had one failure that was not captured and did not recur, which the description records.
| release.set() | ||
| # The worker has run the callbacks of the call once it is shut | ||
| # down, and the event loop runs theirs on the next iterations. | ||
| await asyncio.to_thread(pool.shutdown, True) |
There was a problem hiding this comment.
Independent follow-up review 2 (relayed Codex result). Codex CLI 0.162.1, model gpt-6-astra, codex exec -s read-only, session 01a14212-3565-7000-a754-34dabb368cc7; static review of a clean snapshot of ef7bfd65 for 8b69b920..ef7bfd65 with the range-diff (earlier commits unchanged). Covered: wait_shielded()/run_shielded(), multipart creation, completion and abort, the AioS3StreamedFile commit/discard callers, cancellation semantics, traceback references, tests, docs. Result: FINDINGS (1 × P3); "No actionable production regression or newly introduced failed-future traceback cycle found." Limits: the finding was not reproduced; runtime behavior and the exact version boundary of the shield() logging were not verified by the reviewer.
Finding (P3): "The logging assertions can run before the original wrapper's callbacks. A worker can mark running finished before invoking its callbacks ... Neither gc.collect() nor one sleep(0) guarantees that callback ran ... a regression restoring the late logging can escape detection." (also the two copy tests)
Verified by reading concurrent.futures (state is set before _invoke_callbacks()), and repaired in 9651656: the tests shut the pool down with wait=True before checking the log, drop the futures that keep the asyncio wrapper, and run three loop iterations. Mutation check (wait_shielded() back to shield() without retrieval): the two copy tests fail on 3.11 and 3.15; this unit test fails on 3.11 and passes on 3.15, where shield() did not log in its setup either (checked with a plain ThreadPoolExecutor script), so the copy tests are the guard there. The changed tests passed 10 consecutive runs with -n 8. A third follow-up of this test-only commit is running.
| await asyncio.to_thread(started.wait, EVENT_TIMEOUT) | ||
| # Lets the task start waiting before it is cancelled. | ||
| deadline = time.monotonic() + EVENT_TIMEOUT | ||
| while not running._done_callbacks: |
There was a problem hiding this comment.
Independent follow-up reviews 3-5 (relayed Codex results), Codex CLI 0.162.1, model gpt-6-astra, codex exec -s read-only, static reviews of clean detached snapshots, each of the newest test-only commit:
- Follow-up 3, session
01a14225-87ba-7731-9871-b78f7459153e,ef7bfd65..96516569: FINDINGS (1 × P2): "The executor test can still skip exercising the waiter ...waiting.cancel()cancels the task beforewait_shielded()starts. A regression to bareasyncio.shield()passes because no waiter was created ... This setup race predates this commit." Also: "For an established waiter, the new shutdown, reference release, collection, and loop turns close the reported late-callback observation gap." Verified: with a barrier, theshield()mutant is now caught on 3.15 too (it was not before). Repaired in 9dd4f91. - Follow-up 4, session
01a14239-a6b8-7821-a0d2-81bb95aeecbc,96516569..9dd4f912: FINDINGS (1 × P2): "The barrier can hang indefinitely ... if the event-loop thread is delayed until the worker's 10-secondrelease.wait(EVENT_TIMEOUT)expires ...add_done_callback()invokes the callback immediately without appending it to_done_callbacks." Verified by readingconcurrent.futures; repaired in 4ed748e (worker waits without a timeout, deadline assertion on the barrier, release and pool shutdown infinally). - Follow-up 5, session
01a14243-b22d-7722-9dd8-0bf487d4f4ae,9dd4f912..4ed748e1: CLEAN. "Barrier and cancellation-assertion failures enterfinally, which releases the worker before joining the pool ... Deleting references, collecting garbage, and draining callbacks preserves detection of bareasyncio.shield(asyncio.wrap_future(future)): older Python versions report the unconsumed exception; Python 3.14-3.15 explicitly log the shielded failure."
Author validation of these repairs: the shield() mutant fails test_wait_shielded_cancelled on 3.11 and 3.15 and the copy cancellation tests on both; tests/pyathena/filesystem/test_s3_executor.py passed 10 consecutive runs with -n 8 on 3.15 and once each on 3.11, 3.13, and 3.14. Both self-review perspectives for these test-only commits: no production code or documented claim changed; the description's test list was updated.
4ed748e to
a8ee49b
Compare
laughingman7743
changed the title
S3Executor.submit_to() clears the frames below the callable and the references of its own frame before it sets the error, so the body of a failed part is freed when the part fails. raise_failure() raises the error of a finished future after the caller's set is emptied, and S3AioThreadPool.run_shielded() and AioS3StreamedFile._discard() drop their references to failed futures, so that an S3File, an AioS3File or an AioS3StreamedFile whose part failed is freed with its data by reference counting once the error is dropped. The async multipart copy of AioS3FileSystem keeps the creation and the completion as futures instead of tasks, and keeps the request of each part, so that when the shutdown of the event loop at the end of asyncio.run() cancels the copy, the cleanup still waits for them and aborts an upload created during the cancellation, after its running parts have ended, and not during a completion. Closes #1186 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
- submit_to() also clears the frames of the errors chained to a failed callable's error, such as the ClientError that S3Core.call() raises an OSError from, whose frames held the body of the part. - _upload_chunk() of S3File and AioS3StreamedFile no longer keeps the last submitted part in its frame, which kept a write of several parts in a cycle with the error of a failed part. - A failed creation no longer keeps the file in a cycle: S3File drops it from the frame of _initiate_upload(), and AioS3StreamedFile drops it in _discard(). - The cleanup of an async multipart copy cancels the part copies that wait for a thread before it waits for the parts and aborts. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
- S3File and AioS3StreamedFile submit an uploaded part from _submit_upload(), so that no frame of _upload_chunk() keeps the future of a part whose error a later wait raises, instead of deleting it. - S3FileSystem._finish_multipart_upload() raises a failed part copy with raise_failure(), as the commit of a file does. - submit_to() clears the frames of the error and its chained errors in one walk; clear_frames() skips the frame of run(), which is executing. - The cleanup of an async multipart copy waits for the part copies and their tasks together. - The reference counting test of S3File also covers a write after a failed part copy of an append. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
- submit_to() also clears the frames of the errors that an exception group of the callable holds. - raise_failure() retrieves the errors of all the failed futures in one pass, so that asyncio does not log those that it does not raise, and no longer fails on an empty set. - The append case of the reference counting test holds the part copy until the first write has returned, as the other append tests do. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
The shutdown of the event loop at the end of asyncio.run() cancels every remaining task, including the cleanup task that a cancellation of the copy has started, which then returned without aborting the upload. The cleanup now runs in a thread of the filesystem through run_shielded(), waiting for the requests as concurrent futures, so the event loop can no longer stop it, and _cleanup_tasks is no longer needed. The creation and the part copies that have not started are cancelled before the cleanup is scheduled, as S3FileSystem does, and the part tasks retrieve their errors themselves. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
After a cancellation, shield() leaves the asyncio wrapper of a creation or completion that fails later unretrieved, and Python 3.15 also logs such a failure itself. The new S3AioThreadPool.wait_shielded() waits for a submitted call with asyncio.wait(), which leaves the call alone when the waiting task is cancelled, and retrieves its outcome. The async multipart copy and run_shielded() use it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
The tests of a late failure after a cancellation shut the pool down before they check the event loop's log, so that the worker has run the callbacks of the failed request, and drop the futures that keep its asyncio wrapper. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
test_wait_shielded_cancelled cancels the waiting task once it has added its callback to the future of the call, so that the cancellation always reaches wait_shielded(); otherwise a regression to shield() could pass on Python 3.15. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
The call waits for the test without a timeout, so it cannot finish before the task waits for it, and the test releases the call and shuts the pool down in a finally block. The wait for the task to add its callback fails after EVENT_TIMEOUT instead of looping forever. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
- S3File._fetch_range() waits for the parallel reads with FIRST_EXCEPTION, cancels the reads that have not started when one fails, and drops the futures of the others before raising, instead of keeping them in its frame in a cycle with the error. - AioS3StreamedFile._fetch_range() does the same with its read tasks, which gather() left running after a failure, and close() drops the cache of a read file, as fsspec does, which kept the file in a cycle. - S3AioExecutor.submit() clears the variables of its run() closure when the function fails, as ThreadPoolExecutor's _WorkItem does, and returns the future through a variable that run() does not clear. - retry_api_call() clears the frames of tenacity in the traceback of a raised error; their retry state kept the error, and so every frame of its traceback, in a reference cycle. - raise_failure() accepts a set of any kind of future, such as tasks. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
- S3File.discard() drops the parts once they have ended, before the abort, so that an abort failure, which keeps the upload for a retry, no longer keeps the errors of failed parts and so the file in a reference cycle. - commit() does nothing for a file whose upload discard() kept, instead of uploading the buffer with PutObject. - A failed multipart commit drops the buffer, which holds only the data of the last part; a later commit() uploaded it alone with PutObject. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
- fetchmany() and fetchall() of the result set and its aio variant return the rows that they took before a failed page request to the front of the buffer, instead of dropping them while counting them in rownumber, so that a later fetch returns them. - The Arrow result set skips record batches without rows, which ended the fetches early, such as the batch of an empty result file. - The size clamp to sys.maxsize moves from the base fetchmany() into the islice() of the pandas and Polars result sets, which need it. - The test helper _page is replaced with get_query_results_responses(). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Keep the fixes of root causes and drop the per-site breaking of reference cycles, which only made files and caller buffers collectable without the garbage collector at the cost of an invariant that every caller had to keep: - A private _S3WorkItem owns the future, the callable, and its arguments of S3Executor.submit_to() and S3AioExecutor.submit(), and drops them when it runs or settles, instead of closures whose shared cells run() deleted. - S3Core.call() clears the frames below it in the traceback of a raised error, those of the retries and of botocore, which kept the request and, in tenacity's retry state, the error in a cycle; this replaces walking the chained errors in submit_to() and clearing tenacity frames in retry_api_call(). - raise_failure(), _submit_upload(), the deletions of locals, the resets of AioS3StreamedFile._discard(), the dropping of the parts in discard() with the guard of commit(), and the cache reset of close() are removed. - A failed multipart commit() also closes the file, and the cleanup of an AioS3FileSystem copy also cancels a completion that has not started. - The result set fixes move to a separate change. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
7330c5b to
e466248
Compare
laughingman7743
changed the title
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
- S3File._fetch_range() also cancels the reads and drops their futures when the wait for them is interrupted, as AioS3StreamedFile does. - A failed multipart commit() closes the file with _close_without_commit() instead of repeating it. - The comments of AioS3StreamedFile._fetch_range() state that cancelling a failed read keeps asyncio from logging its error. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
| T = TypeVar("T") | ||
|
|
||
|
|
||
| class _S3WorkItem(Generic[T]): |
There was a problem hiding this comment.
Self-review round 1 of the narrowed PR (behavior, failure paths, resources, simplicity, tests). Base b2d7aa24b00a204e9e150952c708e3bb1126f9e2, head 14b7ab38cecb88e53a6b8874dda97b83c9112d72. The maintainer narrowed the PR after a design consultation (Claude claude-fable-5-1 on the max profile, effort high, and Codex gpt-6-astra, both read-only): the per-site cycle breaking, raise_failure(), the tenacity frame clearing in retry_api_call(), and the discard/commit guard were removed; _S3WorkItem replaces the closures; S3Core.call() clears the request frames; the result-set fixes moved to #1222; the write-session design is #1211. Inputs of this round: my pass over the full diff and /code-review high (6 findings, verified below). Result: FINDINGS, 2 repaired, 1 rejected as a false positive, 3 deferred with reasons.
Verified in my pass: _S3WorkItem claims the future once (lock) for run() and settle(), drops the future, callable, and arguments on every path, and run() keeps no local referencing the future while its error propagates; the S3AioExecutor path now also clears the frames below a failed function, which only removes locals of request frames.
| # The failure of the first read to finish is raised. | ||
| future.result() | ||
| try: | ||
| done = wait(futures, return_when=FIRST_EXCEPTION).done |
There was a problem hiding this comment.
Round 1, /code-review finding (P2-like): S3File._fetch_range() cancelled and dropped the other ranges only on a failure, not when the wait was interrupted (Ctrl-C), so the interrupt's traceback (e.g. sys.last_exc in a REPL) kept every range's data. Repaired in 14b7ab3 with the same except BaseException as AioS3StreamedFile._fetch_range(). Covered by test_interrupted_read_frees_ranges, which fails before the repair.
Rejected: the finding that a cancelled AioS3StreamedFile._fetch_range() leaves the errors of failed read tasks unretrieved. Future.cancel() clears the log flag even on a finished future (checked on 3.11 and 3.15: an unretrieved failed task logs "Task exception was never retrieved" unless it is cancelled), so cancelling them already suppresses the log; the comments now say so.
| f"Failed to abort multipart upload {upload_id} " | ||
| f"to s3://{self.bucket}/{self.key}." | ||
| ) | ||
| self._close_without_commit() |
There was a problem hiding this comment.
Round 1, /code-review cleanup: the failed-commit branch repeated _close_without_commit() (drop the buffer, close, discard, log, shut down the executor) with small differences. Repaired in 14b7ab3 by calling _close_without_commit(); test_commit_failure_and_discard covers the closed file and the no-op second commit.
| cleanup.add_done_callback(_cleanup_tasks.discard) | ||
| await asyncio.shield(cleanup) | ||
| # the cleanup. | ||
| await self._thread_pool.run_shielded(_abort) |
There was a problem hiding this comment.
Round 1, /code-review findings deferred with reasons:
- If the filesystem's pool is shut down during a copy,
run_shielded(_abort)raisesRuntimeErrorin theexceptblock, hiding the original error, and the upload is not aborted. This is the same on master (its cleanup task also submitted the abort to the pool). An inline fallback would block the event loop for as long as the running part copies take, so it is not taken here. - The cleanup holds a pool thread while it waits for the running part copies, which reduces the threads available to other calls during a mass cancellation. It cannot deadlock (it waits only for requests submitted before it; queued ones are cancelled first). An event-driven cleanup chained on the request futures would avoid holding a thread; it belongs with the request ownership of Give multipart writes a single owner of their parts and an explicit state #1211.
| raise S3ClientError(e).os_error from e | ||
| except BaseException as e: | ||
| # clear_frames() skips this frame, which is still executing. | ||
| traceback.clear_frames(e.__traceback__) |
There was a problem hiding this comment.
Round 1, /code-review finding deferred: clearing the frames below call() on every error also runs for expected misses (404 from exists()/info()) and removes the locals of request frames from every S3 error, also without an executor. This is the approved design: call() is the single boundary of S3 requests, which covers direct requests (PutObject of small files, single-range reads) and breaks tenacity's retry-state cycle at its source; the cost is a walk of the frames of an error that already made a network request. The release notes state the debugging change.
| OSError: The translation of an S3 error response (see | ||
| :class:`~pyathena.filesystem.s3_errors.S3ClientError`). | ||
|
|
||
| Note: |
There was a problem hiding this comment.
Self-review round 2 of the narrowed PR (claims, callers, AWS operation, docs, evidence). Base b2d7aa24b00a204e9e150952c708e3bb1126f9e2, head 14b7ab38cecb88e53a6b8874dda97b83c9112d72. Result: FINDINGS in the description, corrected; no code change.
Claims checked against measurements at this head (Python 3.15, without gc.collect()): an upload_part answered with AccessDenied by a botocore Stubber keeps 0.1 MiB while its error is held (master 8.1 MiB); a parallel read of 4 × 8 MiB ranges whose first range fails keeps 0.2 MiB (master 24 MiB). The description's statement that files and caller buffers are left to the garbage collector matches: the #1173 reproduction still keeps 16 MiB (the buffer and the next part) until gc.collect().
Corrections: the release note and the description of the read path now include an interrupted wait (14b7ab3), the full suite count is 1286, the related tests are 94 on 3.11, 3.13, and 3.14, and test_interrupted_read_frees_ranges is listed.
Callers and operation: submit_to()/submit() keep their signatures and outcomes; S3Core.call() raises the same exception types; no request is added (reads that have not started are cancelled, and an AioS3FileSystem copy cancels its queued requests). docs/filesystem.md states the copy cleanup at loop shutdown, the cancellation of requests that have not started, and the closed file after a failed multipart commit; the docstrings of S3Core.call(), submit_to(), _fetch_range(), commit(), and the async copy match the code. Not verified: behavior under real S3 errors other than the stubbed AccessDenied; the AWS jobs run when the PR is Ready.
- An interrupted or cancelled parallel read also clears the frames of the wait in the traceback, whose collections kept the finished reads. - The docstring of submit_to() states that the errors that a callable's error was raised from keep their frames, and that S3Core.call() clears those of a request's ClientError. - The interrupted-read tests wait as concurrent.futures.wait() and asyncio.wait() do, with the futures in their frames; the async read test waits for the threads of the file; the queued-completion test waits for the submission of the completion itself. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
| # The reads are cancelled with the read of the range, as below, | ||
| # and the frames of the wait, which keep their futures, are | ||
| # cleared; clear_frames() skips this frame, which is executing. | ||
| traceback.clear_frames(e.__traceback__) |
There was a problem hiding this comment.
Independent review of the narrowed PR (relayed results). Reviewed snapshot 14b7ab38cecb88e53a6b8874dda97b83c9112d72 against base b2d7aa24b00a204e9e150952c708e3bb1126f9e2, prompt without PR text or prior findings, read-only, static. (1) Codex CLI 0.162.1, model gpt-6-astra, codex exec -s read-only, session 01a14584-c91a-7ce0-a6b0-ffee5901751b: FINDINGS (4 × P2). (2) Claude Code claude-fable-5-1, profile max, effort high, --permission-mode plan with read-only tools (requested by the maintainer as an additional reviewer): FINDINGS (2 × Low, 2 notes; "high・medium はなし。意図された 3 つの挙動は成立している"). Fable verified _S3WorkItem claim ordering on all three executors, S3Core.call clearing, the copy cleanup ordering (no pool wait on queued requests), commit() → _close_without_commit(), and that the regression tests fail on the old code.
Codex P2 #1: "Interrupted waits retain range results. If some ranges finish before a KeyboardInterrupt or task cancellation interrupts the wait, emptying futures/reads leaves the wait helpers' own collections reachable through the exception's traceback ... The synchronous regression test replaces wait entirely, so it misses this path."
Verified (concurrent.futures.wait() and asyncio.wait() keep the futures in a set in their frames). Repaired in 5fe9176: the interrupted paths also clear the frames below _fetch_range() in the traceback; the tests now wait as those functions do, with a set of the finished reads in the frame, and fail before the repair (test_interrupted_read_frees_ranges, test_cancelled_read_frees_ranges).
|
|
||
| The arguments of a failed callable, such as the body of a part, are | ||
| freed when it fails: its error keeps the frames that it was raised | ||
| from, without their local variables. The errors that it was raised |
There was a problem hiding this comment.
Codex P2 #2: "Chained exceptions can retain request arguments ... If helper(body) raises and the submitted callable translates that error with raise OSError(...) from exc, the helper's frame remains reachable through __cause__ ... traverse their tracebacks with cycle protection."
Verified and resolved by narrowing the contract instead of walking chains: in PyAthena the only translation of a request error is S3Core.call() (raise S3ClientError(e).os_error from e), which clears the frames of the ClientError; walking arbitrary chains in the executor would also mutate the tracebacks of exceptions shared with other holders (raised by an earlier /code-review). The docstring of submit_to() now states that the errors a callable's error was raised from keep their frames (5fe9176).
| fs.core.create_multipart_upload.assert_not_called() | ||
|
|
||
| @pytest.mark.asyncio | ||
| async def test_failed_read_frees_other_ranges(self, fs): |
There was a problem hiding this comment.
Codex P2 #3 and Fable Low #2 (same finding): the async read-release test could assert before the worker threads had set their results, because AioS3StreamedFile.close() shuts its pool down with wait=False. Repaired in 5fe9176: the test shuts the pool down with wait=True and runs a few loop iterations before the weak-reference assertions.
| ) | ||
|
|
||
| submit = fs._thread_pool.submit | ||
| completion_queued = threading.Event() |
There was a problem hiding this comment.
| except BaseException as e: | ||
| # The reads are cancelled with the read of the range, and the | ||
| # frames of the wait, which keep their tasks, are cleared; | ||
| # clear_frames() skips this frame, which is executing. Cancelling a |
There was a problem hiding this comment.
Fable Low #1, rejected: "割り込み経路で、既に失敗済みの read タスクの例外が取得されない ... cancel() を呼ぶが、これは no-op なので例外は未取得のまま残り". Future.cancel()/Task.cancel() clear the log flag before checking the state, so cancelling a finished failed task suppresses "Task exception was never retrieved"; checked on Python 3.11 and 3.15 (an unretrieved failed task logs unless it is cancelled). The comment states this.
Fable note #4 (unverified boundary of shield() logging on 3.14): checked the 3.14.5 stdlib, which has the same _log_on_exception as 3.15; 3.13 does not.
| # frames of the wait, which keep their tasks, are cleared; | ||
| # clear_frames() skips this frame, which is executing. Cancelling a | ||
| # failed read also keeps asyncio from logging its error. | ||
| traceback.clear_frames(e.__traceback__) |
There was a problem hiding this comment.
Self-review of the repair and independent follow-up for 14b7ab38..5fe9176a6dd59f64b3c1eb60c66f1ba54e07f50a (base unchanged, b2d7aa24).
Self-review (both perspectives): the new clear_frames() calls run in except BaseException as e of _fetch_range(), whose own frame is executing and skipped; only the finished frames of the wait below are cleared, so the interrupt or cancellation keeps its type, identity, and traceback lines. The docstring of submit_to() now matches what _S3WorkItem and S3Core.call() clear. Validation: the full filesystem suite on 3.15 passed (1287 passed, 9 skipped); the related tests passed on 3.11, 3.13, and 3.14 (95 each); the new interruption tests fail before the repair. The description's counts are updated.
Independent follow-up (relayed): Codex CLI 0.162.1, model gpt-6-astra, codex exec -s read-only, session 01a14704-69ac-72e0-9120-ff240409a4e1, static review of a clean snapshot of this head for the diff 14b7ab38..5fe9176a: CLEAN. "At both new clear_frames() calls, the wait frames have unwound and _fetch_range remains executing, so it is skipped. Caller frames above it are not traversed. ... The revised tests establish the relevant completion boundaries before asserting release or cancelling the queued completion."
1 participant
WHAT
Two cleanup gaps of the S3 filesystem's executors (#1186), fixed where their causes are, with the guarantee restated as: the data that a failed request owns (part bodies, read results) is released; prompt collection of files and caller buffers is not promised, which the cyclic garbage collector still does.
1. A failed request no longer keeps its data.
S3Executor.submit_to()andS3AioExecutor.submit()run each callable through a private_S3WorkItem, which owns the caller's future, the callable, and its arguments, and drops them once it runs or the executor drops it. When the callable fails, it clears the frames below it in the error's traceback (traceback.clear_frames()), so the body of a failed part is freed when the part fails, and the error forms no cycle with the future. This replaces two closures whose shared cells had to be deleted by hand.S3Core.call(), through which every S3 request goes, clears the frames below it in the traceback of a raised error: those ofretry_api_call(), tenacity, and botocore. They kept the request, such as the body of a part, and tenacity'sRetryCallStatekept the error itself in a reference cycle, also as the__cause__of the translatedOSError.S3File/AioS3File(_fetch_range()) waits withFIRST_EXCEPTION; when a range fails or the wait is interrupted, it cancels the reads that have not started and drops the futures of the others before raising, instead of keeping them, and their data, in its frame.AioS3StreamedFile._fetch_range()does the same with its read tasks, whichgather()left running after a failure.Measured (Python 3.15): an
upload_partthat a botocore Stubber answers with AccessDenied keeps 8.1 MiB of an 8 MiB part while its error is held on master, 0.1 MiB with this PR; a parallel read of 4 × 8 MiB ranges whose first range fails keeps 24 MiB of the other ranges on master, none with this PR.2. An
AioS3FileSystemmultipart copy is cleaned up when the event loop shuts down.asyncio.run()cancels every remaining task when it shuts down: the tasks of the copy's requests, whose threads keep running, and the cleanup task that a cancellation of the copy has started. The requests are now concurrent futures of the filesystem's thread pool, and the cleanup runs in a thread of that pool throughS3AioThreadPool.run_shielded()instead of a task (the module-level_cleanup_tasksset is removed), so the event loop cannot stop it. On a cancellation, the creation, the part copies, and the completion that have not started are cancelled; the cleanup waits for the running ones and aborts the upload unless it completed:The copy waits for its creation and completion with the new public
S3AioThreadPool.wait_shielded(future): a cancellation of the waiting task leaves the call running and its late failure retrieved; it waits withasyncio.wait(), sinceasyncio.shield()on Python 3.14+ logs such a failure even when it is retrieved.run_shielded()uses it too.3. A failed multipart
commit()closes the file without its buffer. The buffer then holds only the data of the last part; a latercommit()uploaded it alone with PutObject as the whole object (also on master). It now does nothing, as after a failed write.WHY
Closes #1186. Part 1 is option 2 of #1173; part 2 is one of the event-loop shutdown edges deferred in #1046 (PR #1056).
Earlier revisions of this PR also broke the remaining reference cycles at each call site, so that a failed file was freed without the garbage collector. That relied on an invariant (no frame on an error's path may hold the failed future) that every caller had to keep by hand, and those fixes repeatedly introduced new bugs. The cycles that remain hold no request data and are collected by the garbage collector; a single owner of multipart part management with an explicit write state is proposed in #1211, and the result-set fixes found during the review move to a separate PR.
Behavior changes for the release notes:
S3Executor, and of an S3 request (S3Core.call()), keeps its frames and lines, but those below the executor orS3Core.call()(the request methods, the retries, botocore) no longer have local variables (for example inpdb.pm(), or tools that capturef_locals).AioS3FileSystemcopy over 5 GiB also cleans up when the cancellation comes from the shutdown of the event loop; its requests that have not started are cancelled.asyncio.run()returns after its running requests end, or, if the copy had been cancelled before, while the abort runs in a thread of the filesystem.commit()ofS3File/AioS3File, the file is closed and a latercommit()does nothing.TEST
Tested at the head of this PR on Python 3.15 (macOS).
just lint,just docs lint: passed.uv run --env-file .env pytest -n 4 tests/pyathena/filesystem/: 1287 passed, 9 skipped.TestS3Executor.test_submit_to_failure_frees_arguments(S3ThreadPoolExecutor,S3AioThreadPool)TestS3Core.test_call_failure_frees_request[True|False],test_call_failure_through_executor_frees_requestTestS3FileSystem.test_failed_read_frees_other_ranges,TestS3FileSystem.test_interrupted_read_frees_ranges,TestAioS3StreamedFile.test_cancelled_read_frees_ranges,TestAioS3File.test_failed_read_frees_other_ranges,TestAioS3StreamedFile.test_failed_read_frees_other_rangesTestS3File.test_commit_failure_and_discard(closed file, no second commit)TestAioS3FileSystem.test_copy_object_with_multipart_upload_loop_shutdown[create|part|complete], also with a cancellation before the shutdown;..._cancelled_queued_parts;..._cancelled_queued_completion; log assertions in..._cancelled_creation|completion[True]TestS3AioThreadPool.test_wait_shielded_cancelledThe AWS jobs run when the PR is marked Ready.
🤖 Generated with Claude Code