Repository navigation
Conversation
The generated BidiReadObject stub parses every response with proto-plus over upb, which copies each data chunk twice on the event-loop thread before it reaches the caller's buffer. Add a wire-format parser for plain data responses that exposes `checksummed_data.content` as a memoryview of the received message, and let `_AsyncReadObjectStream` use a stub built with it (wrapped like the generated method: error mapping, no default retry or timeout, same client-info header). Responses carrying object metadata, or anything that does not parse cleanly, are handed to the generated parser, so callers keep seeing the real message or its usual DecodeError. CRC32C verification in the read path is unchanged: the memoryview flows through google_crc32c.value(), the user buffer write and the rolling checksum. Because google-crc32c releases up to 1.9.0 only accept bytes there, the fast path is used only when the installed google-crc32c can checksum a memoryview; GOOGLE_CLOUD_STORAGE_FAST_BIDI_READ=0 forces the generated parser as well.
…sages and invalid tags The zero-copy BidiReadObjectResponse parser is meant to be indistinguishable from the generated one apart from the type of ``content``; a comparison against upb found a few places where it was not: - ReadRange.read_offset/read_length/read_id are int64, so a ten-byte two's-complement varint must come back negative, and bits above 64 in an over-long varint are dropped rather than kept. - A singular embedded message that occurs twice in its parent is merged; the sub-message parsers now fill in the existing object, which also removes two throw-away allocations per ObjectRangeData. - Field number 0, tags above 32 bits and tags encoded in more than five bytes are malformed and now reach the generated parser's DecodeError. Tests cover each case against the generated parser.
…thread update_state_from_response hashed every chunk inline on the asyncio loop thread, serialized with receive and parse. Chunks of at least 512 KiB are now submitted to a small shared thread pool (google_crc32c releases the GIL for read-only buffers of 1 MiB or more) and written immediately; download_ranges() awaits the pending results after each response (bounded to 32 outstanding) and fully before returning, raising DataCorruption on the first mismatch. Small chunks are still verified inline. On the 8-rank DCP load benchmark (test-c4-highmem, zonal bucket) read_data throughput per rank went from 1.08/1.10 GB/s to 1.39/1.46 GB/s with verification on; 1.57 GB/s is the same stack with verification off.
…l-object reads inline - Drain pending offloaded chunk checksums at the start of `recover_state_on_failure` so a corrupt chunk already written to the user buffer surfaces as `DataCorruption` before reopening the stream from the advanced offset. - Skip the worker-thread CRC32C offload when `rolling_checksum` is present (`is_full_object_read=True`), since full-object reads already fold every chunk into `rolling_checksum` on the event-loop thread. - Store `pending_checksums` in a `collections.deque` for O(1) `popleft` and cancel remaining queued futures on the first checksum mismatch. - Parse `GOOGLE_CLOUD_STORAGE_CRC32C_OFFLOAD_MIN_BYTES` via `_int_from_env` so malformed values fall back to the 512 KiB default and document the 1 MiB GIL-release threshold in `google_crc32c`. - Always acquire `_crc32c_executor_lock` in `_get_crc32c_executor`.
yuxin00j
force-pushed
the
storage-zero-copy-read-tput
branch
from
October 2, 2026 11:57
75fd34e to
f4d73cb
Compare
Chunks checksummed on the worker pool were written to the caller's buffer and counted in bytes_written before their CRC32C was known. A mismatch left the corrupt bytes in the buffer, a retry resumed after them, and an error or cancellation that arrived first hid the mismatch. Hold received chunks in arrival order and write each one only once its checksum matches. next_expected_offset still advances on receipt, so offsets are validated as before. On a mismatch, drop every held chunk and rewind each read to the end of its written bytes, so a retry asks for the dropped bytes again. download_ranges() writes or rejects the held chunks before an attempt ends and before another error propagates, and drops them on cancellation.
A forked child inherits the module-level pool but none of its worker threads, so offloaded checksums never finished there and the download hung. Reset the pool and its lock in the child after fork(). Once interpreter shutdown starts, ThreadPoolExecutor.submit() raises RuntimeError, which failed downloads still running in non-daemon threads. Hash those chunks inline instead.
google-crc32c releases the GIL only for chunks of 1 MiB or more that are backed by bytes, not for any read-only buffer. Received chunks are bytes or memoryview slices of bytes, so offloaded hashes still run in parallel.
… GOOGLE_CLOUD_STORAGE_ZERO_COPY_BIDI_READ=1
yuxin00j
force-pushed
the
storage-zero-copy-read-tput
branch
from
October 6, 2026 13:54
db3146c to
b203695
Compare
…stants Explain the opt-in switch's accepted spellings, the four protobuf wire types, the 64-bit varint mask and the 32-bit tag limit.
A ThreadPoolExecutor starts no threads until its first submit(), so creating it at import costs nothing and removes the lazy getter and its lock. The at-fork hook now just installs a fresh pool in the child, and the shutdown fallback in _submit_crc32c is unchanged.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Avoid copying payload bytes out of the gRPC receive buffer when parsing
BidiReadObjectResponsemessages in_AsyncReadObjectStream, and offload per-chunk CRC32C verification for large read chunks (>= 1 MiB) from theasyncioevent-loop thread to a background worker pool.Changes
BidiReadObjectResponseparser (google/cloud/storage/asyncio/_fast_bidi_read.py,async_read_object_stream.py):BidiReadObjectResponsewire frames directly overmemoryview(buf)so eachobject_data_ranges[].checksummed_data.contentis a zero-copymemoryviewslice of the gRPC frame buffer instead of a newly allocatedbytescopy.storage_v2.BidiReadObjectResponse.deserializewhenmetadata(field 1, normally present only on the initial open response) or an unrecognized/malformed tag is encountered.is_supported()(GOOGLE_CLOUD_STORAGE_FAST_BIDI_READenv toggle and a one-timegoogle_crc32c.value(memoryview(b"\x00"))capability probe, falling back to the generated protobuf parser when an oldergoogle-crc32cbuild rejectsmemoryview).google/cloud/storage/asyncio/retry/reads_resumption_strategy.py,async_multi_range_downloader.py):google_crc32c.value(data)calls for range reads of at least_CRC32C_OFFLOAD_MIN_BYTES(default 1 MiB, configurable viaGOOGLE_CLOUD_STORAGE_CRC32C_OFFLOAD_MIN_BYTES) to a lazily initialized 2-workerThreadPoolExecutor(gcs-crc32c).AsyncMultiRangeDownloader.download_ranges(bounding in-flight unverified chunks to_CRC32C_MAX_PENDING = 32), before returning fromdownload_ranges, and at the start ofrecover_state_on_failurebefore reopening a stream.Benchmark
Concurrent multi-range reads (
AsyncMultiRangeDownloader.download_ranges, 8 processes x 16 concurrent streams reading 1 GiB/process = 8 GiB total in 16 MiB ranges with CRC32C verification enabled) from a GCS Zonal bucket onc4-highmem-96(us-central1-b):BidiReadObjectResponse.deserialize+ inline CRC32C)BidiReadObjectResponse+ inline CRC32CBidiReadObjectResponse+ off-loop CRC32C