Repository navigation
Align AioS3FileSystem with S3FileSystem for cancellation, recursive put, expand_path, and arguments #1337
Copy link
Copy link
Open
Labels
Description
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
- A cancelled
_put_file,_pipe_fileor_get_filekeeps running and commits. They run the synchronous method in a pool thread (s3_async.py:271-273), andS3AioThreadPool.runonly stops retries of a running call. Withasyncio.wait_for(fs._put_file(...), 0.1)on a 21 MiB file (5 MiB blocks, one worker), the caller gotTimeoutErrorafter CreateMultipartUpload and one UploadPart; the thread then sent four more UploadPart requests and CompleteMultipartUpload, and the object existed. fsspec's fail-fast_put/_getcancel the other coroutines, whose transfers continue the same way. A cancelled multipart_cp_fileand a cancelledAioS3StreamedFilewrite abort instead.docs/aio.mdsays a request already sent runs to its end, not that new requests and the commit are sent. - Recursive
put()creates buckets. fsspec'sAsyncFileSystem._putcalls_makedirs(d, exist_ok=True)for every local directory, which runsS3FileSystem.mkdir(s3.py:1199-1259). Forput(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 andallow_bucket_creation=True;PermissionError: Bucket creation is disabledbefore any upload by default. The synchronousputsends only the PutObject requests. expand_path(..., **kwargs)raisesTypeError.AioS3FileSystem._expand_pathforwards keyword arguments to fsspec'sAsyncFileSystem._expand_path(path, recursive, maxdepth, assume_literal), which accepts none (s3_async.py:828-830);expand_path("s3://bucket/dir", recursive=True, refresh=True)orprefix=...work with the synchronous filesystem. The missing**kwargsis fsspec's and should be reported there; delegating_expand_pathto the synchronousexpand_path, as_findalready is, is consistent delegation rather than a workaround.- Arguments that one filesystem accepts are sent to S3 by the other, raising
ParamValidationError:rm(..., max_workers=N)(sync_delete_objectstakes 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 documentsmax_workersforrm. S3FileSystem(max_workers=None)is accepted, against itsintannotation, and a multi-block read then raisesTypeError: '>' not supported between instances of 'NoneType' and 'int'(S3File._get_ranges);AioS3FileSystemresolvesNoneto 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.
Activity