timetest.py

import time
import random
import uuid
from functools import partial
from google.cloud import storage

def time_test(func):
  start_time = time.monotonic()
  func()
  end_time = time.monotonic()
  time_taken = end_time-start_time
  print(f"{func.args[-1]}kIB / {time_taken:.3f}s: {func.args[-1]/time_taken:.1f}kIB/s")
  

BUCKET_NAME = 'ag-test-turbo-replication'

def _create_block(desired_kib):
  line = "abc123XYZ" * 14 + "!" + "\n"
  return int(desired_kib / len(line)) * line

def upload(bucket, blob_name, size):
  blob = bucket.blob(blob_name)
  payload = _create_block(size)
  blob.upload_from_string(payload)

def download(bucket, blob_name, size):
  blob = bucket.blob(blob_name)
  _ = blob.download_as_text()

def generate_func_list(bucket_name, min_size, max_size):
  size = random.randint(min_size, max_size)
  blob_name = uuid.uuid4().hex
  func_list = [
                partial(upload, storage.Client().bucket(bucket_name), blob_name, size), 
                *[partial(download, storage.Client().bucket(bucket_name), blob_name, size) for i in range(3)]
              ]
  return func_list


num_iters = 3
for i in range(num_iters):
  for func in generate_func_list(BUCKET_NAME, 0, 5120000):
    # generate random size and blob name
    time_test(func)