[raj-prince]

Description

This PR adds optional multi-stream support to AsyncMultiRangeDownloader (MRD), enabling downstream consumers like gcsfs to scale throughput beyond single-stream limits on high-bandwidth storage classes like RAPID.

By default, existing single-stream behavior is fully preserved with zero pooling or load-tracking overhead (no lock acquisition or pending range/byte accounting). Multi-stream is strictly opt-in via the stream_config parameter.

Summary of changes:

• MRDStreamConfig: Added a configuration dataclass to bundle multi-stream settings (min_streams, max_streams, target_io_depth, target_bytes).
• Strictly opt-in (stream_config=None by default): When stream_config is omitted, the pool is completely bypassed (_pool = None). Requests execute directly on the primary stream with no byte/range counting, locking, or load calculation.
• Dynamic stream pool (_StreamPool): When enabled, range requests are distributed across streams using a hybrid selection strategy (least-loaded across base min_streams, first-fit across dynamically scaled streams). If load exceeds target capacity, additional streams are opened in the background in proportion to overall load without blocking active reads on already-open streams.
• Routing state reuse: Newly opened streams reuse the object's existing read_handle and routing_token (with redirect-aware retry handling) to connect directly to the same backend without repeating lookup overhead.
• Clean lifecycle management: Background scale-up tasks are tracked and cancelled when the pool or MRD is closed.

Benchmark Results (20 GiB download on RAPID)

Streams (min_streams = max_streams)	I/O Size	I/O Depth	Elapsed Time	Throughput (MiB/s)	Throughput (GB/s)
1	8 MB	8	19.45 s	1,053.02 MiB/s	1.10 GB/s
2	8 MB	8	11.20 s	1,827.91 MiB/s	1.92 GB/s
4	8 MB	8	10.10 s	2,027.68 MiB/s	2.13 GB/s
8	8 MB	8	10.72 s	1,910.87 MiB/s	2.00 GB/s

test_reads.py execution

All 6 benchmark cases in test_reads.py executed and PASSED (total elapsed time: 3m 49.37s):

Benchmark Case	Bucket Type	Pattern	Processes	Coroutines	File Size	Chunk Size	Ranges	Avg Throughput (MiB/s)	Status
seq_regional_96p_1c_10240MiB_64KiB_1ranges	Regional (JSON)	Sequential	96	1	10 GiB	64 KiB	1	37.56 MiB/s	PASSED
seq_rapid_96p_1c_10240MiB_64KiB_1ranges	RAPID (MRD)	Sequential	96	1	10 GiB	64 KiB	1	82.51 MiB/s	PASSED
rand_regional_1p_1c_10240MiB_64KiB_1ranges	Regional (JSON)	Random	1	1	10 GiB	64 KiB	1	0.25 MiB/s	PASSED
rand_regional_1p_16c_10240MiB_64KiB_1ranges	Regional (JSON)	Random	1	16	10 GiB	64 KiB	1	0.26 MiB/s	PASSED
rand_rapid_1p_1c_10240MiB_64KiB_1ranges	RAPID (MRD)	Random	1	1	10 GiB	64 KiB	1	1.06 MiB/s	PASSED
rand_rapid_1p_16c_10240MiB_64KiB_1ranges	RAPID (MRD)	Random	1	16	10 GiB	64 KiB	1	22.27 MiB/s	PASSED

[gemini-code-assist]

Code Review

This pull request introduces multi-stream scaling and unfinalized object support in AsyncMultiRangeDownloader (MRD) by implementing an internal stream pool (_StreamPool) and stream manager (_ManagedStream). Feedback on these changes focuses on optimizing concurrency and resource management. Key recommendations include releasing the lock before awaiting the stream factory in acquire_stream to prevent blocking concurrent tasks, tracking background tasks to avoid garbage collection, initializing and updating self._routing_token to prevent AttributeError and preserve routing state, and properly handling pool closure to avoid resource leaks.

[chandra-siri]

/gcbrun(b183f7f)

[chandra-siri]

@raj-prince

some comments on descriptions

• scale throughput beyond single-connection TCP/HTTP2 limits

Just mention single-stream (because each stream maps to 1 grpc channel , the underlying TCP connections per channel depends on the gRPC's xDS for direct path)

• existing single-connection behavior

change this to single-stream

• what do you mean by "tracking overhead" ?

• "dispatched immediately to the least-loaded active stream."

Aren't we going with the hybrid approach ?

[raj-prince]

/gcbrun

[chandra-siri]

these 2 lines are not necessary.

[raj-prince]

Special handling was for performance purpose where all ranges will be in a single task. Added comment.

[chandra-siri]

should we use acquire_stream instead ?

[raj-prince]

acquire_stream is not required not given distribute_ranges does the work of allocating stream for a given range. Renamed the _select_stream to _acquire_stream and removed the original acquire_stream (with lock).