Skip to content

Consolidate the filesystem executors before parallel batch operations #1332

Open

Description

@laughingman7743

Part of #1327 (design problem 1 of the audit report). Related to #1322 and #1323, which need a decision here.

Summary

Consolidate the filesystem's executors into one thread-pool type with one sizing and ownership policy, and one batch primitive, before #1322 adds concurrency to the batch operations.

Current state (766d79d)

  1. Four mechanisms. S3ThreadPoolExecutor (a ThreadPoolExecutor wrapper), S3AioThreadPool (a superset: lazy threads, release(), after-fork reset, run/run_shielded), S3AioExecutor (an S3AioThreadPool reached through run_coroutine_threadsafe on fsspec's loop), and ReadAheadStream's own ThreadPoolExecutor.
  2. The loop hop of S3AioExecutor adds failure modes and no function: it needs a running loop, hangs after a fork (Make files usable after fork with either filesystem #1336), a blocking wait on the loop thread never completes, and futures stay pending when the loop stops, while the write session waits on them; every call is wrapped in two _S3WorkItems. AioS3File overrides nothing and exists for this executor.
  3. Pools per call. The synchronous filesystem creates a pool per open file and per rm, recursive chmod, clear_multipart_uploads and multipart copy; the asyncio filesystem has one pool plus one per open() and per open_async(). Total threads are unbounded across concurrent files. A file shuts down an executor that its caller passed in, so pools cannot be shared.
  4. Sizing. max_workers defaults to cpu_count() * 5 in S3FileSystem, S3File and S3AioExecutor, and to min(32, cpu + 4) in S3AioThreadPool and AioS3FileSystem; neither is reconciled with botocore's max_pool_connections of 10 (Run the batch operations of S3FileSystem concurrently #1322 question 5).
  5. Nesting. AioS3FileSystem's delegated put_file/pipe_file/get_file run in a filesystem thread and open a file with its own pool: up to max_workers ** 2 parts in flight (16 concurrent UploadPart and 22 threads at max_workers=4). A call in an S3AioThreadPool thread that waits for work submitted to the same pool starves once every thread waits (max_workers=1: 0 of 2 done after 2 s). Run the batch operations of S3FileSystem concurrently #1322 must avoid both.
  6. Fan-out by hand. Track-before-schedule, cancel-before-stop and clear_frames are written at five sites (S3File._fetch_range, recursive chmod, AioS3StreamedFile._fetch_range, the asyncio annotation copies, the write session's _stop_requests), with different failure semantics; rm and clear_multipart_uploads wait for everything and stop nothing.

Proposal (for decision)

  1. One thread pool type (merge S3ThreadPoolExecutor into S3AioThreadPool under a neutral name, with aliases), used directly by both file classes; deprecate S3AioExecutor; keep AioS3File as a plain subclass. File work would then run to its end when fsspec's loop stops instead of being cancelled with it.
  2. One bounded pool per filesystem, shared by its files and batch operations, with max_workers as a per-call cap; files stop owning or shutting down executors. Size the pool against the client's max_pool_connections, or set the latter from max_workers when the filesystem builds the client.
  3. One batch primitive (for example map(fn, items, *, fail_fast) with an asyncio counterpart) that implements the fan-out rules once, for Run the batch operations of S3FileSystem concurrently #1322's batch operations and the five sites above.
  4. A range reader shared by both file classes and a pipelined get_file (Download in parallel ranges in get_file() #1323), with the range splitter on S3Core (merging the short tail, as part_ranges does).

Risks: S3AioExecutor and S3ThreadPoolExecutor are public and documented; head-of-line blocking between files sharing a pool; error order of batch operations.

Metadata

Metadata

Assignees

No one assigned

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions