docs(storage): add zonal bucket pre-warmed writer pool sample - #14606
chandra-siri merged 1 commit into
Conversation
|
Here is the summary of changes. You are about to add 1 region tag.
This comment is generated by snippet-bot.
|
There was a problem hiding this comment.
Code Review
This pull request introduces a new code snippet and documentation demonstrating how to optimize write latency in zonal buckets using a pre-warmed writer pool, along with corresponding integration tests. The review feedback suggests two key improvements: concurrently initializing the writer pool using asyncio.gather to reduce startup latency, and wrapping the operations in a try...finally block to prevent resource leaks of the writers and the AsyncGrpcClient in case of exceptions.
| if grpc_client is None: | ||
| grpc_client = AsyncGrpcClient() | ||
|
|
||
| next_object_name = f"{key_prefix}_{pool_size}" | ||
|
|
||
| async def new_prewarmed_writer(name: str) -> AsyncAppendableObjectWriter: | ||
| w = AsyncAppendableObjectWriter( | ||
| client=grpc_client, | ||
| bucket_name=bucket_name, | ||
| object_name=name, | ||
| generation=0, | ||
| ) | ||
| await w.open() | ||
| await w.flush() # Forces 0-byte object creation in the background. | ||
| return w | ||
|
|
||
| # 1. Init pool: Flushing incurs operation charges, so size the pool | ||
| # carefully. | ||
| pool = [ | ||
| await new_prewarmed_writer(f"{key_prefix}_{i}") | ||
| for i in range(pool_size) | ||
| ] | ||
|
|
||
| # 2. Write: Pop a pre-warmed writer; append() writes and flushes data | ||
| # (~1-2 ms). | ||
| writer = pool.pop(0) | ||
| await writer.append(b"0123456789") | ||
|
|
||
| # 3. Pool maintenance (run asynchronously off the critical write path): | ||
| # Close the used writer without finalizing, refill the pool, and discard | ||
| # stale writers. | ||
| async def maintain_pool(used: AsyncAppendableObjectWriter, next_name: str): | ||
| await used.close(finalize_on_close=False) | ||
| pool.append(await new_prewarmed_writer(next_name)) | ||
|
|
||
| maintenance_task = asyncio.create_task( | ||
| maintain_pool(writer, next_object_name) | ||
| ) | ||
|
|
||
| # 4. Read: Unfinalized objects are readable after flush(). | ||
| mrd = AsyncMultiRangeDownloader( | ||
| grpc_client, bucket_name, f"{key_prefix}_0" | ||
| ) | ||
| await mrd.open() | ||
| buf = BytesIO() | ||
| await mrd.download_ranges([(0, 0, buf)]) | ||
| await mrd.close() | ||
|
|
||
| await maintenance_task | ||
| for rem in pool: | ||
| await rem.close(finalize_on_close=False) |
There was a problem hiding this comment.
There are two main areas of improvement in this implementation:
- Performance/Efficiency: The pre-warmed pool is currently initialized sequentially using a list comprehension with
await. Since opening and flushing writers are I/O-bound operations, they can be executed concurrently usingasyncio.gather. This will significantly reduce the pool startup latency (e.g., from ~860ms down to ~300ms for 3 writers). - Resource Leak / Exception Safety: If an exception occurs during the read operation (step 4) or if
maintenance_taskfails, the remaining writers in the pool are never closed, and the locally createdAsyncGrpcClientis leaked. Wrapping the execution in atry...finallyblock ensures that all resources are reliably cleaned up.
grpc_client_created = False
if grpc_client is None:
grpc_client = AsyncGrpcClient()
grpc_client_created = True
next_object_name = f"{key_prefix}_{pool_size}"
async def new_prewarmed_writer(name: str) -> AsyncAppendableObjectWriter:
w = AsyncAppendableObjectWriter(
client=grpc_client,
bucket_name=bucket_name,
object_name=name,
generation=0,
)
await w.open()
await w.flush() # Forces 0-byte object creation in the background.
return w
# 1. Init pool concurrently: Flushing incurs operation charges, so size the pool
# carefully. Concurrently initializing the pool avoids sequential latency overhead.
pool = await asyncio.gather(
*(new_prewarmed_writer(f"{key_prefix}_{i}") for i in range(pool_size))
)
try:
# 2. Write: Pop a pre-warmed writer; append() writes and flushes data
# (~1-2 ms).
writer = pool.pop(0)
await writer.append(b"0123456789")
# 3. Pool maintenance (run asynchronously off the critical write path):
# Close the used writer without finalizing, refill the pool, and discard
# stale writers.
async def maintain_pool(used: AsyncAppendableObjectWriter, next_name: str):
await used.close(finalize_on_close=False)
pool.append(await new_prewarmed_writer(next_name))
maintenance_task = asyncio.create_task(
maintain_pool(writer, next_object_name)
)
# 4. Read: Unfinalized objects are readable after flush().
mrd = AsyncMultiRangeDownloader(
grpc_client, bucket_name, f"{key_prefix}_0"
)
await mrd.open()
buf = BytesIO()
await mrd.download_ranges([(0, 0, buf)])
await mrd.close()
await maintenance_task
finally:
# Ensure all remaining writers in the pool are closed even if an exception occurs.
for rem in pool:
await rem.close(finalize_on_close=False)
# Close the gRPC client if it was created locally to prevent resource leaks.
if grpc_client_created:
await grpc_client.close()22a2c87 to
243ad1a
Compare
243ad1a to
43fdda4
Compare
43fdda4 to
8b687d9
Compare
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
8b687d9 to
8dd1f99
Compare
|
/gcbrun(8dd1f99) |
15652be
into
GoogleCloudPlatform:main
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=, 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
Description
Fixes #
Checklist
Testing
Compliance & Style
Post-Approval Actions