Share one request pool per filesystem among its files and operations (GH-1332)

benchmarks/README.md

@@ -238,8 +238,8 @@ 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.
 A `get` also writes the object to the trial's temporary directory, which the parent process measures as the peak temp dir and removes; on the Amazon Linux 2023 hosts, that directory is on the tmpfs `/tmp`, so the object stays in memory outside the process's RSS. A `put` reads a sparse file, which no disk bounds.
 
-The `sync` API calls the filesystem on the trial's thread, and a file's parallel requests run in an `S3ThreadPool`.
-The `aio` API runs a read, get, write, or put in `asyncio.to_thread()`, as `AioS3FSCursor` reads result files, and a file's parallel requests run in a pool of `max_workers` threads of the file, as in the `sync` API; its `get_file()` and `put_file()` run the synchronous filesystem's in the filesystem's thread pool.
+The `sync` API calls the filesystem on the trial's thread, and a file's parallel requests run in the request pool of the filesystem, which its files and operations share.
+The `aio` API runs a read, get, write, or put in `asyncio.to_thread()`, as `AioS3FSCursor` reads result files, and a file's parallel requests run in the request pool of the filesystem, as in the `sync` API; its `get_file()` and `put_file()` run the synchronous filesystem's in the filesystem's thread pool.
 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.

benchmarks/pyathena_bench/filesystem.py

@@ -136,7 +136,7 @@ async def transfer(
     for each operation, a coroutine on fsspec's IO loop. As ``AioS3FSCursor``
     reads result files, the asyncio filesystem's file and local-file transfers
     run in ``asyncio.to_thread()``: its files run their range requests and
-    part uploads in a pool of ``max_workers`` threads of the file, and its
+    part uploads in the request pool of the filesystem, and its
     ``get_file()`` and ``put_file()`` run the synchronous filesystem's in the
     filesystem's thread pool. Its copy is a coroutine awaited on the
     trial's event loop.

docs/aio.md

@@ -294,23 +294,29 @@ default executor of the event loop, so `loop.set_default_executor()` neither lim
 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.
+  coroutines, such as `_info()`, `_cat_file()`, `_put_file()`, and the completion and
+  cleanup of a multipart copy, and for the commit and abort of the files of
+  `open_async()`.
+- Its internal `S3FileSystem` has a request pool of `max_workers` threads for the
+  requests that these calls and the files run in parallel: the part copies of multipart
+  copies and their annotation copies, the DeleteObjects batches of `_rm()`, and the
+  range reads and parts of the files of `open()` and `open_async()`. A file or a copy
+  runs at most its own `max_workers` requests at once, and all of them together run at
+  most the `max_workers` of the filesystem.
+
+`max_workers` defaults to the `max_pool_connections` of the S3 client, 10 unless
+configured, as for `S3FileSystem`.
 
 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.
 
-`close()` releases the threads of the filesystem once no call runs or waits in them, and
+`close()` releases the threads of both pools once no call runs or waits in them, and
 so does the end of a `with` or `async with` block. Until then, calls, including those made
 after `close()`, run in those threads within `max_workers`, and the calls in progress,
 including the cleanup of a multipart copy, run to their end. The filesystem stays usable,
 also when other callers share it through fsspec's instance cache: a call after the release
-creates new threads. Files opened before keep their own threads. Without `close()`, the
+creates new threads. The requests of open files run to their end. Without `close()`, the
 threads are released when the filesystem is garbage collected, which a filesystem in
 fsspec's instance cache is not until `clear_instance_cache()`.
 
@@ -339,7 +345,8 @@ creates new threads. `S3ThreadPoolExecutor` and `S3AioThreadPool` are aliases of
 `loop` argument, with a `DeprecationWarning` if one is given.
 
 The file handles of `AioS3FileSystem.open()` run their multipart uploads and parallel
-range reads in a pool of `max_workers` threads of each file, as `S3File` does.
+range reads in the request pool of the filesystem, at most the `max_workers` of the
+file at once, as `S3File` does.
 
 In a forked child process, create a new filesystem. A filesystem created in the parent
 keeps the parent's boto3 client, which is not meant to be shared across processes, and
@@ -386,7 +393,10 @@ async with await fs.open_async("s3://my-bucket/data/file.bin", "rb") as f:
   upload; leaving the block with an exception after a cancelled final flush, while that
   abort runs, waits for it. An interrupt while the commit is being queued, such as a
   repeated Ctrl+C under `asyncio.run()`, closes the file without committing it and
-  aborts its multipart upload, if any.
+  aborts its multipart upload, if any. A further interrupt during that cleanup can leave
+  the multipart upload incomplete; the close still does not wait forever.
+- Closing a file, or leaving its `async with` block, waits for its range reads, commit,
+  and abort, also if the task is cancelled, and raises the cancellation after they end.
 - 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.

docs/filesystem.md

@@ -100,6 +100,33 @@ transaction commits, and the multipart upload is aborted.
 `max_workers` is an argument of `S3FileSystem`, `AioS3FileSystem`, `open`, `put`, and
 `pipe`.
 
+The requests that the files and operations of a filesystem run in parallel (range
+reads, parts, part copies, the ACLs of a recursive `chmod`, the DeleteObjects batches of
+`rm`, and the aborts of `clear_multipart_uploads`) share one request pool of the
+filesystem, with `max_workers` threads. Each file or operation runs at most its own
+`max_workers` requests at once in that pool. The `max_workers` of the filesystem
+defaults to the `max_pool_connections` of its S3 client: 10 unless configured, or that
+of the client of the connection given with `connection`. A client that the filesystem
+builds takes its `max_pool_connections` from `max_workers` unless `config_kwargs` sets
+it. For more parallel requests, raise `max_workers`, and with `connection`, also the
+`max_pool_connections` of its S3 client, such as with its `s3_config`. `close()`, or the end of a `with` block, releases the threads of the pool once they
+are idle, and the filesystem stays usable.
+
+In a forked child process, create a new filesystem. A file that the parent opened raises
+`RuntimeError` there from its reads, writes, and flushes, `commit()`, and `discard()`,
+and closing it only marks it closed, without uploading or aborting anything.
+
+An interrupt, such as Ctrl+C, while a file reads ranges in parallel cancels the range
+reads that have not started. Those that are running finish, and their data is released
+once they end. Closing the file waits for its running requests; an interrupt of that
+wait stops it, and the requests still finish. An interrupt while a
+request is being submitted can let one more request than `max_workers` run at once
+until the request pool is idle. These guarantees hold for one interrupt at a time. If another interrupt arrives while the cleanup of an interrupt or
+an error runs, the call still does not wait forever, but the cleanup may stop early:
+the data of finished requests can stay referenced until the garbage collector frees it,
+and a multipart upload can be left incomplete. `clear_multipart_uploads` removes such
+uploads.
+
 A file returned by `open` whose write or commit has failed, or which has been
 discarded, cannot be committed: its `commit()` raises `S3MultipartStateError`. Inside a
 transaction, a write error that the caller catches therefore makes the transaction fail
@@ -642,6 +669,7 @@ for source, destination in pairing.copy_pairs(sources):
 ## Async filesystem
 
 `AioS3FileSystem` provides the same functionality on top of fsspec's
-`AsyncFileSystem`, running its blocking calls in thread pools sized by `max_workers`.
+`AsyncFileSystem`, running its blocking calls in a thread pool sized by `max_workers`,
+and its parallel requests in the request pool of its internal `S3FileSystem`.
 Its `open_async()` returns a file whose reads and writes are coroutines.
 See {ref}`aio-s3-filesystem` for details.

docs/usage.md

@@ -59,11 +59,10 @@ The reader rejects an unknown name when it reads the query results, which PyAthe
 
 ## Worker settings in PyAthena 4.0
 
-The pandas and Polars cursors take the number of threads that read each result file from S3 as `s3_max_workers`.
+The pandas and Polars cursors take, as `s3_max_workers`, the size of the request pool of the S3 filesystem that reads the result files; a file reads at most that many ranges at once.
 It can be set in the cursor constructor and overridden for one query in `execute()`.
-The default is `(cpu_count() or 1) * 5`.
-Polars uses it only for CSV results read without `chunksize`.
-Chunked CSV reads and UNLOAD results use Polars' native readers, which do not use this setting.
+The default is None, which takes the `max_pool_connections` of the connection's S3 client: 10, unless set with `s3_config`.
+Polars uses it for CSV results read without `chunksize`, and for UNLOAD results read in chunks through PyArrow without read options; its other readers do not use this setting.
 
 The thread-pool cursors, such as `AsyncCursor`, `AsyncPandasCursor`, and `AsyncPolarsCursor`, keep `max_workers` for the size of the thread pool that waits for queries and collects their results.
 

pyathena/aio/pandas/cursor.py

@@ -6,7 +6,6 @@
 import logging
 import operator
 from collections.abc import AsyncGenerator, Callable, Iterable
-from multiprocessing import cpu_count
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -65,7 +64,7 @@ def __init__(
         chunksize: int | None = None,
         block_size: int | None = None,
         cache_type: str | None = None,
-        s3_max_workers: int = (cpu_count() or 1) * 5,
+        s3_max_workers: int | None = None,
         result_reuse_enable: bool = False,
         result_reuse_minutes: int = CursorIterator.DEFAULT_RESULT_REUSE_MINUTES,
         auto_optimize_chunksize: bool = False,
@@ -95,7 +94,10 @@ def __init__(
                 the next block while pandas parses the current one.
                 Without ``cache_type``, PyAthena's filesystem reads a Parquet file of a chunked
                 UNLOAD result that is larger than 32 MiB without a cache.
-            s3_max_workers: Maximum number of workers of the S3 filesystem.
+            s3_max_workers: The size of the request pool of the S3 filesystem
+                that reads the result files. If None, the
+                ``max_pool_connections`` of the connection's S3 client (10
+                unless set with ``s3_config``).
             result_reuse_enable: Whether to enable Athena query result reuse.
             result_reuse_minutes: Maximum age of a reused query result in minutes.
             auto_optimize_chunksize: Whether to choose a chunk size from the size of the

pyathena/aio/polars/cursor.py

@@ -6,7 +6,6 @@
 import logging
 import operator
 from collections.abc import AsyncGenerator, Callable
-from multiprocessing import cpu_count
 from typing import TYPE_CHECKING, Any
 
 from pyathena.aio.common import WithAsyncFetch
@@ -62,7 +61,7 @@ def __init__(
         result_reuse_minutes: int = CursorIterator.DEFAULT_RESULT_REUSE_MINUTES,
         block_size: int | None = None,
         cache_type: str | None = None,
-        s3_max_workers: int = (cpu_count() or 1) * 5,
+        s3_max_workers: int | None = None,
         chunksize: int | None = None,
         **kwargs,
     ) -> None:
@@ -87,7 +86,10 @@ def __init__(
                 ``"background"``; see ``S3FileSystem``. Without ``cache_type``, PyAthena's
                 filesystem reads a Parquet file of a chunked UNLOAD result that is larger than
                 32 MiB without a cache.
-            s3_max_workers: Maximum number of workers of the S3 filesystem.
+            s3_max_workers: The size of the request pool of the S3 filesystem
+                that reads the result files. If None, the
+                ``max_pool_connections`` of the connection's S3 client (10
+                unless set with ``s3_config``).
             chunksize: Number of rows per chunk. If set, result files in S3 are read
                 lazily in chunks of this size.
             **kwargs: Other cursor arguments, such as ``connection`` and ``arraysize``,