Addressed gemini comments

- Introduced RandomBytesIO for generating random bytes on-the-fly, reducing memory usage during uploads.
- Updated upload logic in test_writes.py to utilize RandomBytesIO instead of os.urandom.
- Adjusted download logic in test_reads.py to correctly handle byte ranges.
- Cleaned up config.py and conftest.py by removing commented-out code and unnecessary lines.
- Minor adjustments in config_writes.yaml for clarity.

Author: chandra-siri

tests/perf/microbenchmarks/_utils.py

@@ -1,5 +1,7 @@
 from typing import Any, List
 import statistics
+import io
+import os
 
 
 def publish_benchmark_extra_info(
@@ -83,3 +85,67 @@ def publish_benchmark_extra_info(
     print(header)
     print(row)
     print("-" * 125)
+
+class RandomBytesIO(io.RawIOBase):
+    """
+    A file-like object that generates random bytes using os.urandom.
+    It enforces a fixed size and an upper safety cap.
+    """
+    # 10 GiB default safety cap
+    DEFAULT_CAP = 10 * 1024 * 1024 * 1024 
+
+    def __init__(self, size, max_size=DEFAULT_CAP):
+        """
+        Args:
+            size (int): The exact size of the virtual file in bytes.
+            max_size (int): The maximum allowed size to prevent safety issues.
+        """
+        if size is None:
+            raise ValueError("Size must be defined (cannot be infinite).")
+        
+        if size > max_size:
+            raise ValueError(f"Requested size {size} exceeds the maximum limit of {max_size} bytes (10 GiB).")
+
+        self._size = size
+        self._pos = 0
+
+    def read(self, n=-1):
+        # 1. Handle "read all" (n=-1)
+        if n is None or n < 0:
+            n = self._size - self._pos
+        
+        # 2. Handle EOF (End of File)
+        if self._pos >= self._size:
+            return b""
+            
+        # 3. Clamp read amount to remaining size
+        # This ensures we stop exactly at `size` bytes.
+        n = min(n, self._size - self._pos)
+        
+        # 4. Generate data
+        data = os.urandom(n)
+        self._pos += len(data)
+        return data
+
+    def readable(self):
+        return True
+
+    def seekable(self):
+        return True
+
+    def tell(self):
+        return self._pos
+
+    def seek(self, offset, whence=io.SEEK_SET):
+        if whence == io.SEEK_SET:
+            new_pos = offset
+        elif whence == io.SEEK_CUR:
+            new_pos = self._pos + offset
+        elif whence == io.SEEK_END:
+            new_pos = self._size + offset
+        else:
+            raise ValueError(f"Invalid whence: {whence}")
+
+        # Clamp position to valid range [0, size]
+        self._pos = max(0, min(new_pos, self._size))
+        return self._pos

tests/perf/microbenchmarks/config.py

@@ -73,7 +73,6 @@ def _get_params() -> Dict[str, List[ReadParameters]]:
 
             # Create a descriptive name for the parameter set
             name = f"{pattern}_{bucket_type}_{num_processes}p_{num_coros}c"
-            # name = f"{workload_name}"
 
             params[workload_name].append(
                 ReadParameters(
@@ -90,7 +89,6 @@ def _get_params() -> Dict[str, List[ReadParameters]]:
                     file_size_bytes=file_size_bytes,
                 )
             )
-    # print(params)
     return params
 
 
@@ -168,22 +166,3 @@ def get_write_params() -> Dict[str, List[WriteParameters]]:
                 )
             )
     return params
-
-
-if __name__ == "__main__":
-    import sys
-    # params = _get_params()
-    # print("Read params:")
-    # print('keys (num_workload in params', len(params), sys.getsizeof(params))
-    # if 'read_seq' in params:
-    #     print(params['read_seq'], len(params['read_seq']))
-
-    write_params = get_write_params()
-
-    print(write_params)
-    print("\nWrite params:")
-    print(
-        "keys (num_workload in params", len(write_params), sys.getsizeof(write_params)
-    )
-    # if 'write_seq' in write_params:
-    #     print(write_params['write_seq'], len(write_params['write_seq']))

tests/perf/microbenchmarks/config_writes.yaml

@@ -3,8 +3,7 @@ common:
     - "regional"
     - "zonal"
   file_sizes_mib:
-    - 1024 # 1GB
-  # chunk_sizes_mib: [1, 2, 8, 64, 100, 128]
+    - 1024 # 1GiB
   chunk_sizes_mib: [100]
   rounds: 10
 

tests/perf/microbenchmarks/conftest.py

@@ -1,10 +1,3 @@
-# from tests.system.conftest import blobs_to_delete
-
-# __all__ = [
-
-# "blobs_to_delete",
-# ]
-
 import contextlib
 from typing import Any
 from tests.perf.microbenchmarks.resource_monitor import ResourceMonitor
@@ -66,8 +59,11 @@ def publish_resource_metrics(benchmark: Any, monitor: ResourceMonitor) -> None:
 
 
 async def upload_appendable_object(bucket_name, object_name, object_size, chunk_size):
+    # flush interval set to little over 1GiB to minimize number of flushes.
+    # this method is to write "appendable" objects which will be used for 
+    # benchmarking reads, hence not concerned performance of writes here.
     writer = AsyncAppendableObjectWriter(
-        AsyncGrpcClient().grpc_client, bucket_name, object_name, writer_options={"FLUSH_INTERVAL_BYTES": 1026 * 1024 * 1024}
+        AsyncGrpcClient().grpc_client, bucket_name, object_name, writer_options={"FLUSH_INTERVAL_BYTES": 1026 * 1024 ** 2}
     )
     await writer.open()
     uploaded_bytes = 0

tests/perf/microbenchmarks/test_reads.py

@@ -71,7 +71,7 @@ def download_chunks_using_json(_, json_client, filename, other_params, chunks):
     blob = bucket.blob(filename)
     start_time = time.monotonic_ns()
     for offset, size in chunks:
-        _ = blob.download_as_bytes(start=offset, end=offset + size)
+        _ = blob.download_as_bytes(start=offset, end=offset + size - 1)
     return (time.monotonic_ns() - start_time) / 1_000_000_000
 
 

tests/perf/microbenchmarks/test_writes.py

@@ -16,7 +16,7 @@
 from google.cloud.storage._experimental.asyncio.async_grpc_client import AsyncGrpcClient
 from google.cloud.storage._experimental.asyncio.async_appendable_object_writer import AsyncAppendableObjectWriter
 
-from tests.perf.microbenchmarks._utils import publish_benchmark_extra_info
+from tests.perf.microbenchmarks._utils import publish_benchmark_extra_info, RandomBytesIO
 from tests.perf.microbenchmarks.conftest import publish_resource_metrics
 import tests.perf.microbenchmarks.config as config
 
@@ -63,9 +63,11 @@ def upload_using_json(_, json_client, filename, other_params):
     bucket = json_client.bucket(other_params.bucket_name)
     blob = bucket.blob(filename)
     upload_size = other_params.file_size_bytes
-    data = os.urandom(upload_size)
-    
-    blob.upload_from_string(data)
+    # Don't use BytesIO because it'll report high memory usage for large files.
+    # `RandomBytesIO` generates random bytes on the fly.
+    in_mem_file = RandomBytesIO(upload_size)
+    # data = os.urandom(upload_size)
+    blob.upload_from_file(in_mem_file)
 
     end_time = time.monotonic_ns()
     elapsed_time = end_time - start_time