Using the Python buffer protocol in `pipe`

[tomwhite]

Is it possible to avoid memory copies when writing to S3 by taking advantage of the Python buffer protocol?

In particular, it would be great if we could use a memoryview object in calls to pipe().

Using bytes and bytearray work fine:

>>> import numpy as np
>>> import fsspec
>>> fs = fsspec.filesystem('s3')
>>> fs.pipe("s3://cubed-unittest/mem-array/s3test", b"hello") # works
[{'ResponseMetadata': {'RequestId': 'E54C5V2H9M1ZE1V1', 'HostId': 'pcEW+k8KZHBbYuohqfgGNZOac8Y7QzRAHaAxOJmMWQdEll75umxh+On2DXxWi8qKMcG2+aZmJ7I=', 'HTTPStatusCode': 200, 'HTTPHeaders': {'x-amz-id-2': 'pcEW+k8KZHBbYuohqfgGNZOac8Y7QzRAHaAxOJmMWQdEll75umxh+On2DXxWi8qKMcG2+aZmJ7I=', 'x-amz-request-id': 'E54C5V2H9M1ZE1V1', 'date': 'Thu, 10 Apr 2025 13:19:42 GMT', 'x-amz-expiration': 'expiry-date="Sat, 12 Apr 2025 00:00:00 GMT", rule-id="gc"', 'x-amz-server-side-encryption': 'AES256', 'etag': '"5d41402abc4b2a76b9719d911017c592"', 'x-amz-checksum-crc32': 'NhCmhg==', 'x-amz-checksum-type': 'FULL_OBJECT', 'content-length': '0', 'server': 'AmazonS3'}, 'RetryAttempts': 0}, 'Expiration': 'expiry-date="Sat, 12 Apr 2025 00:00:00 GMT", rule-id="gc"', 'ETag': '"5d41402abc4b2a76b9719d911017c592"', 'ChecksumCRC32': 'NhCmhg==', 'ChecksumType': 'FULL_OBJECT', 'ServerSideEncryption': 'AES256'}]
>>> fs.pipe("s3://cubed-unittest/mem-array/s3test", bytearray(b"hello")) # works
[{'ResponseMetadata': {'RequestId': 'C58QYEAVJTBW96Y6', 'HostId': 'lwi6otZoaiTsPhLYApOdMmhFXphWEfhoYh/sRO4txe88uUnj8wLNHTgaleq7agiipYlyd8GHlzrxrZ8sAF1qQIuyXQUwm5Nq', 'HTTPStatusCode': 200, 'HTTPHeaders': {'x-amz-id-2': 'lwi6otZoaiTsPhLYApOdMmhFXphWEfhoYh/sRO4txe88uUnj8wLNHTgaleq7agiipYlyd8GHlzrxrZ8sAF1qQIuyXQUwm5Nq', 'x-amz-request-id': 'C58QYEAVJTBW96Y6', 'date': 'Thu, 10 Apr 2025 13:20:04 GMT', 'x-amz-expiration': 'expiry-date="Sat, 12 Apr 2025 00:00:00 GMT", rule-id="gc"', 'x-amz-server-side-encryption': 'AES256', 'etag': '"5d41402abc4b2a76b9719d911017c592"', 'x-amz-checksum-crc32': 'NhCmhg==', 'x-amz-checksum-type': 'FULL_OBJECT', 'content-length': '0', 'server': 'AmazonS3'}, 'RetryAttempts': 0}, 'Expiration': 'expiry-date="Sat, 12 Apr 2025 00:00:00 GMT", rule-id="gc"', 'ETag': '"5d41402abc4b2a76b9719d911017c592"', 'ChecksumCRC32': 'NhCmhg==', 'ChecksumType': 'FULL_OBJECT', 'ServerSideEncryption': 'AES256'}]

But using memoryview or just a numpy array (which supports the buffer protocol too) fail:

>>> fs.pipe("s3://cubed-unittest/mem-array/s3test", memoryview(np.frombuffer(bytearray(b"hello"), dtype="uint8")))
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/Users/tom/miniforge3/envs/memray-array/lib/python3.11/site-packages/fsspec/asyn.py", line 118, in wrapper
    return sync(self.loop, func, *args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/tom/miniforge3/envs/memray-array/lib/python3.11/site-packages/fsspec/asyn.py", line 103, in sync
    raise return_result
  File "/Users/tom/miniforge3/envs/memray-array/lib/python3.11/site-packages/fsspec/asyn.py", line 56, in _runner
    result[0] = await coro
                ^^^^^^^^^^
  File "/Users/tom/miniforge3/envs/memray-array/lib/python3.11/site-packages/fsspec/asyn.py", line 418, in _pipe
    return await _run_coros_in_chunks(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/tom/miniforge3/envs/memray-array/lib/python3.11/site-packages/fsspec/asyn.py", line 268, in _run_coros_in_chunks
    result, k = await done.pop()
                ^^^^^^^^^^^^^^^^
  File "/Users/tom/miniforge3/envs/memray-array/lib/python3.11/site-packages/fsspec/asyn.py", line 245, in _run_coro
    return await asyncio.wait_for(coro, timeout=timeout), i
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/tom/miniforge3/envs/memray-array/lib/python3.11/asyncio/tasks.py", line 452, in wait_for
    return await fut
           ^^^^^^^^^
  File "/Users/tom/miniforge3/envs/memray-array/lib/python3.11/site-packages/s3fs/core.py", line 1185, in _pipe_file
    out = await self._call_s3(
          ^^^^^^^^^^^^^^^^^^^^
  File "/Users/tom/miniforge3/envs/memray-array/lib/python3.11/site-packages/s3fs/core.py", line 371, in _call_s3
    return await _error_wrapper(
           ^^^^^^^^^^^^^^^^^^^^^
  File "/Users/tom/miniforge3/envs/memray-array/lib/python3.11/site-packages/s3fs/core.py", line 146, in _error_wrapper
    raise err
  File "/Users/tom/miniforge3/envs/memray-array/lib/python3.11/site-packages/s3fs/core.py", line 114, in _error_wrapper
    return await func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/tom/miniforge3/envs/memray-array/lib/python3.11/site-packages/aiobotocore/client.py", line 369, in _make_api_call
    request_dict = await self._convert_to_request_dict(
                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/tom/miniforge3/envs/memray-array/lib/python3.11/site-packages/aiobotocore/client.py", line 440, in _convert_to_request_dict
    request_dict = self._serializer.serialize_to_request(
                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/tom/miniforge3/envs/memray-array/lib/python3.11/site-packages/botocore/validate.py", line 381, in serialize_to_request
    raise ParamValidationError(report=report.generate_report())
botocore.exceptions.ParamValidationError: Parameter validation failed:
Invalid type for parameter Body, value: <memory at 0x102f99a80>, type: <class 'memoryview'>, valid types: <class 'bytes'>, <class 'bytearray'>, file-like object

This may be a limitation in the underlying boto library, but perhaps there is a way to avoid copying the source data (in this case a numpy array) that I'm missing.

cc @jakirkham who commented about this in zarr-developers/zarr-python#2972 (comment)

[jakirkham]

Thanks Tom! 🙏

It seems like the error is coming from botocore

The traceback cuts short of the indicating the specific function, but reading the tea leaves leads me to conclude this botocore input validation function is involved

Now fixing that could resolve the issue. OTOH there may be other assumptions made in the code that are less clear. That said, there usually is a path forward with memoryviews (usually with some tweaks)

Looking around found a similar upstream issue: boto/boto3#3423

There is some work to fix that in PR: boto/botocore#3107

Unfortunately there has not been a lot of engagement as of yet. Tried to do a rough review of the PR to see if that might help move things along

[tomwhite]

Thanks for digging into this @jakirkham!

I tried rebasing the botocore PR on main then running the fsspec test in this issue again. I got:

Traceback (most recent call last):
  File "/Users/tom/workspace/botocore/botocore/awsrequest.py", line 537, in reset_stream
    self.body.seek(0)
  File "/Users/tom/workspace/botocore/botocore/httpchecksum.py", line 183, in seek
    self._raw.seek(0)
    ^^^^^^^^^^^^^^
AttributeError: 'memoryview' object has no attribute 'seek'

This is because the body is a aiobotocore.httpchecksum.AioAwsChunkedWrapper, not just bytes. So I worked around that, then got:

Traceback (most recent call last):
  File "/Users/tom/miniforge3/envs/memray-array/lib/python3.11/site-packages/aiohttp/client_reqrep.py", line 658, in write_bytes
    await self.body.write(writer)
  File "/Users/tom/miniforge3/envs/memray-array/lib/python3.11/site-packages/aiohttp/payload.py", line 494, in write
    chunk = await self._iter.__anext__()
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/tom/miniforge3/envs/memray-array/lib/python3.11/site-packages/aiobotocore/httpchecksum.py", line 68, in __anext__
    return await self._make_chunk()
           ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/tom/miniforge3/envs/memray-array/lib/python3.11/site-packages/aiobotocore/httpchecksum.py", line 49, in _make_chunk
    raw_chunk = await resolve_awaitable(self._raw.read(self._chunk_size))
                                        ^^^^^^^^^^^^^^
AttributeError: 'memoryview' object has no attribute 'read'

So it looks like memoryview is not sufficient.

BTW I also stumbled across this article: https://pythonspeed.com/articles/bytesio-reduce-memory-usage/, with the intriguing suggestion that using BytesIO can avoid memory copies (if you start with a BytesIO object, at least). But afaict there's no way to go from a numpy array to BytesIO without a copy, which is what we need in the Zarr case that started this discussion...

[martindurant]

I have a couple of suggestions to try:

• subclassing like

class ErsatzBytes(memoryview, bytes):

which might just "do the right thing", depending on what botocore actually does with it.
This is enough to pass the isinstance check.

• using one of the classes like RustyBuffer in cramjam ( https://docs.rs/cramjam/latest/cramjam/io/struct.RustyBuffer.html ), which are designed to reflect numpy/memoryviews without copy, be buffer protocol compatible, and also have seek/read methods.

[tomwhite]

Thank you for the suggestions @martindurant!

I tried using cramjam's Buffer to wrap a NumPy array, but it seems to copy the bytes on construction. This also happens when using a NumPy array wrapped in a memoryview.

So that leaves ErsatzBytes which may be possible, but needs more investigation.

[martindurant]

@milesgranger - is there a correct invocation to use one of the cramjam buffer types to pass numpy/memoryview data to a third party with zero copy? That third party officially accepts bytes, bytearray and BytesIO.

[milesgranger]

Cramjam's (only) Buffer type owns its data https://github.com/milesgranger/cramjam/blob/ebc9ce798c8f86871a32cb8bf9e515e45468f77c/src/io.rs#L342, but of course implements buffer protocol should one want to pass its data to a third-party with zero copy.

Anyhow, interesting and a very plausible implementation detail to support this behavior, but not directly applicable for cramjam as all the compression bits (which is cramjam's...jam.) implements zero-copy input and optionally zero-copy output w/ its _into functions, so not directly relevant for cramjam.Buffer to implement a zero copy overlay to numpy arrays or anything else for that matter.

[martindurant]

Yeah, it seems like the pyo3 ecosystem is now at the point where it can provide some nice behind-the-scenes benefits like zero-copy where otherwise it wouldn't work. Not sure, then, how to actually make that happen!

I believe the rationale for having bytes (and str) own its memory and not be mutable, is that they han then be hashed in dict/sets. Many applications don't care about that, though.

[milesgranger]

I agree, and partly why I'm quite unmotivated to remove cramjam's current ability to overwrite bytes as an output destination... I should clarify, I'm definitely open to adding this functionality! Seems most of the pieces are already there and if even a few people would find it helpful, I think it's relatively low overhead to add it.

[martindurant]

Well that's great! It still might not solve the original problem, though, depending on exactly what isinstance or similar checks are done within botocore. Still, once such a thing is possible, people ought to start using it.

[kylebarron]

FWIW the obstore.Bytes type allows externally-allocated memory via the buffer protocol (and via pyo3-bytes) and is slowly progressing to support all the methods on bytes.

We could probably mix and match that and the RustyBuffer methods to get you what you want.

[martindurant]

Good to hear, @kylebarron . Does it inherit from Bytes too?

[kylebarron]

Does it inherit from Bytes too?

You mean subclass? That's not possible because it doesn't use builtins.bytes internally. But it can match the bytes methods as a protocol as closely as possible.

[martindurant]

For the purposes of the use here, it might be useful to have a pretend subclass (obstore.Bytes, builtins.bytes) for the purposes of third-party libraries that use issubclass to make decisions.

[kylebarron]

I think that's a valid choice for downstream users but not something that obstore.Bytes should implement itself.