From c024951fea03dbc2ce347b2de0709899769598cb Mon Sep 17 00:00:00 2001 From: yuxin00j Date: Wed, 4 Nov 2026 03:08:35 +0000 Subject: [PATCH 1/4] test(storage): add Rapid system test for zero-copy BidiReadObject reads Exercise the opt-in zero-copy BidiReadObjectResponse parser (GCS_PYTHON_CLIENT_ZERO_COPY_BIDI_READ=1) against a Rapid bucket over a single AsyncMultiRangeDownloader stream: zero-copy RPC selection, bytes chunks for default write() destinations, memoryview chunks with accepts_memoryview=True when google-crc32c accepts memoryview, multi-message reads into bytearray and writable memoryview buffers with guard bytes, and BufferError on a read clamped at EOF without closing the stream. --- .../tests/system/test_rapid.py | 103 ++++++++++++++++++ 1 file changed, 103 insertions(+) diff --git a/packages/google-cloud-storage/tests/system/test_rapid.py b/packages/google-cloud-storage/tests/system/test_rapid.py index 038203cb5103..caa3ce53a2b7 100644 --- a/packages/google-cloud-storage/tests/system/test_rapid.py +++ b/packages/google-cloud-storage/tests/system/test_rapid.py @@ -27,6 +27,7 @@ from google.cloud import kms from google.cloud.storage._helpers import _parse_bool_env +from google.cloud.storage.asyncio import _zero_copy_bidi_read from google.cloud.storage.asyncio.async_appendable_object_writer import ( _DEFAULT_FLUSH_INTERVAL_BYTES, AsyncAppendableObjectWriter, @@ -1374,6 +1375,108 @@ async def _run(): event_loop.run_until_complete(_run()) +def test_mrd_zero_copy_bidi_read( + storage_client, + blobs_to_delete, + event_loop, + grpc_client_direct, + monkeypatch, +): + """Tests zero-copy BidiReadObject reads across write() and buffer destinations.""" + monkeypatch.setenv("GCS_PYTHON_CLIENT_ZERO_COPY_BIDI_READ", "1") + object_size = 5 * 1024 * 1024 + object_name = f"test_mrd_zero_copy-{uuid.uuid4()}" + + class _RecordingSink: + def __init__(self): + self.chunks = [] + + def write(self, chunk): + self.chunks.append(chunk) + return len(chunk) + + async def _run(): + object_data = os.urandom(object_size) + + writer = AsyncAppendableObjectWriter( + grpc_client_direct, + _BUCKET_UNDER_TEST, + object_name, + storage_class="RAPID" if RCU_SYSTEM_TESTS else None, + ) + await writer.open() + await writer.append(object_data) + await writer.close(finalize_on_close=True) + blobs_to_delete.append( + storage_client.bucket(_BUCKET_UNDER_TEST).blob(object_name) + ) + + async with AsyncMultiRangeDownloader( + grpc_client_direct, _BUCKET_UNDER_TEST, object_name + ) as mrd: + transport = grpc_client_direct.grpc_client._client._transport + assert mrd.read_obj_str.rpc is _zero_copy_bidi_read.wrapped_rpc(transport) + + # Default write() destination receives bytes. + default_sink = _RecordingSink() + await mrd.download_ranges([(0, 0, default_sink)]) + assert all(type(chunk) is bytes for chunk in default_sink.chunks) + assert b"".join(default_sink.chunks) == object_data + + # accepts_memoryview=True receives memoryview when crc32c supports it. + sink = _RecordingSink() + await mrd.download_ranges([(1024 * 1024, 0, sink)], accepts_memoryview=True) + non_empty_chunks = [chunk for chunk in sink.chunks if chunk] + assert len(non_empty_chunks) >= 2 + expected_type = memoryview if _zero_copy_bidi_read.is_supported() else bytes + assert all(isinstance(chunk, expected_type) for chunk in non_empty_chunks) + assert b"".join(sink.chunks) == object_data[1024 * 1024 :] + + # Multi-chunk (> 2 MiB) writable buffer destinations. + range_len = 2 * 1024 * 1024 + 512 * 1024 + raw_buf = bytearray(range_len) + sentinel = b"Z" * 16 + backing = bytearray(sentinel + bytes(range_len) + sentinel) + slice_view = memoryview(backing)[16 : 16 + range_len] + try: + await mrd.download_ranges( + [ + (256 * 1024, range_len, raw_buf), + (2 * 1024 * 1024, range_len, slice_view), + ] + ) + finally: + slice_view.release() + + assert bytes(raw_buf) == object_data[256 * 1024 : 256 * 1024 + range_len] + assert bytes(backing[:16]) == sentinel + assert ( + bytes(backing[16 : 16 + range_len]) + == object_data[2 * 1024 * 1024 : 2 * 1024 * 1024 + range_len] + ) + assert bytes(backing[16 + range_len :]) == sentinel + raw_buf.clear() + backing.clear() + + # Short read past EOF into a buffer raises BufferError without closing mrd. + short_buf = bytearray(4096) + with pytest.raises(BufferError) as exc_info: + await mrd.download_ranges([(object_size - 1024, 4096, short_buf)]) + + assert exc_info.tb is not None + assert mrd.is_stream_open + short_buf.clear() + + tail_buf = bytearray(1024) + await mrd.download_ranges([(object_size - 1024, 1024, tail_buf)]) + assert bytes(tail_buf) == object_data[-1024:] + + del writer + gc.collect() + + event_loop.run_until_complete(_run()) + + def test_finalize_with_correct_checksum( storage_client, blobs_to_delete, From 72c680866ea1c996e0130a99b5a072f53ebca3b3 Mon Sep 17 00:00:00 2001 From: yuxin00j Date: Wed, 4 Nov 2026 03:22:17 +0000 Subject: [PATCH 2/4] test(storage): use isinstance and check the short-read error message --- packages/google-cloud-storage/tests/system/test_rapid.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/google-cloud-storage/tests/system/test_rapid.py b/packages/google-cloud-storage/tests/system/test_rapid.py index caa3ce53a2b7..37cdb8c7ecf4 100644 --- a/packages/google-cloud-storage/tests/system/test_rapid.py +++ b/packages/google-cloud-storage/tests/system/test_rapid.py @@ -1420,7 +1420,7 @@ async def _run(): # Default write() destination receives bytes. default_sink = _RecordingSink() await mrd.download_ranges([(0, 0, default_sink)]) - assert all(type(chunk) is bytes for chunk in default_sink.chunks) + assert all(isinstance(chunk, bytes) for chunk in default_sink.chunks) assert b"".join(default_sink.chunks) == object_data # accepts_memoryview=True receives memoryview when crc32c supports it. @@ -1463,7 +1463,7 @@ async def _run(): with pytest.raises(BufferError) as exc_info: await mrd.download_ranges([(object_size - 1024, 4096, short_buf)]) - assert exc_info.tb is not None + assert str(exc_info.value).endswith("got 1024") assert mrd.is_stream_open short_buf.clear() From 7fad3410f94ce5b10d38ab7465904d61ca13044d Mon Sep 17 00:00:00 2001 From: yuxin00j Date: Wed, 4 Nov 2026 04:35:07 +0000 Subject: [PATCH 3/4] test(storage): check the object size after open and explain the exc_info binding --- packages/google-cloud-storage/tests/system/test_rapid.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/packages/google-cloud-storage/tests/system/test_rapid.py b/packages/google-cloud-storage/tests/system/test_rapid.py index 37cdb8c7ecf4..a97cf983a8a8 100644 --- a/packages/google-cloud-storage/tests/system/test_rapid.py +++ b/packages/google-cloud-storage/tests/system/test_rapid.py @@ -1416,6 +1416,7 @@ async def _run(): ) as mrd: transport = grpc_client_direct.grpc_client._client._transport assert mrd.read_obj_str.rpc is _zero_copy_bidi_read.wrapped_rpc(transport) + assert mrd.persisted_size == object_size # Default write() destination receives bytes. default_sink = _RecordingSink() @@ -1460,6 +1461,8 @@ async def _run(): # Short read past EOF into a buffer raises BufferError without closing mrd. short_buf = bytearray(4096) + # exc_info keeps download_ranges' frame alive, so short_buf.clear() only + # succeeds if the SDK released its view of the buffer itself. with pytest.raises(BufferError) as exc_info: await mrd.download_ranges([(object_size - 1024, 4096, short_buf)]) From e4dbd542e37e81f8137b42ca00185edeebf6ec6f Mon Sep 17 00:00:00 2001 From: yuxin00j Date: Wed, 4 Nov 2026 05:53:54 +0000 Subject: [PATCH 4/4] test(storage): run the Rapid read system tests with the zero-copy parser and accepts_memoryview on and off --- .../tests/system/test_rapid.py | 148 +++++++++++++++--- 1 file changed, 124 insertions(+), 24 deletions(-) diff --git a/packages/google-cloud-storage/tests/system/test_rapid.py b/packages/google-cloud-storage/tests/system/test_rapid.py index a97cf983a8a8..2f77fb05d577 100644 --- a/packages/google-cloud-storage/tests/system/test_rapid.py +++ b/packages/google-cloud-storage/tests/system/test_rapid.py @@ -169,6 +169,21 @@ def grpc_client_direct(grpc_clients): return grpc_clients[True] +@pytest.fixture(params=[False, True], ids=["zero_copy_off", "zero_copy_on"]) +def zero_copy_bidi_read(request, monkeypatch): + """Turns the zero-copy BidiReadObject parser off or on for the test.""" + monkeypatch.setenv( + "GCS_PYTHON_CLIENT_ZERO_COPY_BIDI_READ", "1" if request.param else "0" + ) + return request.param + + +@pytest.fixture(params=[False, True], ids=["accepts_mv_false", "accepts_mv_true"]) +def accepts_memoryview(request): + """The ``accepts_memoryview`` argument passed to ``download_ranges``.""" + return request.param + + def _get_equal_dist(a: int, b: int) -> tuple[int, int]: step = (b - a) // 3 return a + step, a + 2 * step @@ -191,6 +206,8 @@ def test_basic_wrd_x_region( object_size, event_loop, grpc_client, + zero_copy_bidi_read, + accepts_memoryview, ): object_name = f"test_basic_wrd-{str(uuid.uuid4())}" @@ -212,7 +229,9 @@ async def _run(): async with mrd: assert mrd._open_retries == 1 # (0, 0) means read the whole object - await mrd.download_ranges([(0, 0, buffer)]) + await mrd.download_ranges( + [(0, 0, buffer)], accepts_memoryview=accepts_memoryview + ) assert mrd.persisted_size == object_size assert buffer.getvalue() == object_data @@ -246,6 +265,8 @@ def test_basic_wrd( object_size, event_loop, grpc_clients, + zero_copy_bidi_read, + accepts_memoryview, ): object_name = f"test_basic_wrd-{str(uuid.uuid4())}" @@ -270,7 +291,9 @@ async def _run(): grpc_client, _BUCKET_UNDER_TEST, object_name ) as mrd: # (0, 0) means read the whole object - await mrd.download_ranges([(0, 0, buffer)]) + await mrd.download_ranges( + [(0, 0, buffer)], accepts_memoryview=accepts_memoryview + ) assert mrd.persisted_size == object_size assert buffer.getvalue() == object_data @@ -294,7 +317,13 @@ async def _run(): ], ) def test_basic_wrd_in_slices( - storage_client, blobs_to_delete, object_size, event_loop, grpc_client + storage_client, + blobs_to_delete, + object_size, + event_loop, + grpc_client, + zero_copy_bidi_read, + accepts_memoryview, ): object_name = f"test_basic_wrd-{str(uuid.uuid4())}" @@ -321,7 +350,9 @@ async def _run(): buffer = BytesIO() await mrd.open() # (0, 0) means read the whole object - await mrd.download_ranges([(0, 0, buffer)]) + await mrd.download_ranges( + [(0, 0, buffer)], accepts_memoryview=accepts_memoryview + ) await mrd.close() assert buffer.getvalue() == object_data assert mrd.persisted_size == object_size @@ -352,6 +383,8 @@ def test_wrd_with_non_default_flush_interval( flush_interval, event_loop, grpc_client, + zero_copy_bidi_read, + accepts_memoryview, ): object_name = f"test_basic_wrd-{str(uuid.uuid4())}" object_size = 9 * 1024 * 1024 @@ -380,7 +413,9 @@ async def _run(): buffer = BytesIO() await mrd.open() # (0, 0) means read the whole object - await mrd.download_ranges([(0, 0, buffer)]) + await mrd.download_ranges( + [(0, 0, buffer)], accepts_memoryview=accepts_memoryview + ) await mrd.close() assert buffer.getvalue() == object_data assert mrd.persisted_size == object_size @@ -539,7 +574,12 @@ async def _run(): def test_read_unfinalized_appendable_object( - storage_client, blobs_to_delete, event_loop, grpc_client_direct + storage_client, + blobs_to_delete, + event_loop, + grpc_client_direct, + zero_copy_bidi_read, + accepts_memoryview, ): object_name = f"read_unfinalized_appendable_object-{str(uuid.uuid4())[:4]}" @@ -560,7 +600,9 @@ async def _run(): await mrd.open() assert mrd.persisted_size == len(_BYTES_TO_UPLOAD) # (0, 0) means read the whole object - await mrd.download_ranges([(0, 0, buffer)]) + await mrd.download_ranges( + [(0, 0, buffer)], accepts_memoryview=accepts_memoryview + ) await mrd.close() assert buffer.getvalue() == _BYTES_TO_UPLOAD @@ -576,7 +618,9 @@ async def _run(): @pytest.mark.skip(reason="Flaky test b/478129078") -def test_mrd_open_with_read_handle(event_loop, grpc_client_direct): +def test_mrd_open_with_read_handle( + event_loop, grpc_client_direct, zero_copy_bidi_read, accepts_memoryview +): object_name = f"test_read_handl-{str(uuid.uuid4())[:4]}" async def _run(): @@ -605,7 +649,9 @@ async def _run(): # persisted_size not set when opened with read_handle assert new_mrd.persisted_size is None buffer = BytesIO() - await new_mrd.download_ranges([(0, 0, buffer)]) + await new_mrd.download_ranges( + [(0, 0, buffer)], accepts_memoryview=accepts_memoryview + ) await new_mrd.close() assert buffer.getvalue() == _BYTES_TO_UPLOAD del mrd @@ -615,7 +661,9 @@ async def _run(): event_loop.run_until_complete(_run()) -def test_mrd_open_with_read_handle_over_cloud_path(event_loop, grpc_client): +def test_mrd_open_with_read_handle_over_cloud_path( + event_loop, grpc_client, zero_copy_bidi_read, accepts_memoryview +): object_name = f"test_read_handl-{str(uuid.uuid4())[:4]}" async def _run(): @@ -643,7 +691,9 @@ async def _run(): # because read_handle won't work in CLOUD_PATH. assert new_mrd.persisted_size == len(_BYTES_TO_UPLOAD) buffer = BytesIO() - await new_mrd.download_ranges([(0, 0, buffer)]) + await new_mrd.download_ranges( + [(0, 0, buffer)], accepts_memoryview=accepts_memoryview + ) await new_mrd.close() assert buffer.getvalue() == _BYTES_TO_UPLOAD del mrd @@ -654,7 +704,12 @@ async def _run(): def test_wrd_open_with_write_handle( - event_loop, grpc_client_direct, storage_client, blobs_to_delete + event_loop, + grpc_client_direct, + storage_client, + blobs_to_delete, + zero_copy_bidi_read, + accepts_memoryview, ): object_name = f"test_write_handl-{str(uuid.uuid4())[:4]}" @@ -695,7 +750,9 @@ async def _run(): ) buffer = BytesIO() await mrd.open() - await mrd.download_ranges([(0, 0, buffer)]) + await mrd.download_ranges( + [(0, 0, buffer)], accepts_memoryview=accepts_memoryview + ) await mrd.close() assert buffer.getvalue() == test_data @@ -712,7 +769,12 @@ async def _run(): def test_read_unfinalized_appendable_object_with_generation( - storage_client, blobs_to_delete, event_loop, grpc_client_direct + storage_client, + blobs_to_delete, + event_loop, + grpc_client_direct, + zero_copy_bidi_read, + accepts_memoryview, ): object_name = f"read_unfinalized_appendable_object-{str(uuid.uuid4())[:4]}" grpc_client = grpc_client_direct @@ -727,7 +789,9 @@ async def _read_and_verify(expected_content, generation=None): await mrd.open() try: assert mrd.persisted_size == len(expected_content) - await mrd.download_ranges([(0, 0, buffer)]) + await mrd.download_ranges( + [(0, 0, buffer)], accepts_memoryview=accepts_memoryview + ) assert buffer.getvalue() == expected_content finally: await mrd.close() @@ -1094,6 +1158,8 @@ def test_mrd_concurrent_download( grpc_client_direct, ranges_desc, chunk_ranges, + zero_copy_bidi_read, + accepts_memoryview, ): """ Test that mrd can handle concurrent `download_ranges` calls correctly. @@ -1137,7 +1203,10 @@ async def _run(): for idx, (start, length) in enumerate(ranges_to_fetch): tasks.append( asyncio.create_task( - mrd.download_ranges([(start, length, buffers[idx])]) + mrd.download_ranges( + [(start, length, buffers[idx])], + accepts_memoryview=accepts_memoryview, + ) ) ) @@ -1161,7 +1230,12 @@ async def _run(): def test_mrd_concurrent_download_cancellation( - storage_client, blobs_to_delete, event_loop, grpc_client_direct + storage_client, + blobs_to_delete, + event_loop, + grpc_client_direct, + zero_copy_bidi_read, + accepts_memoryview, ): """ Test task cancellation / abort mid-stream. @@ -1196,7 +1270,10 @@ async def _run(): start = i * chunk_size tasks.append( asyncio.create_task( - mrd.download_ranges([(start, chunk_size, buffers[i])]) + mrd.download_ranges( + [(start, chunk_size, buffers[i])], + accepts_memoryview=accepts_memoryview, + ) ) ) @@ -1227,7 +1304,12 @@ async def _run(): def test_mrd_concurrent_download_out_of_bounds( - storage_client, blobs_to_delete, event_loop, grpc_client_direct + storage_client, + blobs_to_delete, + event_loop, + grpc_client_direct, + zero_copy_bidi_read, + accepts_memoryview, ): """ Test out-of-bounds & edge ranges concurrent with valid requests. @@ -1255,12 +1337,17 @@ async def _run(): ) as mrd: valid_buffer = BytesIO() valid_task = asyncio.create_task( - mrd.download_ranges([(0, 100, valid_buffer)]) + mrd.download_ranges( + [(0, 100, valid_buffer)], accepts_memoryview=accepts_memoryview + ) ) oob_buffer = BytesIO() oob_task = asyncio.create_task( - mrd.download_ranges([(object_size + 1000, 100, oob_buffer)]) + mrd.download_ranges( + [(object_size + 1000, 100, oob_buffer)], + accepts_memoryview=accepts_memoryview, + ) ) results = await asyncio.gather(valid_task, oob_task, return_exceptions=True) @@ -1296,6 +1383,8 @@ def test_mrd_checksum_validation( read_start, read_length, enable_checksum, + zero_copy_bidi_read, + accepts_memoryview, ): """ Tests full downloads with specified offset, length, and enable_checksum toggle on finalized objects. @@ -1321,7 +1410,9 @@ async def _run(): ) as mrd: buffer = BytesIO() await mrd.download_ranges( - [(read_start, read_length, buffer)], enable_checksum=enable_checksum + [(read_start, read_length, buffer)], + enable_checksum=enable_checksum, + accepts_memoryview=accepts_memoryview, ) assert buffer.getvalue() == object_data @@ -1336,7 +1427,12 @@ async def _run(): def test_mrd_checksum_unfinalized_appendable_skipped( - storage_client, blobs_to_delete, event_loop, grpc_client_direct + storage_client, + blobs_to_delete, + event_loop, + grpc_client_direct, + zero_copy_bidi_read, + accepts_memoryview, ): """ Verifies that live, unfinalized appendable objects skip the full-object checksum check @@ -1361,7 +1457,11 @@ async def _run(): ) as mrd: buffer = BytesIO() # Since it's unfinalized, it should skip the checksum check without raising - await mrd.download_ranges([(0, 0, buffer)], enable_checksum=True) + await mrd.download_ranges( + [(0, 0, buffer)], + enable_checksum=True, + accepts_memoryview=accepts_memoryview, + ) assert buffer.getvalue() == _BYTES_TO_UPLOAD # cleanup