Skip to content

Align AioS3FileSystem with S3FileSystem for cancellation, recursive put, expand_path, and arguments #1337

Open

Description

@laughingman7743

Part of #1327 (defects 14 to 18 of the audit report).

Summary

AioS3FileSystem behaves differently from S3FileSystem for the same calls in five ways. Each was reproduced offline at 766d79d.

Defects

  1. A cancelled _put_file, _pipe_file or _get_file keeps running and commits. They run the synchronous method in a pool thread (s3_async.py:271-273), and S3AioThreadPool.run only stops retries of a running call. With asyncio.wait_for(fs._put_file(...), 0.1) on a 21 MiB file (5 MiB blocks, one worker), the caller got TimeoutError after CreateMultipartUpload and one UploadPart; the thread then sent four more UploadPart requests and CompleteMultipartUpload, and the object existed. fsspec's fail-fast _put/_get cancel the other coroutines, whose transfers continue the same way. A cancelled multipart _cp_file and a cancelled AioS3StreamedFile write abort instead. docs/aio.md says a request already sent runs to its end, not that new requests and the commit are sent.
  2. Recursive put() creates buckets. fsspec's AsyncFileSystem._put calls _makedirs(d, exist_ok=True) for every local directory, which runs S3FileSystem.mkdir (s3.py:1199-1259). For put(localdir/, "s3://bucket/dst/", recursive=True) with a subdirectory: an extra HeadBucket when the bucket exists; two concurrent HeadBucket and CreateBucket when it is missing and allow_bucket_creation=True; PermissionError: Bucket creation is disabled before any upload by default. The synchronous put sends only the PutObject requests.
  3. expand_path(..., **kwargs) raises TypeError. AioS3FileSystem._expand_path forwards keyword arguments to fsspec's AsyncFileSystem._expand_path(path, recursive, maxdepth, assume_literal), which accepts none (s3_async.py:828-830); expand_path("s3://bucket/dir", recursive=True, refresh=True) or prefix=... work with the synchronous filesystem. The missing **kwargs is fsspec's and should be reported there; delegating _expand_path to the synchronous expand_path, as _find already is, is consistent delegation rather than a workaround.
  4. Arguments that one filesystem accepts are sent to S3 by the other, raising ParamValidationError: rm(..., max_workers=N) (sync _delete_objects takes it, aio passes it to DeleteObjects, s3_async.py:372-386); cp_file(..., recursive=..., maxdepth=..., on_error=...) (documented as unused by sync, passed to CopyObject by aio, s3_async.py:535-549); cat(paths, batch_size=N) (accepted by aio, passed to GetObject by sync). Neither docstring documents max_workers for rm.
  5. S3FileSystem(max_workers=None) is accepted, against its int annotation, and a multi-block read then raises TypeError: '>' not supported between instances of 'NoneType' and 'int' (S3File._get_ranges); AioS3FileSystem resolves None to its default.

Fix direction

Abort a delegated upload when its coroutine is cancelled (or drive the write from the coroutine through the write session, as the multipart copy does), override _put so directories are not passed to _makedirs, delegate _expand_path and report the signature upstream, agree on the keyword arguments of rm, cp_file and the batch operations for both filesystems, and resolve max_workers=None in S3FileSystem.

Metadata

Metadata

Assignees

No one assigned

    Labels

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions