rust/neon
1
0
Fork 0
mirror of https://github.com/neondatabase/neon.git synced 2026-07-07 14:10:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

replace env var with config variable; add test suite fixture env var to override default

Browse Source
This commit is contained in:
Christian Schwarz
2025-01-20 23:35:47 +01:00
parent 0eff09e35f
commit 0a37164c29
13 changed files with 118 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
.github/workflows/_build-and-test-locally.yml vendored
View File

@@ -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

25
libs/pageserver_api/src/config.rs
View File

@@ -120,6 +120,7 @@ pub struct ConfigToml {
pub no_sync: Option<bool>,
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
},
}
}
}

3
pageserver/src/basebackup.rs
View File

@@ -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()

1
pageserver/src/bin/pageserver.rs
View File

@@ -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.

4
pageserver/src/config.rs
View File

@@ -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

26
pageserver/src/page_service.rs
View File

@@ -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<TenantManager>,
auth: Option<Arc<SwappableJwtAuth>>,
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<TenantManager>,
auth: Option<Arc<SwappableJwtAuth>>,
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<Arc<SwappableJwtAuth>>,
claims: Option<Claims>,
@@ -655,6 +662,7 @@ impl BatchedFeMessage {
impl PageServerHandler {
pub fn new(
conf: &'static PageServerConf,
tenant_manager: Arc<TenantManager>,
auth: Option<Arc<SwappableJwtAuth>>,
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()

4
pageserver/src/tenant.rs
View File

@@ -6548,8 +6548,8 @@ mod tests {
let gate = Gate::default();
let io_concurrency_levels: Vec<Box<dyn Fn() -> 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

53
pageserver/src/tenant/storage_layer.rs
View File

@@ -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<IoFuture>,
},
}
@@ -213,15 +214,15 @@ pub(crate) enum IoConcurrency {
type IoFuture = Pin<Box<dyn Send + Future<Output = ()>>>;
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<String> = 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<Output = ()> + 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.

13
test_runner/deep_layers_with_delta.py
View File

@@ -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

18
test_runner/fixtures/neon_fixtures.py
View File

@@ -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,

5
test_runner/fixtures/parametrize.py
View File

@@ -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:

4
test_runner/performance/pageserver/pagebench/test_pageserver_max_throughput_getpage_at_latest_lsn.py
View File

@@ -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(

15
test_runner/performance/pageserver/test_page_service_batching.py
View File

@@ -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)
#