From 0a37164c29831207de55a97d21297600dd26c8c3 Mon Sep 17 00:00:00 2001 From: Christian Schwarz Date: Mon, 20 Jan 2025 23:35:47 +0100 Subject: [PATCH] replace env var with config variable; add test suite fixture env var to override default --- .github/workflows/_build-and-test-locally.yml | 3 +- libs/pageserver_api/src/config.rs | 25 +++++++++ pageserver/src/basebackup.rs | 3 +- pageserver/src/bin/pageserver.rs | 1 + pageserver/src/config.rs | 4 ++ pageserver/src/page_service.rs | 26 ++++++--- pageserver/src/tenant.rs | 4 +- pageserver/src/tenant/storage_layer.rs | 53 +++++++++---------- test_runner/deep_layers_with_delta.py | 13 +++-- test_runner/fixtures/neon_fixtures.py | 18 +++++++ test_runner/fixtures/parametrize.py | 5 ++ ...er_max_throughput_getpage_at_latest_lsn.py | 4 +- .../pageserver/test_page_service_batching.py | 15 ++---- 13 files changed, 118 insertions(+), 56 deletions(-) diff --git a/.github/workflows/_build-and-test-locally.yml b/.github/workflows/_build-and-test-locally.yml index a1d661a59b..61223ee77a 100644 --- a/.github/workflows/_build-and-test-locally.yml +++ b/.github/workflows/_build-and-test-locally.yml @@ -230,7 +230,7 @@ jobs: # run pageserver tests with different settings for io_engine in std-fs tokio-epoll-uring ; do - NEON_PAGESERVER_VALUE_RECONSTRUCT_IO_CONCURRENCY=futures-unordered NEON_PAGESERVER_UNIT_TEST_VIRTUAL_FILE_IOENGINE=$io_engine ${cov_prefix} cargo nextest run $CARGO_FLAGS $CARGO_FEATURES -E 'package(pageserver)' + NEON_PAGESERVER_UNIT_TEST_VIRTUAL_FILE_IOENGINE=$io_engine ${cov_prefix} cargo nextest run $CARGO_FLAGS $CARGO_FEATURES -E 'package(pageserver)' done # Run separate tests for real S3 @@ -314,6 +314,7 @@ jobs: CHECK_ONDISK_DATA_COMPATIBILITY: nonempty BUILD_TAG: ${{ inputs.build-tag }} PAGESERVER_VIRTUAL_FILE_IO_ENGINE: tokio-epoll-uring + PAGESERVER_GET_VECTORED_CONCURRENT_IO: sidecar-task USE_LFC: ${{ matrix.lfc_state == 'with-lfc' && 'true' || 'false' }} # Temporary disable this step until we figure out why it's so flaky diff --git a/libs/pageserver_api/src/config.rs b/libs/pageserver_api/src/config.rs index 09cfbc55fd..a0f3c00c3c 100644 --- a/libs/pageserver_api/src/config.rs +++ b/libs/pageserver_api/src/config.rs @@ -120,6 +120,7 @@ pub struct ConfigToml { pub no_sync: Option, pub wal_receiver_protocol: PostgresClientProtocol, pub page_service_pipelining: PageServicePipeliningConfig, + pub get_vectored_concurrent_io: GetVectoredConcurrentIo, } #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] @@ -158,6 +159,25 @@ pub enum PageServiceProtocolPipelinedExecutionStrategy { Tasks, } +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(tag = "mode", rename_all = "kebab-case")] +#[serde(deny_unknown_fields)] +pub enum GetVectoredConcurrentIo { + /// The read path is fully sequential: layers are visited + /// one after the other and IOs are issued and waited upon + /// from the same task that traverses the layers. + Sequential, + /// The read path still traverses layers sequentially, and + /// index blocks will be read into the PS PageCache from + /// that task, with waiting. + /// But data IOs are dispatched and waited upon from a sidecar + /// task so that the traversing task can continue to traverse + /// layers while the IOs are in flight. + /// If the PS PageCache miss rate is low, this improves + /// throughput dramatically. + SidecarTask, +} + pub mod statvfs { pub mod mock { #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] @@ -450,6 +470,11 @@ impl Default for ConfigToml { execution: PageServiceProtocolPipelinedExecutionStrategy::ConcurrentFutures, }) }, + get_vectored_concurrent_io: if !cfg!(test) { + GetVectoredConcurrentIo::Sequential + } else { + GetVectoredConcurrentIo::SidecarTask + }, } } } diff --git a/pageserver/src/basebackup.rs b/pageserver/src/basebackup.rs index f0ee20b639..c18c78c3b3 100644 --- a/pageserver/src/basebackup.rs +++ b/pageserver/src/basebackup.rs @@ -301,7 +301,8 @@ where let mut slru_builder = SlruSegmentsBuilder::new(&mut self.ar); - let io_concurrency = IoConcurrency::spawn_from_env( + let io_concurrency = IoConcurrency::spawn_from_conf( + self.timeline.conf, self.timeline .gate .enter() diff --git a/pageserver/src/bin/pageserver.rs b/pageserver/src/bin/pageserver.rs index 921c6a5092..5764728505 100644 --- a/pageserver/src/bin/pageserver.rs +++ b/pageserver/src/bin/pageserver.rs @@ -135,6 +135,7 @@ fn main() -> anyhow::Result<()> { info!(?conf.virtual_file_io_mode, "starting with virtual_file IO mode"); info!(?conf.wal_receiver_protocol, "starting with WAL receiver protocol"); info!(?conf.page_service_pipelining, "starting with page service pipelining config"); + info!(?conf.get_vectored_concurrent_io, "starting with get_vectored IO concurrency config"); // The tenants directory contains all the pageserver local disk state. // Create if not exists and make sure all the contents are durable before proceeding. diff --git a/pageserver/src/config.rs b/pageserver/src/config.rs index 1651db8500..ce480c70a0 100644 --- a/pageserver/src/config.rs +++ b/pageserver/src/config.rs @@ -191,6 +191,8 @@ pub struct PageServerConf { pub wal_receiver_protocol: PostgresClientProtocol, pub page_service_pipelining: pageserver_api::config::PageServicePipeliningConfig, + + pub get_vectored_concurrent_io: pageserver_api::config::GetVectoredConcurrentIo, } /// Token for authentication to safekeepers @@ -352,6 +354,7 @@ impl PageServerConf { no_sync, wal_receiver_protocol, page_service_pipelining, + get_vectored_concurrent_io, } = config_toml; let mut conf = PageServerConf { @@ -396,6 +399,7 @@ impl PageServerConf { import_pgdata_aws_endpoint_url, wal_receiver_protocol, page_service_pipelining, + get_vectored_concurrent_io, // ------------------------------------------------------------ // fields that require additional validation or custom handling diff --git a/pageserver/src/page_service.rs b/pageserver/src/page_service.rs index a7af386520..e5063b7fc2 100644 --- a/pageserver/src/page_service.rs +++ b/pageserver/src/page_service.rs @@ -113,6 +113,7 @@ pub fn spawn( let task = COMPUTE_REQUEST_RUNTIME.spawn(task_mgr::exit_on_panic_or_error( "libpq listener", libpq_listener_main( + conf, tenant_manager, pg_auth, tcp_listener, @@ -166,7 +167,9 @@ impl Connections { /// Returns Ok(()) upon cancellation via `cancel`, returning the set of /// open connections. /// +#[allow(clippy::too_many_arguments)] pub async fn libpq_listener_main( + conf: &'static PageServerConf, tenant_manager: Arc, auth: Option>, listener: tokio::net::TcpListener, @@ -204,6 +207,7 @@ pub async fn libpq_listener_main( let connection_ctx = listener_ctx .detached_child(TaskKind::PageRequestHandler, DownloadBehavior::Download); connection_handler_tasks.spawn(page_service_conn_main( + conf, tenant_manager.clone(), local_auth, socket, @@ -235,6 +239,7 @@ type ConnectionHandlerResult = anyhow::Result<()>; #[instrument(skip_all, fields(peer_addr))] #[allow(clippy::too_many_arguments)] async fn page_service_conn_main( + conf: &'static PageServerConf, tenant_manager: Arc, auth: Option>, socket: tokio::net::TcpStream, @@ -292,6 +297,7 @@ async fn page_service_conn_main( // But it's in a shared crate, so, we store connection_ctx inside PageServerHandler // and create the per-query context in process_query ourselves. let mut conn_handler = PageServerHandler::new( + conf, tenant_manager, auth, pipelining_config, @@ -329,6 +335,7 @@ async fn page_service_conn_main( } struct PageServerHandler { + conf: &'static PageServerConf, auth: Option>, claims: Option, @@ -655,6 +662,7 @@ impl BatchedFeMessage { impl PageServerHandler { pub fn new( + conf: &'static PageServerConf, tenant_manager: Arc, auth: Option>, pipelining_config: PageServicePipeliningConfig, @@ -663,6 +671,7 @@ impl PageServerHandler { gate_guard: GateGuard, ) -> Self { PageServerHandler { + conf, auth, claims: None, connection_ctx, @@ -1313,13 +1322,16 @@ impl PageServerHandler { } } - let io_concurrency = IoConcurrency::spawn_from_env(match self.gate_guard.try_clone() { - Ok(guard) => guard, - Err(_) => { - info!("shutdown request received in page handler"); - return Err(QueryError::Shutdown); - } - }); + let io_concurrency = IoConcurrency::spawn_from_conf( + self.conf, + match self.gate_guard.try_clone() { + Ok(guard) => guard, + Err(_) => { + info!("shutdown request received in page handler"); + return Err(QueryError::Shutdown); + } + }, + ); let pgb_reader = pgb .split() diff --git a/pageserver/src/tenant.rs b/pageserver/src/tenant.rs index 444a439dff..b494af59ab 100644 --- a/pageserver/src/tenant.rs +++ b/pageserver/src/tenant.rs @@ -6548,8 +6548,8 @@ mod tests { let gate = Gate::default(); let io_concurrency_levels: Vec SelectedIoConcurrency>> = vec![ - Box::new(|| SelectedIoConcurrency::Serial), - Box::new(|| SelectedIoConcurrency::FuturesUnordered(gate.enter().unwrap())), + Box::new(|| SelectedIoConcurrency::Sequential), + Box::new(|| SelectedIoConcurrency::SidecarTask(gate.enter().unwrap())), ]; for (io_concurrency_level_idx, io_concurrency_level) in diff --git a/pageserver/src/tenant/storage_layer.rs b/pageserver/src/tenant/storage_layer.rs index 0c9ce3d2cf..2369213ddf 100644 --- a/pageserver/src/tenant/storage_layer.rs +++ b/pageserver/src/tenant/storage_layer.rs @@ -10,6 +10,7 @@ mod layer_desc; mod layer_name; pub mod merge_iterator; +use crate::config::PageServerConf; use crate::context::{AccessStatsBehavior, RequestContext}; use bytes::Bytes; use futures::stream::FuturesUnordered; @@ -204,8 +205,8 @@ pub(crate) struct ValuesReconstructState { /// This struct and the dispatching in the impl will be removed once /// we've built enough confidence. pub(crate) enum IoConcurrency { - Serial, - FuturesUnordered { + Sequential, + SidecarTask { ios_tx: tokio::sync::mpsc::UnboundedSender, }, } @@ -213,15 +214,15 @@ pub(crate) enum IoConcurrency { type IoFuture = Pin>>; pub(crate) enum SelectedIoConcurrency { - Serial, - FuturesUnordered(GateGuard), + Sequential, + SidecarTask(GateGuard), } impl std::fmt::Debug for IoConcurrency { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { - IoConcurrency::Serial => write!(f, "Serial"), - IoConcurrency::FuturesUnordered { .. } => write!(f, "FuturesUnordered"), + IoConcurrency::Sequential => write!(f, "Sequential"), + IoConcurrency::SidecarTask { .. } => write!(f, "SidecarTask"), } } } @@ -229,8 +230,8 @@ impl std::fmt::Debug for IoConcurrency { impl std::fmt::Debug for SelectedIoConcurrency { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { - SelectedIoConcurrency::Serial => write!(f, "Serial"), - SelectedIoConcurrency::FuturesUnordered(_) => write!(f, "FuturesUnordered"), + SelectedIoConcurrency::Sequential => write!(f, "Sequential"), + SelectedIoConcurrency::SidecarTask(_) => write!(f, "SidecarTask"), } } } @@ -240,28 +241,24 @@ impl IoConcurrency { /// Requires finding a long-lived root for the IoConcurrency and funneling it through /// to the place that does the get_values_reconstruct_data call. pub(crate) fn serial() -> Self { - Self::spawn(SelectedIoConcurrency::Serial) + Self::spawn(SelectedIoConcurrency::Sequential) } - pub(crate) fn spawn_from_env(gate_guard: GateGuard) -> IoConcurrency { - static IO_CONCURRENCY: once_cell::sync::Lazy = once_cell::sync::Lazy::new(|| { - std::env::var("NEON_PAGESERVER_VALUE_RECONSTRUCT_IO_CONCURRENCY") - .unwrap_or_else(|_| "serial".to_string()) - }); - let selected = match IO_CONCURRENCY.as_str() { - "serial" => SelectedIoConcurrency::Serial, - "futures-unordered" => SelectedIoConcurrency::FuturesUnordered(gate_guard), - x => panic!( - "Invalid value for NEON_PAGESERVER_VALUE_RECONSTRUCT_IO_CONCURRENCY: {}", - x - ), + pub(crate) fn spawn_from_conf( + conf: &'static PageServerConf, + gate_guard: GateGuard, + ) -> IoConcurrency { + use pageserver_api::config::GetVectoredConcurrentIo; + let selected = match conf.get_vectored_concurrent_io { + GetVectoredConcurrentIo::Sequential => SelectedIoConcurrency::Sequential, + GetVectoredConcurrentIo::SidecarTask => SelectedIoConcurrency::SidecarTask(gate_guard), }; Self::spawn(selected) } pub(crate) fn spawn(io_concurrency: SelectedIoConcurrency) -> Self { match io_concurrency { - SelectedIoConcurrency::Serial => IoConcurrency::Serial, - SelectedIoConcurrency::FuturesUnordered(gate_guard) => { + SelectedIoConcurrency::Sequential => IoConcurrency::Sequential, + SelectedIoConcurrency::SidecarTask(gate_guard) => { let (ios_tx, ios_rx) = tokio::sync::mpsc::unbounded_channel(); static TASK_ID: AtomicUsize = AtomicUsize::new(0); let task_id = TASK_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed); @@ -362,15 +359,15 @@ impl IoConcurrency { } drop(gate_guard); // drop it right before we exitlast }.instrument(span)); - IoConcurrency::FuturesUnordered { ios_tx } + IoConcurrency::SidecarTask { ios_tx } } } } pub(crate) fn clone(&self) -> Self { match self { - IoConcurrency::Serial => IoConcurrency::Serial, - IoConcurrency::FuturesUnordered { ios_tx } => IoConcurrency::FuturesUnordered { + IoConcurrency::Sequential => IoConcurrency::Sequential, + IoConcurrency::SidecarTask { ios_tx } => IoConcurrency::SidecarTask { ios_tx: ios_tx.clone(), }, } @@ -430,8 +427,8 @@ impl IoConcurrency { F: std::future::Future + Send + 'static, { match self { - IoConcurrency::Serial => fut.await, - IoConcurrency::FuturesUnordered { ios_tx, .. } => { + IoConcurrency::Sequential => fut.await, + IoConcurrency::SidecarTask { ios_tx, .. } => { let fut = Box::pin(fut); // NB: experiments showed that doing an opportunistic poll of `fut` here was bad for throughput // while insignificant for latency. diff --git a/test_runner/deep_layers_with_delta.py b/test_runner/deep_layers_with_delta.py index 2ae9f84fcc..8550a74c63 100644 --- a/test_runner/deep_layers_with_delta.py +++ b/test_runner/deep_layers_with_delta.py @@ -17,20 +17,23 @@ Some commands from shell history used in combination with this script: ``` export NEON_REPO_DIR=$PWD/.neon - ./target/release/neon_local stop + export NEON_BIN_DIR=$PWD/target/release + $NEON_BIN_DIR/neon_local stop rm -rf $NEON_REPO_DIR - ./target/release/neon_local init + $NEON_BIN_DIR/neon_local init cat >> $NEON_REPO_DIR/pageserver_1/pageserver.toml <<"EOF" # customizations virtual_file_io_mode = "direct" page_service_pipelining={mode="pipelined", max_batch_size=32, execution="concurrent-futures"} + get_vectored_concurrent_io={mode="sidecar-task"} EOF - ./target/release/neon_local start + $NEON_BIN_DIR/neon_local start psql 'postgresql://localhost:1235/storage_controller' -c 'DELETE FROM tenant_shards' - NEON_PAGESERVER_VALUE_RECONSTRUCT_IO_CONCURRENCY=futures-unordered ./target/release/neon_local pageserver restart + sed 's/.*get_vectored_concurrent_io.*/get_vectored_concurrent_io={mode="sidecar-task"}/' -i $NEON_REPO_DIR/pageserver_1/pageserver.toml + $NEON_BIN_DIR/neon_local pageserver restart sleep 2 - ./target/release/neon_local tenant create --set-default + $NEON_BIN_DIR/neon_local tenant create --set-default ./target/debug/neon_local endpoint stop foo rm -rf $NEON_REPO_DIR/endpoints/foo ./target/debug/neon_local endpoint create foo diff --git a/test_runner/fixtures/neon_fixtures.py b/test_runner/fixtures/neon_fixtures.py index cc0dc71c76..f30a5d5078 100644 --- a/test_runner/fixtures/neon_fixtures.py +++ b/test_runner/fixtures/neon_fixtures.py @@ -391,6 +391,7 @@ class NeonEnvBuilder: storage_controller_port_override: int | None = None, pageserver_virtual_file_io_mode: str | None = None, pageserver_wal_receiver_protocol: PageserverWalReceiverProtocol | None = None, + pageserver_get_vectored_concurrent_io: str | None = None, ): self.repo_dir = repo_dir self.rust_log_override = rust_log_override @@ -430,6 +431,9 @@ class NeonEnvBuilder: self.storage_controller_config: dict[Any, Any] | None = None self.pageserver_virtual_file_io_engine: str | None = pageserver_virtual_file_io_engine + self.pageserver_get_vectored_concurrent_io: str | None = ( + pageserver_get_vectored_concurrent_io + ) self.pageserver_default_tenant_config_compaction_algorithm: dict[str, Any] | None = ( pageserver_default_tenant_config_compaction_algorithm @@ -1066,6 +1070,7 @@ class NeonEnv: self.pageserver_virtual_file_io_engine = config.pageserver_virtual_file_io_engine self.pageserver_virtual_file_io_mode = config.pageserver_virtual_file_io_mode self.pageserver_wal_receiver_protocol = config.pageserver_wal_receiver_protocol + self.pageserver_get_vectored_concurrent_io = config.pageserver_get_vectored_concurrent_io # Create the neon_local's `NeonLocalInitConf` cfg: dict[str, Any] = { @@ -1127,6 +1132,15 @@ class NeonEnv: "max_batch_size": 32, } + # Concurrent IO (https://github.com/neondatabase/neon/issues/9378): + # enable concurrent IO by default in tests and benchmarks. + # Compat tests are exempt because old versions fail to parse the new config. + if not config.compatibility_neon_binpath: + if self.pageserver_get_vectored_concurrent_io is not None: + ps_cfg["get_vectored_concurrent_io"] = { + "mode": self.pageserver_get_vectored_concurrent_io, + } + if self.pageserver_virtual_file_io_engine is not None: ps_cfg["virtual_file_io_engine"] = self.pageserver_virtual_file_io_engine if config.pageserver_default_tenant_config_compaction_algorithm is not None: @@ -1468,6 +1482,7 @@ def neon_simple_env( pageserver_virtual_file_io_engine: str, pageserver_default_tenant_config_compaction_algorithm: dict[str, Any] | None, pageserver_virtual_file_io_mode: str | None, + pageserver_get_vectored_concurrent_io: str | None, ) -> Iterator[NeonEnv]: """ Simple Neon environment, with 1 safekeeper and 1 pageserver. No authentication, no fsync. @@ -1500,6 +1515,7 @@ def neon_simple_env( pageserver_virtual_file_io_engine=pageserver_virtual_file_io_engine, pageserver_default_tenant_config_compaction_algorithm=pageserver_default_tenant_config_compaction_algorithm, pageserver_virtual_file_io_mode=pageserver_virtual_file_io_mode, + pageserver_get_vectored_concurrent_io=pageserver_get_vectored_concurrent_io, combination=combination, ) as builder: env = builder.init_start() @@ -1526,6 +1542,7 @@ def neon_env_builder( pageserver_default_tenant_config_compaction_algorithm: dict[str, Any] | None, record_property: Callable[[str, object], None], pageserver_virtual_file_io_mode: str | None, + pageserver_get_vectored_concurrent_io: str | None, ) -> Iterator[NeonEnvBuilder]: """ Fixture to create a Neon environment for test. @@ -1568,6 +1585,7 @@ def neon_env_builder( test_overlay_dir=test_overlay_dir, pageserver_default_tenant_config_compaction_algorithm=pageserver_default_tenant_config_compaction_algorithm, pageserver_virtual_file_io_mode=pageserver_virtual_file_io_mode, + pageserver_get_vectored_concurrent_io=pageserver_get_vectored_concurrent_io, ) as builder: yield builder # Propogate `preserve_database_files` to make it possible to use in other fixtures, diff --git a/test_runner/fixtures/parametrize.py b/test_runner/fixtures/parametrize.py index f57c0f801f..3404c16f55 100644 --- a/test_runner/fixtures/parametrize.py +++ b/test_runner/fixtures/parametrize.py @@ -44,6 +44,11 @@ def pageserver_virtual_file_io_mode() -> str | None: return os.getenv("PAGESERVER_VIRTUAL_FILE_IO_MODE") +@pytest.fixture(scope="function", autouse=True) +def pageserver_get_vectored_concurrent_io() -> str | None: + return os.getenv("PAGESERVER_GET_VECTORED_CONCURRENT_IO") + + def get_pageserver_default_tenant_config_compaction_algorithm() -> dict[str, Any] | None: toml_table = os.getenv("PAGESERVER_DEFAULT_TENANT_CONFIG_COMPACTION_ALGORITHM") if toml_table is None: diff --git a/test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py b/test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py index 1e1734fd82..52e0df93ec 100644 --- a/test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py +++ b/test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py @@ -105,7 +105,7 @@ def setup_and_run_pagebench_benchmark( pgbench_scale: int, duration: int, n_clients: int, - io_concurrency: str = "serial", + io_concurrency: str = "sequential", ps_direct_io_mode: str = "buffered", ): def record(metric, **kwargs): @@ -141,6 +141,7 @@ def setup_and_run_pagebench_benchmark( f"page_cache_size={page_cache_size}; max_file_descriptors={max_file_descriptors}" ) neon_env_builder.pageserver_virtual_file_io_mode = ps_direct_io_mode + neon_env_builder.pageserver_get_vectored_concurrent_io = io_concurrency params.update( { "pageserver_config_override.page_cache_size": ( @@ -171,7 +172,6 @@ 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( diff --git a/test_runner/performance/pageserver/test_page_service_batching.py b/test_runner/performance/pageserver/test_page_service_batching.py index 49ee939790..7624bc4d83 100644 --- a/test_runner/performance/pageserver/test_page_service_batching.py +++ b/test_runner/performance/pageserver/test_page_service_batching.py @@ -32,7 +32,7 @@ class PageServicePipeliningConfigPipelined(PageServicePipeliningConfig): PS_DIRECT_IO = ["direct"] -PS_IO_CONCURRENCY = ["serial", "futures-unordered"] +PS_IO_CONCURRENCY = ["sequential", "sidecar-task"] EXECUTION = ["concurrent-futures"] NON_BATCHABLE: list[PageServicePipeliningConfig] = [PageServicePipeliningConfigSerial()] @@ -256,12 +256,10 @@ def test_throughput( { "page_service_pipelining": dataclasses.asdict(pipelining_config), "virtual_file_io_mode": ps_direct_io_mode, + "get_vectored_concurrent_io": {"mode": ps_io_concurrency}, } ) - env.pageserver.stop() - env.pageserver.start( - extra_env_vars={"NEON_PAGESERVER_VALUE_RECONSTRUCT_IO_CONCURRENCY": ps_io_concurrency} - ) + env.pageserver.restart() metrics = workload() log.info("Results: %s", metrics) @@ -327,6 +325,8 @@ def test_latency( def patch_ps_config(ps_config): if pipelining_config is not None: ps_config["page_service_pipelining"] = dataclasses.asdict(pipelining_config) + ps_config["get_vectored_concurrent_io"] = {"mode": ps_io_concurrency} + # FIXME: parametrize over direct IO mode neon_env_builder.pageserver_config_override = patch_ps_config @@ -358,11 +358,6 @@ 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) #