[yuxin00j]

Summary

An opt-in parser for BidiReadObjectResponse that reads the message in place over the bytes object gRPC hands to the deserializer, so _AsyncReadObjectStream passes each chunk's checksummed_data.content to the read path as a memoryview of the received message instead of the two copies the generated parser makes on the event-loop thread (the upb parse, then .content into a new bytes; 257 us per 2 MiB message, 5 us with this parser). Off by default. Requires a google-crc32c release that accepts memoryview (#18626); with an older release the generated parser is used automatically.

Changes

• asyncio/_zero_copy_bidi_read.py (new): deserialize(buf) walks the wire format of data responses (object_data_ranges, read_handle) over memoryview(buf) and returns stand-ins with the attributes the read path uses. A response carrying metadata, bytes that do not parse cleanly, or a buffer that is not an immutable bytes object is handed to BidiReadObjectResponse.deserialize, which returns the real message or raises its usual DecodeError. Varint, int64 and tag handling match upb.
• Opt-in per downloader: AsyncMultiRangeDownloader.create_mrd(..., zero_copy=None). True asks for the parser, False keeps the generated one, None (default) defers to GOOGLE_CLOUD_STORAGE_ZERO_COPY_BIDI_READ (1/true/yes/on). Either way the parser is used only when the installed google-crc32c can hash a memoryview (is_supported() probe). The flag reaches every stream the downloader opens, including on resumption.
• wrapped_rpc(transport, zero_copy, client_info): the transport's BidiReadObject stub with the deserializer swapped in, wrapped like the generated _wrapped_methods entry (same error mapping and x-goog-api-client header, built from the client's ClientInfo), cached per transport.
• download_ranges contract: with zero_copy on, writeable_buffer.write() receives a read-only memoryview of the received message, which is immutable and stays alive as long as the view does; a buffer may keep the view and copy from it after write() returns, and a buffer that needs bytes must call bytes(data). io.BytesIO and binary files work unchanged.

Tests

tests/unit/asyncio/test_zero_copy_bidi_read.py: field-by-field comparison with the generated parser on encoder-built and proto-plus-serialized messages (including unknown fields of every wire type at each nesting level), 16 malformed inputs on which both parsers raise DecodeError, fallback cases (metadata, non-bytes buffers), is_enabled() precedence (argument over environment, both gated by the probe), wrapped_rpc caching, and an in-process grpc.aio server driving _AsyncReadObjectStream with the parser on and off and a custom ClientInfo (user agent and gapic version reach the server). test_async_multi_range_downloader.py / test_async_read_object_stream.py cover the argument plumbing. 100% line and branch coverage of the new module, on the upb and the pure-Python protobuf backends.

nox -s unit lint cover

Benchmark

c4-standard-192 (us-central1-c), Rapid bucket in the same zone, 16 concurrent streams per process, 4 x 16 MiB per download_ranges call into preallocated buffers, CRC32C verification on, google-crc32c from the companion PR; median of 3 interleaved runs, aggregate GB/s:

Processes	main	this PR
1	1.55	2.62
8	9.76	11.29

[gemini-code-assist]

Code Review

This pull request introduces a zero-copy deserializer for BidiReadObjectResponse messages in _zero_copy_bidi_read.py to optimize performance by avoiding extra data copies on the event-loop thread. It integrates this parser into the async read path (async_multi_range_downloader.py and async_read_object_stream.py) with a configurable zero_copy option, and adds comprehensive unit tests. Feedback on the changes highlights a potential IndexError in the _varint decoding loop when parsing truncated inputs, suggesting a robust bounds check at the start of each iteration.

[raj-prince]

I see, we are parsing all the fields by own, should we re-use the standard parser for few fields like parsing read_range, BidiReadHandle one?

_ReadRangePb = _storage_v2.ReadRange.pb()
_BidiReadHandlePb = _storage_v2.BidiReadHandle.pb()
elif tag == _TAG_RANGE_READ_RANGE and read_range is None:
    start, i = _length_prefixed(buf, i, end)
    read_range = _ReadRangePb.FromString(buf[start:i])

[raj-prince]

Should we narrow it down to not go with the slow path for all exception like (Attribute Error, TypeError...)?
Maybe (_Malformed, IndexError, DecodeError) should be sufficient

[raj-prince]

Should we merge all the test_deserialize_... as a single parameterized test? Like, this?

@pytest.mark.parametrize(
    "buf",
    [
        pytest.param(_serialize(_range(crc32c=google_crc32c.value(PAYLOAD))), id="data"),
        pytest.param(_serialize(_range()), id="no_crc32c"),
        pytest.param(_serialize(_range(), handle=b"opaque"), id="read_handle"),
        pytest.param(
            _serialize(
                _range(b"first", offset=0, read_id=1, range_end=False),
                _range(b"second", offset=5, read_id=2),
            ),
            id="multiple_ranges",
        ),
        pytest.param(b"", id="empty_message"),
        pytest.param(_response(_range_data(_checksummed())), id="no_read_range"),
        pytest.param(_response(_range_data(read_range=_read_range())), id="no_checksummed"),
        pytest.param(_data_response() + _UNKNOWN_FIELDS, id="unknown_top_level_fields"),
        pytest.param(
            _response(
                _range_data(
                    _checksummed(extra=_UNKNOWN_FIELDS),
                    _read_range(extra=_UNKNOWN_FIELDS),
                    extra=_UNKNOWN_FIELDS,
                ),
                handle=_len_field(_HANDLE_HANDLE, b"opaque") + _UNKNOWN_FIELDS,
            ),
            id="unknown_nested_fields",
        ),
        pytest.param(_serialize(_range(offset=-1, read_id=-1)), id="int64_minus_one"),
        pytest.param(
            _serialize(_range(offset=-(1 << 63), read_id=(1 << 63) - 1)),
            id="int64_extremes",
        ),
        pytest.param(
            # A ten-byte varint may carry 70 bits; the generated parser keeps 64.
            _response(_range_data(read_range=_tag(_READ_RANGE_ID, 0) + b"\xff" * 9 + b"\x7f")),
            id="over_long_varint",
        ),
        pytest.param(_SPLIT_SUB_MESSAGES, id="split_sub_messages_are_merged"),
    ],
)
def test_deserialize_matches_generated_parser(buf):
    expected = _storage_v2.BidiReadObjectResponse.deserialize(buf)
    # The zero-copy parser skips unknown fields; the generated one keeps them.
    expected._pb.DiscardUnknownFields()

    actual = _zero_copy_bidi_read.deserialize(buf)

    assert isinstance(actual, _zero_copy_bidi_read._ZeroCopyBidiReadObjectResponse)
    assert all(
        isinstance(r.checksummed_data.content, memoryview)
        for r in actual.object_data_ranges
    )
    assert _as_generated(actual) == expected

[raj-prince]

/gcbrun