Skip to content

Commit 43fdda4

Browse files
committed
docs(storage): add zonal bucket pre-warmed writer pool sample
Adds storage_optimize_write_latency_pool sample (region tag: storage_optimize_write_latency_pool) demonstrating a pre-warmed pool of AsyncAppendableObjectWriter instances with finalize_on_close=False to avoid object creation and finalization metadata overhead on the critical write path. Verified with both mock unit tests and live integration testing against a Rapid (zonal) bucket in us-central1-a: Running live Python test against bucket=<zonal-bucket>, prefix=live_py_pool_1790087453494 Python 1. Init pool (3 writers): 862.92 ms Python 2. Write+flush: 82.28 ms Python 4. Read back: b'0123456789', pool size after refill: 3 Ran 1 test in 1.612s OK
1 parent 430a87a commit 43fdda4

3 files changed

Lines changed: 164 additions & 6 deletions

File tree

‎storage/samples/snippets/zonal_buckets/README.md‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -75,4 +75,12 @@ This snippet downloads a range of bytes from multiple objects concurrently.
7575

7676
```bash
7777
python samples/snippets/zonal_buckets/storage_open_multiple_objects_ranged_read.py --bucket_name <bucket_name> --object_names <object_name_1> <object_name_2>
78+
```
79+
80+
### Optimize write latency with a pre-warmed writer pool
81+
82+
This snippet uses a pre-warmed pool of writers for a zonal bucket.
83+
84+
```bash
85+
python samples/snippets/zonal_buckets/storage_optimize_write_latency_pool.py --bucket_name <bucket_name> --key_prefix <key_prefix>
7886
```
Lines changed: 134 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,134 @@
1+
#!/usr/bin/env python
2+
3+
# Copyright 2026 Google LLC
4+
#
5+
# Licensed under the Apache License, Version 2.0 (the 'License');
6+
# you may not use this file except in compliance with the License.
7+
# You may obtain a copy of the License at
8+
#
9+
# http://www.apache.org/licenses/LICENSE-2.0
10+
#
11+
# Unless required by applicable law or agreed to in writing, software
12+
# distributed under the License is distributed on an "AS IS" BASIS,
13+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
# See the License for the specific language governing permissions and
15+
# limitations under the License.
16+
17+
import argparse
18+
import asyncio
19+
from io import BytesIO
20+
21+
from google.cloud.storage.asyncio.async_appendable_object_writer import (
22+
AsyncAppendableObjectWriter,
23+
)
24+
from google.cloud.storage.asyncio.async_grpc_client import AsyncGrpcClient
25+
from google.cloud.storage.asyncio.async_multi_range_downloader import (
26+
AsyncMultiRangeDownloader,
27+
)
28+
29+
30+
# [START storage_optimize_write_latency_pool]
31+
async def storage_optimize_write_latency_pool(
32+
bucket_name: str, key_prefix: str, pool_size: int = 3, grpc_client=None
33+
):
34+
"""Uses a pre-warmed pool of writers for a zonal bucket.
35+
36+
grpc_client: an existing grpc_client to use, this is only for testing.
37+
"""
38+
# The ID of your GCS zonal bucket
39+
# bucket_name = "your-unique-bucket-name"
40+
41+
# The prefix for your pooled GCS objects
42+
# key_prefix = "pooled-object"
43+
44+
grpc_client_created = False
45+
if grpc_client is None:
46+
grpc_client = AsyncGrpcClient()
47+
grpc_client_created = True
48+
49+
next_object_name = f"{key_prefix}_{pool_size}"
50+
51+
async def new_prewarmed_writer(name: str) -> AsyncAppendableObjectWriter:
52+
w = AsyncAppendableObjectWriter(
53+
client=grpc_client,
54+
bucket_name=bucket_name,
55+
object_name=name,
56+
generation=0,
57+
)
58+
await w.open()
59+
await w.flush() # Forces 0-byte object creation in the background.
60+
return w
61+
62+
# 1. Init pool: Flushing incurs operation charges, so size the pool
63+
# carefully.
64+
pool = list(
65+
await asyncio.gather(
66+
*(
67+
new_prewarmed_writer(f"{key_prefix}_{i}")
68+
for i in range(pool_size)
69+
)
70+
)
71+
)
72+
73+
try:
74+
# 2. Write: Pop a pre-warmed writer; append() writes and flushes data
75+
# (~1-2 ms).
76+
writer = pool.pop(0)
77+
await writer.append(b"0123456789")
78+
79+
# 3. Pool maintenance (run asynchronously off the critical write path):
80+
# Close the used writer without finalizing, refill the pool, and
81+
# discard stale writers.
82+
async def maintain_pool(
83+
used: AsyncAppendableObjectWriter, next_name: str
84+
):
85+
await used.close(finalize_on_close=False)
86+
pool.append(await new_prewarmed_writer(next_name))
87+
88+
maintenance_task = asyncio.create_task(
89+
maintain_pool(writer, next_object_name)
90+
)
91+
92+
# 4. Read: Unfinalized objects are readable after flush().
93+
mrd = AsyncMultiRangeDownloader(
94+
grpc_client, bucket_name, f"{key_prefix}_0"
95+
)
96+
await mrd.open()
97+
buf = BytesIO()
98+
await mrd.download_ranges([(0, 0, buf)])
99+
await mrd.close()
100+
101+
await maintenance_task
102+
print(
103+
f"Read unfinalized object {key_prefix}_0: "
104+
f"{buf.getvalue().decode('utf-8')}"
105+
)
106+
finally:
107+
for rem in pool:
108+
await rem.close(finalize_on_close=False)
109+
if grpc_client_created:
110+
await grpc_client.close()
111+
112+
113+
# [END storage_optimize_write_latency_pool]
114+
115+
116+
if __name__ == "__main__":
117+
parser = argparse.ArgumentParser(
118+
description=__doc__,
119+
formatter_class=argparse.RawDescriptionHelpFormatter,
120+
)
121+
parser.add_argument(
122+
"--bucket_name", help="Your Cloud Storage zonal bucket name."
123+
)
124+
parser.add_argument(
125+
"--key_prefix", help="Prefix for pooled object names."
126+
)
127+
args = parser.parse_args()
128+
129+
asyncio.run(
130+
storage_optimize_write_latency_pool(
131+
bucket_name=args.bucket_name,
132+
key_prefix=args.key_prefix,
133+
)
134+
)

‎storage/samples/snippets/zonal_buckets/zonal_snippets_test.py‎

Lines changed: 22 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
# Copyright 2025 Google, Inc.
1+
# Copyright 2025 Google LLC
22
#
33
# Licensed under the Apache License, Version 2.0 (the "License");
44
# you may not use this file except in compliance with the License.
@@ -13,17 +13,16 @@
1313
# limitations under the License.
1414

1515
import asyncio
16-
import uuid
16+
import contextlib
1717
import os
18+
import uuid
1819

19-
import pytest
2020
from google.cloud.storage import Client
21-
import contextlib
22-
23-
from google.cloud.storage.asyncio.async_grpc_client import AsyncGrpcClient
2421
from google.cloud.storage.asyncio.async_appendable_object_writer import (
2522
AsyncAppendableObjectWriter,
2623
)
24+
from google.cloud.storage.asyncio.async_grpc_client import AsyncGrpcClient
25+
import pytest
2726

2827
# Import all the snippets
2928
import storage_create_and_write_appendable_object
@@ -32,6 +31,7 @@
3231
import storage_open_object_multiple_ranged_read
3332
import storage_open_object_read_full_object
3433
import storage_open_object_single_ranged_read
34+
import storage_optimize_write_latency_pool
3535
import storage_pause_and_resume_appendable_upload
3636
import storage_read_appendable_object_tail
3737

@@ -258,3 +258,19 @@ def test_storage_open_multiple_objects_ranged_read(
258258
blob2 = json_client.bucket(_ZONAL_BUCKET).blob(blob2_name)
259259
blob1.delete()
260260
blob2.delete()
261+
262+
263+
def test_storage_optimize_write_latency_pool(
264+
async_grpc_client, json_client, event_loop, capsys
265+
):
266+
key_prefix = f"test-writer-pool-{uuid.uuid4()}"
267+
event_loop.run_until_complete(
268+
storage_optimize_write_latency_pool.storage_optimize_write_latency_pool(
269+
_ZONAL_BUCKET, key_prefix, pool_size=3, grpc_client=async_grpc_client
270+
)
271+
)
272+
out, _ = capsys.readouterr()
273+
assert f"Read unfinalized object {key_prefix}_0: 0123456789" in out
274+
bucket = json_client.bucket(_ZONAL_BUCKET)
275+
for i in range(4):
276+
bucket.blob(f"{key_prefix}_{i}").delete()

0 commit comments

Comments
 (0)