Add configurable block_size to write microbenchmarks and fix parallel test race conditions

Author: yuxin00j

gcsfs/tests/conftest.py

@@ -330,12 +330,13 @@ def _create_extended_gcsfs(gcs_factory, buckets_to_delete, populate_bucket, **kw
     extended_gcsfs = gcs_factory(**kwargs)
     # Only create/delete/populate the bucket if we are NOT using the real GCS endpoint.
     if not is_real_gcs:
-        try:
-            extended_gcsfs.rm(TEST_ZONAL_BUCKET, recursive=True)
-        except FileNotFoundError:
-            pass
-        extended_gcsfs.mkdir(TEST_ZONAL_BUCKET)
-        buckets_to_delete.add(TEST_ZONAL_BUCKET)
+        if not extended_gcsfs.exists(TEST_ZONAL_BUCKET):
+            extended_gcsfs.mkdir(TEST_ZONAL_BUCKET)
+            buckets_to_delete.add(TEST_ZONAL_BUCKET)
+
+        if not extended_gcsfs.exists(TEST_BUCKET):
+            extended_gcsfs.mkdir(TEST_BUCKET)
+            buckets_to_delete.add(TEST_BUCKET)
     try:
         if populate_bucket:
             # To avoid hitting object mutation limits, only pipe files if they

gcsfs/tests/perf/microbenchmarks/conftest.py

@@ -136,6 +136,7 @@ def gcsfs_benchmark_write(extended_gcs_factory, request):
         params,
         "benchmark-write",
         create_files=False,
+        gcs_kwargs={"block_size": params.block_size_bytes},
     )
 
 

gcsfs/tests/perf/microbenchmarks/test_configs.py

@@ -107,6 +107,7 @@ def test_write_configurator(mock_config_dependencies):
     common = {
         "bucket_types": ["regional"],
         "chunk_sizes_mb": [10],
+        "block_sizes_mb": [32],
         "rounds": 1,
         "runtime": 30,
     }
@@ -117,9 +118,13 @@ def test_write_configurator(mock_config_dependencies):
 
     assert len(cases) == 1
     case = cases[0]
-    assert case.name == "write_test_2procs_1threads_10MB_chunk_regional_30s_duration"
+    assert (
+        case.name
+        == "write_test_2procs_1threads_10MB_chunk_32MB_block_regional_30s_duration"
+    )
     assert case.file_size_bytes == 0
     assert case.chunk_size_bytes == 10 * MB
+    assert case.block_size_bytes == 32 * MB
     assert case.processes == 2
     assert case.files == 2  # threads * processes
 

gcsfs/tests/perf/microbenchmarks/write_fixed_duration/configs.py

@@ -12,22 +12,32 @@ def build_cases(self, scenario, common_config):
         threads_list = scenario.get("threads", [1])
         bucket_types = common_config.get("bucket_types", ["regional"])
         chunk_sizes_mb = common_config.get("chunk_sizes_mb", [64, 100])
+        block_sizes_mb = common_config.get("block_sizes_mb", [16])
+        scenario_block_sizes_mb = scenario.get("block_sizes_mb")
+        if scenario_block_sizes_mb:
+            block_sizes_mb = scenario_block_sizes_mb
         runtime = common_config.get("runtime", 30)
         rounds = common_config.get("rounds", 1)
 
         cases = []
         param_combinations = itertools.product(
-            procs_list, threads_list, chunk_sizes_mb, bucket_types
+            procs_list, threads_list, chunk_sizes_mb, block_sizes_mb, bucket_types
         )
 
-        for procs, threads, chunk_size_mb, bucket_type in param_combinations:
+        for (
+            procs,
+            threads,
+            chunk_size_mb,
+            block_size_mb,
+            bucket_type,
+        ) in param_combinations:
             bucket_name = self.get_bucket_name(bucket_type)
             if not bucket_name:
                 continue
 
             name = (
                 f"{scenario['name']}_{procs}procs_{threads}threads_"
-                f"{chunk_size_mb}MB_chunk_{bucket_type}_{runtime}s_duration"
+                f"{chunk_size_mb}MB_chunk_{block_size_mb}MB_block_{bucket_type}_{runtime}s_duration"
             )
 
             params = WriteFixedDurationBenchmarkParameters(
@@ -38,7 +48,8 @@ def build_cases(self, scenario, common_config):
                 processes=procs,
                 files=threads * procs,
                 rounds=rounds,
-                chunk_size_bytes=chunk_size_mb * MB,
+                chunk_size_bytes=int(chunk_size_mb * MB),
+                block_size_bytes=int(block_size_mb * MB),
                 file_size_bytes=0,
                 runtime=runtime,
             )

gcsfs/tests/perf/microbenchmarks/write_fixed_duration/parameters.py

@@ -9,5 +9,8 @@ class WriteFixedDurationBenchmarkParameters(IOBenchmarkParameters):
     Defines the parameters for a write benchmark test cases with runtime.
     """
 
+    # The block size for gcsfs file buffering.
+    block_size_bytes: int
+
     # Time in seconds the test should run.
     runtime: int

gcsfs/tests/perf/microbenchmarks/write_fixed_duration/test_write.py

@@ -127,5 +127,6 @@ def args_builder(gcs_instance, i, shared_arr):
         worker_target=_process_worker_fixed_duration,
         args_builder=args_builder,
         benchmark_group=BENCHMARK_GROUP,
+        gcs_kwargs={"block_size": params.block_size_bytes},
         request=request,
     )