Merge the filesystem thread pools into one S3ThreadPool type (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 `S3ThreadPoolExecutor`.
-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 through `S3AioExecutor`, dispatched on the filesystem's event loop, fsspec's IO thread, to a pool of `max_workers` threads of the file; 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 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.
 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

@@ -135,10 +135,10 @@ async def transfer(
     The synchronous filesystem and s3fs run on the calling thread; s3fs waits
     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 dispatch their range requests
-    and part uploads through ``S3AioExecutor`` to the filesystem's event loop,
-    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
+    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
+    ``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.
 
     Args:

docs/aio.md

@@ -282,14 +282,14 @@ its coroutines run the blocking boto3 calls in threads that PyAthena owns, sized
 |---|---|---|
 | **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()` |
+| **File handles** | `S3File` with thread pool | `AioS3File` with thread pool; `AioS3StreamedFile` from `open_async()` |
 | **Bulk delete** | Thread pool per batch | `asyncio.gather` per batch |
 | **Multipart copy** | Thread pool per copy | Thread pool of the filesystem |
 | **Best for** | Synchronous applications | Async frameworks (FastAPI, aiohttp, etc.) |
 
 ### Threads
 
-The threads of `AioS3FileSystem` belong to `S3AioThreadPool` instances, not to the
+The threads of `AioS3FileSystem` belong to `S3ThreadPool` instances, not to the
 default executor of the event loop, so `loop.set_default_executor()` neither limits nor
 blocks them:
 
@@ -332,22 +332,19 @@ multipart uploads.
 ### Executor strategy
 
 `S3FileSystem` and `S3File` use a pluggable executor abstraction (`S3Executor`) for
-parallel operations. Two implementations are provided:
-
-- `S3ThreadPoolExecutor` — runs the calls in a pool of `max_workers` threads (default
-  for sync usage); in a forked child process, the pool creates new threads
-- `S3AioExecutor` — dispatches work via `asyncio.run_coroutine_threadsafe` to an
-  `S3AioThreadPool` of its own
-
-`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`.
-
-In a forked child process, the event loop of the parent does not run.
-An `AioS3FileSystem` created in the parent without `asynchronous=True` raises
-`RuntimeError` from `open()` there, as fsspec's synchronous methods do, and so does
-`S3AioExecutor.submit()`; a filesystem created in the child works.
+parallel operations. `S3ThreadPool` queues the calls in the order they were submitted
+and runs them in a pool of `max_workers` threads; in a forked child process, the pool
+creates new threads. `S3ThreadPoolExecutor` and `S3AioThreadPool` are aliases of `S3ThreadPool`.
+`S3AioExecutor` is kept for compatibility: it is an `S3ThreadPool` that ignores its
+`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.
+
+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
+the event loop of the parent does not run in the child, so fsspec's synchronous methods
+of an `AioS3FileSystem` created in the parent, such as `ls()`, raise `RuntimeError` there.
 
 ### Async file handles
 

docs/api/filesystem.rst

@@ -49,13 +49,14 @@ S3 Executor
 .. autoclass:: pyathena.filesystem.s3_executor.S3Executor
    :members:
 
-.. autoclass:: pyathena.filesystem.s3_executor.S3ThreadPoolExecutor
+.. autoclass:: pyathena.filesystem.s3_executor.S3ThreadPool
    :members:
 
-.. autoclass:: pyathena.filesystem.s3_executor.S3AioExecutor
-   :members:
+.. autodata:: pyathena.filesystem.s3_executor.S3ThreadPoolExecutor
 
-.. autoclass:: pyathena.filesystem.s3_executor.S3AioThreadPool
+.. autodata:: pyathena.filesystem.s3_executor.S3AioThreadPool
+
+.. autoclass:: pyathena.filesystem.s3_executor.S3AioExecutor
    :members:
 
 S3 Paths

pyathena/filesystem/s3.py

@@ -36,7 +36,7 @@
 from pyathena.connection import Connection
 from pyathena.filesystem.s3_core import S3Core, S3DeleteBatch, S3DeleteResult
 from pyathena.filesystem.s3_errors import S3ClientError
-from pyathena.filesystem.s3_executor import S3Executor, S3ThreadPoolExecutor
+from pyathena.filesystem.s3_executor import S3Executor, S3ThreadPool
 from pyathena.filesystem.s3_object import (
     S3Metadata,
     S3MultipartUpload,
@@ -1099,18 +1099,19 @@ def _delete_paths(
         )
 
     def _create_executor(self, max_workers: int) -> S3Executor:
-        """Create an executor strategy for parallel operations.
+        """Create the executor of the parallel requests of a file or an operation.
 
-        Subclasses can override to provide alternative execution strategies
-        (e.g., asyncio-based execution).
+        The files of ``S3FileSystem`` and ``AioS3FileSystem`` and the
+        parallel operations of ``S3FileSystem`` take their executors from
+        this method.
 
         Args:
             max_workers: Maximum number of parallel workers.
 
         Returns:
-            An S3Executor instance.
+            A new ``S3ThreadPool``.
         """
-        return S3ThreadPoolExecutor(max_workers=max_workers)
+        return S3ThreadPool(max_workers=max_workers)
 
     def _delete_objects(self, paths: list[str], max_workers: int | None = None, **kwargs) -> None:
         """Delete objects with DeleteObjects requests grouped by bucket.
@@ -2882,7 +2883,7 @@ def __init__(
                 in flight, waiting for one of them to finish before it submits
                 another.
             executor: The executor for parallel operations. If None, a new
-                ``S3ThreadPoolExecutor`` is created.
+                ``S3ThreadPool`` is created.
             block_size: The block size for reads and writes. Must be between
                 ``S3Core.MULTIPART_UPLOAD_MIN_PART_SIZE`` and
                 ``S3Core.MULTIPART_UPLOAD_MAX_PART_SIZE``, inclusive, unless reading.
@@ -2928,7 +2929,7 @@ def __init__(
         self.s3_additional_kwargs: dict[str, Any] = target.request_kwargs
         self._details: S3Object | dict[str, Any] = target.details
 
-        self._executor: S3Executor = executor or S3ThreadPoolExecutor(max_workers=max_workers)
+        self._executor: S3Executor = executor or S3ThreadPool(max_workers=max_workers)
         # The requests and the state of the write; None when reading.
         self._session: _S3WriteSession | None = None
         if mode != "rb":

pyathena/filesystem/s3_async.py

@@ -32,12 +32,7 @@
 from pyathena.aio.util import create_in_thread
 from pyathena.filesystem.s3 import CompressedBuffer, S3File, S3FileSystem, S3FileTarget
 from pyathena.filesystem.s3_core import S3Core
-from pyathena.filesystem.s3_executor import (
-    S3AioExecutor,
-    S3AioThreadPool,
-    S3Executor,
-    S3ThreadPoolExecutor,
-)
+from pyathena.filesystem.s3_executor import S3ThreadPool
 from pyathena.filesystem.s3_object import (
     S3CompleteMultipartUpload,
     S3Metadata,
@@ -63,7 +58,7 @@ class AioS3FileSystem(AsyncFileSystem):
     """An async filesystem interface for Amazon S3 using fsspec's AsyncFileSystem.
 
     This class wraps ``S3FileSystem`` to provide native asyncio support. Its
-    coroutines run the blocking boto3 calls in an ``S3AioThreadPool`` with
+    coroutines run the blocking boto3 calls in an ``S3ThreadPool`` with
     ``max_workers`` threads that the filesystem owns, instead of the default
     executor of the event loop, and run parallel requests with
     ``asyncio.gather``.
@@ -72,11 +67,9 @@ class AioS3FileSystem(AsyncFileSystem):
     all boto3 calls, while this class delegates to it in the threads of its pool.
     This avoids diamond inheritance issues and keeps all boto3 logic in one place.
 
-    File handles created by ``_open`` use ``S3AioExecutor`` so that parallel
-    operations (range reads, multipart uploads) 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 a ``ThreadPoolExecutor``. :meth:`open_async` returns an
+    File handles created by ``_open`` run their parallel operations (range
+    reads, multipart uploads) in an ``S3ThreadPool`` of ``max_workers``
+    threads of each file, as ``S3File`` does. :meth:`open_async` returns an
     ``AioS3StreamedFile``, whose reads and writes are coroutines.
     :meth:`close`, or the end of a ``with`` or ``async with`` block, releases
     the threads of the filesystem, which stays usable.
@@ -146,7 +139,7 @@ def __init__(
         # The blocking calls of the coroutines run in threads of this pool,
         # not in the default executor of the event loop. Created first, so
         # that None takes its default for the files and copies too.
-        self._thread_pool = S3AioThreadPool(max_workers)
+        self._thread_pool = S3ThreadPool(max_workers)
         max_workers = self._thread_pool.max_workers
         self._sync_fs = S3FileSystem(
             connection=connection,
@@ -172,8 +165,8 @@ def core(self) -> S3Core:
     def close(self) -> None:
         """Release the threads of the filesystem, which stays usable, once they are idle.
 
-        See :meth:`S3AioThreadPool.release()
-        <pyathena.filesystem.s3_executor.S3AioThreadPool.release>`. Files opened
+        See :meth:`S3ThreadPool.release()
+        <pyathena.filesystem.s3_executor.S3ThreadPool.release>`. Files opened
         before have threads of their own and are not affected.
         """
         self._thread_pool.release()
@@ -836,28 +829,6 @@ async def _expand_path(self, path, recursive=False, maxdepth=None, **kwargs) ->
             raise FileNotFoundError(path)
         return sorted(out)
 
-    def _create_executor(self, max_workers: int) -> S3Executor:
-        """Create the executor for the parallel operations of a file.
-
-        An instance created with ``asynchronous=True`` has no event loop of
-        its own, so its files run the operations in a thread pool.
-
-        Args:
-            max_workers: The maximum number of operations that run at once.
-
-        Returns:
-            An ``S3AioExecutor`` on the event loop of this filesystem, or an
-            ``S3ThreadPoolExecutor`` if it has none.
-
-        Raises:
-            RuntimeError: If this instance has an event loop and the call is
-                made in a forked child process of the process that created
-                it, where the loop does not run.
-        """
-        if self._loop is None:
-            return S3ThreadPoolExecutor(max_workers=max_workers)
-        return S3AioExecutor(loop=self.loop, max_workers=max_workers)
-
     def _open(
         self,
         path: str,
@@ -888,7 +859,7 @@ def _open(
             path,
             mode,
             max_workers=max_workers,
-            executor=self._create_executor(max_workers=max_workers),
+            executor=self._sync_fs._create_executor(max_workers=max_workers),
             block_size=block_size,
             cache_type=cache_type,
             autocommit=autocommit,
@@ -1194,25 +1165,20 @@ def touch(self, path: str, truncate: bool = True, **kwargs) -> dict[str, Any]:
 
 
 class AioS3File(S3File):
-    """Async-aware S3 file handle using ``S3AioExecutor``.
-
-    Functionally identical to ``S3File``; exists as a distinct type for
-    ``isinstance`` checks and to document the async execution model.
-    All parallel operations (range reads, multipart uploads) are dispatched
-    through the ``S3Executor`` interface — the ``S3AioExecutor``
-    provided by ``AioS3FileSystem`` dispatches them through the event loop to
-    a pool of ``max_workers`` threads of the file.
-    For an ``AioS3FileSystem`` created with ``asynchronous=True``, it is an
-    ``S3ThreadPoolExecutor``. For reads and writes that are coroutines, see
-    ``AioS3StreamedFile``.
+    """The file handle that ``AioS3FileSystem.open()`` returns.
+
+    Functionally identical to ``S3File``, whose parallel operations (range
+    reads, multipart uploads) it runs in a pool of ``max_workers`` threads of
+    the file; it exists as a distinct type for ``isinstance`` checks. For
+    reads and writes that are coroutines, see ``AioS3StreamedFile``.
     """
 
 
 class AioS3StreamedFile(AbstractAsyncStreamedFile):
     """An S3 file whose reads and writes are coroutines.
 
     Returned by :meth:`AioS3FileSystem.open_async`. The requests of the file
-    run in an ``S3AioThreadPool`` of ``max_workers`` threads of its own, and
+    run in an ``S3ThreadPool`` of ``max_workers`` threads of its own, and
     the coroutines wait for them without blocking the event loop:
 
     - A read splits the requested range into ranges of the block size, read
@@ -1298,7 +1264,7 @@ def __init__(
         if mode not in {"rb", "wb", "ab", "xb"}:
             raise NotImplementedError("File mode not supported")
         # Created first, so that an invalid max_workers raises before the lookups.
-        self._pool = S3AioThreadPool(max_workers)
+        self._pool = S3ThreadPool(max_workers)
         self.max_workers = self._pool.max_workers
         target = S3FileTarget.resolve(
             fs._sync_fs,