Repository navigation
Conversation
…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`.
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.
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. (cherry picked from commit 49d5e29)
…d drop held chunks on worker errors
GOOGLE_CLOUD_STORAGE_CRC32C_OFFLOAD_MIN_BYTES=0 (or any value below 1) meant
"offload every chunk", the opposite of what an operator setting it to zero
expects, and there was no documented way to verify every chunk inline. The
threshold now only applies when it is positive, so 0 turns the offload off;
download_ranges() documents the knob and that chunks still waiting for their
checksum are discarded (and re-requested on retry) when the call is cancelled
or a mismatch is found.
If a checksum raised on the worker, verify_pending_checksums() re-raised it
without discarding the queue, so the drain in download_ranges() re-raised the
same error again and the queued futures were left behind. The exception is
now re-raised after discard_pending_checksums(), like a mismatch.
Test plan: from packages/google-cloud-storage:
PYTHONPATH=$PWD python -m pytest tests/unit/asyncio/retry/test_reads_resumption_strategy.py \
tests/unit/asyncio/test_async_multi_range_downloader.py
78 pass, including a worker exception with a queued future (queue emptied,
future cancelled, offset rewound) and a zero threshold (nothing submitted).
Authored with an AI assistant.
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces offloaded CRC32C checksum verification for large chunks (1 MiB or more) using a background thread pool, preventing blocking of the main event loop during multi-range downloads. It implements a bounded queue to hold pending chunks and write them in arrival order, with robust handling for cancellations, stream failures, and process forks. The review feedback suggests catching AttributeError alongside RuntimeError when submitting tasks to the thread pool to gracefully handle interpreter shutdown when globals might be cleared.
In addition to summary section - Can you also add a Also clearly call out the memory implications. |
|
/gcbrun(ea73968) |
|
/gcbrun |
Updated the PR description. |
| # (received chunks are) always run with the GIL released. | ||
| _CRC32C_OFFLOAD_MIN_BYTES = _int_from_env( | ||
| "GCS_PYTHON_CLIENT_CRC32C_OFFLOAD_MIN_BYTES", | ||
| getattr(google_crc32c, "GIL_RELEASE_THRESHOLD", 1024 * 1024), |
There was a problem hiding this comment.
The speedup needs google-crc32c >= 1.9.0, but setup.py allows >= 1.6.0.
Offloading only helps if google_crc32c.value() releases the GIL. In my test 1.8.0 doesn't, and 1.9.0 does for inputs of 1 MiB or more. Timing for 4 threads doing 4x the hashing work, relative to 1 thread (64 cores; ~1x means they ran in parallel):
| input | 1.8.0 | 1.9.0 |
|---|---|---|
| 1 MiB - 1 B | 4.09x | 3.97x |
| 1 MiB | 4.01x | 1.10x |
| 2 MiB | 4.09x | 1.07x |
setup.py still allows google-crc32c >= 1.6.0 and testing/constraints-3.10.txt pins 1.6.0. On versions before 1.9.0 the worker holds the GIL while it hashes, so offloading only adds a thread hop, latency and held-chunk memory. Also, GIL_RELEASE_THRESHOLD doesn't exist in 1.9.0, so this getattr always uses the 1 MiB fallback.
Suggestion: raise the minimum to google-crc32c >= 1.9.0, < 2.0.0 and update the 3.10 constraints file. Or default _CRC32C_OFFLOAD_MIN_BYTES to 0 on versions before 1.9.0.
| except DataCorruption: | ||
| raise |
There was a problem hiding this comment.
A later DataCorruption skips chunks still waiting for their checksum.
The comment below says the drain makes an earlier corrupt chunk raise DataCorruption "matching inline verification". That only holds for errors that aren't DataCorruption. If a >= 1 MiB chunk is still waiting on the worker and the next message fails inline (a small tail with a bad CRC, or an offset mismatch), update_state_from_response raises right away and this branch skips the drain. Then:
- If the earlier chunk is valid, it's dropped. With
GCS_PYTHON_CLIENT_CRC32C_OFFLOAD_MIN_BYTES=0it would have been written. - If the earlier chunk is also corrupt, the error names the later chunk instead.
Suggestion: remove except DataCorruption: raise so every exception goes through the drain. When the DataCorruption came from verify_pending_checksums, the queue is already empty and the drain does nothing. This needs the _write_chunk fix in my other comment. test_download_ranges_closes_on_datacorruption then needs mock_strategy_cls.return_value.verify_pending_checksums = AsyncMock().
Tests: an offloaded valid chunk followed by a corrupt tail should leave the first chunk in the buffer. An offloaded corrupt chunk followed by a corrupt tail should report the first chunk.
| except BaseException: | ||
| # Keep every read resumable right after its written bytes. | ||
| self.discard_pending_checksums(state) | ||
| raise | ||
|
|
||
| # Final Byte Count & Full Object Checksum Verification | ||
| if object_data_range.range_end: | ||
| read_state.is_complete = True | ||
| if ( | ||
| read_state.initial_length != 0 | ||
| and read_state.bytes_written > read_state.initial_length | ||
| ): | ||
| # Final Byte Count & Full Object Checksum Verification |
There was a problem hiding this comment.
The end-of-range checks don't discard the other held chunks.
The try/except BaseException only covers the buffer write. The byte-count and full-object checksum DataCorruption raises below it are outside it. When either fires from verify_pending_checksums, chunks for other reads stay queued: their futures aren't cancelled and their next_expected_offset isn't rewound. That breaks the guarantee in the verify_pending_checksums docstring. Today the finally in download_ranges cleans up, but only after await self.close(). If the outer drain also handles DataCorruption, those leftover chunks would be written after the corruption.
Suggestion: move the if chunk.range_end: block inside the try.
Test: extend test_verify_pending_checksums_checks_byte_count_of_held_chunk with a second read's chunk queued behind it. Assert that its future is cancelled, the queue is empty, and its offset is rewound.
This branch has not been deployed
3 participants
Summary
Hash read chunks of 1 MiB or more on a small worker pool instead of the asyncio event-loop thread, and write each chunk to its buffer only after its checksum matches. On
main,google_crc32c.value()is 10-12% of the event-loop thread's samples at 2 GB/s per process (py-spy, 1 process x 16 streams), time during which no other stream on that loop is served. The releasedgoogle-crc32calready releases the GIL forbytesof 1 MiB or more, so the workers run in parallel with the loop thread and no other change is required.Changes
asyncio/retry/reads_resumption_strategy.py:_CRC32C_OFFLOAD_MIN_BYTES(defaults togetattr(google_crc32c, "GIL_RELEASE_THRESHOLD", 1024 * 1024); configurable viaGCS_PYTHON_CLIENT_CRC32C_OFFLOAD_MIN_BYTES, where0verifies every chunk inline) that are not part of a full-object read are hashed on a 2-threadThreadPoolExecutor; smaller chunks and full-object reads (which feed the rolling checksum) are hashed inline as before.state["pending_checksums"]and written in arrival order once their checksum matches (verify_pending_checksums).next_expected_offsetadvances on receipt andbytes_writtenonly on write, so a retry resumes right after the last verified byte.discard_pending_checksumsdrops every held chunk, cancels queued hashes and rewinds each read to its written end;DataCorruptionis raised as it is for an inline mismatch.recover_state_on_failureverifies held chunks before a stream is reopened.os.register_at_fork); once interpreter shutdown refuses new work, chunks are hashed inline.asyncio/async_multi_range_downloader.py:download_rangeswrites verified chunks after each received message (holding at most_CRC32C_MAX_PENDING = 32), drains fully when an attempt ends or fails, discards held chunks on cancellation, and documents all of this.Usage
Enabled automatically for non-full-object read chunks of at least 1 MiB (
getattr(google_crc32c, "GIL_RELEASE_THRESHOLD", 1024 * 1024)). SetGCS_PYTHON_CLIENT_CRC32C_OFFLOAD_MIN_BYTESto override the threshold in bytes, or0to verify every chunk inline on the event-loop thread.Memory
download_rangesboundspending_checksumsto_CRC32C_MAX_PENDING = 32chunks per call before draining on the event loop. For 2 MiB chunks, each held_PendingChunkretains both thedataslice and the enclosingBidiReadObjectResponse, bounding pending memory perdownload_rangescall to at most ~128 MiB with the generatedproto-plus/upbparser (~4 MiB per chunk across theupbarena copy and.contentbytescopy) or ~64 MiB with the zero-copy parser in #18627 (~2 MiB per chunk sharing the gRPC buffer).Tests
tests/unit/asyncio/retry/test_reads_resumption_strategy.pyandtests/unit/asyncio/test_async_multi_range_downloader.py: offload threshold, zero threshold and full-object exclusion; in-order writes behind a pending chunk; mismatch on an offloaded chunk (nothing from it on is written, offsets rewound, the retry requests the bytes again); a worker exception (queue emptied, queued futures cancelled); drain before a retry and at the end of an attempt; discard on cancellation; fork and shutdown behavior of the pool.Benchmark
Setup: GCE
c4-standard-192, Rapid bucket in the same zone, 8 processes x 16 concurrentAsyncMultiRangeDownloaderstreams (128 streams total),4 x 16 MiBranges perdownload_ranges()call into preallocated buffers, 5 s warmup + 15 s timed window per round, 8 interleaved rounds per arm.1. Off-Loop CRC32C Only vs.
main(Neutral Alone)main(Mean / Median)main(generated parser + inline CRC32C)offlooponly (GCS_PYTHON_CLIENT_ZERO_COPY_BIDI_READ=0)With the generated
proto-plus/upbparser active, each 2 MiB chunk is still copied twice on the event-loop thread under the GIL (~254 us per message), so the loop remains saturated and offloading CRC32C alone is neutral.2. Zero-Copy Only vs. Zero-Copy + Off-Loop CRC32C (Additive on Top of #18627)
main(Mean / Median)main(generated parser + inline CRC32C)zero-copyonly (#18626 + #18627, inline CRC32C)zero-copy + offloop(#18626 + #18627 + this PR)Once #18627 removes the two
upb/proto-pluscopies from the event-loop thread (254 us -> 4.7 us per 2 MiB message, +18.6% mean / +16.7% median overmain), synchronousgoogle_crc32c.value()(~100-200 us per 2 MiB chunk) becomes the largest remaining per-message work on the loop thread; offloading >= 1 MiB chunks onto the worker pool adds another +18.3% mean / +17.1% median (+40.4% mean / +36.7% median total overmain).