Size the asyncio filesystem's threads by max_workers and add open_async()

[laughingman7743]

WHAT

The asyncio filesystem runs its blocking calls in thread pools that PyAthena owns, sized by max_workers, instead of the default executor of the event loop. It also gains open_async(), a file whose reads and writes are coroutines.

• S3AioThreadPool (new, pyathena/filesystem/s3_executor.py): an S3Executor for the blocking calls of asyncio code.
  • run() / run_shielded() / submit() / submit_to() / shutdown().
  • Threads are created lazily, and calls keep the caller's context variables (as asyncio.to_thread() does).
  • A forked child resets the pools through os.register_at_fork() and creates new threads.
  • max_workers=None takes ThreadPoolExecutor's default.
• AioS3FileSystem: every former asyncio.to_thread() call runs in a pool of max_workers threads of the filesystem.
  • This includes the requests of multipart copies (> 5 GiB). The max_workers of the call bounds each copy through its semaphore, and the filesystem's pool bounds all copies together, so a bulk copy does not start max_workers threads per object. The Cancelling an async multipart copy leaves its multipart upload behind #1046 task/shield/cleanup design is unchanged.
• S3AioExecutor (option A of the design discussion): it still dispatches through the event loop, but to a pool of its own, so AioS3File parts and range reads follow max_workers.
  • shutdown() now releases that pool. A function reaches the pool only when the loop runs its dispatch task, so shutdown() schedules a task that shuts the pool down after the dispatches submitted before it. With wait=True, it waits for that task and the pool, except on the loop's own thread, where waiting would stop the dispatches, and it stops waiting if the loop stops first or the interpreter is exiting.
  • A submission after shutdown() raises RuntimeError, as with ThreadPoolExecutor.
• AioS3FileSystem.open_async() → AioS3StreamedFile (new, on fsspec's AbstractAsyncStreamedFile).
  • Modes rb/wb/ab/xb; a pool of max_workers threads per file.
  • Reads: parallel range GETs, conditional on the ETag/version as in S3File; no cache.
  • Writes: at most max_workers parts in flight, waiting with asyncio.wait(FIRST_COMPLETED) so the event loop keeps running. A part found failed by that wait stops the write.
  • Commit on close(), as S3File.
  • Leaving an async with block with an exception, a failed write, or a cancelled write aborts the upload after the running parts end. The commit and the abort run as blocking functions in the file's pool, so a repeated cancellation does not stop them.
  • The creation and the parts are tracked before they are scheduled (submit_to()), as in S3File.
  • Once the commit is submitted, it completes the upload or aborts it on a failure, and closing the file without committing does not abort it too.
  • An unclosed write file that is garbage collected emits a ResourceWarning and aborts its multipart upload. A synchronous with block raises TypeError.
  • Write modes inside a transaction, and autocommit=False, raise NotImplementedError, since the file always commits on close.
• S3FileTarget (new, pyathena/filesystem/s3.py): the argument checks and the lookups of S3File.__init__ (version pinning, IfMatch, append lookup, xb existence check), moved without behavior changes so that both file types share them.
• Docs: docs/aio.md (AioS3FileSystem section: threads, executor strategy, async file handles), docs/filesystem.md, docs/api/filesystem.rst, and the filesystem suite's description in benchmarks/README.md.

Release notes (4.2.0):

• The coroutines of AioS3FileSystem, including its multipart copies, run in a pool of up to max_workers threads of the filesystem. The parts and range reads of each of its files run in a pool of up to max_workers threads of the file. The default executor of the event loop no longer limits or blocks them, and loop.set_default_executor() does not size them. Processes may use more threads than before when max_workers exceeds the default executor's size (min(32, cpu + 4)).
• A multipart write of an AioS3FileSystem file no longer deadlocks when it runs in the only thread of the loop's default executor.
• S3AioExecutor.shutdown() now releases its threads, and a later submit() raises RuntimeError.
• AioS3FileSystem raises ValueError for a max_workers that is not positive when it is created, instead of when a file is opened. max_workers=None takes ThreadPoolExecutor's default, min(32, cpu + 4), for the filesystem's threads and as the default of its files and copies; files and copies used to fail with it.
• New: AioS3FileSystem.open_async(), AioS3StreamedFile, S3AioThreadPool, S3FileTarget.

WHY

Refs #1159. The 4.1.0 filesystem benchmark showed that with max_workers=40, aio writes reached 0.75–0.83× and multipart copies 0.30–0.35× of the sync throughput. The aio processes peaked at 24–25 threads, against 51 for sync, because asyncio.to_thread() is capped by the default executor (12 threads on c8gn.2xlarge).

The pre-existing deadlock deferred from #1165 reproduces on master: with a supplied loop whose default executor has one thread, a write run through that loop's asyncio.to_thread() waits for its multipart creation, which needs the same thread.

The design was agreed with the maintainer after consulting Claude claude-fable-5-1 and Codex:

• owned pools instead of the default executor;
• option A for S3AioExecutor;
• a filesystem-owned pool for the single calls;
• open_async() with all four modes, abort on an exception, no writes in a transaction, and parallel range reads.

#1173 option 2 (releasing the references of a failed part in submit_to()) is left to a separate issue.

The re-measurement and the decision on the async default for writes and copies stay in #1159, for the combined 4.2.0 benchmark run, so this PR does not close it.

TEST

Tested at 324d868 (rebased onto 38dc8f8), and the self-review repairs at 0f4ff60, on macOS with CPython 3.15.0:

• just lint: passed (ruff, format, mypy, cfn-lint, license headers).
• just docs lint: 0 errors. just docs build builds master and the tags, not this branch, so the API pages of the new classes are left to the docs CI.
• uv run --env-file .env pytest -n 4 tests/pyathena/filesystem/: 1251 passed, 9 skipped (before the rebase; the rebase brought no filesystem changes). This includes the live S3 tests of open_async(): 1 KiB and 10 MiB write/read, xb/ab.
• New regression tests that fail on master's executor and pass here (checked by swapping in master's s3_executor.py):
  • TestS3AioExecutor::test_submit_beyond_default_executor
  • TestS3AioExecutor::test_submit_from_default_executor (the 1-thread deadlock)
  • TestS3AioThreadPool::test_run_beyond_default_executor
• Other new tests:
  • executor: test_shutdown_before_dispatch (reproduced the transaction failure found during development), pool cancellation/shield/shutdown/fork;
  • filesystem: test_copy_object_with_multipart_upload_max_workers, test_copy_object_with_multipart_upload_shares_threads, test_blocking_calls_bypass_default_executor;
  • TestAioS3StreamedFile: parts in flight released from the event loop, a failed part while waiting, exit with an exception, cancelled writes (1 and 2 cancellations), a cancelled commit, an exception after close, small and empty commits, parallel reads, invalid arguments, transactions, xb, ab.
• At 0f4ff60: uv run --env-file .env pytest -n 4 tests/pyathena/filesystem/test_s3_async.py tests/pyathena/filesystem/test_s3_executor.py: 201 passed, 6 skipped. At the round-two commit, TestAioS3StreamedFile: 22 passed; just lint and just docs lint passed.
• After the /code-review and /simplify repairs, at de807b1: just lint passed; uv run --env-file .env pytest -n 4 tests/pyathena/filesystem/: 1259 passed, 9 skipped.
• After the independent review's repairs, at f475a9a: just lint and just docs lint passed; uv run --env-file .env pytest -n 4 tests/pyathena/filesystem/test_s3_async.py tests/pyathena/filesystem/test_s3_executor.py: 209 passed, 6 skipped. test_shutdown_loop_stopped and test_flush_cancelled_then_exit_with_exception fail with de807b1's sources.
• At 35cc4f3 (the independent follow-ups' test and docs repairs): uv run --env-file .env pytest -n 4 tests/pyathena/filesystem/test_s3_executor.py tests/pyathena/filesystem/test_s3_async.py: 209 passed, 6 skipped; the second Codex follow-up was CLEAN.
• Additional review by Claude claude-fable-5-1 (requested by the maintainer; a Claude model, so it is not counted as the independent review): 6 findings. 4 are repaired in 3df48b4, 1 is already release-noted, and 1 docs nit is fixed. At 3df48b4: uv run --env-file .env pytest -n 4 tests/pyathena/filesystem/: 1262 passed, 9 skipped.
• Rebased onto 27ceea3 (Stop appends, sync multipart copies, and commits at a failed part #1183). AioS3StreamedFile follows Stop appends, sync multipart copies, and commits at a failed part #1183's S3File behavior: a write stops at a failed part copy of an append, and the commit raises a failed part without waiting for the parts before it. At 55b8c76: uv run --env-file .env pytest -n 4 tests/pyathena/filesystem/: 1268 passed, 9 skipped. At 5f32a53, test_s3_async.py and test_s3_executor.py: 212 passed, 6 skipped. Codex follow-ups: the rebase and port review was repaired, and the last one was CLEAN.
• Deferred (pre-existing, from the independent review): when asyncio.run() shuts down during the CreateMultipartUpload of an AioS3FileSystem multipart copy, the created upload is not aborted. This is one of the loop-shutdown edges deferred in Cancelling an async multipart copy leaves its multipart upload behind #1046. Filed as Free failed part bodies without the GC and abort an async copy's upload created during loop shutdown #1186, together with Failed parts of an S3File multipart write keep their data and do not stop the write #1173 option 2.
• Not run locally: the full just test pyathena suite (CI runs it once the PR is Ready) and any throughput measurement, which belongs to the combined 4.2.0 benchmark.

🤖 Generated with Claude Code

[laughingman7743]

Self-review round one (behavior, failure paths, resources, framework contracts, simplicity and tests; performed by the author, not an independent review).

Scope: git diff 38dc8f860ece434b36779f0da0291d505f4e42a1..324d868b1a0fd2f812c053993559b636366cec1a (all 9 files). Inventory: S3AioThreadPool; S3AioExecutor dispatch, settle and shutdown; every former asyncio.to_thread() site of AioS3FileSystem; the multipart copy and its #1046 cancellation design; S3FileTarget against the removed S3File.__init__ block; open_async() and AioS3StreamedFile (write, flush, close, __aexit__, commit, abort, reads, append, xb, transactions, __del__); docs and tests.

Result: FINDINGS, 2 repaired in 0f4ff60.

1. Introduced, this function: each multipart copy had its own pool of max_workers threads. _mv() gathers all its copies and fsspec's _copy() runs them in batches, so a bulk copy of N objects larger than 5 GiB could start N x max_workers threads (for example 50 x 40). Master capped the total at the default executor, and sync copies run one at a time. Repaired: the part copies run in the filesystem's pool. The call's max_workers still bounds each copy through its semaphore, and the filesystem's max_workers bounds all copies together. This also removes the per-copy pool shutdown after a detached cleanup. New test: test_copy_object_with_multipart_upload_shares_threads (two copies, max_workers=2, peak 2).

[laughingman7743]

Self-review round one, finding 2 (introduced): async with f: await f.close(); raise KeyError(...) went into _close_without_commit() on a file that was already closed. That submitted an abort to the pool, which close() had already shut down, and the resulting RuntimeError: cannot schedule new futures after shutdown replaced the caller's KeyError. Reproduced on 324d868. A committed multipart upload would also have been sent an abort. Repaired in 0f4ff60: __aexit__ and _close_without_commit() do nothing for a closed file. New test: TestAioS3StreamedFile::test_exit_with_exception_after_close.

Validation after the repairs: just lint passed; uv run --env-file .env pytest -n 4 tests/pyathena/filesystem/test_s3_async.py tests/pyathena/filesystem/test_s3_executor.py: 201 passed, 6 skipped.

[laughingman7743]

Self-review round one, checked and kept (no change):

• shutdown(wait=True) now waits for the submitted futures, including those whose dispatch is still queued on the event loop. Called on that loop's own thread, it would block. A file whose methods run on its dispatch loop already blocks that loop at its first wait for a part, which is a pre-existing misuse, so no guard was added.
• A future cancelled before its dispatch counts as done for _release(). If the pool is then shut down, its dispatch raises inside the task, and settle() leaves the cancelled future as it is.
• submit() after shutdown() now raises, as with ThreadPoolExecutor. S3File never submits after close() or _close_without_commit(); this was traced through commit() and discard() in a transaction.
• S3AioThreadPool._get_executor() holds a lock only while it creates the executor. A fork during that moment could leave the child's lock held, the same narrow window as other lock-guarded state. Not addressed.
• AioS3FileSystem(max_workers=0) now raises ValueError at construction instead of at the first file. This is added to the release notes.

[laughingman7743]

Self-review round two (claims, existing callers, operators, docs readers, evidence; by the author, not independent). Full claim pass over the PR body, commit messages, changed docstrings, docs/aio.md, docs/filesystem.md, docs/api/filesystem.rst, and the issue premises, at git diff 38dc8f860ece434b36779f0da0291d505f4e42a1..0f4ff606802e1c2a749de048e9c6d3e35002da6c.

Result: FINDINGS, 2 repaired in 1ade9fa; the other claims hold.

Claims checked, with their evidence:

• Benchmark numbers in WHY (0.75–0.83 for writes, 0.30–0.35 for copies, 51 vs 24–25 threads, 12 threads on 8 vCPUs): match benchmarks/history/2026-10-13-437276e1-filesystem.md:28,50-52,100-108.
• The deadlock reproduces on master: /tmp repro at c0cc145. The branch base 38dc8f8 only adds a Polars change. test_submit_from_default_executor fails on master's executor.
• Every former to_thread() call of AioS3FileSystem uses the filesystem pool: no asyncio.to_thread remains in s3_async.py.
• The calls keep the caller's context variables (this doc line):
  • The filesystem pool's submit() copies the context of the awaiting task.
  • For S3AioExecutor, run_coroutine_threadsafe → call_soon_threadsafe copies the context of the submitting thread into the dispatch task, as the former to_thread() did.
  • test_run_context covers the pool.
• The reads of AioS3StreamedFile are conditional like S3File:
  • S3FileTarget.resolve sets IfMatch and pins the version when the filesystem is version-aware.
  • _fetch_range passes S3Path(bucket, key, version_id).
  • test_read_parallel asserts IfMatch.
• Existing callers:
  • The signatures of S3File.__init__, S3AioExecutor.__init__, and AioS3FileSystem.__init__ are unchanged.
  • New behaviors are release-noted: S3AioExecutor.shutdown() now blocks with wait=True and later submissions raise; AioS3FileSystem(max_workers<=0) raises at construction.
  • The full filesystem suite passed at 324d868.
• AWS operator: the request parallelism rises to max_workers, as sync already does. Botocore retries are unchanged and no application retries were added. Throughput and thread counts on S3 are not measured here; they are left to the combined 4.2.0 benchmark (Size the asyncio filesystem's parallel requests by max_workers instead of the default executor #1159).

Finding 1, evidence gap: this doc line promised that a started commit completes despite a cancellation, but no test covered it. Repaired: TestAioS3StreamedFile::test_close_cancelled (the completion runs on, and there is no abort).

[laughingman7743]

Self-review round two, finding 2 (stale prose outside the diff): benchmarks/README.md:221-223 still said that the aio filesystem suite's part uploads and copies run in asyncio.to_thread(), bounded by the default executors. Since this PR, that no longer holds. Repaired in 1ade9fa. The paragraph now describes the file pools and the filesystem pool, and states the pre-#1159 behavior that the 4.1.0 history entry measured. benchmarks/RESULTS.md:35 stays as it is: it records the measured 4.1.0 runs and points to #1159, and the combined 4.2.0 run will update it.

Validation: just lint passed; just docs lint: 0 errors; pytest -k TestAioS3StreamedFile: 22 passed.

Deferred, with reasons:

• RSS and threads on S3 are unmeasured. They are left to the planned combined benchmark, which is the rule for Size the asyncio filesystem's parallel requests by max_workers instead of the default executor #1159.
• The Sphinx pages of the new classes are not built locally (just docs build builds master and the tags). They are left to the docs CI.

[laughingman7743]

Self-review addendum: /code-review (high), relayed. This was a Claude review fork run by the author, so it counts as self-review, not the independent review. It ran on a detached snapshot of 1ade9fa, scope 38dc8f860ece434b36779f0da0291d505f4e42a1..1ade9fa3, static. It returned 10 candidates; each was verified against the code before acting. 8 are repaired in 1fce178, and 2 are rejected or deferred.

Repaired:

1. This method: a write file collected without await close() dropped its data silently and left the multipart upload incomplete. It now emits a ResourceWarning and submits the abort to the file's threads. Test: test_garbage_collected_unclosed.
2. Sync with on the streamed file: __exit__ called the coroutine close() without awaiting it, so the file was never committed. __enter__ now raises TypeError. Test: test_sync_with.
3. Tracking: the creation and the parts were tracked only after submit() returned. An interrupt in between could leave a part untracked, which the abort then neither waits for nor cancels. They are now created as futures first and run with submit_to(), as in S3File (Fix interrupted multipart creation scheduling and flaky SIGINT tests #1113/Track multipart parts before scheduling and own multipart copy cleanup #1122); for that, S3AioThreadPool became an S3Executor.
4. open_async(autocommit=False) was ignored silently; it now raises NotImplementedError. A compression=None keyword is popped instead of being passed on as an S3 parameter.
5. AioS3FileSystem(max_workers=None): it used to construct, and its info/ls worked; the eager pool made it a TypeError. S3AioThreadPool(None) now takes ThreadPoolExecutor's default, min(32, cpu + 4). Test: test_max_workers_none.

See the next comment for 6–10.

[laughingman7743]

/code-review addendum, continued.

Repaired (continued):
6. This line: S3AioExecutor.shutdown(wait=True) used to be a no-op. Run on its event loop's thread, for example when a file is garbage collected on fsspec's IO thread, it now blocked on functions that only that loop dispatches. On the loop thread it now does not wait, and the last pending function shuts the pool down. Test: test_shutdown_on_loop_thread.
7. submit()/shutdown() race: submit() checked the shutdown flag outside the lock. It now checks under the same lock that registers the pending future.
8. Fork: a lock copied while a parent thread held it would hang the child. _get_executor() now replaces the lock in a forked child. Test: test_fork holds the lock.

Rejected or deferred:
9. Duplication with S3File (_get_request_kwargs, append setup, range split, abort order): deferred. The shared logic already lives in S3FileTarget and S3MultipartWriter, and _get_ranges is reused. The remaining code differs by sync/async control flow, so merging it would need a shared base, a structural change outside this PR.
10. A filesystem pool shared by AioS3File instances, with a per-file semaphore: rejected. _pipe_file_in_transaction runs in the filesystem pool and writes through open(), so a file waiting on parts in the same pool could deadlock when the pool is saturated. A pool per file matches S3File, whose files also each own a pool.

Validation at 1fce178:

• just lint passed.
• uv run --env-file .env pytest -n 4 tests/pyathena/filesystem/test_s3_async.py tests/pyathena/filesystem/test_s3_executor.py: the files passed apart from one test that I then fixed (the GC test's placement); the rerun of -k TestAioS3StreamedFile gave 25 passed.

[laughingman7743]

Self-review addendum: /simplify, relayed, plus the self-review of the repairs. /simplify ran 4 Claude cleanup agents (reuse, simplification, efficiency, altitude), static, over 38dc8f860ece434b36779f0da0291d505f4e42a1..1fce178faf4c3490265f818f710a2f5dfe4f3611. Its findings were applied in de807b1:

• This method (altitude, efficiency): S3AioExecutor tracked every pending future in a set, with a done-callback, so it could shut its pool down after the last dispatch. The event loop starts tasks in scheduling order, so shutdown() now schedules this task, under the lock that also orders submit(). The task shuts the pool down after the earlier dispatches. That removes the per-part set, callback, and lock round trip.
• Fork (altitude, efficiency): the pool compared os.getpid() on every call. It now resets through os.register_at_fork(), as concurrent.futures.thread does, and the existing executor is read without the lock.
• Large append (efficiency): the streamed file sent a second HeadObject for an existing object over 5 GiB. It now uses the size that S3FileTarget.resolve() looked up with refresh=True.
• Per flush (efficiency): the request parameters of the parts are computed once per flush, not once per part.
• Simplification: _abort_upload was merged into _discard(). open_async() no longer repeats the file's mode check. The tests import the shared EVENT_TIMEOUT.

Skipped, with reasons:

• Moving the commit/abort flow, or the post-resolve block, into a shared base or S3MultipartWriter would be a structural change to S3File (see the /code-review note, item 9).
• Merging S3AioExecutor.submit into submit_to() would change the settle paths for little gain.
• Shutting the file pool down in __del__ was kept: explicit, and harmless.
• Guarding the other synchronous AbstractBufferedFile methods: __enter__ covers the silent data-loss case, and the others fail visibly.
• The PutObject buffer copy is the same as in S3File.
• The duplicated mock setup in the tests, and test_open_parallel_requests (outside the diff), were left as they are.

Self-review of the repairs (git range-diff-sized follow-up over 1ade9fa3..de807b1b), both perspectives:

• Behavior:
  • Submissions and the shutdown task are both scheduled with call_soon_threadsafe under one lock, so their order holds.
  • If the loop shuts down before the shutdown task runs, the task is cancelled. result() raises CancelledError, which is suppressed, and the pool is shut down directly.
  • A forked child resets _executor to None, so the lock-free path never returns the parent's executor.
  • The large-append change chooses the whole-object copy from the existing size, not from tell(). That gives the same parts for an existing object up to 5 GiB, and skips the lookup above it.
• Claims: the PR body now states the shutdown() behavior, the register_at_fork reset, the max_workers=None default, the GC warning and abort, and the autocommit=False rejection.
• Validation:
  • just lint passed.
  • uv run --env-file .env pytest -n 4 tests/pyathena/filesystem/: 1259 passed, 9 skipped, on the tree committed as de807b1.
  • This includes the live S3 open_async and transaction-write tests, which exercise the S3AioExecutor shutdown path.

Result: CLEAN after the repairs. The independent review is next.

[laughingman7743]

Independent review (relayed Codex result). Codex CLI 0.162.1, codex exec -s read-only, model gpt-6-astra, session 01a13b66-1c13-7072-9e7d-175ce1ad9f31. It ran on a clean detached snapshot of de807b1b76483d017d01337227baa5e46dc665a2, scope git diff 38dc8f860ece434b36779f0da0291d505f4e42a1..de807b1b76483d017d01337227baa5e46dc665a2. The prompt carried no PR number, description, commit messages, or prior findings. Static review only: no builds, tests, network, history, or memory. The snapshot stayed unchanged (git status clean, HEAD de807b1).

Covered, in Codex's words: "executor lifecycle, cancellation and repeated cancellation, loop shutdown, fork reset, garbage collection, thread/memory bounds, file compatibility, multipart ordering, S3 parameters, regression tests, and the scoped documentation." Result: FINDINGS (4).

Finding 1 (P2, introduced; this line): "If another thread stops the loop after is_running() succeeds but before that coroutine executes, the future remains pending… shutdown(wait=True) — including through S3File.close() — can hang." Verified. Repaired in f475a9a: the wait polls the task and stops waiting once the loop no longer runs, then shuts the pool down. New test: TestS3AioExecutor::test_shutdown_loop_stopped; it fails on de807b1 and passes here.

[laughingman7743]

Independent review, finding 2 (P2, introduced; this line): "Inside async with, cancel await f.flush(force=True) after CompleteMultipartUpload starts… context exit schedules _discard(), which waits only for creation and parts before sending abort. Completion and abort can therefore race." Verified. Repaired in f475a9a: once the commit is submitted (_committing), it owns the upload: it completes it, or aborts it if it fails. _close_without_commit() and __del__ then only close the file. The docs and the class docstring say so. New test: test_flush_cancelled_then_exit_with_exception; it fails on de807b1 and passes here.

[laughingman7743]

Independent review, finding 3 (P2, pre-existing; pyathena/filesystem/s3_async.py:638): "When asyncio.run() shuts down during CreateMultipartUpload, it cancels both the copy task and its creation task. The blocking request can still finish successfully, but cleanup returns here without recovering its upload ID." Verified as present on master: the same check exists before this PR. It is one of the event-loop shutdown edges recorded as deferred in #1046/#1056. Deferred: the copy's orchestration is unchanged here apart from the thread pool, and recovering the upload ID independently of task cancellation is a separate change to that design.

[laughingman7743]

Independent review, finding 4 (P3, introduced, test; this line): "gives both the blocker and wait_for() five seconds… An _info() implementation still using that executor can then return successfully." Verified. Repaired in f475a9a: the blocker now waits up to 60 seconds and is released only in the finally block, so a lookup through the default executor fails the 5-second wait_for().

Self-review of the repairs (both perspectives, de807b1b..f475a9a8):

• Behavior:
  • The commit of a small file has no upload to abort.
  • A commit that fails calls _discard() itself.
  • The buffer stays in place for a commit that has not started.
  • The 0.1 s poll runs only while shutdown(wait=True) waits.
• Claims: docs/aio.md and the AioS3StreamedFile docstring now say that leaving an async with block with an exception after a started commit does not abort it.
• Validation:
  • just lint and just docs lint passed.
  • uv run --env-file .env pytest -n 4 tests/pyathena/filesystem/test_s3_async.py tests/pyathena/filesystem/test_s3_executor.py: 209 passed, 6 skipped.
  • The two new tests fail with de807b1's s3_executor.py/s3_async.py swapped in.

An independent follow-up on the repair range comes next.

[laughingman7743]

Independent follow-up (relayed Codex result). Codex CLI 0.162.1, codex exec -s read-only, model gpt-6-astra, session 01a13b77-3224-7d32-8648-fdf703fef096. It ran on a clean detached snapshot of f475a9a, scope git diff de807b1b76483d017d01337227baa5e46dc665a2..f475a9a8c2cfefb181d6e3274d58b3a2fe8acfaa plus the three earlier issues, static only. The snapshot stayed unchanged.

Covered, in Codex's words: "the exact repair diff; executor dispatch, cancellation, shutdown and S3File callers; streamed-file flush, close, exceptional exit, garbage collection and failed-completion cleanup; tests and documentation." Codex's assessment: "The runtime changes address the two reported failure paths by inspection." Result: FINDINGS (2).

Finding 1 (P2, test; this line): "The 0.2-second wait does not establish that the waiter has scheduled _shutdown_pool()… The old implementation then observes is_running() == False… and passes the test." Verified. Repaired in 35cc4f3: the test wraps asyncio.run_coroutine_threadsafe and releases the loop only after the shutdown task is scheduled. With de807b1's s3_executor.py, it now fails (1 failed); with the repair, it passes.

[laughingman7743]

Second independent follow-up (relayed Codex result): CLEAN. Codex CLI 0.162.1, codex exec -s read-only, session 01a13b80-9cd8-7321-ab13-32bd51aeac92. It ran on a clean detached snapshot of 35cc4f3, scope git diff f475a9a8c2cfefb181d6e3274d58b3a2fe8acfaa..35cc4f3ef91f7ecb6876cdd410c56f08fc57173a plus the two follow-up issues, static only. The snapshot stayed unchanged.

Covered, in Codex's words: "the repair diff, executor shutdown and test cleanup, async file commit/destructor/discard paths, related lifecycle tests, and both documentation changes."

On the test: "scheduled is signalled after real coroutine submission. The blocked callback then stops the loop before the coroutine runs. A shutdown that waits indefinitely fails the assertion at line 243. Cleanup may subsequently execute the coroutine and unblock shutdown, but it cannot suppress that assertion failure."

On the docs: "the warning/abort promise now excludes files whose commit has started, matching the destructor's not closed and not _committing guards."

[laughingman7743]

Independent follow-up, finding 2 (P3, docs; this line and the AioS3StreamedFile docstring): "After a final flush starts committing, dropping an unclosed handle now produces neither the promised ResourceWarning nor an abort." Verified. Repaired in 35cc4f3: the guarantee is now limited to a file collected before it is closed or before its commit has started, in docs/aio.md, the class docstring, and the __del__ comment.

Self-review of these repairs: a test and docs only, with no runtime change. just lint and just docs lint passed; uv run --env-file .env pytest -n 4 tests/pyathena/filesystem/test_s3_executor.py tests/pyathena/filesystem/test_s3_async.py: 209 passed, 6 skipped. A second independent follow-up on f475a9a8..35cc4f3e is next.

[laughingman7743]

Additional review by Claude claude-fable-5-1 (relayed; requested by the maintainer). It ran with Claude Code, profile max (first-party claude.ai authentication, checked with claude auth status), --model claude-fable-5-1 --effort high --permission-mode plan, read-only tools, session 7856740a-515e-4218-afbd-bf6b85115ea5. The target was a clean detached snapshot of 35cc4f3, scope git diff 38dc8f860ece434b36779f0da0291d505f4e42a1..35cc4f3ef91f7ecb6876cdd410c56f08fc57173a. The prompt was the same neutral package as the Codex review; the snapshot stayed unchanged. Because the change is Claude-authored, this review is recorded as additional, not as the independent review (that was Codex, above). Static review only.

Covered: the executor pool and shutdown, every delegation, the copy, open_async()/AioS3StreamedFile, S3FileTarget against the former S3File.__init__, the tests, the docs, and AioS3FSCursor's use through pyathena/s3fs/result_set.py. Result: FINDINGS (6). Fable also confirmed:

• the loop-ordered shutdown and its lock ordering;
• no deadlock between _complete_upload and _discard on one worker, thanks to the FIFO start order;
• exactly one abort or commit on every cancellation path;
• the S3FileTarget semantics;
• the memory bound;
• the Size the asyncio filesystem's parallel requests by max_workers instead of the default executor #1159 tests failing on the old code.

1. Medium, introduced (this line): "Files that are still open at interpreter shutdown… are closed by finalizers… fsspec's IO-loop thread is a daemon and stops when finalization begins, but loop.is_running() stays True… so it polls closing forever and the process never exits." Verified by reading the code; not reproduced at runtime. Repaired in 3df48b4: the wait also stops when sys.is_finalizing(). Test: test_shutdown_interpreter_finalizing, which fails with 35cc4f3's sources.
2. Low: S3AioExecutor.shutdown() is now terminal and blocking, which affects an executor shared through executor=. No change: this is already in the release notes ("S3AioExecutor.shutdown() now releases its threads, and a later submit() raises RuntimeError"), and a shared S3ThreadPoolExecutor already behaves the same way.
3. Low, introduced: "the fork hook drops the parent's executor in the child… its weakref callback puts to a queue whose copied mutex a parent worker may have held." Repaired: _reset_after_fork() keeps the copied executors in S3AioThreadPool._forked_executors. test_fork asserts it.

[laughingman7743]

Additional claude-fable-5-1 review, continued.

2. Low (this line): "AioS3FileSystem(max_workers=None) stores None on the sync filesystem, so files from open() hit TypeError… and the multipart copy creates asyncio.Semaphore(None)." Verified. Repaired in 3df48b4: the pool is created before the internal S3FileSystem, and its default is passed on. test_max_workers_none asserts the files' and copies' default too, and fails with 35cc4f3's sources.
3. Nit (test): "test_write_parts_in_flight would produce the same trace if all four parts were submitted at once, since the 2-thread pool caps them anyway… it does not prove the memory bound." Verified. Repaired: the test counts the parts submitted and not finished at each submit_to(). That count is [1, 2, 2, 2] only with the bound; without it, it would be [1, 2, 3, 4].
4. Nit (docs): "a commit queued behind parts also completes; run_shielded never drops the call." Repaired: docs/aio.md and the class docstring now say that the commit completes once it has been submitted, including while it waits for a thread.

Self-review of the repairs (both perspectives):

• Finalization: at interpreter finalization, the pool's own shutdown(wait=True) returns, because threading._shutdown has already joined the workers.
• Invalid max_workers: max_workers<=0 still raises ValueError at construction, now before the internal filesystem is created.
• The PR body states the normalized default and the shutdown at interpreter exit.

Validation:

• just lint and just docs lint passed.
• uv run --env-file .env pytest -n 4 tests/pyathena/filesystem/: 1262 passed, 9 skipped.

An independent (Codex) follow-up on 35cc4f3e..3df48b44 is next.

[laughingman7743]

Independent follow-up (relayed Codex result) on the claude-fable-5-1 repairs. Codex CLI 0.162.1, codex exec -s read-only, session 01a13ca1-abb1-7c51-833d-4c4163e0bcbe. It ran on a clean detached snapshot of 3df48b4, scope git diff 35cc4f3ef91f7ecb6876cdd410c56f08fc57173a..3df48b44e44a9853f7b784e27e4ab114e405724d plus the five issues, static only. The snapshot stayed unchanged.

Covered, in Codex's words: "shutdown during finalization and loop stopping; fork retention and recreation; construction, fsspec caching, and max_workers propagation into files and copies; queued-commit cancellation; documentation and changed tests." Codex also reported: "No additional defect found in the traced repair paths… the revised in-flight test detects queued excess parts." Result: FINDINGS (2).

1. P2, test (this line): "The loop blocker's ten-second timeout starts before the measured shutdown interval. Without the sys.is_finalizing() check, the blocker can expire… and the measured duration can still be under ten seconds." Verified. Repaired in e15f5b7: the loop stays blocked for up to 60 seconds, the shutdown runs in a thread, and the test asserts that it returns within EVENT_TIMEOUT while the loop is still blocked. With 35cc4f3's s3_executor.py swapped in, the test fails.

[laughingman7743]

Narrow independent follow-up (relayed Codex result): CLEAN. Codex CLI 0.162.1, codex exec -s read-only, session 01a13caa-e4b2-7360-aad9-cd3235635e6d. It ran on a clean detached snapshot of e15f5b7, scope git diff 3df48b44e44a9853f7b784e27e4ab114e405724d..e15f5b73161c0f56e79c85d91eada50ce366eb60 plus the two issues, static only. The snapshot stayed unchanged.

On the test: "Removing sys.is_finalizing() leaves shutdown waiting, causing the assertion to fail. Failure cleanup releases the blocker, stops and joins the loop thread, and drains the queued shutdown coroutine."

On the default: "matches ThreadPoolExecutor: os.cpu_count() on Python 3.11–3.12 and os.process_cpu_count() on 3.13–3.15… Both repaired docstrings agree with this behavior."

[laughingman7743]

Independent follow-up, finding 2 (P3, docs and default; this line): "The added 'as with ThreadPoolExecutor' claim is inaccurate on supported Python 3.13+, which uses os.process_cpu_count()." Verified. Repaired in e15f5b7: the default now uses os.process_cpu_count() when it is available, and os.cpu_count() before 3.13, so it matches ThreadPoolExecutor. The docstrings say "the default of ThreadPoolExecutor: min(32, CPUs + 4)", and the tests compute the expectation the same way.

Validation:

• just lint passed.
• uv run --env-file .env pytest -n 4 tests/pyathena/filesystem/test_s3_executor.py tests/pyathena/filesystem/test_s3_async.py: 210 passed, 6 skipped.

A narrow independent follow-up on 3df48b44..e15f5b73 is next.

[laughingman7743]

Rebase onto 27ceea3 (#1183), with the port of #1183 to the streamed file.

Rebase: #1183 conflicted in S3File.__init__; the resolution keeps its _copies_in_flight alongside S3FileTarget's target.appended. A test import also conflicted. git range-diff 38dc8f86..e15f5b73 origin/master..aef5dc23 shows commits 2–7 unchanged; 1, 8 and 9 differ only by those resolutions and by context.

Port (55b8c76): AioS3StreamedFile now behaves like S3File after #1183. The docs said that its appends and commits follow S3File:

• This line: a write raises a part copy of an append found failed before it submits a part.
• _complete_upload() raises a failed part with wait(FIRST_EXCEPTION), without waiting for the parts before it.
• Tests: test_append_failed_part_copy and test_commit_failed_part_first. Both fail with aef5dc2's s3_async.py; the commit test first hung there, and was changed to release its futures in a finally block.

Self-review, both perspectives: _complete_upload() runs in the file's pool and waits only on parts submitted before it, which have all started in FIFO order, so the new wait cannot deadlock. The release notes need no change, because the streamed file is new.

Independent follow-up (relayed Codex result): Codex CLI 0.162.1, codex exec -s read-only, session 01a13cb8-cc03-7773-837b-0d740e722820, on a clean detached snapshot of 55b8c76. The scope was #1183's s3.py diff, this change's s3.py diff against the new base, and the port diff, static only.

In Codex's words: "The synchronous behavior is preserved, and I found no new deadlock, cancellation, or buffer-growth regression." Result: FINDINGS (2), both repaired in 5f32a53; see the next comment.

Validation:

• uv run --env-file .env pytest -n 4 tests/pyathena/filesystem/ at 55b8c76: 1268 passed, 9 skipped.
• At 5f32a53, test_s3_async.py and test_s3_executor.py: 212 passed, 6 skipped.

[laughingman7743]

Rebase follow-up, finding 1 (P2, test; this line): "The copy mock fails immediately. If its worker finishes before the first write reaches the new copy-failure check, this unguarded write raises." Verified. Repaired: the copy is held until the first write returns, as in the synchronous test, and the test then waits for the copy before the second write.

Finding 2 (P3, docs, docs/aio.md): "Completion detects that failure immediately, but _discard() waits for the earlier running part before aborting and re-raising." Verified. Repaired: the docs and the _complete_upload() docstring now say that the commit cancels the parts that have not started as soon as one fails, and waits for the running ones before it aborts and raises.

A narrow independent follow-up on 55b8c761..5f32a53b is next.

[laughingman7743]

Narrow independent follow-up (relayed Codex result): CLEAN. Codex CLI 0.162.1, codex exec -s read-only, session 01a13cbe-1b13-7f90-9281-cd28ad6737f5. It ran on a clean detached snapshot of 5f32a53, scope git diff 55b8c7616ebe30887a0a0b741a80f6392b60e98b..5f32a53b97a0161309ffed2024b27b51430a86de plus the two issues, static only. The snapshot stayed unchanged.

Codex's findings:

• On the test: "the event removes the immediate-failure race; finally releases the worker… removing [the copy-failure guard] lets the second write return without raising, so pytest.raises(PermissionError) fails."
• On the docs: "both match the code's sequence—detect failure, cancel pending parts, wait for running parts, abort, then propagate the error."