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.
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
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()accounts for 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 released google-crc32c already 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(1 MiB,GOOGLE_CLOUD_STORAGE_CRC32C_OFFLOAD_MIN_BYTES) that are not part of a full-object read are hashed on a lazily created 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, and discards held chunks on cancellation.Tests
tests/unit/asyncio/retry/test_reads_resumption_strategy.pyandtests/unit/asyncio/test_async_multi_range_downloader.py: offload 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); drain before a retry and at the end of an attempt; discard on cancellation; fork and shutdown behavior of the pool.Benchmark
c4-standard-192(us-central1-c), Rapid bucket, 16 concurrent streams per process, 4 x 16 MiB perdownload_rangescall into preallocated buffers, CRC32C verification on, google-crc32c 1.9.0 (mrd_bench.py --sink offload, 9 interleaved runs per arm, aggregate GB/s, median [min-max]):mainThis PR led in 7 of the 9 paired runs. Single-process throughput is unchanged (about 2.0 GB/s for both arms when the bucket delivers at full rate).