From 6f4bafd51898250301c772556d8db0b5656ea7bc Mon Sep 17 00:00:00 2001 From: Chandra Sirimala Date: Tue, 30 Sep 2025 14:18:07 +0000 Subject: [PATCH 1/4] samples: Add samples for `multi_range_downloader` --- .../storage_async_download_ranges.py | 46 +++++++++++++++++++ 1 file changed, 46 insertions(+) create mode 100644 samples/snippets/experimental/storage_async_download_ranges.py diff --git a/samples/snippets/experimental/storage_async_download_ranges.py b/samples/snippets/experimental/storage_async_download_ranges.py new file mode 100644 index 000000000..461b93f1e --- /dev/null +++ b/samples/snippets/experimental/storage_async_download_ranges.py @@ -0,0 +1,46 @@ +from google.cloud.storage._experimental.asyncio.async_multi_range_downloader import ( + AsyncMultiRangeDownloader, +) +from google.cloud.storage._experimental.asyncio.async_grpc_client import ( + AsyncGrpcClient, +) +from io import BytesIO +import asyncio +import argparse + + +async def test_mrd(bucket_name, object_name, generation_number=None): + client = AsyncGrpcClient()._grpc_client + + mrd = AsyncMultiRangeDownloader(client, bucket_name, object_name, generation_number) + await mrd.open() + # create buffers. + # Make sure buffer desitnation has enough space to accomodate bytes requetsed. + # Buffers could be in-memory or on disk + my_buff1 = open(f"sample_file_to_write_contents.txt", "wb") + my_buff2 = BytesIO() + my_buff3 = BytesIO() + my_buff4 = BytesIO() + results_arr = await mrd.download_ranges( + [ + (0, 100, my_buff1), + (100, 20, my_buff2), + (200, 123, my_buff3), + (300, 789, my_buff4), + ] + ) + + for result in results_arr: + print("downloaded bytes", result) + + # close MRD + await mrd.close() + + +if __name__ == "__main__": + parser = argparse.ArgumentParser() + parser.add_argument("--bucket_name", type=str, required=True) + parser.add_argument("--object_name", type=str, required=True) + parser.add_argument("--generation_number", type=int, default=None) + args = parser.parse_args() + asyncio.run(test_mrd(args.bucket_name, args.object_name, args.generation_number)) From 95186ecaafa0b40dff477a88a5c9323d691be4c6 Mon Sep 17 00:00:00 2001 From: Chandra Sirimala Date: Tue, 30 Sep 2025 18:12:34 +0000 Subject: [PATCH 2/4] fix linting issue, missing placeholder --- samples/snippets/experimental/storage_async_download_ranges.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/samples/snippets/experimental/storage_async_download_ranges.py b/samples/snippets/experimental/storage_async_download_ranges.py index 461b93f1e..06a0e0697 100644 --- a/samples/snippets/experimental/storage_async_download_ranges.py +++ b/samples/snippets/experimental/storage_async_download_ranges.py @@ -17,7 +17,7 @@ async def test_mrd(bucket_name, object_name, generation_number=None): # create buffers. # Make sure buffer desitnation has enough space to accomodate bytes requetsed. # Buffers could be in-memory or on disk - my_buff1 = open(f"sample_file_to_write_contents.txt", "wb") + my_buff1 = open("sample_file_to_write_contents.txt", "wb") my_buff2 = BytesIO() my_buff3 = BytesIO() my_buff4 = BytesIO() From 6e6229e388dab92a6ae3b012418f8a60531d9597 Mon Sep 17 00:00:00 2001 From: Chandra Sirimala Date: Sun, 12 Oct 2025 18:59:43 +0000 Subject: [PATCH 3/4] code to reproduce gRPC stalling issue --- .../asyncio/async_multi_range_downloader.py | 53 ++++++++++----- .../asyncio/async_read_object_stream.py | 6 +- .../storage_async_download_ranges.py | 66 ++++++++++++++----- 3 files changed, 89 insertions(+), 36 deletions(-) diff --git a/google/cloud/storage/_experimental/asyncio/async_multi_range_downloader.py b/google/cloud/storage/_experimental/asyncio/async_multi_range_downloader.py index b9ac0f88d..ad5ac0f32 100644 --- a/google/cloud/storage/_experimental/asyncio/async_multi_range_downloader.py +++ b/google/cloud/storage/_experimental/asyncio/async_multi_range_downloader.py @@ -26,6 +26,8 @@ from io import BytesIO from google.cloud import _storage_v2 +import asyncio +import traceback _MAX_READ_RANGES_PER_BIDI_READ_REQUEST = 100 @@ -160,6 +162,9 @@ def __init__( self.read_handle = read_handle self.read_obj_str: Optional[_AsyncReadObjectStream] = None self._is_stream_open: bool = False + self.read_id_to_writable_buffer_dict = {} + self.read_id_to_func_offset = {} + self.func_id_to_pending_read_ids = {} async def open(self) -> None: """Opens the bidi-gRPC connection to read from the object. @@ -189,8 +194,8 @@ async def open(self) -> None: return async def download_ranges( - self, read_ranges: List[Tuple[int, int, BytesIO]] - ) -> List[Result]: + self, read_ranges: List[Tuple[int, int, BytesIO]], func_offest: int + ): """Downloads multiple byte ranges from the object into the buffers provided by user. @@ -214,8 +219,7 @@ async def download_ranges( if not self._is_stream_open: raise ValueError("Underlying bidi-gRPC stream is not open") - read_id_to_writable_buffer_dict = {} - results = [] + read_ids_in_this_func = set() for i in range(0, len(read_ranges), _MAX_READ_RANGES_PER_BIDI_READ_REQUEST): read_ranges_segment = read_ranges[ i : i + _MAX_READ_RANGES_PER_BIDI_READ_REQUEST @@ -223,10 +227,12 @@ async def download_ranges( read_ranges_for_bidi_req = [] for j, read_range in enumerate(read_ranges_segment): - read_id = i + j - read_id_to_writable_buffer_dict[read_id] = read_range[2] + read_id = i + j + func_offest + self.read_id_to_func_offset[read_id] = func_offest + self.read_id_to_writable_buffer_dict[read_id] = read_range[2] + read_ids_in_this_func.add(read_id) bytes_requested = read_range[1] - results.append(Result(bytes_requested)) + # results.append(Result(bytes_requested)) read_ranges_for_bidi_req.append( _storage_v2.ReadRange( read_offset=read_range[0], @@ -234,12 +240,27 @@ async def download_ranges( read_id=read_id, ) ) + print("sending read_ranges", read_ranges_for_bidi_req) await self.read_obj_str.send( _storage_v2.BidiReadObjectRequest(read_ranges=read_ranges_for_bidi_req) ) - while len(read_id_to_writable_buffer_dict) > 0: - response = await self.read_obj_str.recv() + self.func_id_to_pending_read_ids[func_offest] = read_ids_in_this_func + # if func_offest == 100: + # await asyncio.sleep(15) + # print("woke up from sleep, recveing to start") + + while len(self.func_id_to_pending_read_ids[func_offest]) > 0: + try: + async with asyncio.timeout(30): + response = await self.read_obj_str.recv() + except TimeoutError as exc: + print("timeout error occurred, Traceback:", traceback.format_exc()) + print("in funcId", func_offest) + for read_id, buffer in self.read_id_to_writable_buffer_dict.items(): + print("read_id", read_id, "buffer size", buffer.getbuffer().nbytes) + print("*" * 40) + continue if response is None: raise Exception("None response received, something went wrong.") @@ -248,18 +269,16 @@ async def download_ranges( if object_data_range.read_range is None: raise Exception("Invalid response, read_range is None") - data = object_data_range.checksummed_data.content read_id = object_data_range.read_range.read_id - buffer = read_id_to_writable_buffer_dict[read_id] + print("received read_id", read_id) + data = object_data_range.checksummed_data.content + buffer = self.read_id_to_writable_buffer_dict[read_id] buffer.write(data) - results[read_id].bytes_written += len(data) if object_data_range.range_end: - del read_id_to_writable_buffer_dict[ - object_data_range.read_range.read_id - ] - - return results + tmp_func_offset = self.read_id_to_func_offset[read_id] + self.func_id_to_pending_read_ids[tmp_func_offset].remove(read_id) + return async def close(self): """ diff --git a/google/cloud/storage/_experimental/asyncio/async_read_object_stream.py b/google/cloud/storage/_experimental/asyncio/async_read_object_stream.py index b14c55227..0c2cfd82b 100644 --- a/google/cloud/storage/_experimental/asyncio/async_read_object_stream.py +++ b/google/cloud/storage/_experimental/asyncio/async_read_object_stream.py @@ -145,7 +145,11 @@ async def recv(self) -> _storage_v2.BidiReadObjectResponse: """ if not self._is_stream_open: raise ValueError("Stream is not open") - return await self.socket_like_rpc.recv() + res = await self.socket_like_rpc.recv() + if hasattr(res, "object_data_ranges"): + for object_data_range in res.object_data_ranges: + print("readid in read_obj_str", object_data_range.read_range.read_id) + return res @property def is_stream_open(self) -> bool: diff --git a/samples/snippets/experimental/storage_async_download_ranges.py b/samples/snippets/experimental/storage_async_download_ranges.py index 06a0e0697..4a58d65dd 100644 --- a/samples/snippets/experimental/storage_async_download_ranges.py +++ b/samples/snippets/experimental/storage_async_download_ranges.py @@ -7,34 +7,61 @@ from io import BytesIO import asyncio import argparse +import logging + +logging.basicConfig( + level=logging.DEBUG, + format="%(asctime)s - %(name)s:%(lineno)d - %(levelname)s - %(message)s", +) +logging.getLogger("grpc.aio._call").setLevel(logging.DEBUG) -async def test_mrd(bucket_name, object_name, generation_number=None): +async def test_mrd_by_tasks(bucket_name, object_name, generation_number=None): client = AsyncGrpcClient()._grpc_client mrd = AsyncMultiRangeDownloader(client, bucket_name, object_name, generation_number) await mrd.open() - # create buffers. - # Make sure buffer desitnation has enough space to accomodate bytes requetsed. - # Buffers could be in-memory or on disk - my_buff1 = open("sample_file_to_write_contents.txt", "wb") + + my_buff1 = BytesIO() my_buff2 = BytesIO() my_buff3 = BytesIO() my_buff4 = BytesIO() - results_arr = await mrd.download_ranges( - [ - (0, 100, my_buff1), - (100, 20, my_buff2), - (200, 123, my_buff3), - (300, 789, my_buff4), - ] - ) + my_buff5 = BytesIO() + my_buff6 = BytesIO() + my_buff7 = BytesIO() + my_buff8 = BytesIO() + ranges1 = [ + (0, 100, my_buff1), + (100, 20, my_buff2), + (200, 123, my_buff3), + (300, 789, my_buff4), + ] - for result in results_arr: - print("downloaded bytes", result) + ranges2 = [ + (200, 34, my_buff7), + (300, 73, my_buff8), + (0, 100, my_buff5), + (100, 543, my_buff6), + (1, 4324, BytesIO()), + (343, 78, BytesIO()), + ] - # close MRD - await mrd.close() + # This works fine ! + # _ = await mrd.download_ranges(ranges1, 1000) + # _ = await mrd.download_ranges(ranges2, 2000) + + # This doesn't work, hangs in `self._cython_call.status()` + # in `grpc/aio/_call.py` + # but when kept under `with asyncio.timeout(30) .... ` it works + # see implementation of `download_ranges` in `async_multi_range_downloader` + # how it's working with asyncio.timeout + task1 = asyncio.create_task(mrd.download_ranges(ranges1, 1000)) + task2 = asyncio.create_task(mrd.download_ranges(ranges2, 2000)) + _ = await asyncio.gather(task1, task2) + + print("downloading complete: Buffer details") + for read_id, buffer in mrd.read_id_to_writable_buffer_dict.items(): + print(read_id, buffer.getbuffer().nbytes) if __name__ == "__main__": @@ -43,4 +70,7 @@ async def test_mrd(bucket_name, object_name, generation_number=None): parser.add_argument("--object_name", type=str, required=True) parser.add_argument("--generation_number", type=int, default=None) args = parser.parse_args() - asyncio.run(test_mrd(args.bucket_name, args.object_name, args.generation_number)) + asyncio.run( + test_mrd_by_tasks(args.bucket_name, args.object_name, args.generation_number), + debug=True, + ) From 50815fd05828e306672a71cec71e9d8c2b1b5a0e Mon Sep 17 00:00:00 2001 From: Chandra Sirimala Date: Mon, 27 Oct 2025 16:04:19 +0000 Subject: [PATCH 4/4] save working copy --- .../asyncio/async_multi_range_downloader.py | 56 +++++++++++++++++-- .../asyncio/async_read_object_stream.py | 6 +- .../storage_async_download_ranges.py | 45 +++++++++++---- setup.py | 1 + 4 files changed, 88 insertions(+), 20 deletions(-) diff --git a/google/cloud/storage/_experimental/asyncio/async_multi_range_downloader.py b/google/cloud/storage/_experimental/asyncio/async_multi_range_downloader.py index ad5ac0f32..cb9db3c9a 100644 --- a/google/cloud/storage/_experimental/asyncio/async_multi_range_downloader.py +++ b/google/cloud/storage/_experimental/asyncio/async_multi_range_downloader.py @@ -28,8 +28,30 @@ import asyncio import traceback +import logging +import datetime +import time _MAX_READ_RANGES_PER_BIDI_READ_REQUEST = 100 +logging.basicConfig( + level=logging.DEBUG, + format="%(asctime)s - %(name)s:%(lineno)d - %(levelname)s - %(message)s", +) +logger = logging.getLogger(__name__) + + +def my_time(time_seconds: float) -> str: + """ + this function takes outupt of time.time() in seconds and returns a human readable timestamp with accuracy up to nanoseconds. + For example, "2023-10-05 14:23:45.123" + :return: str: formatted timestamp + + """ + + ts = datetime.datetime.fromtimestamp(time_seconds) + return ts.strftime("%Y-%m-%d %H:%M:%S.%f") + + return class Result: @@ -194,7 +216,7 @@ async def open(self) -> None: return async def download_ranges( - self, read_ranges: List[Tuple[int, int, BytesIO]], func_offest: int + self, read_ranges: List[Tuple[int, int, BytesIO]], func_offest: int, lock ): """Downloads multiple byte ranges from the object into the buffers provided by user. @@ -240,7 +262,14 @@ async def download_ranges( read_id=read_id, ) ) - print("sending read_ranges", read_ranges_for_bidi_req) + # logger.debug("sending read_ranges in func_id %d: ", func_offest) + + print( + my_time(time.time()), + "sending read_ranges in funcId: ", + func_offest, + # read_ranges_for_bidi_req, + ) await self.read_obj_str.send( _storage_v2.BidiReadObjectRequest(read_ranges=read_ranges_for_bidi_req) ) @@ -249,11 +278,27 @@ async def download_ranges( # if func_offest == 100: # await asyncio.sleep(15) # print("woke up from sleep, recveing to start") - + # async with lock: + recv_count = 0 while len(self.func_id_to_pending_read_ids[func_offest]) > 0: try: - async with asyncio.timeout(30): + async with asyncio.timeout(10): + # async with lock: + print( + my_time(time.time()), + "receiving read_ranges in func_id: ", + func_offest, + "recv_count:", + recv_count, + ) response = await self.read_obj_str.recv() + print( + my_time(time.time()), + "received read_ranges in func_id: ", + func_offest, + "recv_count:", + recv_count, + ) except TimeoutError as exc: print("timeout error occurred, Traceback:", traceback.format_exc()) print("in funcId", func_offest) @@ -270,7 +315,7 @@ async def download_ranges( raise Exception("Invalid response, read_range is None") read_id = object_data_range.read_range.read_id - print("received read_id", read_id) + # print("received read_id", read_id) data = object_data_range.checksummed_data.content buffer = self.read_id_to_writable_buffer_dict[read_id] buffer.write(data) @@ -278,6 +323,7 @@ async def download_ranges( if object_data_range.range_end: tmp_func_offset = self.read_id_to_func_offset[read_id] self.func_id_to_pending_read_ids[tmp_func_offset].remove(read_id) + recv_count += 1 return async def close(self): diff --git a/google/cloud/storage/_experimental/asyncio/async_read_object_stream.py b/google/cloud/storage/_experimental/asyncio/async_read_object_stream.py index 0c2cfd82b..243ca4f94 100644 --- a/google/cloud/storage/_experimental/asyncio/async_read_object_stream.py +++ b/google/cloud/storage/_experimental/asyncio/async_read_object_stream.py @@ -146,9 +146,9 @@ async def recv(self) -> _storage_v2.BidiReadObjectResponse: if not self._is_stream_open: raise ValueError("Stream is not open") res = await self.socket_like_rpc.recv() - if hasattr(res, "object_data_ranges"): - for object_data_range in res.object_data_ranges: - print("readid in read_obj_str", object_data_range.read_range.read_id) + # if hasattr(res, "object_data_ranges"): + # for object_data_range in res.object_data_ranges: + # print("readid in read_obj_str", object_data_range.read_range.read_id) return res @property diff --git a/samples/snippets/experimental/storage_async_download_ranges.py b/samples/snippets/experimental/storage_async_download_ranges.py index 4a58d65dd..379bd02af 100644 --- a/samples/snippets/experimental/storage_async_download_ranges.py +++ b/samples/snippets/experimental/storage_async_download_ranges.py @@ -8,20 +8,34 @@ import asyncio import argparse import logging +from grpc._cython import cygrpc + logging.basicConfig( level=logging.DEBUG, format="%(asctime)s - %(name)s:%(lineno)d - %(levelname)s - %(message)s", ) logging.getLogger("grpc.aio._call").setLevel(logging.DEBUG) +# logging.getLogger( +# "google.cloud.storage._experimental.asyncio.async_multi_range_downloader" +# ).setLevel(logging.DEBUG) async def test_mrd_by_tasks(bucket_name, object_name, generation_number=None): + # loop1 = asyncio.get_event_loop() + # print("loop1 id ", id(loop1)) + # loop2 = cygrpc.get_working_loop() + # print("loop2 id ", id(loop2)) client = AsyncGrpcClient()._grpc_client mrd = AsyncMultiRangeDownloader(client, bucket_name, object_name, generation_number) await mrd.open() + # mrd2 = AsyncMultiRangeDownloader( + # client, bucket_name, object_name, generation_number + # ) + # await mrd2.open() + my_buff1 = BytesIO() my_buff2 = BytesIO() my_buff3 = BytesIO() @@ -37,14 +51,14 @@ async def test_mrd_by_tasks(bucket_name, object_name, generation_number=None): (300, 789, my_buff4), ] - ranges2 = [ - (200, 34, my_buff7), - (300, 73, my_buff8), - (0, 100, my_buff5), - (100, 543, my_buff6), - (1, 4324, BytesIO()), - (343, 78, BytesIO()), - ] + # ranges2 = [ + # (200, 34, my_buff7), + # (300, 73, my_buff8), + # (0, 100, my_buff5), + # (100, 543, my_buff6), + # (1, 4324, BytesIO()), + # (343, 78, BytesIO()), + # ] # This works fine ! # _ = await mrd.download_ranges(ranges1, 1000) @@ -55,14 +69,21 @@ async def test_mrd_by_tasks(bucket_name, object_name, generation_number=None): # but when kept under `with asyncio.timeout(30) .... ` it works # see implementation of `download_ranges` in `async_multi_range_downloader` # how it's working with asyncio.timeout - task1 = asyncio.create_task(mrd.download_ranges(ranges1, 1000)) - task2 = asyncio.create_task(mrd.download_ranges(ranges2, 2000)) - _ = await asyncio.gather(task1, task2) + lock = asyncio.Lock() + task1 = asyncio.create_task(mrd.download_ranges(ranges1, 1000, lock)) + # task2 = asyncio.create_task(mrd.download_ranges(ranges2, 2000, lock)) + print("task1 loopid ", id(task1._loop)) + # print("task2 loopid ", id(task2._loop)) + # _ = await asyncio.gather(task1, task2) print("downloading complete: Buffer details") for read_id, buffer in mrd.read_id_to_writable_buffer_dict.items(): print(read_id, buffer.getbuffer().nbytes) + # print("Buffer details for mrd2") + # for read_id, buffer in mrd2.read_id_to_writable_buffer_dict.items(): + # print(read_id, buffer.getbuffer().nbytes) + if __name__ == "__main__": parser = argparse.ArgumentParser() @@ -72,5 +93,5 @@ async def test_mrd_by_tasks(bucket_name, object_name, generation_number=None): args = parser.parse_args() asyncio.run( test_mrd_by_tasks(args.bucket_name, args.object_name, args.generation_number), - debug=True, + # debug=True, ) diff --git a/setup.py b/setup.py index 43e3404f6..d39e44759 100644 --- a/setup.py +++ b/setup.py @@ -31,6 +31,7 @@ "google-auth >= 2.26.1, < 3.0.0", "google-api-core >= 2.15.0, < 3.0.0", "google-cloud-core >= 2.4.2, < 3.0.0", + "grpcio==1.77.0.dev0" # The dependency "google-resumable-media" is no longer used. However, the # dependency is still included here to accommodate users who may be # importing exception classes from the google-resumable-media without