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

benchmarks/README.md

@@ -218,9 +218,10 @@ The `[filesystem]` section of `config.toml` sets the swept values, 8 and 40 work
 The suite's scales are the object sizes in bytes in `[filesystem] sizes`, not the row counts in `[scales]`: 64 MiB (small), 256 MiB (medium), 1 GiB (large), and 6 GiB (xlarge) by default.
 
 The `sync` API calls the filesystem on the trial's thread, and a file's parallel requests run in an `S3ThreadPoolExecutor`.
-The `aio` API runs a read or write in `asyncio.to_thread()`, as `AioS3FSCursor` reads result files, and a file's parallel requests run through `S3AioExecutor` in `asyncio.to_thread()` on the filesystem's event loop, fsspec's IO thread.
-Its copy is a coroutine awaited on the trial's event loop, whose part copies run in `asyncio.to_thread()`.
-Both event loops keep Python's default executor, whose `min(32, os.cpu_count() + 4)` threads bound the parallel requests of the `aio` API.
+The `aio` API runs a read or write in `asyncio.to_thread()`, as `AioS3FSCursor` reads result files, and a file's parallel requests run through `S3AioExecutor`, dispatched on the filesystem's event loop, fsspec's IO thread, to a pool of `max_workers` threads of the file.
+Its copy is a coroutine awaited on the trial's event loop, whose part copies run in the filesystem's pool of `max_workers` threads.
+Both event loops keep Python's default executor, which runs only the trial's `asyncio.to_thread()` call.
+Before #1159, the parallel requests of the `aio` API ran in the default executors, whose `min(32, os.cpu_count() + 4)` threads bound them.
 
 The source objects, of zero bytes, are uploaded with boto3 below the run's scratch prefix and recorded under `objects` in the manifest.
 `prepare --filesystem-scale` uploads them with the manifest, so that every fleet host reads the same objects.

docs/aio.md

@@ -249,43 +249,92 @@ See each cursor's documentation page for detailed usage examples.
 ## AioS3FileSystem
 
 `AioS3FileSystem` is a native asyncio filesystem interface for Amazon S3, built on
-fsspec's `AsyncFileSystem`. It provides the same functionality as `S3FileSystem` but
-uses `asyncio.gather` with `asyncio.to_thread` for parallel operations instead of
-`ThreadPoolExecutor`.
-
-### Why AioS3FileSystem?
-
-The synchronous `S3FileSystem` uses `ThreadPoolExecutor` for parallel S3 operations
-(batch deletes, multipart uploads, range reads). When used from within an asyncio
-application via `AioS3FSCursor`, this creates a thread-in-thread pattern:
-the cursor wraps calls in `asyncio.to_thread()`, and inside that thread
-`S3FileSystem` spawns additional threads via `ThreadPoolExecutor`.
-
-`AioS3FileSystem` eliminates this inefficiency by dispatching all parallel
-operations through the asyncio event loop.
+fsspec's `AsyncFileSystem`. It provides the same functionality as `S3FileSystem`, and
+its coroutines run the blocking boto3 calls in threads that PyAthena owns, sized by
+`max_workers`, instead of the default executor of the event loop.
 
 | | S3FileSystem | AioS3FileSystem |
 |---|---|---|
-| **Parallelism** | `ThreadPoolExecutor` | `asyncio.gather` + `asyncio.to_thread` |
-| **File handles** | `S3File` with thread pool | `AioS3File` with `S3AioExecutor` |
+| **Blocking calls** | Caller's thread | Thread pool of the filesystem |
+| **Parallelism** | `ThreadPoolExecutor` | `asyncio.gather` on a thread pool |
+| **File handles** | `S3File` with thread pool | `AioS3File` with `S3AioExecutor`; `AioS3StreamedFile` from `open_async()` |
 | **Bulk delete** | Thread pool per batch | `asyncio.gather` per batch |
-| **Multipart copy** | Thread pool per part | `asyncio.gather` per part |
+| **Multipart copy** | Thread pool per part | Asyncio task per part |
 | **Best for** | Synchronous applications | Async frameworks (FastAPI, aiohttp, etc.) |
 
+### Threads
+
+The threads of `AioS3FileSystem` belong to `S3AioThreadPool` instances, not to the
+default executor of the event loop, so `loop.set_default_executor()` neither limits nor
+blocks them:
+
+- The filesystem has a pool of `max_workers` threads for the blocking calls of its
+  coroutines, such as `_info()`, `_cat_file()`, the requests of `_rm()`, and the part
+  copies of multipart copies. A multipart copy runs at most the `max_workers` of the
+  call of its part copies at once, and the copies together run at most the
+  `max_workers` of the filesystem.
+- Each file opened with `open()` or `open_async()` has a pool of its own, sized by the
+  `max_workers` of the call.
+
+The pools create threads as the calls need them, and each call runs with the caller's
+context variables, as with `asyncio.to_thread()`.
+In a forked child process, a pool creates new threads.
+
 ### Executor strategy
 
 `S3FileSystem` and `S3File` use a pluggable executor abstraction (`S3Executor`) for
 parallel operations. Two implementations are provided:
 
 - `S3ThreadPoolExecutor` — wraps `ThreadPoolExecutor` (default for sync usage)
-- `S3AioExecutor` — dispatches work via `asyncio.run_coroutine_threadsafe` + `asyncio.to_thread`
+- `S3AioExecutor` — dispatches work via `asyncio.run_coroutine_threadsafe` to an
+  `S3AioThreadPool` of its own
 
-`AioS3FileSystem` automatically uses `S3AioExecutor` for file handles, so multipart
-uploads and parallel range reads are dispatched through the event loop with
-`asyncio.to_thread()` instead of a separate `ThreadPoolExecutor` per file.
-At most `max_workers` of them run at once.
+`AioS3FileSystem` automatically uses `S3AioExecutor` for the file handles of `open()`,
+so multipart uploads and parallel range reads are dispatched through the event loop to
+a pool of `max_workers` threads of each file.
 An instance created with `asynchronous=True` has no event loop of its own, so its file handles use `S3ThreadPoolExecutor`.
 
+### Async file handles
+
+`open_async()` returns an `AioS3StreamedFile`, whose `read()`, `write()`, and `close()`
+are coroutines. It supports the `rb`, `wb`, `ab`, and `xb` modes:
+
+```python
+from pyathena.filesystem.s3_async import AioS3FileSystem
+
+fs = AioS3FileSystem(asynchronous=True)
+
+async with await fs.open_async("s3://my-bucket/data/file.bin", "wb") as f:
+    await f.write(data)
+
+async with await fs.open_async("s3://my-bucket/data/file.bin", "rb") as f:
+    data = await f.read()
+```
+
+- A read splits the requested range into ranges of the block size, and reads them in
+  parallel. The file has no read cache.
+- A write uploads each full block as a part of a multipart upload, with at most
+  `max_workers` parts in flight. A write that would upload another part awaits one of
+  them, so the event loop keeps running. A part found failed by that wait, or a part
+  copied from the existing object in an append that has failed before the write uploads
+  another part, stops the write with its error, and the multipart upload is aborted.
+  A part that fails later raises its error when the file is closed. The commit cancels
+  the parts that have not started as soon as one fails, and waits for the running ones
+  before it aborts the upload and raises the error.
+- Closing the file commits the written data, as `S3File` does.
+- Leaving an `async with` block with an exception, or cancelling a write, closes the file
+  without committing it, and aborts the multipart upload after the running parts finish.
+  The abort completes even if the task is cancelled again.
+- Once closing or the final flush has begun the commit, the commit completes even if
+  the task that closes or flushes the file is cancelled, and leaving an `async with`
+  block with an exception afterward does not abort it.
+- A file opened for writing must be closed with `await f.close()` or an `async with`
+  block. A file that is garbage collected before it is closed or its commit has begun
+  is not committed: it emits a `ResourceWarning`, and its multipart upload is aborted.
+  A synchronous `with` block raises `TypeError`.
+- `open_async()` raises `NotImplementedError` for the write modes inside a transaction,
+  and for `autocommit=False`, because the file commits when it is closed.
+
 ### Usage with AioS3FSCursor
 
 `AioS3FSCursor` automatically uses `AioS3FileSystem` internally. No additional

docs/api/filesystem.rst

@@ -22,6 +22,9 @@ S3 FileSystem
 .. autoclass:: pyathena.filesystem.s3.S3File
    :members:
 
+.. autoclass:: pyathena.filesystem.s3.S3FileTarget
+   :members:
+
 Async S3 FileSystem
 -------------------
 
@@ -31,6 +34,9 @@ Async S3 FileSystem
 .. autoclass:: pyathena.filesystem.s3_async.AioS3File
    :members:
 
+.. autoclass:: pyathena.filesystem.s3_async.AioS3StreamedFile
+   :members:
+
 Read-ahead Stream
 -----------------
 
@@ -49,6 +55,9 @@ S3 Executor
 .. autoclass:: pyathena.filesystem.s3_executor.S3AioExecutor
    :members:
 
+.. autoclass:: pyathena.filesystem.s3_executor.S3AioThreadPool
+   :members:
+
 S3 Paths
 --------
 

docs/filesystem.md

@@ -625,5 +625,6 @@ for source, destination in pairing.copy_pairs(sources):
 ## Async filesystem
 
 `AioS3FileSystem` provides the same functionality on top of fsspec's
-`AsyncFileSystem`, dispatching parallel operations through the asyncio event loop.
+`AsyncFileSystem`, running its blocking calls in thread pools sized by `max_workers`.
+Its `open_async()` returns a file whose reads and writes are coroutines.
 See {ref}`aio-s3-filesystem` for details.