Add fs-api layer benchmark

Signed-off-by: Christian Hagemeier <[email protected]>

Author: c-hagem

benchmark/benchmark.py

@@ -18,6 +18,7 @@
 from benchmarks.client_benchmark import ClientBenchmark
 from benchmarks.crt_benchmark import CrtBenchmark
 from benchmarks.fio_benchmark import FioBenchmark
+from benchmarks.filesystem_api_benchmark import FilesystemApiBenchmark
 from benchmarks.prefetch_benchmark import PrefetchBenchmark
 
 logging.basicConfig(level=os.environ.get('LOGLEVEL', 'INFO').upper())
@@ -225,6 +226,8 @@ def run_experiment(cfg: DictConfig) -> None:
         benchmark = ClientBenchmark(cfg, metadata)
     elif benchmark_type == "client-bp":
         benchmark = ClientBenchmark(cfg, metadata, backpressure=True)
+    elif benchmark_type == "filesystem_api":
+        benchmark = FilesystemApiBenchmark(cfg, metadata)
     else:
         raise ValueError(f"Unsupported benchmark type: {benchmark_type}")
 

benchmark/benchmarks/benchmark_config_parser.py

@@ -84,3 +84,13 @@ def get_client_config(self) -> Dict[str, Any]:
         return {
             'read_window_size': getattr(client_cfg, 'read_window_size', 2147483648),  # Reaslitic default value 8M/2G?
         }
+
+    def get_filesystem_api_config(self) -> Dict[str, Any]:
+        filesystem_api_cfg = self.cfg.benchmarks.filesystem_api
+        return {
+            'max_memory_target': getattr(filesystem_api_cfg, 'max_memory_target', None),
+            'crt_memory_limit_gib': getattr(filesystem_api_cfg, 'crt_memory_limit_gib', None),
+            'threads_per_file': getattr(filesystem_api_cfg, 'threads_per_file', 1),
+            'iterations': getattr(filesystem_api_cfg, 'iterations', None),
+            'mock_client': getattr(filesystem_api_cfg, 'mock_client', False),
+        }

benchmark/benchmarks/filesystem_api_benchmark.py

@@ -0,0 +1,95 @@
+import logging
+import os
+import subprocess
+from typing import Dict, Any
+
+from benchmarks.base_benchmark import BaseBenchmark
+from omegaconf import DictConfig
+
+from benchmarks.benchmark_config_parser import BenchmarkConfigParser
+
+log = logging.getLogger(__name__)
+
+
+class FilesystemApiBenchmark(BaseBenchmark):
+    def __init__(self, cfg: DictConfig, metadata: Dict[str, Any]):
+        self.config_parser = BenchmarkConfigParser(cfg)
+        self.common_config = self.config_parser.get_common_config()
+        self.filesystem_api_config = self.config_parser.get_filesystem_api_config()
+        self.mountpoint_config = self.config_parser.get_mountpoint_config()
+
+    def setup(self) -> None:
+        pass
+
+    def run_benchmark(self) -> None:
+        subprocess_args = [
+            "cargo",
+            "run",
+            "--example",
+            "filesystem_api_benchmark",
+            self.common_config['s3_bucket'],
+        ]
+
+        object_size = self.common_config['object_size_in_gib']
+        size_gib = str(object_size)
+        app_workers = self.common_config['application_workers']
+
+        # Check if objects are specified or if we have to fall back to objects
+        # generated by fio.
+        objects = self.common_config['s3_keys']
+        if not objects:
+            objects = self.config_parser.default_object_keys(app_workers, size_gib)
+
+        if len(objects) >= app_workers:
+            for i in range(app_workers):
+                subprocess_args.append(objects[i])
+        else:
+            raise ValueError("Seeing fewer objects than app workers. So cannot proceed with the run.")
+
+        region = self.common_config["region"]
+        subprocess_args.extend(["--region", region])
+
+        if (max_throughput := self.common_config['max_throughput_gbps']) is not None:
+            subprocess_args.extend(["--maximum-throughput-gbps", str(max_throughput)])
+
+        if (max_memory_target := self.filesystem_api_config['max_memory_target']) is not None:
+            subprocess_args.extend(["--max-memory-target", str(max_memory_target)])
+
+        if (crt_memory_limit_gib := self.filesystem_api_config['crt_memory_limit_gib']) is not None:
+            subprocess_args.extend(["--crt-memory-limit-gib", str(crt_memory_limit_gib)])
+
+        if (read_part_size := self.common_config['read_part_size']) is not None:
+            subprocess_args.extend(["--part-size", str(read_part_size)])
+
+        read_size = self.common_config['read_size']
+        subprocess_args.extend(["--read-size", str(read_size)])
+
+        threads_per_file = self.filesystem_api_config['threads_per_file']
+        subprocess_args.extend(["--threads-per-file", str(threads_per_file)])
+
+        if (iterations := self.filesystem_api_config['iterations']) is not None:
+            subprocess_args.extend(["--iterations", str(iterations)])
+
+        if self.filesystem_api_config.get('mock_client', False) or self.mountpoint_config.get('stub_mode') == 's3_client':
+            subprocess_args.append("--mock-client")
+
+        for interface in self.common_config['network_interfaces']:
+            subprocess_args.extend(["--bind", interface])
+
+        if (run_time := self.common_config['run_time']) is not None:
+            subprocess_args.extend(["--max-duration", str(run_time)])
+
+        subprocess_args.extend(["--output-file", "filesystem-api-output.json"])
+
+        log.info("Running filesystem API benchmark with args: %s", subprocess_args)
+
+        # Set up environment variables
+        env = os.environ.copy()
+        if self.mountpoint_config.get('use_download_checksums', True) is False:
+            env["MOUNTPOINT_EXPERIMENTAL_DISABLE_CHECKSUMS"] = "1"
+
+        subprocess.run(subprocess_args, check=True, capture_output=True, text=True, env=env)
+        log.info("Filesystem API benchmark completed successfully.")
+
+    def post_process(self) -> None:
+        pass

benchmark/conf/config.yaml

@@ -2,80 +2,86 @@
 # as well as some which will be 'swept' over for experiments.
 
 defaults:
-  - _self_
+    - _self_
 
 # ===== Common parameters for all benchmarks =====
 s3_bucket: ???
 s3_result_bucket: !!null
 application_workers: 1
 iteration: 0
 iterations: 1
-run_time: 30  # Default run time in seconds
+run_time: 30 # Default run time in seconds
 read_size: 262144 # Defaults to 256KiB, can go up to 1MiB.
 read_part_size: !!null
 region: "us-east-1"
-write_part_size: 16777216  # 16 MiB, to allow for uploads of large files
-object_size_in_gib: 100  # Size of the object to benchmark
+write_part_size: 16777216 # 16 MiB, to allow for uploads of large files
+object_size_in_gib: 100 # Size of the object to benchmark
 benchmark_type: "fio" # fio, prefetch, client, client-bp, crt
-s3_keys: !!null
+s3_keys: !!null # Network configuration
 
-# Network configuration
 network:
-  interface_names: []
-  maximum_throughput_gbps: !!null
+    interface_names: []
+    maximum_throughput_gbps:
+        !!null # Monitoring options (common to all benchmarks)
 
-# Monitoring options (common to all benchmarks)
 monitoring:
-  with_bwm: false
-  with_perf_stat: false
+    with_bwm: false
+    with_perf_stat: false
 
 # ===== Mountpoint configuration =====
 mountpoint:
-  fuse_threads: !!null
-  prefix: !!null
-  metadata_ttl: "indefinite"
-  mountpoint_max_background: !!null
-  mountpoint_congestion_threshold: !!null
-  mountpoint_binary: !!null
-  upload_checksums: !!null
-  stub_mode: "off"  # Options: "off", "fs_handler", "s3_client"
-  mountpoint_debug: false
-  mountpoint_debug_crt: false
+    fuse_threads: !!null
+    prefix: !!null
+    metadata_ttl: "indefinite"
+    mountpoint_max_background: !!null
+    mountpoint_congestion_threshold: !!null
+    mountpoint_binary: !!null
+    upload_checksums: !!null
+    stub_mode: "off" # Options: "off", "fs_handler", "s3_client"
+    mountpoint_debug: false
+    mountpoint_debug_crt: false
 
 # ===== Benchmark-specific configurations =====
 benchmarks:
-  fio:
-    direct_io: false
-    fio_benchmarks:
-      - sequential_read
-    fio_benchmark: "${benchmarks.fio.fio_benchmarks[0]}"
-    fio_io_engine: "psync"
-  
-  prefetch:
-    max_memory_target: !!null
-  
-  crt:
-    crt_benchmarks_path: !!null
+    fio:
+        direct_io: false
+        fio_benchmarks:
+            - sequential_read
+        fio_benchmark: "${benchmarks.fio.fio_benchmarks[0]}"
+        fio_io_engine: "psync"
 
-  client:  
-    # None 
+    prefetch:
+        max_memory_target: !!null
+
+    crt:
+        crt_benchmarks_path: !!null
+
+    client:
+        # None
+    filesystem_api:
+        max_memory_target: !!null
+        crt_memory_limit_gib: !!null
+        threads_per_file: 1
+        iterations: !!null
+        mock_client: false
+
+    client_backpressure:
+        read_window_size: !!null #2147483648
 
-  client_backpressure:
-    read_window_size: !!null #2147483648 
 
 hydra:
-  help:
-    app_name: "Mountpoint benchmark runner"
-  mode: MULTIRUN
-  job:
-    chdir: true
-  sweeper:
-    # Global sweeper params - use this for common parameters across all benchmarks
-    # For specific parameter use sweep params under the specific benchmark type config
-    params:
-      'application_workers': 1, 4, 16, 64, 128
-      'iteration': "range(${iterations})"
-      'mountpoint.fuse_threads': 1, 16, 64
-      'benchmarks.fio.direct_io': false, true
-      # 'benchmarks.prefetch.max_memory_target': !!null, 1073741824, 2147483648  # null, 1GB, 2GB
-      #'benchmarks.client_backpressure.read_window_size': 8388608, 2147483648
+    help:
+        app_name: "Mountpoint benchmark runner"
+    mode: MULTIRUN
+    job:
+        chdir: true
+    sweeper:
+        # Global sweeper params - use this for common parameters across all benchmarks
+        # For specific parameter use sweep params under the specific benchmark type config
+        params:
+            "application_workers": 1, 4, 16, 64, 128
+            "iteration": "range(${iterations})"
+            "mountpoint.fuse_threads": 1, 16, 64
+            "benchmarks.fio.direct_io": false, true
+            # 'benchmarks.prefetch.max_memory_target': !!null, 1073741824, 2147483648  # null, 1GB, 2GB
+            #'benchmarks.client_backpressure.read_window_size': 8388608, 2147483648