perf: update microbenchmarks and use concurrent download_ranges

Author: zhixiangli

packages/google-cloud-storage/tests/perf/microbenchmarks/time_based/reads/config.py

@@ -83,7 +83,7 @@ def _get_params() -> Dict[str, List[TimeBasedReadParameters]]:
             num_files = num_processes * num_coros
 
             # Create a descriptive name for the parameter set
-            name = f"{pattern}_{bucket_type}_{num_processes}p_{file_size_mib}MiB_{chunk_size_kib}KiB_{num_ranges_val}ranges"
+            name = f"{pattern}_{bucket_type}_{num_processes}p_{num_coros}c_{file_size_mib}MiB_{chunk_size_kib}KiB_{num_ranges_val}ranges"
 
             params[workload_name].append(
                 TimeBasedReadParameters(

packages/google-cloud-storage/tests/perf/microbenchmarks/time_based/reads/config.yaml

@@ -23,6 +23,11 @@ workload:
     coros: [1]
     processes: [1]
 
+  - name: "read_rand_multi_coro"
+    pattern: "rand"
+    coros: [16]
+    processes: [1]
+
 defaults:
   DEFAULT_RAPID_ZONAL_BUCKET: "chandrasiri-benchmarks-zb"
   DEFAULT_STANDARD_BUCKET: "chandrasiri-benchmarks-rb"

packages/google-cloud-storage/tests/perf/microbenchmarks/time_based/reads/test_reads.py

@@ -115,47 +115,53 @@ def _download_time_based_json(client, filename, params):
 
 
 async def _download_time_based_async(client, filename, params):
-    total_bytes_downloaded = 0
-
     mrd = AsyncMultiRangeDownloader(client, params.bucket_name, filename)
     await mrd.open()
 
-    offset = 0
-    is_warming_up = True
-    start_time = time.monotonic()
-    warmup_end_time = start_time + params.warmup_duration
-    test_end_time = warmup_end_time + params.duration
+    shared_lock = asyncio.Lock()
 
-    while time.monotonic() < test_end_time:
-        current_time = time.monotonic()
-        if is_warming_up and current_time >= warmup_end_time:
-            is_warming_up = False
-            total_bytes_downloaded = 0  # Reset counter after warmup
+    async def _worker_coro():
+        total_bytes_downloaded = 0
+        offset = 0
+        is_warming_up = True
+        start_time = time.monotonic()
+        warmup_end_time = start_time + params.warmup_duration
+        test_end_time = warmup_end_time + params.duration
 
-        ranges = []
-        if params.pattern == "rand":
-            for _ in range(params.num_ranges):
-                offset = random.randint(
-                    0, params.file_size_bytes - params.chunk_size_bytes
-                )
-                ranges.append((offset, params.chunk_size_bytes, BytesIO()))
-        else:  # seq
-            for _ in range(params.num_ranges):
-                ranges.append((offset, params.chunk_size_bytes, BytesIO()))
-                offset += params.chunk_size_bytes
-                if offset + params.chunk_size_bytes > params.file_size_bytes:
-                    offset = 0  # Reset offset if end of file is reached
-
-        await mrd.download_ranges(ranges)
+        while time.monotonic() < test_end_time:
+            current_time = time.monotonic()
+            if is_warming_up and current_time >= warmup_end_time:
+                is_warming_up = False
+                total_bytes_downloaded = 0  # Reset counter after warmup
 
-        bytes_in_buffers = sum(r[2].getbuffer().nbytes for r in ranges)
-        assert bytes_in_buffers == params.chunk_size_bytes * params.num_ranges
-
-        if not is_warming_up:
-            total_bytes_downloaded += params.chunk_size_bytes * params.num_ranges
+            ranges = []
+            if params.pattern == "rand":
+                for _ in range(params.num_ranges):
+                    offset = random.randint(
+                        0, params.file_size_bytes - params.chunk_size_bytes
+                    )
+                    ranges.append((offset, params.chunk_size_bytes, BytesIO()))
+            else:  # seq
+                for _ in range(params.num_ranges):
+                    ranges.append((offset, params.chunk_size_bytes, BytesIO()))
+                    offset += params.chunk_size_bytes
+                    if offset + params.chunk_size_bytes > params.file_size_bytes:
+                        offset = 0  # Reset offset if end of file is reached
+
+            await mrd.download_ranges(ranges, lock=shared_lock)
+
+            bytes_in_buffers = sum(r[2].getbuffer().nbytes for r in ranges)
+            assert bytes_in_buffers == params.chunk_size_bytes * params.num_ranges
+
+            if not is_warming_up:
+                total_bytes_downloaded += params.chunk_size_bytes * params.num_ranges
+        return total_bytes_downloaded
+
+    tasks = [asyncio.create_task(_worker_coro()) for _ in range(params.num_coros)]
+    results = await asyncio.gather(*tasks)
 
     await mrd.close()
-    return total_bytes_downloaded
+    return sum(results)
 
 
 def _download_files_worker(process_idx, filename, params, bucket_type):
@@ -176,7 +182,9 @@ def download_files_mp_mc_wrapper(pool, files_names, params, bucket_type):
 
 @pytest.mark.parametrize(
     "workload_params",
-    all_params["read_seq_multi_process"] + all_params["read_rand_multi_process"],
+    all_params["read_seq_multi_process"]
+    + all_params["read_rand_multi_process"]
+    + all_params.get("read_rand_multi_coro", []),
     indirect=True,
     ids=lambda p: p.name,
 )