fix: don't flush at every append, results in bad perf

google/cloud/storage/asyncio/async_appendable_object_writer.py

@@ -211,6 +211,7 @@ def __init__(
         self.bytes_appended_since_last_flush = 0
         self._routing_token: Optional[str] = None
         self.object_resource: Optional[_storage_v2.Object] = None
+        self._flush_count = 0
 
     async def state_lookup(self) -> int:
         """Returns the persisted_size
@@ -318,6 +319,8 @@ async def _do_open():
                 self.write_handle = self.write_obj_stream.write_handle
             if self.write_obj_stream.persisted_size is not None:
                 self.persisted_size = self.write_obj_stream.persisted_size
+                # set offset while opening
+                self.offset = self.persisted_size
 
             self._is_stream_open = True
             self._routing_token = None
@@ -400,39 +403,44 @@ async def generator():
                     write_state.write_handle = self.write_handle
                     write_state.routing_token = None
 
-                    write_state.user_buffer.seek(write_state.persisted_size)
-                    write_state.bytes_sent = write_state.persisted_size
+                    write_state.user_buffer.seek(write_state.bytes_sent)
+                    # write_state.bytes_sent =
+                    # TODO: why ?
                     write_state.bytes_since_last_flush = 0
+                    self.bytes_appended_since_last_flush = 0
 
                     requests = strategy.generate_requests(state)
 
-                num_requests = len(requests)
-                for i, chunk_req in enumerate(requests):
-                    if i == num_requests - 1:
-                        chunk_req.state_lookup = True
-                        chunk_req.flush = True
+                # print('len(requests)', len(requests))
+                for chunk_req in requests:
                     await self.write_obj_stream.send(chunk_req)
+                    if chunk_req.flush:
+                        self._flush_count += 1
 
-                resp = await self.write_obj_stream.recv()
-                if resp:
-                    if resp.persisted_size is not None:
-                        self.persisted_size = resp.persisted_size
-                        state["write_state"].persisted_size = resp.persisted_size
-                        self.offset = self.persisted_size
-                    if resp.write_handle:
-                        self.write_handle = resp.write_handle
-                        state["write_state"].write_handle = resp.write_handle
-                    self.bytes_appended_since_last_flush = 0
+                    if chunk_req.state_lookup:
+                        # TODO: if there's error, it'll raise error
+                        # and will be handled by `recover_state_on_failure`
+                        resp = await self.write_obj_stream.recv()
 
-                yield resp
+                    if resp:
+                        if resp.persisted_size is not None:
+                            self.persisted_size = resp.persisted_size
+                            state["write_state"].persisted_size = resp.persisted_size
+                            self.offset = self.persisted_size
+                        if resp.write_handle:
+                            self.write_handle = resp.write_handle
+                            state["write_state"].write_handle = resp.write_handle
+
+                yield None
 
             return generator()
 
         # State initialization
         write_state = _WriteState(_MAX_CHUNK_SIZE_BYTES, buffer, self.flush_interval)
         write_state.write_handle = self.write_handle
         write_state.persisted_size = self.persisted_size
-        write_state.bytes_sent = self.persisted_size
+        # TODO: what if you open a object in b/w?
+        write_state.bytes_sent = self.offset or 0
         write_state.bytes_since_last_flush = self.bytes_appended_since_last_flush
 
         retry_manager = _BidiStreamRetryManager(
@@ -442,11 +450,12 @@ async def generator():
         await retry_manager.execute({"write_state": write_state}, retry_policy)
 
         # Sync local markers
+        # TODO: looks like a blunder ?(why??)
         self.write_obj_stream.persisted_size = write_state.persisted_size
+        # TODO: why update write hand to write_obj_stream ?
         self.write_obj_stream.write_handle = write_state.write_handle
         self.bytes_appended_since_last_flush = write_state.bytes_since_last_flush
-        self.persisted_size = write_state.persisted_size
-        self.offset = write_state.persisted_size
+        self.offset = write_state.bytes_sent
 
     async def simple_flush(self) -> None:
         """Flushes the data to the server.
@@ -516,7 +525,7 @@ async def close(self, finalize_on_close=False) -> Union[int, _storage_v2.Object]
         await self.write_obj_stream.close()
 
         self._is_stream_open = False
-        self.offset = None
+        # self.offset = None
         return self.persisted_size
 
     async def finalize(self) -> _storage_v2.Object:

google/cloud/storage/asyncio/async_write_object_stream.py

@@ -190,6 +190,8 @@ async def requests_done(self):
         _utils.update_write_handle_if_exists(self, first_resp)
 
         if first_resp != grpc.aio.EOF:
+            # this persisted_size will not be upto date., also what if response
+            # doesn't have persisted_size? , it'll throw error.
             self.persisted_size = first_resp.persisted_size
             second_resp = await self.socket_like_rpc.recv()
             assert second_resp == grpc.aio.EOF

google/cloud/storage/asyncio/retry/bidi_stream_retry_manager.py

@@ -54,6 +54,7 @@ async def attempt():
             stream = self._send_and_recv(requests, state)
             try:
                 async for response in stream:
+                    # print('coming here')
                     self._strategy.update_state_from_response(response, state)
                 return
             except Exception as e:

google/cloud/storage/asyncio/retry/writes_resumption_strategy.py

@@ -89,9 +89,11 @@ def generate_requests(
             write_state.bytes_sent += chunk_len
             write_state.bytes_since_last_flush += chunk_len
 
+            count = 0
             if write_state.bytes_since_last_flush >= write_state.flush_interval:
+                count += 1
                 request.flush = True
-                # reset counter after marking flush
+                request.state_lookup = True
                 write_state.bytes_since_last_flush = 0
 
             requests.append(request)

samples/snippets/zonal_buckets/benchmark_ranged_read.py

@@ -0,0 +1,216 @@
+#!/usr/bin/env python
+
+# Copyright 2026 Google Inc. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the 'License');
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import argparse
+import asyncio
+import time
+import statistics
+from io import BytesIO
+
+from google.cloud.storage.asyncio.async_grpc_client import AsyncGrpcClient
+from google.cloud.storage.asyncio.async_multi_range_downloader import (
+    AsyncMultiRangeDownloader,
+)
+import random
+
+OBJECT_SIZE = 1024**3  # 1 GiB
+
+
+async def benchmark_single_stream(bucket_name, object_name, num_ranges, range_size):
+    """Benchmark downloading n ranges using 1 stream."""
+    grpc_client = AsyncGrpcClient()
+    mrd = AsyncMultiRangeDownloader(grpc_client, bucket_name, object_name)
+
+    total_downloaded_size = 0
+    start_time = time.monotonic()
+    try:
+        await mrd.open()
+
+        buffers = [BytesIO() for _ in range(num_ranges)]
+        # ranges = [(i * range_size, range_size, buffers[i]) for i in range(num_ranges)]
+        ranges = []
+        for i in range(num_ranges):
+            offset = random.randint(0, OBJECT_SIZE - range_size)
+            ranges.append((offset, range_size, buffers[i]))
+
+        # start_time = time.monotonic()
+        await mrd.download_ranges(ranges)
+        end_time = time.monotonic()
+
+        for output_buffer in buffers:
+            total_downloaded_size += output_buffer.getbuffer().nbytes
+
+    finally:
+        await mrd.close()
+
+    latency = end_time - start_time
+    throughput = total_downloaded_size / (1024 * 1024) / latency
+    # print(f"Total downloaded size: {total_downloaded_size} bytes")
+    # print(f"Time taken: {latency:.4f} seconds")
+    # print(f"Throughput: {throughput:.4f} MiB/s")
+    return latency, throughput
+
+
+async def download_one_range(mrd, start_byte, range_size, buffer):
+    """Helper coroutine for multi-stream benchmark"""
+    await mrd.download_ranges([(start_byte, range_size, buffer)])
+
+
+async def benchmark_multi_stream(
+    bucket_name, object_name, num_ranges, range_size, num_workers
+):
+    """Benchmark downloading n ranges in n streams."""
+    grpc_client = AsyncGrpcClient()
+    buffers = [BytesIO() for _ in range(num_ranges)]
+    mrds = [
+        AsyncMultiRangeDownloader(grpc_client, bucket_name, object_name)
+        for _ in range(num_ranges)
+    ]
+
+    total_downloaded_size = 0
+    start_time = time.monotonic()
+    try:
+        await asyncio.gather(*(mrd.open() for mrd in mrds))
+
+        tasks = []
+        for i in range(num_ranges):
+            offset = random.randint(0, OBJECT_SIZE - range_size)
+            task = asyncio.create_task(
+                download_one_range(
+                    mrds[i],
+                    offset,
+                    range_size,
+                    buffers[i],
+                )
+            )
+            tasks.append(task)
+
+        # start_time = time.monotonic()
+        await asyncio.gather(*tasks)
+        end_time = time.monotonic()
+
+        for output_buffer in buffers:
+            total_downloaded_size += output_buffer.getbuffer().nbytes
+    finally:
+        await asyncio.gather(*(mrd.close() for mrd in mrds))
+
+    latency = end_time - start_time
+    throughput = total_downloaded_size / (1024 * 1024) / latency
+    # print(f"Total downloaded size: {total_downloaded_size} bytes")
+    # print(f"Time taken: {latency:.4f} seconds")
+    # print(f"Throughput: {throughput:.4f} MiB/s")
+    return latency, throughput
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(
+        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
+    )
+    parser.add_argument(
+        "--bucket_name", help="Your Cloud Storage bucket name.", required=True
+    )
+    parser.add_argument(
+        "--object_name", help="Your Cloud Storage object name.", required=True
+    )
+    parser.add_argument(
+        "--num_ranges", help="Number of ranges to download.", type=int, default=10
+    )
+    parser.add_argument(
+        "--range_size",
+        help="Size of each range in bytes.",
+        type=int,
+        default=1024 * 1024,
+    )
+    parser.add_argument(
+        "--num_workers",
+        help="Number of concurrent workers for multi-stream download.",
+        type=int,
+        default=10,
+    )
+    parser.add_argument(
+        "--num_iterations",
+        help="Number of iterations to run the benchmark.",
+        type=int,
+        default=1,
+    )
+    parser.add_argument(
+        "--scenario",
+        choices=["single-stream", "multi-stream", "all"],
+        default="all",
+        help="Which benchmark scenario to run.",
+    )
+
+    args = parser.parse_args()
+
+    if args.scenario == "single-stream" or args.scenario == "all":
+        latencies = []
+        throughputs = []
+        for i in range(args.num_iterations):
+            # print(f"\n--- Running single-stream iteration {i+1}/{args.num_iterations} ---")
+            latency, throughput = asyncio.run(
+                benchmark_single_stream(
+                    args.bucket_name, args.object_name, args.num_ranges, args.range_size
+                )
+            )
+            latencies.append(latency)
+            throughputs.append(throughput)
+
+        print("\n--- single-stream Benchmark Summary ---")
+        if latencies:
+            print(f"Latencies (s):")
+            print(f"  Mean: {statistics.mean(latencies):.4f}")
+            print(f"  Median: {statistics.median(latencies):.4f}")
+            print(f"  Min: {min(latencies):.4f}")
+            print(f"  Max: {max(latencies):.4f}")
+
+        if throughputs:
+            print(f"Throughputs (MiB/s):")
+            print(f"  Mean: {statistics.mean(throughputs):.4f}")
+            print(f"  Median: {statistics.median(throughputs):.4f}")
+            print(f"  Min: {min(throughputs):.4f}")
+            print(f"  Max: {max(throughputs):.4f}")
+
+    if args.scenario == "multi-stream" or args.scenario == "all":
+        latencies = []
+        throughputs = []
+        for i in range(args.num_iterations):
+            # print(f"\n--- Running multi-stream iteration {i+1}/{args.num_iterations} ---")
+            latency, throughput = asyncio.run(
+                benchmark_multi_stream(
+                    args.bucket_name,
+                    args.object_name,
+                    args.num_ranges,
+                    args.range_size,
+                    args.num_workers,
+                )
+            )
+            latencies.append(latency)
+            throughputs.append(throughput)
+
+        print("\n--- multi-stream Benchmark Summary ---")
+        if latencies:
+            print(f"Latencies (s):")
+            print(f"  Mean: {statistics.mean(latencies):.4f}")
+            print(f"  Median: {statistics.median(latencies):.4f}")
+            print(f"  Min: {min(latencies):.4f}")
+            print(f"  Max: {max(latencies):.4f}")
+
+        if throughputs:
+            print(f"Throughputs (MiB/s):")
+            print(f"  Mean: {statistics.mean(throughputs):.4f}")
+            print(f"  Median: {statistics.median(throughputs):.4f}")
+            print(f"  Min: {min(throughputs):.4f}")
+            print(f"  Max: {max(throughputs):.4f}")

samples/snippets/zonal_buckets/storage_read_appendable_object_tail.py

@@ -28,7 +28,7 @@
     AsyncMultiRangeDownloader,
 )
 
-BYTES_TO_APPEND = b"fav_bytes."
+BYTES_TO_APPEND = b"fav_bytes." * 100 * 1024 * 1024
 NUM_BYTES_TO_APPEND_EVERY_SECOND = len(BYTES_TO_APPEND)
 
 
@@ -37,14 +37,16 @@ async def appender(writer: AsyncAppendableObjectWriter, duration: int):
     """Appends 10 bytes to the object every second for a given duration."""
     print("Appender started.")
     bytes_appended = 0
-    for i in range(duration):
+    start_time = time.monotonic()
+    # Run the appender for the specified duration.
+    while time.monotonic() - start_time < duration:
         await writer.append(BYTES_TO_APPEND)
         now = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
         bytes_appended += NUM_BYTES_TO_APPEND_EVERY_SECOND
         print(
             f"[{now}] Appended {NUM_BYTES_TO_APPEND_EVERY_SECOND} new bytes. Total appended: {bytes_appended} bytes."
         )
-        await asyncio.sleep(1)
+        await asyncio.sleep(0.1)
     print("Appender finished.")
 
 
@@ -67,9 +69,7 @@ async def tailer(
             bytes_downloaded = output_buffer.getbuffer().nbytes
             if bytes_downloaded > 0:
                 now = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
-                print(
-                    f"[{now}] Tailer read {bytes_downloaded} new bytes: {output_buffer.getvalue()}"
-                )
+                print(f"[{now}] Tailer read {bytes_downloaded} new bytes: ")
                 start_byte += bytes_downloaded
 
             await asyncio.sleep(0.1)  # Poll for new data every 0.1 seconds.

tests/perf/microbenchmarks/writes/config.py

@@ -86,7 +86,7 @@ def get_write_params() -> Dict[str, List[WriteParameters]]:
             num_files = num_processes * num_coros
 
             # Create a descriptive name for the parameter set
-            name = f"{workload_name}_{bucket_type}_{num_processes}p_{num_coros}c"
+            name = f"{workload_name}_{bucket_type}_{num_processes}p_{num_coros}c_{chunk_size_mib}csize"
 
             params[workload_name].append(
                 WriteParameters(