Skip to content
This repository was archived by the owner on Mar 31, 2026. It is now read-only.

Commit 970b162

Browse files
chandra-sirichandra-siri
committed
working copy
1 parent 3c7e7af commit 970b162

11 files changed

Lines changed: 145 additions & 86 deletions

File tree

‎benchmarks/download_zb_n_workers.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,7 @@ def main():
4343
parser = argparse.ArgumentParser()
4444
parser.add_argument("--bucket_name", type=str, default='chandrasiri-rs')
4545
parser.add_argument("--download_size", type=int, default=1024 * 1024 * 1024) # 1 GiB
46-
parser.add_argument("--chunk_size", type=int, default=100 * 1024 * 1024) # 100 MiB
46+
parser.add_argument("--chunk_size", type=int, default=64 * 1024 * 1024) # 100 MiB
4747
parser.add_argument("--count", type=int, default=100)
4848
parser.add_argument("--start_object_num", type=int, default=0)
4949
parser.add_argument("-n", "--num_workers", type=int, default=2, help="Number of worker threads or processes.")

‎benchmarks/parallel_uploader.py‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ async def upload_object_async(bucket_name, object_name):
2828

2929
def upload_worker(object_name):
3030
"""A synchronous wrapper to be called by multiprocessing."""
31-
bucket_name = "chandrasiri-rs"
31+
bucket_name = "chandrasiri-benchmarks-zb" # Replace with your bucket name
3232
try:
3333
asyncio.run(upload_object_async(bucket_name, object_name))
3434
return f"Successfully uploaded {object_name}"
@@ -40,9 +40,9 @@ def upload_worker(object_name):
4040
def main():
4141
"""Main function to orchestrate parallel uploads."""
4242
num_objects = 3000
43-
num_processes = 24
43+
num_processes = 64
4444

45-
object_names = [f"para_64-{i}" for i in range(num_objects)]
45+
object_names = [f"high_mem_long_running-{i}" for i in range(num_objects)]
4646

4747
with Pool(processes=num_processes) as pool:
4848
for result in pool.imap_unordered(upload_worker, object_names):

‎benchmarks/read_appendable_object.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -55,4 +55,4 @@ def get_persisted_size_sync(bucket_name, object_name):
5555
# future.result()
5656

5757
if __name__ == "__main__":
58-
get_persisted_size_sync("chandrasiri-rs", sys.argv[1])
58+
get_persisted_size_sync(sys.argv[1], sys.argv[2])

‎benchmarks/upload_zb_n_workers.py‎

Lines changed: 83 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -11,14 +11,21 @@
1111
)
1212
import math
1313
import uuid
14+
import sys
15+
import os
16+
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
17+
from tests.perf.microbenchmarks.resource_monitor import ResourceMonitor
1418

1519
async def upload_one_async(bucket_name, object_name, upload_size, chunk_size):
1620
"""Uploads a single object of size `upload_size`, in chunks of `chunk_size`"""
1721
print(f"Uploading {object_name} of size {upload_size} in chunks of {chunk_size} to {bucket_name} from process {os.getpid()} and thread {threading.get_ident()}")
1822
client = AsyncGrpcClient().grpc_client
1923
start_time = time.perf_counter()
2024
writer = AsyncAppendableObjectWriter(
21-
client=client, bucket_name=bucket_name, object_name=object_name
25+
client=client,
26+
bucket_name=bucket_name,
27+
object_name=object_name,
28+
writer_options={"FLUSH_INTERVAL_BYTES": 1026 * 1024**2},
2229
)
2330

2431
await writer.open()
@@ -30,7 +37,7 @@ async def upload_one_async(bucket_name, object_name, upload_size, chunk_size):
3037
await writer.append(data)
3138
uploaded_bytes += bytes_to_upload
3239
count += 1
33-
await writer.close()
40+
await writer.close(finalize_on_close=True)
3441
assert uploaded_bytes == upload_size
3542
assert count == math.ceil(upload_size / chunk_size)
3643

@@ -42,54 +49,87 @@ async def upload_one_async(bucket_name, object_name, upload_size, chunk_size):
4249
print(f"Latency: {latency:.2f} seconds")
4350
print(f"Throughput: {throughput:.2f} MB/s")
4451

45-
# def upload_one_sync(bucket_name, object_name, upload_size, chunk_size):
46-
# """Wrapper to run the async upload_one in a new event loop."""
47-
# asyncio.run(upload_one_async(bucket_name, object_name, upload_size, chunk_size))
48-
4952

50-
# def main():
51-
# parser = argparse.ArgumentParser()
52-
# parser.add_argument("--bucket_name", type=str, default='chandrasiri-rs')
53-
# parser.add_argument("--upload_size", type=int, default=1024 * 1024 * 1024) # 1 GiB
54-
# parser.add_argument("--chunk_size", type=int, default=100 * 1024 * 1024) # 100 MiB
55-
# parser.add_argument("--count", type=int, default=100)
56-
# parser.add_argument("--start_object_num", type=int, default=0)
57-
# parser.add_argument("-n", "--num_workers", type=int, default=2, help="Number of worker threads or processes.")
58-
# parser.add_argument("--executor", type=str, choices=['thread', 'process'], default='process', help="Executor to use: 'thread' for ThreadPoolExecutor, 'process' for ProcessPoolExecutor")
59-
# args = parser.parse_args()
53+
def upload_one_sync(bucket_name, object_name, upload_size, chunk_size):
54+
"""Wrapper to run the async upload_one in a new event loop."""
55+
asyncio.run(upload_one_async(bucket_name, object_name, upload_size, chunk_size))
6056

61-
# total_start_time = time.perf_counter()
6257

63-
# ExecutorClass = ThreadPoolExecutor if args.executor == 'thread' else ProcessPoolExecutor
58+
def main():
59+
parser = argparse.ArgumentParser()
60+
parser.add_argument("--bucket_name", type=str, default="chandrasiri-rs")
61+
parser.add_argument("--upload_size_mib", type=int, default=1024) # 1 GiB
62+
parser.add_argument("--chunk_size_mib", type=int, default=100) # 100 MiB
63+
parser.add_argument("--count", type=int, default=100)
64+
parser.add_argument("--start_object_num", type=int, default=0)
65+
parser.add_argument(
66+
"-n",
67+
"--num_workers",
68+
type=int,
69+
default=2,
70+
help="Number of worker threads or processes.",
71+
)
72+
parser.add_argument(
73+
"--executor",
74+
type=str,
75+
choices=["thread", "process"],
76+
default="process",
77+
help="Executor to use: 'thread' for ThreadPoolExecutor, 'process' for ProcessPoolExecutor",
78+
)
79+
args = parser.parse_args()
6480

65-
# with ExecutorClass(max_workers=args.num_workers) as executor:
66-
# futures = []
67-
# for i in range(args.start_object_num, args.start_object_num + args.count):
68-
# object_name = f"py-sdk-mb-mt-{i}"
69-
# future = executor.submit(upload_one_sync, args.bucket_name, object_name, args.upload_size, args.chunk_size)
70-
# futures.append(future)
81+
total_start_time = time.perf_counter()
7182

72-
# for future in futures:
73-
# future.result() # wait for all workers to complete
83+
ExecutorClass = (
84+
ThreadPoolExecutor if args.executor == "thread" else ProcessPoolExecutor
85+
)
7486

75-
# total_end_time = time.perf_counter()
76-
# total_latency = total_end_time - total_start_time
77-
# total_uploaded_bytes = args.upload_size * args.count
78-
# aggregate_throughput = (total_uploaded_bytes / total_latency) / (1000 * 1000) # MB/s
87+
with ResourceMonitor() as m:
88+
with ExecutorClass(max_workers=args.num_workers) as executor:
89+
futures = []
90+
for i in range(args.start_object_num, args.start_object_num + args.count):
91+
object_name = f"py-sdk-mb-mt-{i}"
92+
future = executor.submit(
93+
upload_one_sync,
94+
args.bucket_name,
95+
object_name,
96+
args.upload_size_mib * 1024 * 1024,
97+
args.chunk_size_mib * 1024 * 1024,
98+
)
99+
futures.append(future)
100+
101+
for future in futures:
102+
future.result() # wait for all workers to complete
103+
104+
105+
total_end_time = time.perf_counter()
106+
total_latency = total_end_time - total_start_time
107+
total_uploaded_bytes = args.upload_size_mib * 1024 * 1024 * args.count
108+
aggregate_throughput = (total_uploaded_bytes / total_latency) / (
109+
1000 * 1000
110+
) # MB/s
111+
112+
print("\n--- Aggregate Results ---")
113+
print(f"Total objects uploaded: {args.count}")
114+
print(f"Total data uploaded: {total_uploaded_bytes / (1024*1024*1024):.2f} GiB")
115+
print(f"Total time taken: {total_latency:.2f} seconds")
116+
print(f"Aggregate throughput: {aggregate_throughput:.2f} MB/s")
117+
118+
print("\n--- Resource Monitor Results ---")
119+
print(f"Max CPU: {m.max_cpu}")
120+
print(f"Max Memory: {m.max_mem}")
121+
print(f"Throughput (MB/s): {m.throughput_mb_s}")
122+
print(f"vCPUs: {m.vcpus}")
79123

80-
# print("\n--- Aggregate Results ---")
81-
# print(f"Total objects uploaded: {args.count}")
82-
# print(f"Total data uploaded: {total_uploaded_bytes / (1024*1024*1024):.2f} GiB")
83-
# print(f"Total time taken: {total_latency:.2f} seconds")
84-
# print(f"Aggregate throughput: {aggregate_throughput:.2f} MB/s")
85124

86125

87126
if __name__ == "__main__":
88-
parser = argparse.ArgumentParser()
89-
parser.add_argument("--bucket_name", type=str, default='chandrasiri-rs')
90-
# parser.add_argument("--object_suffix", type=str, required=True)
91-
parser.add_argument("--upload_size", type=int, default=1024 * 1024 * 1024) # 1 GiB
92-
parser.add_argument("--chunk_size", type=int, default=100 * 1024 * 1024) # 100 MiB
93-
args = parser.parse_args()
94-
object_name = f'upload-test-{str(uuid.uuid4())[:4]}'
95-
asyncio.run(upload_one_async(args.bucket_name, object_name, args.upload_size, args.chunk_size))
127+
main()
128+
# parser = argparse.ArgumentParser()
129+
# parser.add_argument("--bucket_name", type=str, default='chandrasiri-rs')
130+
# # parser.add_argument("--object_suffix", type=str, required=True)
131+
# parser.add_argument("--upload_size", type=int, default=1024 * 1024 * 1024) # 1 GiB
132+
# parser.add_argument("--chunk_size", type=int, default=100 * 1024 * 1024) # 100 MiB
133+
# args = parser.parse_args()
134+
# object_name = f'upload-test-{str(uuid.uuid4())[:4]}'
135+
# asyncio.run(upload_one_async(args.bucket_name, object_name, args.upload_size, args.chunk_size))

‎benchmarks/upload_zb_one_coro.py‎

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,11 @@
77
from google.cloud.storage._experimental.asyncio.async_appendable_object_writer import (
88
AsyncAppendableObjectWriter,
99
)
10+
import logging
11+
import sys
12+
13+
14+
logging.basicConfig(level=logging.DEBUG, stream=sys.stdout)
1015

1116
async def upload_one(client, bucket_name, object_name, upload_size, chunk_size):
1217
"""Uploads a single object of size `upload_size`, in chunks of `chunk_size`"""
@@ -33,7 +38,7 @@ async def upload_one(client, bucket_name, object_name, upload_size, chunk_size):
3338
latency = end_time - start_time
3439
throughput = (upload_size / latency) / (10**6) # MB/s
3540

36-
print(f"Finished uploading {object_name}")
41+
print(f"Finished uploading {object_name}, with generation, {writer.generation}")
3742
print(f"Latency: {latency:.2f} seconds")
3843
print(f"Throughput: {throughput:.2f} MB/s")
3944

@@ -45,9 +50,11 @@ async def main():
4550
args = parser.parse_args()
4651

4752
client = AsyncGrpcClient().grpc_client
48-
object_name = f"py-sdk-mb-1GiB-1"
53+
# object_name = f"test-half-close-current-code" # generation 1767887565143753
54+
# object_name = f"test-half-close-with-good-close"
55+
object_name = f"test-logs"
4956

5057
await upload_one(client, args.bucket_name, object_name, args.upload_size, args.chunk_size)
5158

5259
if __name__ == "__main__":
53-
asyncio.run(main())
60+
asyncio.run(main(), debug=True)

‎google/cloud/storage/_experimental/asyncio/async_appendable_object_writer.py‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -293,8 +293,8 @@ async def close(self, finalize_on_close=False) -> Union[int, _storage_v2.Object]
293293

294294
if finalize_on_close:
295295
await self.finalize()
296-
else:
297-
await self.flush()
296+
# else:
297+
# await self.flush()
298298

299299
await self.write_obj_stream.close()
300300

‎samples/snippets/storage_list_files_with_prefix.py‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -66,10 +66,14 @@ def list_blobs_with_prefix(bucket_name, prefix, delimiter=None):
6666

6767
# Note: The call returns a response only when the iterator is consumed.
6868
print("Blobs:")
69+
count = 0
6970
for blob in blobs:
70-
print(blob.name, "deleting it")
71+
if count % 500 == 0:
72+
print(blob.name)
73+
count += 1
74+
print(f"Total blobs with prefix {prefix}: {count}")
7175

72-
blob.delete()
76+
# blob.delete()
7377

7478
# if delimiter:
7579
# print("Prefixes:")

‎tests/perf/microbenchmarks/config.yaml‎

Lines changed: 14 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -1,16 +1,15 @@
11
common:
22
bucket_types:
3-
- "regional"
3+
# - "regional"
44
- "zonal"
55
file_sizes_mib:
6-
- 10 # 1GB
7-
# chunk_sizes_mib: [1, 2, 8, 64, 100, 128]
8-
chunk_sizes_mib: [6]
6+
- 1024 # 1GiB
7+
chunk_sizes_mib: [100]
98
rounds: 10
109

1110
workload:
1211

13-
############# single proc single coroutines #########
12+
############# single process single coroutine #########
1413
- name: "read_seq"
1514
pattern: "seq"
1615
coros: [1]
@@ -21,30 +20,30 @@ workload:
2120
coros: [1]
2221
processes: [1]
2322

24-
############# single proc multiple coroutines #########
23+
############# single process multi coroutine #########
2524

2625
- name: "read_seq_multi_coros"
2726
pattern: "seq"
28-
coros: [2, 4]
27+
coros: [2, 4, 8, 16]
2928
processes: [1]
3029

3130
- name: "read_rand_multi_coros"
3231
pattern: "rand"
33-
coros: [2, 4]
32+
coros: [2, 4, 8, 16]
3433
processes: [1]
3534

36-
############# multiple proc multiple coroutines #########
35+
############# multi process multi coroutine #########
3736
- name: "read_seq_multi_process"
3837
pattern: "seq"
39-
coros: [1, 2]
40-
processes: [2]
38+
coros: [4]
39+
processes: [16]
4140

4241
- name: "read_rand_multi_process"
4342
pattern: "rand"
44-
coros: [1,2]
45-
processes: [2,4]
43+
coros: [4]
44+
processes: [16]
4645

4746

4847
defaults:
49-
DEFAULT_RAPID_ZONAL_BUCKET: "chandrasiri-rs"
50-
DEFAULT_STANDARD_BUCKET: "gcs-pytest-benchmark-standard-bucket"
48+
DEFAULT_RAPID_ZONAL_BUCKET: "chandrasiri-benchmarks-zb"
49+
DEFAULT_STANDARD_BUCKET: "chandrasiri-benchmarks-rb"

‎tests/perf/microbenchmarks/config_writes.yaml‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -5,8 +5,8 @@ common:
55
file_sizes_mib:
66
- 1024 # 1GB
77
# chunk_sizes_mib: [1, 2, 8, 64, 100, 128]
8-
chunk_sizes_mib: [64]
9-
rounds: 3
8+
chunk_sizes_mib: [100]
9+
rounds: 10
1010

1111
workload:
1212

@@ -20,16 +20,16 @@ workload:
2020

2121
- name: "write_seq_multi_coros"
2222
pattern: "seq"
23-
coros: [2, 4]
23+
coros: [2, 4, 8, 16]
2424
processes: [1]
2525

2626
############# multiple proc multiple coroutines #########
2727
- name: "write_seq_multi_process"
2828
pattern: "seq"
2929
coros: [1, 2]
30-
processes: [2]
30+
processes: [8, 16, 32, 64]
3131

3232

3333
defaults:
34-
DEFAULT_RAPID_ZONAL_BUCKET: "chandrasiri-rs"
35-
DEFAULT_STANDARD_BUCKET: "gcs-pytest-benchmark-standard-bucket"
34+
DEFAULT_RAPID_ZONAL_BUCKET: "chandrasiri-benchmarks-zb"
35+
DEFAULT_STANDARD_BUCKET: "chandrasiri-benchmarks-rb"

‎tests/perf/microbenchmarks/conftest.py‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -67,7 +67,7 @@ def publish_resource_metrics(benchmark: Any, monitor: ResourceMonitor) -> None:
6767

6868
async def upload_appendable_object(bucket_name, object_name, object_size, chunk_size):
6969
writer = AsyncAppendableObjectWriter(
70-
AsyncGrpcClient().grpc_client, bucket_name, object_name
70+
AsyncGrpcClient().grpc_client, bucket_name, object_name, writer_options={"FLUSH_INTERVAL_BYTES": 1026 * 1024 * 1024}
7171
)
7272
await writer.open()
7373
uploaded_bytes = 0
@@ -101,7 +101,7 @@ def _upload_worker(args):
101101
return object_name, uploaded_bytes
102102

103103

104-
def _create_files(num_files, bucket_name, bucket_type, object_size, chunk_size=128 * 1024 * 1024):
104+
def _create_files(num_files, bucket_name, bucket_type, object_size, chunk_size=1024 * 1024 * 1024):
105105
"""
106106
1. using upload_appendable_object implement this and return a list of file names.
107107
TODO: adapt this to REGIONAL BUCKETS as well.

0 commit comments

Comments
 (0)