hacky parametrization of relevant benchmarks

test_runner/fixtures/neon_fixtures.py

@@ -1180,7 +1180,11 @@ class NeonEnv:
             force=config.config_init_force,
         )
 
-    def start(self, timeout_in_seconds: int | None = None):
+    def start(
+        self,
+        timeout_in_seconds: int | None = None,
+        extra_ps_env_vars: dict[str, str] | None = None,
+    ):
         # Storage controller starts first, so that pageserver /re-attach calls don't
         # bounce through retries on startup
         self.storage_controller.start(timeout_in_seconds=timeout_in_seconds)
@@ -1199,7 +1203,10 @@ class NeonEnv:
             for pageserver in self.pageservers:
                 futs.append(
                     executor.submit(
-                        lambda ps=pageserver: ps.start(timeout_in_seconds=timeout_in_seconds)  # type: ignore[misc]
+                        lambda ps=pageserver: ps.start(
+                            extra_env_vars=extra_ps_env_vars or {},
+                            timeout_in_seconds=timeout_in_seconds,
+                        )
                     )
                 )
 

test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py

@@ -56,13 +56,14 @@ def test_pageserver_characterize_throughput_with_n_tenants(
 # For reference, the space usage of the snapshots:
 # sudo du -hs /instance_store/neon/test_output/shared-snapshots/*
 # 19G	/instance_store/neon/test_output/shared-snapshots/max_throughput_latest_lsn-1-136
-@pytest.mark.parametrize("duration", [20 * 60])
+@pytest.mark.parametrize("duration", [20])
 @pytest.mark.parametrize("pgbench_scale", [get_scale_for_db(2048)])
 # we use 1 client to characterize latencies, and 64 clients to characterize throughput/scalability
 # we use 64 clients because typically for a high number of connections we recommend the connection pooler
 # which by default uses 64 connections
 @pytest.mark.parametrize("n_clients", [1, 64])
 @pytest.mark.parametrize("n_tenants", [1])
+@pytest.mark.parametrize("io_concurrency", ["serial", "parallel"])
 @pytest.mark.timeout(2400)
 @skip_on_ci(
     "This test needs lot of resources and should run on dedicated HW, not in github action runners as part of CI"
@@ -75,9 +76,17 @@ def test_pageserver_characterize_latencies_with_1_client_and_throughput_with_man
     pgbench_scale: int,
     duration: int,
     n_clients: int,
+    io_concurrency: str,
 ):
     setup_and_run_pagebench_benchmark(
-        neon_env_builder, zenbenchmark, pg_bin, n_tenants, pgbench_scale, duration, n_clients
+        neon_env_builder,
+        zenbenchmark,
+        pg_bin,
+        n_tenants,
+        pgbench_scale,
+        duration,
+        n_clients,
+        io_concurrency,
     )
 
 
@@ -89,6 +98,7 @@ def setup_and_run_pagebench_benchmark(
     pgbench_scale: int,
     duration: int,
     n_clients: int,
+    io_concurrency: str = "serial",
 ):
     def record(metric, **kwargs):
         zenbenchmark.record(
@@ -104,6 +114,14 @@ def setup_and_run_pagebench_benchmark(
             "pgbench_scale": (pgbench_scale, {"unit": ""}),
             "duration": (duration, {"unit": "s"}),
             "n_clients": (n_clients, {"unit": ""}),
+            "config": (
+                0,
+                {
+                    "labels": {
+                        "io_concurrency": io_concurrency,
+                    }
+                },
+            ),
         }
     )
 
@@ -124,7 +142,14 @@ def setup_and_run_pagebench_benchmark(
     )
 
     for param, (value, kwargs) in params.items():
-        record(param, metric_value=value, report=MetricReport.TEST_PARAM, **kwargs)
+        record(
+            param,
+            metric_value=value,
+            report=MetricReport.TEST_PARAM,
+            labels=kwargs.pop("labels", None),
+            unit=kwargs.pop("unit", ""),
+            **kwargs,
+        )
 
     def setup_wrapper(env: NeonEnv):
         return setup_tenant_template(env, pg_bin, pgbench_scale)
@@ -136,6 +161,7 @@ def setup_and_run_pagebench_benchmark(
         setup_wrapper,
         # https://github.com/neondatabase/neon/issues/8070
         timeout_in_seconds=60,
+        extra_ps_env_vars={"NEON_PAGESERVER_VALUE_RECONSTRUCT_IO_CONCURRENCY": io_concurrency},
     )
 
     env.pageserver.allowed_errors.append(
@@ -145,6 +171,10 @@ def setup_and_run_pagebench_benchmark(
         r".*query handler for.*pagestream.*failed: unexpected message: CopyFail during COPY.*"
     )
 
+    import pdb
+
+    pdb.set_trace()
+
     run_pagebench_benchmark(env, pg_bin, record, duration, n_clients)
 
 

test_runner/performance/pageserver/test_page_service_batching.py

@@ -11,7 +11,7 @@ from fixtures.log_helper import log
 from fixtures.neon_fixtures import NeonEnvBuilder, PgBin, wait_for_last_flush_lsn
 from fixtures.utils import humantime_to_ms
 
-TARGET_RUNTIME = 30
+TARGET_RUNTIME = 5
 
 
 @dataclass
@@ -31,6 +31,7 @@ class PageServicePipeliningConfigPipelined(PageServicePipeliningConfig):
     mode: str = "pipelined"
 
 
+PS_IO_CONCURRENCY = ["serial", "parallel"]
 EXECUTION = ["concurrent-futures", "tasks"]
 
 NON_BATCHABLE: list[PageServicePipeliningConfig] = [PageServicePipeliningConfigSerial()]
@@ -45,7 +46,7 @@ for max_batch_size in [1, 2, 4, 8, 16, 32]:
 
 
 @pytest.mark.parametrize(
-    "tablesize_mib, pipelining_config, target_runtime, effective_io_concurrency, readhead_buffer_size, name",
+    "tablesize_mib, pipelining_config, target_runtime, ps_io_concurrency, effective_io_concurrency, readhead_buffer_size, name",
     [
         # non-batchable workloads
         # (A separate benchmark will consider latency).
@@ -54,11 +55,13 @@ for max_batch_size in [1, 2, 4, 8, 16, 32]:
                 50,
                 config,
                 TARGET_RUNTIME,
+                ps_io_concurrency,
                 1,
                 128,
                 f"not batchable {dataclasses.asdict(config)}",
             )
             for config in NON_BATCHABLE
+            for ps_io_concurrency in PS_IO_CONCURRENCY
         ],
         # batchable workloads should show throughput and CPU efficiency improvements
         *[
@@ -66,11 +69,13 @@ for max_batch_size in [1, 2, 4, 8, 16, 32]:
                 50,
                 config,
                 TARGET_RUNTIME,
+                ps_io_concurrency,
                 100,
                 128,
                 f"batchable {dataclasses.asdict(config)}",
             )
             for config in BATCHABLE
+            for ps_io_concurrency in PS_IO_CONCURRENCY
         ],
     ],
 )
@@ -80,6 +85,7 @@ def test_throughput(
     tablesize_mib: int,
     pipelining_config: PageServicePipeliningConfig,
     target_runtime: int,
+    ps_io_concurrency: str,
     effective_io_concurrency: int,
     readhead_buffer_size: int,
     name: str,
@@ -117,7 +123,19 @@ def test_throughput(
         }
     )
     # For storing configuration as a metric, insert a fake 0 with labels with actual data
-    params.update({"pipelining_config": (0, {"labels": dataclasses.asdict(pipelining_config)})})
+    params.update(
+        {
+            "config": (
+                0,
+                {
+                    "labels": {
+                        "pipelining_config": dataclasses.asdict(pipelining_config),
+                        "ps_io_concurrency": ps_io_concurrency,
+                    }
+                },
+            )
+        }
+    )
 
     log.info("params: %s", params)
 
@@ -231,7 +249,10 @@ def test_throughput(
     env.pageserver.patch_config_toml_nonrecursive(
         {"page_service_pipelining": dataclasses.asdict(pipelining_config)}
     )
-    env.pageserver.restart()
+    env.pageserver.stop()
+    env.pageserver.start(
+        extra_env_vars={"NEON_PAGESERVER_VALUE_RECONSTRUCT_IO_CONCURRENCY": ps_io_concurrency}
+    )
     metrics = workload()
 
     log.info("Results: %s", metrics)
@@ -266,14 +287,19 @@ for max_batch_size in [1, 32]:
 
 
 @pytest.mark.parametrize(
-    "pipelining_config,name",
-    [(config, f"{dataclasses.asdict(config)}") for config in PRECISION_CONFIGS],
+    "pipelining_config,ps_io_concurrency,name",
+    [
+        (config, ps_io_concurrency, f"{dataclasses.asdict(config)}")
+        for config in PRECISION_CONFIGS
+        for ps_io_concurrency in PS_IO_CONCURRENCY
+    ],
 )
 def test_latency(
     neon_env_builder: NeonEnvBuilder,
     zenbenchmark: NeonBenchmarker,
     pg_bin: PgBin,
     pipelining_config: PageServicePipeliningConfig,
+    ps_io_concurrency: str,
     name: str,
 ):
     """
@@ -323,6 +349,11 @@ def test_latency(
     for sk in env.safekeepers:
         sk.stop()
 
+    env.pageserver.stop()
+    env.pageserver.start(
+        extra_env_vars={"NEON_PAGESERVER_VALUE_RECONSTRUCT_IO_CONCURRENCY": ps_io_concurrency}
+    )
+
     #
     # Run single-threaded pagebench (TODO: dedup with other benchmark code)
     #

test_runner/performance/pageserver/util.py

@@ -48,6 +48,7 @@ def setup_pageserver_with_tenants(
     n_tenants: int,
     setup: Callable[[NeonEnv], tuple[TenantId, TimelineId, dict[str, Any]]],
     timeout_in_seconds: int | None = None,
+    extra_ps_env_vars: dict[str, str] | None = None,
 ) -> NeonEnv:
     """
     Utility function to set up a pageserver with a given number of identical tenants.
@@ -57,6 +58,6 @@ def setup_pageserver_with_tenants(
         return many_tenants.single_timeline(neon_env_builder, setup, n_tenants)
 
     env = neon_env_builder.build_and_use_snapshot(name, doit)
-    env.start(timeout_in_seconds=timeout_in_seconds)
+    env.start(timeout_in_seconds=timeout_in_seconds, extra_ps_env_vars=extra_ps_env_vars)
     ensure_pageserver_ready_for_benchmarking(env, n_tenants)
     return env