This is an automated email from the ASF dual-hosted git repository. hubcio pushed a commit to branch feat/segment-write-rwf-uncached in repository https://gitbox.apache.org/repos/asf/iggy.git
commit b46b8a81b7a8d9c8ec1f70ab28321d404c8596d0 Author: Hubert Gruszecki <[email protected]> AuthorDate: Thu Aug 27 19:23:51 2026 +0200 feat(server): add uncached (RWF_DONTCACHE) segment writes Segment .log and .index writes always went through the buffered compio path, so every byte a broker appended stayed in the page cache until the kernel decided otherwise. On a write-heavy node that cache holds data no reader will ask for again, and it competes with the pages that do get read. system.segment.write_io now selects buffered (the default, unchanged behaviour) or uncached. Uncached routes both writers through new server_common io_uring ops: compio 0.12.4 builds its file ops without rw_flags and keeps their fields crate-private, so WriteAtFlags and WriteVectoredAtFlags mirror compio's own ops, hold the descriptor through SharedFd so a dropped future cannot outlive it, and build iovecs in init because the driver boxes the op after construction. Boot fails closed rather than discovering the problem on the first flush. The probe submits the same op the writers use, since a kernel and filesystem pair can accept pwritev2 with the flag and still reject it through the ring, and it probes the deepest existing ancestor of the streams path because segments do not live under system.path and a submount can only attach to a directory that already exists. It runs before --fresh wipes anything and writes the config key, kernel release and filesystem to stderr directly, since tracing before late_init goes to a null writer. Kernels 6.14 to 7.1 submit the writeback inline and only get a warning; off Linux the validator rejects uncached outright. A submount created after startup cannot be probed at all, so the writers name the knob once per process on EOPNOTSUPP. The error stays CannotWriteToFile because error codes are replicated state. The knob is not a free page-cache saving and the config comment now says what a kernel-level micro-bench on ext4 measured: only whole pages written by a single flush are dropped, small appends stay resident, each fresh 4 KiB block costs one synchronous read-back, a paced small writer without fsync multiplies device writes about 9x, and the drop lands when writeback completes rather than when the write returns. Tests that passed with the feature inert were the main risk. The integration suite measures page residency with mincore after forcing writeback (1 of 263 pages resident uncached against 263 of 263 buffered on ext4), covers the boot refusal on tmpfs and a restart round-trip, and IGGY_REQUIRE_UNCACHED_IO=1 turns a previously silent skip into a failure. The writer unit tests require the uncached write to be refused where the buffered one succeeds, and a tmpfs rejection test guards the scalar op's rw_flags, which could be deleted with every test still green. iggy-bench gains the topic segment size, flush and preallocation knobs the write-io matrix needs, and its warmup loop now shares the measured loop's rate limiter so each arm enters its window with comparable page-cache and FTL state. Co-Authored-By: Claude Opus 5 (1M context) <[email protected]> Claude-Session: https://claude.ai/code/session_01UbkxH2rbvWdjtB38nJhMYk --- Cargo.lock | 3 + Cargo.toml | 1 + .../src/actors/producer/benchmark_producer.rs | 13 +- core/bench/src/args/common.rs | 23 + core/bench/src/benchmarks/benchmark.rs | 10 +- core/configs/src/common/defaults.rs | 1 + core/configs/src/common/displays.rs | 6 +- core/configs/src/common/system.rs | 5 + core/configs/src/common/validators.rs | 8 + .../configs/src/configs_impl/typed_env_provider.rs | 3 + core/configs/src/server_config/validators.rs | 30 + core/integration/Cargo.toml | 5 + core/integration/tests/config_provider/mod.rs | 28 + core/integration/tests/server/mod.rs | 4 + core/integration/tests/server/segment_write_io.rs | 472 ++++++++++++++ core/partitions/src/iggy_index_writer.rs | 136 +++- core/partitions/src/iggy_partition.rs | 244 +++++++- core/partitions/src/lib.rs | 150 +++++ core/partitions/src/messages_writer.rs | 201 +++++- core/partitions/src/state_transfer.rs | 2 + core/partitions/src/types.rs | 4 + core/server/config.toml | 29 + core/server/src/boot/mod.rs | 61 +- core/server/src/boot/recovery.rs | 1 + core/server/src/dispatch/mod.rs | 1 + core/server/src/dispatch/partition.rs | 3 + core/server/src/dispatch/session_ops.rs | 1 + core/server/src/dispatch/test_support.rs | 1 + core/server/src/partition_helpers.rs | 84 ++- core/server/src/partition_reconciler.rs | 1 + core/server/src/server_error.rs | 10 + core/server_common/Cargo.toml | 4 + core/server_common/src/diagnostics.rs | 2 +- core/server_common/src/lib.rs | 3 + core/server_common/src/segment_io.rs | 119 ++++ core/server_common/src/uncached_io.rs | 688 +++++++++++++++++++++ core/simulator/src/replica.rs | 1 + 37 files changed, 2263 insertions(+), 95 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 56ecb71dd..67d56552d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7506,6 +7506,7 @@ dependencies = [ "serde", "serde_json", "serial_test", + "server_common", "sqlparser 0.62.0", "sqlx", "sysinfo 0.39.6", @@ -12265,6 +12266,8 @@ dependencies = [ "human-repr", "iggy_binary_protocol", "iggy_common", + "io-uring", + "libc", "nix", "opentelemetry", "opentelemetry-appender-tracing", diff --git a/Cargo.toml b/Cargo.toml index af7a697ca..53abb5fd5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -207,6 +207,7 @@ iggy_connector_doris_sink = { path = "core/connectors/sinks/doris_sink" } iggy_connector_sdk = { path = "core/connectors/sdk", version = "0.4.0-edge.3" } indexmap = "2.14.1" integration = { path = "core/integration" } +io-uring = "0.7.13" ipnet = "2.12.1" journal = { path = "core/journal" } js-sys = "0.3" diff --git a/core/bench/src/actors/producer/benchmark_producer.rs b/core/bench/src/actors/producer/benchmark_producer.rs index c51c45c5f..5de7a148c 100644 --- a/core/bench/src/actors/producer/benchmark_producer.rs +++ b/core/bench/src/actors/producer/benchmark_producer.rs @@ -86,12 +86,22 @@ impl<P: BenchmarkProducerClient> BenchmarkProducer<P> { let mut batch_generator = BenchmarkBatchGenerator::new(self.config.message_size, self.config.messages_per_batch); + // One limiter shared by warmup and the measured loop, so warmup cannot run at + // drive speed and hand the measured window a different cache / FTL state per arm. + let rate_limiter = self.limit_bytes_per_second.map(BenchmarkRateLimiter::new); + if self.config.warmup_time.get_duration() != Duration::from_millis(0) { self.log_warmup_info(); let warmup_end = Instant::now() + self.config.warmup_time.get_duration(); while Instant::now() < warmup_end { - let _ = self.client.produce_batch(&mut batch_generator).await?; + if let Some(batch) = self.client.produce_batch(&mut batch_generator).await? + && let Some(rate_limiter) = &rate_limiter + { + rate_limiter + .wait_until_necessary(batch.user_data_bytes) + .await; + } } } @@ -104,7 +114,6 @@ impl<P: BenchmarkProducerClient> BenchmarkProducer<P> { let mut user_data_bytes_processed = 0; let mut total_bytes_processed = 0; - let rate_limiter = self.limit_bytes_per_second.map(BenchmarkRateLimiter::new); let start_timestamp = Instant::now(); while !self.finish_condition.is_done() { diff --git a/core/bench/src/args/common.rs b/core/bench/src/args/common.rs index c11a32a8e..5c1f7060d 100644 --- a/core/bench/src/args/common.rs +++ b/core/bench/src/args/common.rs @@ -113,6 +113,21 @@ pub struct IggyBenchArgs { /// Topic option at creation, so it has no effect with `--reuse-streams`. #[arg(long)] pub messages_required_to_save: Option<NonZeroU32>, + + /// Reserve each segment's bytes on disk when it is created (server + /// default: false). Admission rejects the topic when + /// `segment_size * partitions` exceeds 64 GiB. + /// Topic option at creation, so it has no effect with `--reuse-streams`. + #[arg(long)] + pub preallocate_segments: Option<bool>, + + /// Bytes a segment holds before it is closed and a new one started + /// (server default: 1 GiB). Accepts human-readable formats like "256MB" + /// or "1GiB". Admission requires at least 1 MiB, at most 1 GiB, and a + /// multiple of 512 B. + /// Topic option at creation, so it has no effect with `--reuse-streams`. + #[arg(long)] + pub segment_size: Option<IggyByteSize>, } impl IggyBenchArgs { @@ -348,6 +363,14 @@ impl IggyBenchArgs { self.messages_required_to_save } + pub const fn preallocate_segments(&self) -> Option<bool> { + self.preallocate_segments + } + + pub const fn segment_size(&self) -> Option<IggyByteSize> { + self.segment_size + } + pub fn username(&self) -> &str { &self.username } diff --git a/core/bench/src/benchmarks/benchmark.rs b/core/bench/src/benchmarks/benchmark.rs index 356b81279..0a92374cd 100644 --- a/core/bench/src/benchmarks/benchmark.rs +++ b/core/bench/src/benchmarks/benchmark.rs @@ -126,15 +126,19 @@ pub trait Benchmarkable: Send { let enforce_fsync = self.args().enforce_fsync(); let messages_required_to_save = self.args().messages_required_to_save().map(NonZeroU32::get); + let preallocate_segments = self.args().preallocate_segments(); + let segment_size = self.args().segment_size(); info!( - "Creating the test topic '{}' for stream '{}' with max topic size: {:?}, message expiry: {}, enforce fsync: {}, messages required to save: {:?}", + "Creating the test topic '{}' for stream '{}' with max topic size: {:?}, message expiry: {}, enforce fsync: {}, messages required to save: {:?}, preallocate segments: {:?}, segment size: {:?}", topic_name, stream_name, max_topic_size, message_expiry, enforce_fsync, - messages_required_to_save + messages_required_to_save, + preallocate_segments, + segment_size ); client @@ -147,8 +151,10 @@ pub trait Benchmarkable: Send { .then_some(message_expiry), max_topic_size: (max_topic_size != MaxTopicSize::ServerDefault) .then_some(max_topic_size), + segment_size, enforce_fsync: enforce_fsync.then_some(true), messages_required_to_save, + preallocate_segments, ..TopicCreateOptions::default() }, ) diff --git a/core/configs/src/common/defaults.rs b/core/configs/src/common/defaults.rs index 1dddfcc54..4ca2f26d4 100644 --- a/core/configs/src/common/defaults.rs +++ b/core/configs/src/common/defaults.rs @@ -276,6 +276,7 @@ impl Default for SegmentConfig { fn default() -> SegmentConfig { SegmentConfig { archive_expired: SERVER_CONFIG.system.segment.archive_expired, + write_io: SERVER_CONFIG.system.segment.write_io.parse().unwrap(), } } } diff --git a/core/configs/src/common/displays.rs b/core/configs/src/common/displays.rs index 32af51798..735473066 100644 --- a/core/configs/src/common/displays.rs +++ b/core/configs/src/common/displays.rs @@ -154,7 +154,11 @@ impl Display for PartitionConfig { impl Display for SegmentConfig { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { - write!(f, "{{ archive_expired: {} }}", self.archive_expired,) + write!( + f, + "{{ archive_expired: {}, write_io: {} }}", + self.archive_expired, self.write_io + ) } } diff --git a/core/configs/src/common/system.rs b/core/configs/src/common/system.rs index ae0e7eeb2..2d555d471 100644 --- a/core/configs/src/common/system.rs +++ b/core/configs/src/common/system.rs @@ -25,6 +25,8 @@ use serde_with::serde_as; use server_common::bootstrap::SystemPaths; use server_common::log::LoggingSettings; +pub use server_common::segment_io::SegmentIoMode; + pub const INDEX_EXTENSION: &str = "index"; pub const LOG_EXTENSION: &str = "log"; @@ -127,6 +129,9 @@ pub struct RecoveryConfig { #[derive(Debug, Deserialize, Serialize, ConfigEnv)] pub struct SegmentConfig { pub archive_expired: bool, + #[serde(default)] + #[config_env(leaf)] + pub write_io: SegmentIoMode, } impl<S: ConfigEnvMappings> SystemConfig<S> { diff --git a/core/configs/src/common/validators.rs b/core/configs/src/common/validators.rs index de3da58c2..098a5c5d1 100644 --- a/core/configs/src/common/validators.rs +++ b/core/configs/src/common/validators.rs @@ -67,6 +67,14 @@ impl Validatable<ConfigurationError> for SegmentConfig { // Segment size is a per-topic creation option now; its ceiling, floor // and 512 B-multiple rule are enforced by // `iggy_common::validate_topic_segment_size` at admission. + + // `RWF_DONTCACHE` rides on io_uring `rw_flags`, which only the Linux + // driver has; kernel and filesystem support is probed at boot. + #[cfg(not(target_os = "linux"))] + if self.write_io == server_common::segment_io::SegmentIoMode::Uncached { + eprintln!("Configured system.segment.write_io = \"uncached\" requires Linux"); + return Err(ConfigurationError::InvalidConfigurationValue); + } Ok(()) } } diff --git a/core/configs/src/configs_impl/typed_env_provider.rs b/core/configs/src/configs_impl/typed_env_provider.rs index 18c019437..cf23999be 100644 --- a/core/configs/src/configs_impl/typed_env_provider.rs +++ b/core/configs/src/configs_impl/typed_env_provider.rs @@ -60,6 +60,9 @@ const IGNORED_ENV_VARS: &[&str] = &[ "IGGY_SHARD_RUNTIME_CAPACITY", "IGGY_TEST_CLEANUP_DISABLED", "IGGY_TEST_VERBOSE", + // Test-harness knob: forbids the uncached-I/O tests from skipping + // themselves; leaks to spawned servers via the IGGY_ env forwarding. + "IGGY_REQUIRE_UNCACHED_IO", ]; /// Prefixes for env vars handled by separate providers with runtime prefixes. diff --git a/core/configs/src/server_config/validators.rs b/core/configs/src/server_config/validators.rs index a20a48b56..aa83eddf8 100644 --- a/core/configs/src/server_config/validators.rs +++ b/core/configs/src/server_config/validators.rs @@ -482,6 +482,7 @@ fn parse_bind_address(context: &str, address: &str) -> Result<SocketAddr, Config mod tests { use super::super::cluster::{ClusterNodeConfig, TransportPorts}; use super::*; + use crate::common::system::SegmentIoMode; use figment::Figment; use figment::providers::{Format, Toml}; @@ -626,6 +627,35 @@ mod tests { assert!(config.validate().is_err()); } + // Flipping the shipped default to `uncached` would refuse boot on every + // host without RWF_DONTCACHE support; the shipped file must stay buffered. + #[test] + fn given_shipped_default_config_when_deserializing_should_write_buffered() { + let config = config_with_override(""); + assert_eq!(config.system.segment.write_io, SegmentIoMode::Buffered); + } + + #[test] + fn given_segment_write_io_uncached_when_deserializing_should_parse() { + let config = config_with_override("[system.segment]\nwrite_io = \"uncached\"\n"); + assert_eq!(config.system.segment.write_io, SegmentIoMode::Uncached); + config + .validate() + .expect("write_io support is probed at boot, not bounded by config validation"); + } + + #[test] + fn given_unknown_segment_write_io_when_deserializing_should_reject() { + let result: Result<ServerConfig, _> = Figment::new() + .merge(Toml::string(DEFAULT_CONFIG)) + .merge(Toml::string("[system.segment]\nwrite_io = \"direct\"\n")) + .extract(); + assert!( + result.is_err(), + "unknown system.segment.write_io must fail config load" + ); + } + #[test] fn given_recreate_missing_state_enabled_when_validating_should_reject() { let config = config_with_override("[system.recovery]\nrecreate_missing_state = true\n"); diff --git a/core/integration/Cargo.toml b/core/integration/Cargo.toml index f53fc1369..3f2c93286 100644 --- a/core/integration/Cargo.toml +++ b/core/integration/Cargo.toml @@ -110,6 +110,11 @@ zip = { workspace = true } [target.'cfg(any(target_os = "linux", target_os = "freebsd", target_os = "dragonfly", target_os = "netbsd", target_os = "openbsd"))'.dependencies] zbus-secret-service-keyring-store = { workspace = true, optional = true } +[target.'cfg(target_os = "linux")'.dependencies] +# Runs the RWF_DONTCACHE probe the server boots with, so the uncached segment +# tests skip where the kernel or the filesystem cannot take the flag. +server_common = { workspace = true } + [build-dependencies] cfg_aliases = { workspace = true } diff --git a/core/integration/tests/config_provider/mod.rs b/core/integration/tests/config_provider/mod.rs index db7472d7a..58b95a8d2 100644 --- a/core/integration/tests/config_provider/mod.rs +++ b/core/integration/tests/config_provider/mod.rs @@ -16,6 +16,7 @@ // under the License. use configs::server::ServerConfig; +use configs::system::SegmentIoMode; use configs::{ConfigEnvMappings, ConfigProvider, TypedEnvProvider}; use configs_derive::ConfigEnv; use figment::providers::{Format, Toml}; @@ -64,6 +65,33 @@ async fn validate_config_env_override() { } } +#[serial] +#[tokio::test] +async fn validate_segment_write_io_env_override() { + let expected_write_io = SegmentIoMode::Uncached; + + unsafe { + env::set_var( + "IGGY_SYSTEM_SEGMENT_WRITE_IO", + expected_write_io.to_string(), + ); + } + + let config_path = get_root_path().join("../server/config.toml"); + let file_config_provider = + ServerConfig::config_provider(&config_path.as_path().display().to_string()); + let config: ServerConfig = file_config_provider + .load_config() + .await + .expect("Failed to load config.toml config with segment write_io override"); + + assert_eq!(config.system.segment.write_io, expected_write_io); + + unsafe { + env::remove_var("IGGY_SYSTEM_SEGMENT_WRITE_IO"); + } +} + #[serial] #[tokio::test] async fn validate_byte_size_leaf_override() { diff --git a/core/integration/tests/server/mod.rs b/core/integration/tests/server/mod.rs index a1da26b1e..db288c6b8 100644 --- a/core/integration/tests/server/mod.rs +++ b/core/integration/tests/server/mod.rs @@ -77,4 +77,8 @@ mod message_retrieval; // a restarted replica rejoins via the view probe + journal repair. mod purge_delete; mod scenarios; +// `[system.segment] write_io = "uncached"`: boot refusal on a filesystem that +// rejects RWF_DONTCACHE, and uncached-flushed segments read back after a restart. +#[cfg(target_os = "linux")] +mod segment_write_io; mod specific; diff --git a/core/integration/tests/server/segment_write_io.rs b/core/integration/tests/server/segment_write_io.rs new file mode 100644 index 000000000..3c872509e --- /dev/null +++ b/core/integration/tests/server/segment_write_io.rs @@ -0,0 +1,472 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! `[system.segment] write_io = "uncached"` end to end: the boot probe must +//! refuse a data directory whose filesystem rejects `RWF_DONTCACHE`, and +//! segments flushed through the uncached ops must leave the page cache behind +//! them and read back through the regular path, live and after a restart +//! reopens them from disk. + +use std::collections::HashMap; +use std::fs; +use std::os::fd::AsRawFd; +use std::path::{Path, PathBuf}; +use std::time::Duration; + +use iggy::prelude::*; +use integration::harness::{TestBinaryError, TestHarness, TestServerConfig, disk}; +use serial_test::parallel; +use server_common::uncached_io::{ + UncachedIoError, probe_uncached_write, require_uncached_io_tests, +}; +use tempfile::TempDir; +use tokio::time::{sleep, timeout}; + +const SHM_DIR: &str = "/dev/shm"; +const REQUIRE_UNCACHED_IO_ENV: &str = "IGGY_REQUIRE_UNCACHED_IO"; +const STREAM_NAME: &str = "uncached-stream"; +const TOPIC_NAME: &str = "uncached-topic"; +const PARTITION_ID: u32 = 0; +const BATCHES: u32 = 4; +const MESSAGES_PER_BATCH: u32 = 128; +/// Below the batch size, so every batch crosses the flush threshold. +const MESSAGES_REQUIRED_TO_SAVE: u32 = 64; +const PAYLOAD_LEN: usize = 2048; +/// Budget for the live flushes to land in the segment files. +const FLUSH_INSTALL_TIMEOUT: Duration = Duration::from_secs(20); +const POLL_INTERVAL: Duration = Duration::from_millis(250); +/// Bound on one poll round trip: if the (restarted) server dies, the SDK's +/// reconnect loop would otherwise hang the test forever. +const POLL_TIMEOUT: Duration = Duration::from_secs(60); +/// Budget for the pages to leave the cache once writeback has been forced. +const PAGE_DROP_TIMEOUT: Duration = Duration::from_secs(10); +/// Share of the segment's pages allowed to remain resident, in percent. +/// Measured on ext4: 1 of 263 pages with `uncached` against 263 of 263 with +/// `buffered`. The slack covers the only residency `uncached` can leave +/// behind, a block no single flush wrote whole. +const MAX_RESIDENT_PERCENT: usize = 25; + +/// The probe runs once at boot, so a system path on a filesystem that +/// rejects `RWF_DONTCACHE` (tmpfs here; every filesystem on kernels below +/// 6.14) must stop the server before it serves anything, with the reason and +/// the remedy in its output. +#[tokio::test] +#[parallel] +async fn given_uncached_write_io_when_the_system_path_is_on_tmpfs_should_refuse_to_boot() { + let Some((system_path, expected_reason)) = tmpfs_system_path() else { + return; + }; + let mut envs = uncached_envs(); + envs.insert( + "IGGY_SYSTEM_PATH".to_string(), + system_path.path().display().to_string(), + ); + let mut harness = TestHarness::builder() + .cluster_nodes(1) + .server(TestServerConfig::builder().extra_envs(envs).build()) + .build() + .unwrap(); + + let error = harness + .start() + .await + .expect_err("boot with write_io = \"uncached\" over tmpfs must refuse"); + let (exit_code, stdout, stderr) = match error { + TestBinaryError::ProcessCrashed { + exit_code, + stdout, + stderr, + .. + } => (exit_code, stdout, stderr), + other => panic!("the server must exit on its own rather than hang or serve: {other}"), + }; + assert!( + matches!(exit_code, Some(code) if code != 0), + "the refusal must be a non-zero exit, got {exit_code:?}" + ); + if logs_are_captured() { + let diagnostics = format!("{stdout}\n{stderr}"); + assert!( + diagnostics.contains(&expected_reason), + "the refusal must carry the probe's reason {expected_reason:?}, got:\n{diagnostics}" + ); + assert!( + diagnostics.contains("UncachedWriteUnsupported"), + "the refusal must surface as ServerError::UncachedWriteUnsupported, got:\n{diagnostics}" + ); + } +} + +/// Segments flushed through the `RWF_DONTCACHE` ops must leave the page cache +/// behind them and must be what the regular read path serves, while the writer +/// is live and after a restart reopens them from disk. Skipped where the data +/// directory's filesystem or the kernel cannot do uncached writes at all. +#[tokio::test] +#[parallel] +async fn given_uncached_write_io_when_the_server_restarts_should_serve_the_flushed_messages() { + let mut harness = TestHarness::builder() + .cluster_nodes(1) + .server( + TestServerConfig::builder() + .extra_envs(uncached_envs()) + .build(), + ) + .build() + .unwrap(); + let data_path = harness.server().data_path(); + fs::create_dir_all(&data_path).expect("create the data directory for the probe"); + match probe_blocking(data_path.clone()) { + Ok(()) => {} + Err(error @ UncachedIoError::Unsupported { .. }) => { + skip_or_fail(&error.to_string()); + return; + } + Err(error) => panic!( + "probe under {} hit an I/O error: {error}", + data_path.display() + ), + } + + harness.start().await.unwrap(); + assert_eq!( + effective_write_io(&data_path), + "uncached", + "the env override must reach the server" + ); + let client = harness + .tcp_root_client() + .await + .expect("create TCP client for sending messages"); + create_stream_and_topic(&client).await; + let payloads = send_batches(&client).await; + let payload_bytes = payloads.iter().map(|payload| payload.len() as u64).sum(); + wait_until_segments_hold(&data_path, payload_bytes).await; + // Before any poll: reading the segment back would fault its pages in + // again and hide the drop this assertion is about. + assert_segment_pages_left_the_cache(&data_path).await; + assert_polled_payloads(&client, &payloads).await; + drop(client); + + harness.restart_server().await.unwrap(); + let client = harness + .tcp_root_client() + .await + .expect("create TCP client after restart"); + assert_polled_payloads(&client, &payloads).await; +} + +fn uncached_envs() -> HashMap<String, String> { + HashMap::from([( + "IGGY_SYSTEM_SEGMENT_WRITE_IO".to_string(), + "uncached".to_string(), + )]) +} + +/// A fresh directory on tmpfs plus the probe's own reason for refusing it, +/// `None` (reason printed) when `/dev/shm` is not tmpfs or this kernel takes +/// `RWF_DONTCACHE` there. +fn tmpfs_system_path() -> Option<(TempDir, String)> { + let dir = match tempfile::tempdir_in(SHM_DIR) { + Ok(dir) => dir, + Err(error) => { + skip_or_fail(&format!( + "cannot create a directory under {SHM_DIR}: {error}" + )); + return None; + } + }; + match probe_blocking(dir.path().to_path_buf()) { + Err(UncachedIoError::Unsupported { filesystem, .. }) if filesystem != "tmpfs" => { + skip_or_fail(&format!("{SHM_DIR} is {filesystem}, not tmpfs")); + None + } + Err(error @ UncachedIoError::Unsupported { .. }) => Some((dir, error.to_string())), + Ok(()) => { + skip_or_fail("tmpfs takes RWF_DONTCACHE on this kernel"); + None + } + Err(error) => panic!("probe under {SHM_DIR} hit an I/O error: {error}"), + } +} + +/// libtest swallows the output of a passing test, so a test that skips itself +/// for good is invisible. Loud on a box that cannot do uncached writes, fatal +/// on one where they are required to have run. +fn skip_or_fail(reason: &str) { + assert!( + !require_uncached_io_tests(), + "### {REQUIRE_UNCACHED_IO_ENV}=1 forbids skipping: {reason} ###" + ); + eprintln!("######## SKIPPING UNCACHED TEST: {reason} ########"); +} + +/// The probe submits the segment writers' own io_uring op, so it needs a +/// compio runtime; these tests run on tokio. +fn probe_blocking(dir: PathBuf) -> Result<(), UncachedIoError> { + std::thread::spawn(move || { + compio::runtime::Runtime::new() + .expect("compio runtime") + .block_on(probe_uncached_write(&dir)) + }) + .join() + .expect("probe thread") +} + +/// `IGGY_TEST_VERBOSE` inherits the server's stdio, leaving nothing captured +/// to assert on. +fn logs_are_captured() -> bool { + std::env::var("IGGY_TEST_VERBOSE").is_err() +} + +/// `system.segment.write_io` as the server re-serialized it into +/// `current_config.toml`. That file is output only, so this proves the env +/// override reached the configuration and nothing about the write path. +fn effective_write_io(data_path: &Path) -> String { + let path = data_path.join("runtime/current_config.toml"); + let content = fs::read_to_string(&path) + .unwrap_or_else(|error| panic!("read {}: {error}", path.display())); + let config: toml::Value = toml::from_str(&content) + .unwrap_or_else(|error| panic!("parse {}: {error}", path.display())); + config["system"]["segment"]["write_io"] + .as_str() + .expect("system.segment.write_io is a string") + .to_owned() +} + +async fn create_stream_and_topic(client: &IggyClient) { + client + .create_stream(STREAM_NAME) + .await + .expect("create stream"); + client + .create_topic( + &Identifier::named(STREAM_NAME).unwrap(), + TOPIC_NAME, + &TopicCreateOptions { + partitions_count: Some(1), + message_expiry: Some(IggyExpiry::NeverExpire), + messages_required_to_save: Some(MESSAGES_REQUIRED_TO_SAVE), + ..TopicCreateOptions::default() + }, + ) + .await + .expect("create topic"); +} + +/// Sends `BATCHES` batches to the single partition, each past the flush +/// threshold, returning every payload in send order (offset order). +async fn send_batches(client: &IggyClient) -> Vec<String> { + let stream = Identifier::named(STREAM_NAME).unwrap(); + let topic = Identifier::named(TOPIC_NAME).unwrap(); + let partitioning = Partitioning::partition_id(PARTITION_ID); + let mut payloads = Vec::with_capacity((BATCHES * MESSAGES_PER_BATCH) as usize); + for batch in 0..BATCHES { + let mut messages = Vec::with_capacity(MESSAGES_PER_BATCH as usize); + for index in 0..MESSAGES_PER_BATCH { + let payload = payload_for(batch * MESSAGES_PER_BATCH + index); + messages.push( + IggyMessage::builder() + .payload(payload.clone().into()) + .build() + .expect("build message"), + ); + payloads.push(payload); + } + client + .send_messages(&stream, &topic, &partitioning, &mut messages) + .await + .unwrap_or_else(|error| panic!("send batch {batch}: {error}")); + } + payloads +} + +/// `PAYLOAD_LEN` bytes unique to `offset`: a tag, then filler that shifts +/// with the offset so a misplaced or repeated write cannot pass the +/// comparison. +fn payload_for(offset: u32) -> String { + let tag = format!("uncached-{offset:05}-"); + let filler = (tag.len()..PAYLOAD_LEN) + .map(|position| char::from(b'a' + ((position + offset as usize) % 26) as u8)); + tag.chars().chain(filler).collect() +} + +/// Total bytes of partition segment `.log` files under `data_path`; +/// preallocation keeps the logical length at the written bytes. +fn segment_log_bytes(data_path: &Path) -> u64 { + let mut total = 0; + let _ = disk::walk(data_path, &mut |path| { + if disk::is_segment_log(path) { + total += fs::metadata(path).map_or(0, |meta| meta.len()); + } + false + }); + total +} + +/// Waits until the segment files hold at least `min_bytes`. Callers pass the +/// payload bytes alone: headers come on top, so the bound is only met once +/// nearly every batch was flushed while the server was live, by the uncached +/// writer rather than the shutdown flush. +async fn wait_until_segments_hold(data_path: &Path, min_bytes: u64) { + let deadline = tokio::time::Instant::now() + FLUSH_INSTALL_TIMEOUT; + loop { + let installed = segment_log_bytes(data_path); + if installed >= min_bytes { + return; + } + assert!( + tokio::time::Instant::now() < deadline, + "only {installed} of at least {min_bytes} segment bytes reached disk within \ + {FLUSH_INSTALL_TIMEOUT:?}" + ); + sleep(POLL_INTERVAL).await; + } +} + +/// The one behavioural check that `buffered` cannot pass: a segment written +/// with `RWF_DONTCACHE` loses its pages as writeback completes, while a +/// buffered one keeps every page it just wrote even after an fsync. +/// +/// The drop is gated on writeback completing, not on the write returning, so +/// left alone the pages linger until the bdi flusher reaches them, which +/// `dirty_writeback_centisecs` alone can stretch past any sane budget on a +/// loaded box. Forcing the writeback removes that scheduling from the +/// assertion without weakening it: fsync evicts nothing by itself. +async fn assert_segment_pages_left_the_cache(data_path: &Path) { + let log_path = disk::walk(data_path, &mut |path| disk::is_segment_log(path)) + .expect("the flushed batches must have created a segment .log"); + force_writeback(&log_path); + let deadline = tokio::time::Instant::now() + PAGE_DROP_TIMEOUT; + loop { + let (resident, pages) = page_residency(&log_path); + assert!( + pages > 1, + "{} holds {pages} page(s), too few to tell a dropped page from a partial tail", + log_path.display() + ); + if resident * 100 <= pages * MAX_RESIDENT_PERCENT { + return; + } + assert!( + tokio::time::Instant::now() < deadline, + "{resident} of {pages} pages of {} are still in the page cache after \ + {PAGE_DROP_TIMEOUT:?}; uncached segment writes did not drop them", + log_path.display() + ); + sleep(POLL_INTERVAL).await; + } +} + +/// Pushes `path`'s dirty pages to the device and returns once they are clean. +/// Read-only so nothing here can dirty a page; `fsync` needs no write access. +fn force_writeback(path: &Path) { + fs::File::open(path) + .and_then(|file| file.sync_all()) + .unwrap_or_else(|error| panic!("fsync {}: {error}", path.display())); +} + +/// Pages of `path` the page cache still holds, and the file's page count. +/// `mincore` over a read-only mapping reports residency without faulting +/// anything in, so measuring does not change what it measures. +fn page_residency(path: &Path) -> (usize, usize) { + let file = + fs::File::open(path).unwrap_or_else(|error| panic!("open {}: {error}", path.display())); + let len = usize::try_from( + file.metadata() + .unwrap_or_else(|error| panic!("stat {}: {error}", path.display())) + .len(), + ) + .expect("segment length fits in usize"); + if len == 0 { + return (0, 0); + } + // SAFETY: a fresh read-only mapping of a file held open for the whole + // call; the pointer is only handed back to `mincore` and `munmap`. + let (result, error, residency) = unsafe { + let page_size = usize::try_from(libc::sysconf(libc::_SC_PAGESIZE)).expect("page size"); + let mut residency = vec![0u8; len.div_ceil(page_size)]; + let address = libc::mmap( + std::ptr::null_mut(), + len, + libc::PROT_READ, + libc::MAP_SHARED, + file.as_raw_fd(), + 0, + ); + assert!( + address != libc::MAP_FAILED, + "mmap {}: {}", + path.display(), + std::io::Error::last_os_error() + ); + let result = libc::mincore(address, len, residency.as_mut_ptr()); + let error = std::io::Error::last_os_error(); + libc::munmap(address, len); + (result, error, residency) + }; + assert_eq!(result, 0, "mincore {}: {error}", path.display()); + ( + residency.iter().filter(|page| *page & 1 == 1).count(), + residency.len(), + ) +} + +/// Polls the partition from offset 0 in pages and checks every payload sits +/// at its send-order offset. +async fn assert_polled_payloads(client: &IggyClient, expected: &[String]) { + let stream = Identifier::named(STREAM_NAME).unwrap(); + let topic = Identifier::named(TOPIC_NAME).unwrap(); + let mut polled = Vec::with_capacity(expected.len()); + while polled.len() < expected.len() { + let page = timeout( + POLL_TIMEOUT, + client.poll_messages( + &stream, + &topic, + Some(PARTITION_ID), + &Consumer::default(), + &PollingStrategy::offset(polled.len() as u64), + MESSAGES_PER_BATCH, + false, + ), + ) + .await + .unwrap_or_else(|_| { + panic!( + "poll from offset {} did not return within {POLL_TIMEOUT:?}", + polled.len() + ) + }) + .expect("poll messages"); + assert!( + !page.messages.is_empty(), + "poll from offset {} returned nothing, {} of {} messages read back", + polled.len(), + polled.len(), + expected.len() + ); + polled.extend(page.messages); + } + assert_eq!(polled.len(), expected.len(), "more messages than were sent"); + for (offset, (message, payload)) in polled.iter().zip(expected).enumerate() { + assert_eq!(message.header.offset, offset as u64); + assert!( + message.payload.as_ref() == payload.as_bytes(), + "payload at offset {offset} differs from what was sent" + ); + } +} diff --git a/core/partitions/src/iggy_index_writer.rs b/core/partitions/src/iggy_index_writer.rs index 78ab2c0eb..ea3529d1d 100644 --- a/core/partitions/src/iggy_index_writer.rs +++ b/core/partitions/src/iggy_index_writer.rs @@ -18,16 +18,26 @@ use compio::fs::{File, OpenOptions}; use compio::io::AsyncWriteAtExt; use iggy_common::IggyError; +use server_common::segment_io::SegmentIoMode; use std::rc::Rc; use std::sync::atomic::{AtomicU64, Ordering}; use tracing::{error, trace}; +#[cfg(target_os = "linux")] +use crate::report_uncached_write_unsupported; +#[cfg(target_os = "linux")] +use compio::driver::ToSharedFd; +#[cfg(target_os = "linux")] +use server_common::uncached_io::write_all_at_uncached; + #[derive(Debug)] pub struct IggyIndexWriter { file_path: String, file: File, index_size_bytes: Rc<AtomicU64>, fsync: bool, + #[cfg(target_os = "linux")] + write_io: SegmentIoMode, } impl IggyIndexWriter { @@ -41,8 +51,14 @@ impl IggyIndexWriter { file_path: &str, index_size_bytes: Rc<AtomicU64>, fsync: bool, + write_io: SegmentIoMode, file_exists: bool, ) -> Result<Self, IggyError> { + // Config validation rejects uncached writes off Linux (no io_uring + // `rw_flags` there), so the mode is only ever `Buffered` here. + #[cfg(not(target_os = "linux"))] + debug_assert_eq!(write_io, SegmentIoMode::Buffered); + let mut opts = OpenOptions::new(); opts.write(true); if !file_exists { @@ -94,6 +110,8 @@ impl IggyIndexWriter { file, index_size_bytes, fsync, + #[cfg(target_os = "linux")] + write_io, }) } @@ -112,12 +130,13 @@ impl IggyIndexWriter { let len = indexes.len(); let position = self.index_size_bytes.load(Ordering::Relaxed); - let file = &self.file; - (&*file) - .write_all_at(indexes, position) - .await - .0 - .map_err(|_| IggyError::CannotSaveIndexToSegment)?; + self.write_all(indexes, position).await.map_err(|error| { + #[cfg(target_os = "linux")] + report_uncached_write_unsupported(self.write_io, &error, self.file_path.as_str()); + #[cfg(not(target_os = "linux"))] + let _ = error; + IggyError::CannotSaveIndexToSegment + })?; if self.fsync { self.fsync().await?; @@ -156,11 +175,27 @@ impl IggyIndexWriter { .map_err(|_| IggyError::CannotWriteToFile)?; Ok(()) } + + /// Uncached writes take our own `io_uring` op: compio's carries no + /// `rw_flags`, and `RWF_DONTCACHE` is a per-write flag, not an open flag. + async fn write_all(&self, indexes: Vec<u8>, position: u64) -> std::io::Result<()> { + #[cfg(target_os = "linux")] + if self.write_io == SegmentIoMode::Uncached { + return write_all_at_uncached(&self.file.to_shared_fd(), indexes, position) + .await + .0; + } + (&self.file).write_all_at(indexes, position).await.0 + } } #[cfg(test)] mod tests { use super::*; + #[cfg(target_os = "linux")] + use crate::uncached_test_support::{tmpfs_scratch_dir, uncached_scratch_dir}; + + const INDEX_ENTRY_LEN: usize = 24; #[compio::test] async fn given_seeded_size_diverging_from_disk_when_opening_existing_file_should_return_size_mismatch_error() @@ -173,6 +208,7 @@ mod tests { path.to_str().unwrap(), Rc::new(AtomicU64::new(32)), false, + SegmentIoMode::Buffered, true, ) .await; @@ -182,4 +218,92 @@ mod tests { Err(IggyError::SegmentSizeMismatchAtOpen(96, 32)) )); } + + #[cfg(target_os = "linux")] + async fn open_writer( + directory: &tempfile::TempDir, + name: &str, + write_io: SegmentIoMode, + size: &Rc<AtomicU64>, + ) -> IggyIndexWriter { + let path = directory.path().join(name); + IggyIndexWriter::new( + path.to_str().expect("utf-8 path"), + Rc::clone(size), + false, + write_io, + false, + ) + .await + .expect("open index writer") + } + + #[cfg(target_os = "linux")] + #[compio::test] + async fn given_uncached_mode_when_saving_indexes_should_append_entries_and_advance_cursor() { + let Some(directory) = uncached_scratch_dir().await else { + return; + }; + let size = Rc::new(AtomicU64::new(0)); + let writer = open_writer(&directory, "segment.index", SegmentIoMode::Uncached, &size).await; + + // Three entries appended one flush at a time, the way the partition + // writes them: each append lands in a partially filled block. + let mut expected = Vec::new(); + for entry in 1u8..=3 { + let bytes = vec![entry; INDEX_ENTRY_LEN]; + let saved = writer.save_indexes(bytes.clone()).await.unwrap(); + writer.advance(saved); + expected.extend_from_slice(&bytes); + } + + assert_eq!(size.load(Ordering::Relaxed), expected.len() as u64); + assert_eq!( + std::fs::read(directory.path().join("segment.index")).unwrap(), + expected + ); + } + + /// The negative control: tmpfs refuses `RWF_DONTCACHE` while taking the + /// identical buffered write, so only a submission that really carries the + /// flag can fail here. Without this, deleting the uncached branch of + /// `write_all` leaves every other test in this file green. + #[cfg(target_os = "linux")] + #[compio::test] + async fn given_tmpfs_when_saving_indexes_should_fail_uncached_but_succeed_buffered() { + let Some(directory) = tmpfs_scratch_dir() else { + return; + }; + let entry = vec![9u8; INDEX_ENTRY_LEN]; + + let uncached_size = Rc::new(AtomicU64::new(0)); + let uncached = open_writer( + &directory, + "uncached.index", + SegmentIoMode::Uncached, + &uncached_size, + ) + .await; + let error = uncached.save_indexes(entry.clone()).await.expect_err( + "tmpfs must reject RWF_DONTCACHE; a write that succeeds here never carried the flag", + ); + assert!( + matches!(error, IggyError::CannotSaveIndexToSegment), + "{error}" + ); + + let buffered_size = Rc::new(AtomicU64::new(0)); + let buffered = open_writer( + &directory, + "buffered.index", + SegmentIoMode::Buffered, + &buffered_size, + ) + .await; + let saved = buffered + .save_indexes(entry.clone()) + .await + .expect("the same directory takes buffered writes"); + assert_eq!(saved, entry.len() as u64); + } } diff --git a/core/partitions/src/iggy_partition.rs b/core/partitions/src/iggy_partition.rs index 3cbdf3e17..858df6879 100644 --- a/core/partitions/src/iggy_partition.rs +++ b/core/partitions/src/iggy_partition.rs @@ -4095,6 +4095,7 @@ where &messages_path, messages_size_bytes, enforce_fsync, + config.write_io, false, preallocate_segments.then_some(segment_size), ) @@ -4107,9 +4108,15 @@ where .ok_or_else(|| IggyError::CannotCreateSegmentIndexFile(index_path.clone()))? .size_counter(); let index_writer = Rc::new( - IggyIndexWriter::new(&index_path, index_size_bytes, enforce_fsync, false) - .await - .map_err(|_| IggyError::CannotCreateSegmentIndexFile(index_path.clone()))?, + IggyIndexWriter::new( + &index_path, + index_size_bytes, + enforce_fsync, + config.write_io, + false, + ) + .await + .map_err(|_| IggyError::CannotCreateSegmentIndexFile(index_path.clone()))?, ); let old_storage = &mut self.log.storages_mut()[old_segment_index]; @@ -4365,6 +4372,7 @@ where &messages_path, messages_size_bytes, enforce_fsync, + config.write_io, false, preallocate_segments.then_some(segment_size), ) @@ -4377,9 +4385,15 @@ where .ok_or_else(|| IggyError::CannotCreateSegmentIndexFile(index_path.clone()))? .size_counter(); let index_writer = Rc::new( - IggyIndexWriter::new(&index_path, index_size_bytes, enforce_fsync, false) - .await - .map_err(|_| IggyError::CannotCreateSegmentIndexFile(index_path.clone()))?, + IggyIndexWriter::new( + &index_path, + index_size_bytes, + enforce_fsync, + config.write_io, + false, + ) + .await + .map_err(|_| IggyError::CannotCreateSegmentIndexFile(index_path.clone()))?, ); self.log .add_persisted_segment(segment, storage, Some(messages_writer), Some(index_writer)); @@ -5382,6 +5396,7 @@ mod tests { use iggy_binary_protocol::{Command, ReplyHeader, WireConsumer, WireEncode}; use message_bus::{BusMessage, SendError}; use server_common::MESSAGE_ALIGN; + use server_common::segment_io::SegmentIoMode; use server_common::send_messages::{ COMMAND_HEADER_SIZE, IggyMessage, IggyMessageHeader, IggyMessages, SendMessagesOwned, }; @@ -6949,6 +6964,13 @@ mod tests { } pub(super) fn repair_config() -> PartitionsConfig { + repair_config_with_io(SegmentIoMode::Buffered) + } + + /// Most fixtures run on `std::env::temp_dir()`, which is tmpfs on many + /// boxes and refuses `RWF_DONTCACHE`; only a fixture that placed its + /// segments on a supported filesystem may ask for `Uncached`. + pub(super) fn repair_config_with_io(write_io: SegmentIoMode) -> PartitionsConfig { PartitionsConfig { messages_required_to_save: 1, size_of_messages_required_to_save: IggyByteSize::from(1024 * 1024), @@ -6956,6 +6978,7 @@ mod tests { validate_checksum: true, segment_size: IggyByteSize::from(1024 * 1024), preallocate_segments: false, + write_io, encryptor: None, path_layout: crate::PartitionPathLayout::default(), } @@ -7675,13 +7698,25 @@ mod tests { async fn new(log_path: &str, index_path: &str) -> Self { let log_cursor = Rc::new(AtomicU64::new(0)); let index_cursor = Rc::new(AtomicU64::new(0)); - let messages_writer = - MessagesWriter::new(log_path, log_cursor.clone(), true, false, None) - .await - .expect("open segment log writer"); - let index_writer = IggyIndexWriter::new(index_path, index_cursor.clone(), true, false) - .await - .expect("open segment index writer"); + let messages_writer = MessagesWriter::new( + log_path, + log_cursor.clone(), + true, + SegmentIoMode::Buffered, + false, + None, + ) + .await + .expect("open segment log writer"); + let index_writer = IggyIndexWriter::new( + index_path, + index_cursor.clone(), + true, + SegmentIoMode::Buffered, + false, + ) + .await + .expect("open segment index writer"); let mut partition = test_partition(); partition.log.add_persisted_segment( @@ -7875,6 +7910,7 @@ mod tests { log_path.to_str().expect("utf-8 path"), log_cursor, true, + SegmentIoMode::Buffered, false, None, ) @@ -7882,9 +7918,10 @@ mod tests { .expect("open segment log writer"); // The index half writes to a device that is always full, so the // persist fails the way a full disk fails. - let index_writer = IggyIndexWriter::new(DEV_FULL, index_cursor, true, false) - .await - .expect("open segment index writer"); + let index_writer = + IggyIndexWriter::new(DEV_FULL, index_cursor, true, SegmentIoMode::Buffered, false) + .await + .expect("open segment index writer"); let mut partition = test_partition(); partition.log.add_persisted_segment( @@ -7968,6 +8005,7 @@ mod tests { index_path.to_str().expect("utf-8 path"), fixture.index_cursor.clone(), true, + SegmentIoMode::Buffered, false, ) .await @@ -8836,3 +8874,177 @@ mod purge_floor_tests { let _ = std::fs::remove_dir_all(&dir); } } + +/// Partition-level coverage for the uncached (`RWF_DONTCACHE`) segment write +/// path. Every other partition fixture pins `Buffered`, so nothing above the +/// leaf writer methods would otherwise drive rotation or the index-failure +/// rewind with the flag set. +#[cfg(all(test, target_os = "linux"))] +mod uncached_disk_tests { + use super::tests::{journal_send_batch, repair_config_with_io, test_partition}; + use super::*; + use crate::uncached_test_support::{tmpfs_scratch_dir, uncached_scratch_dir}; + use server_common::segment_io::SegmentIoMode; + use tempfile::TempDir; + + /// A partition whose segment 0 is backed by real writers in `write_io` + /// mode. The index path is a parameter: putting it on tmpfs is the only + /// deterministic way to make an uncached index save fail while the + /// uncached batch write next to it succeeds. Returns the shared segment + /// write cursor so a caller can observe a rewind. + async fn disk_partition( + directory: &TempDir, + index_path: &str, + write_io: SegmentIoMode, + max_segment_size: IggyByteSize, + ) -> (IggyPartition<IggyMessageBus>, Rc<AtomicU64>) { + let partition_dir = directory.path().to_string_lossy().into_owned(); + let messages_path = format!("{partition_dir}/{:0>20}.log", 0u64); + + let mut partition = test_partition(); + partition.set_partition_dir(partition_dir); + partition.log.segments_mut()[0].max_size = max_segment_size; + + let storage = SegmentStorage::new(&messages_path, index_path, 0, 0, false) + .await + .expect("create segment storage"); + let messages_size = storage + .messages_writer + .as_ref() + .expect("storage opens a messages writer") + .size_counter(); + let index_size = storage + .index_writer + .as_ref() + .expect("storage opens an index writer") + .size_counter(); + partition.log.storages_mut()[0] = storage; + partition.log.messages_writers_mut()[0] = Some(Rc::new( + MessagesWriter::new( + &messages_path, + Rc::clone(&messages_size), + false, + write_io, + false, + None, + ) + .await + .expect("open messages writer"), + )); + partition.log.index_writers_mut()[0] = Some(Rc::new( + IggyIndexWriter::new(index_path, index_size, false, write_io, false) + .await + .expect("open index writer"), + )); + (partition, messages_size) + } + + fn file_len(path: &std::path::Path) -> u64 { + std::fs::metadata(path) + .unwrap_or_else(|error| panic!("stat {}: {error}", path.display())) + .len() + } + + #[compio::test] + async fn given_uncached_mode_when_committing_past_the_cap_should_rotate_and_append_on_disk() { + let Some(directory) = uncached_scratch_dir().await else { + return; + }; + let index_path = format!("{}/{:0>20}.index", directory.path().display(), 0u64); + // A one-byte cap makes every committed batch fill the segment, so the + // flush loop rotates between the two batches and the second one is + // appended through writers the rotation itself created. + let (mut partition, _) = Box::pin(disk_partition( + &directory, + &index_path, + SegmentIoMode::Uncached, + IggyByteSize::from(1u64), + )) + .await; + + journal_send_batch(&mut partition, 1).await; + journal_send_batch(&mut partition, 2).await; + partition.consensus().advance_commit_max(2); + partition + .commit_journal(&repair_config_with_io(SegmentIoMode::Uncached)) + .await; + + // The cap is read once per flush, so both batches trip the eager + // seal: two written segments plus the empty one the last seal opened. + assert_eq!( + partition.log.segments().len(), + 3, + "each batch must fill and seal its segment" + ); + assert!( + partition.log.segments()[0].sealed && partition.log.segments()[1].sealed, + "rotation must seal the segments it leaves behind" + ); + for start_offset in [0u64, 1] { + let log = directory.path().join(format!("{start_offset:0>20}.log")); + let index = directory.path().join(format!("{start_offset:0>20}.index")); + assert!(file_len(&log) > 0, "segment {start_offset} log is empty"); + assert!( + file_len(&index) > 0, + "segment {start_offset} index is empty" + ); + } + } + + /// The index save is forced to fail by keeping the index file on tmpfs + /// while the log sits on a filesystem that takes the flag: tmpfs refuses + /// an uncached write, the log's filesystem does not. Only the uncached + /// path can produce that split, so this drives the rewind AND proves the + /// branch ran. + #[compio::test] + async fn given_uncached_mode_when_index_save_fails_should_rewind_the_write_cursor() { + let Some(directory) = uncached_scratch_dir().await else { + return; + }; + let Some(tmpfs) = tmpfs_scratch_dir() else { + return; + }; + let index_path = tmpfs + .path() + .join("segment.index") + .to_string_lossy() + .into_owned(); + let (mut partition, messages_size) = Box::pin(disk_partition( + &directory, + &index_path, + SegmentIoMode::Uncached, + IggyByteSize::from(1024 * 1024_u64), + )) + .await; + + journal_send_batch(&mut partition, 1).await; + partition.consensus().advance_commit_max(1); + // The commit path treats a persist failure as fatal and panics, so + // drive the flush directly to observe what it left behind. + partition + .flush_committed_messages(&repair_config_with_io(SegmentIoMode::Uncached)) + .await + .expect_err("the uncached index write on tmpfs must be refused"); + + let log = directory.path().join(format!("{:0>20}.log", 0u64)); + assert!( + file_len(&log) > 0, + "the batch bytes must land before the index save fails" + ); + assert_eq!( + messages_size.load(Ordering::Relaxed), + 0, + "a failed index save must rewind the segment write cursor" + ); + assert_eq!( + partition.log.active_segment().size.as_bytes_u64(), + 0, + "a failed persist must not stamp the segment size" + ); + assert_eq!( + file_len(std::path::Path::new(&index_path)), + 0, + "the refused index write must leave the file empty" + ); + } +} diff --git a/core/partitions/src/lib.rs b/core/partitions/src/lib.rs index 818872029..ddedef21e 100644 --- a/core/partitions/src/lib.rs +++ b/core/partitions/src/lib.rs @@ -51,6 +51,17 @@ pub use types::{ RepairConclusion, RepairSession, SendMessagesResult, }; +#[cfg(target_os = "linux")] +use nix::libc::EOPNOTSUPP; +#[cfg(target_os = "linux")] +use server_common::segment_io::SegmentIoMode; +#[cfg(target_os = "linux")] +use std::io; +#[cfg(target_os = "linux")] +use std::sync::Once; +#[cfg(target_os = "linux")] +use tracing::error; + /// A partition's message log, named so a caller can carry one across a rebuild. /// /// Exists for the simulator, which has no segment files and so must hold the log @@ -105,3 +116,142 @@ pub trait Partition { PartitionOffsets::default() } } + +#[cfg(target_os = "linux")] +const WRITE_IO_CONFIG_KEY: &str = "system.segment.write_io"; + +/// Names the `write_io` knob when a segment write is rejected because the +/// filesystem underneath it cannot serve `RWF_DONTCACHE`. +/// +/// The boot probe only reaches directories that exist at boot, so a per-stream +/// submount added later fails here instead, with an errno that says nothing +/// about the config that asked for the flag. Deliberately not fatal: one +/// stream's mount must not take the node down. Reported once per process, +/// because the remedy is a config edit plus a restart and a per-commit line +/// would only bury it. +#[cfg(target_os = "linux")] +pub(crate) fn report_uncached_write_unsupported( + write_io: SegmentIoMode, + error: &io::Error, + file_path: &str, +) { + static REPORTED: Once = Once::new(); + + if !is_uncached_write_unsupported(write_io, error) { + return; + } + + REPORTED.call_once(|| { + error!( + target: "iggy.partitions.storage", + file = file_path, + config_key = WRITE_IO_CONFIG_KEY, + "segment write rejected RWF_DONTCACHE, so this filesystem cannot serve \ + {WRITE_IO_CONFIG_KEY} = \"uncached\": set {WRITE_IO_CONFIG_KEY} = \"buffered\", \ + or keep segments on ext4/XFS under Linux >= 6.14" + ); + }); +} + +#[cfg(target_os = "linux")] +fn is_uncached_write_unsupported(write_io: SegmentIoMode, error: &io::Error) -> bool { + write_io == SegmentIoMode::Uncached && error.raw_os_error() == Some(EOPNOTSUPP) +} + +/// Shared fixtures for the tests that exercise the uncached (`RWF_DONTCACHE`) +/// segment write path. +#[cfg(all(test, target_os = "linux"))] +mod uncached_test_support { + use nix::sys::statfs; + use server_common::uncached_io::{ + UncachedIoError, probe_uncached_write, require_uncached_io_tests, + }; + use std::path::Path; + use tempfile::TempDir; + + const SHM_DIR: &str = "/dev/shm"; + const REQUIRE_ENV: &str = "IGGY_REQUIRE_UNCACHED_IO"; + + /// A directory whose filesystem takes `RWF_DONTCACHE`, or `None` with the + /// reason reported. `tempdir()` lands on tmpfs on many boxes and tmpfs + /// refuses the flag, so the fixture sits next to the test binary instead. + pub async fn uncached_scratch_dir() -> Option<TempDir> { + let exe = std::env::current_exe().expect("current_exe"); + let base = exe.parent().expect("test binary has a parent directory"); + let directory = tempfile::tempdir_in(base).expect("scratch directory next to the binary"); + match probe_uncached_write(directory.path()).await { + Ok(()) => Some(directory), + Err(error @ UncachedIoError::Unsupported { .. }) => { + skip_or_fail(&error.to_string()); + None + } + Err(error) => panic!("uncached probe hit an I/O error: {error}"), + } + } + + /// A tmpfs directory: the negative control for the uncached path, since + /// tmpfs is the filesystem guaranteed to refuse `RWF_DONTCACHE` while + /// taking the very same buffered write. + pub fn tmpfs_scratch_dir() -> Option<TempDir> { + let shm = Path::new(SHM_DIR); + let is_tmpfs = + statfs::statfs(shm).is_ok_and(|stat| stat.filesystem_type() == statfs::TMPFS_MAGIC); + if !is_tmpfs { + skip_or_fail(&format!("{SHM_DIR} is not tmpfs")); + return None; + } + match tempfile::tempdir_in(shm) { + Ok(directory) => Some(directory), + Err(error) => { + skip_or_fail(&format!( + "cannot create a directory under {SHM_DIR}: {error}" + )); + None + } + } + } + + /// Loud enough to spot in a scrolling CI log, and a hard failure wherever + /// the uncached path is required to have run. A silent skip on a build + /// tree that sits on overlayfs is how this whole suite goes green without + /// ever submitting an uncached write. + pub fn skip_or_fail(reason: &str) { + assert!( + !require_uncached_io_tests(), + "### {REQUIRE_ENV}=1 forbids skipping: {reason} ###" + ); + eprintln!("######## SKIPPING UNCACHED TEST: {reason} ########"); + } +} + +#[cfg(all(test, target_os = "linux"))] +mod tests { + use super::is_uncached_write_unsupported; + use nix::libc::{EIO, EOPNOTSUPP}; + use server_common::segment_io::SegmentIoMode; + use std::io; + + #[test] + fn given_uncached_mode_when_write_is_rejected_as_unsupported_should_blame_the_knob() { + assert!(is_uncached_write_unsupported( + SegmentIoMode::Uncached, + &io::Error::from_raw_os_error(EOPNOTSUPP) + )); + } + + #[test] + fn given_buffered_mode_when_write_is_rejected_as_unsupported_should_stay_silent() { + assert!(!is_uncached_write_unsupported( + SegmentIoMode::Buffered, + &io::Error::from_raw_os_error(EOPNOTSUPP) + )); + } + + #[test] + fn given_uncached_mode_when_write_fails_for_another_reason_should_stay_silent() { + assert!(!is_uncached_write_unsupported( + SegmentIoMode::Uncached, + &io::Error::from_raw_os_error(EIO) + )); + } +} diff --git a/core/partitions/src/messages_writer.rs b/core/partitions/src/messages_writer.rs index 63ad93c11..753087a27 100644 --- a/core/partitions/src/messages_writer.rs +++ b/core/partitions/src/messages_writer.rs @@ -21,14 +21,21 @@ use compio::{ }; use iggy_common::{IggyByteSize, IggyError}; use server_common::iobuf::{Frozen, IOV_MAX}; +use server_common::segment_io::SegmentIoMode; use std::{ rc::Rc, sync::atomic::{AtomicU64, Ordering}, }; use tracing::{error, warn}; +#[cfg(target_os = "linux")] +use crate::report_uncached_write_unsupported; +#[cfg(target_os = "linux")] +use compio::driver::ToSharedFd; #[cfg(target_os = "linux")] use nix::fcntl::{FallocateFlags, fallocate}; +#[cfg(target_os = "linux")] +use server_common::uncached_io::write_vectored_all_at_uncached; #[derive(Debug)] pub struct MessagesWriter { @@ -36,6 +43,8 @@ pub struct MessagesWriter { file: File, messages_size_bytes: Rc<AtomicU64>, fsync: bool, + #[cfg(target_os = "linux")] + write_io: SegmentIoMode, } impl MessagesWriter { @@ -49,9 +58,15 @@ impl MessagesWriter { file_path: &str, messages_size_bytes: Rc<AtomicU64>, fsync: bool, + write_io: SegmentIoMode, file_exists: bool, preallocate_size: Option<IggyByteSize>, ) -> Result<Self, IggyError> { + // Config validation rejects uncached writes off Linux (no io_uring + // `rw_flags` there), so the mode is only ever `Buffered` here. + #[cfg(not(target_os = "linux"))] + debug_assert_eq!(write_io, SegmentIoMode::Buffered); + let mut opts = OpenOptions::new(); opts.write(true); if !file_exists { @@ -99,6 +114,8 @@ impl MessagesWriter { file, messages_size_bytes, fsync, + #[cfg(target_os = "linux")] + write_io, }) } @@ -121,7 +138,7 @@ impl MessagesWriter { } let position = self.messages_size_bytes.load(Ordering::Relaxed); - write_frozen_chunked(&self.file, &self.file_path, position, buffers).await?; + self.write_frozen_chunked(position, buffers).await?; if self.fsync { self.fsync().await?; @@ -158,6 +175,52 @@ impl MessagesWriter { .map_err(|_| IggyError::CannotWriteToFile)?; Ok(()) } + + async fn write_frozen_chunked<const ALIGN: usize>( + &self, + mut position: u64, + buffers: &[Frozen<ALIGN>], + ) -> Result<(), IggyError> { + for chunk in buffers.chunks(IOV_MAX) { + let chunk_size: usize = chunk.iter().map(Frozen::len).sum(); + let chunk_vec: Vec<_> = chunk.to_vec(); + + self.write_vectored_all(chunk_vec, position) + .await + .map_err(|err| { + #[cfg(target_os = "linux")] + report_uncached_write_unsupported(self.write_io, &err, self.file_path.as_str()); + error!( + target: "iggy.partitions.storage", + file = self.file_path.as_str(), + write_position = position, + %err, + "failed to write frozen messages to segment file" + ); + IggyError::CannotWriteToFile + })?; + + position += chunk_size as u64; + } + + Ok(()) + } + + /// Uncached writes take our own `io_uring` op: compio's carries no + /// `rw_flags`, and `RWF_DONTCACHE` is a per-write flag, not an open flag. + async fn write_vectored_all<const ALIGN: usize>( + &self, + chunk: Vec<Frozen<ALIGN>>, + position: u64, + ) -> std::io::Result<()> { + #[cfg(target_os = "linux")] + if self.write_io == SegmentIoMode::Uncached { + return write_vectored_all_at_uncached(&self.file.to_shared_fd(), chunk, position) + .await + .0; + } + (&self.file).write_vectored_all_at(chunk, position).await.0 + } } #[cfg(target_os = "linux")] @@ -209,40 +272,13 @@ fn preallocate_file(_file: &File, file_path: &str, _len: u64) { ); } -async fn write_frozen_chunked<const ALIGN: usize>( - file: &File, - file_path: &str, - mut position: u64, - buffers: &[Frozen<ALIGN>], -) -> Result<(), IggyError> { - for chunk in buffers.chunks(IOV_MAX) { - let chunk_size: usize = chunk.iter().map(Frozen::len).sum(); - let chunk_vec: Vec<_> = chunk.to_vec(); - - let (result, _) = (&*file) - .write_vectored_all_at(chunk_vec, position) - .await - .into(); - result.map_err(|err| { - error!( - target: "iggy.partitions.storage", - file = file_path, - write_position = position, - %err, - "failed to write frozen messages to segment file" - ); - IggyError::CannotWriteToFile - })?; - - position += chunk_size as u64; - } - - Ok(()) -} - #[cfg(test)] mod tests { use super::*; + #[cfg(target_os = "linux")] + use crate::uncached_test_support::{tmpfs_scratch_dir, uncached_scratch_dir}; + #[cfg(target_os = "linux")] + use server_common::iobuf::Owned; #[compio::test] async fn preallocated_file_keeps_logical_length() { @@ -252,6 +288,7 @@ mod tests { path.to_str().unwrap(), Rc::new(AtomicU64::new(0)), false, + SegmentIoMode::Buffered, false, Some(IggyByteSize::from(1024 * 1024_u64)), ) @@ -272,6 +309,7 @@ mod tests { path.to_str().unwrap(), Rc::new(AtomicU64::new(129)), false, + SegmentIoMode::Buffered, true, None, ) @@ -282,4 +320,103 @@ mod tests { Err(IggyError::SegmentSizeMismatchAtOpen(128, 129)) )); } + + /// Unaligned lengths with per-offset content, so a misplaced or repeated + /// write cannot pass a byte-exact comparison. + #[cfg(target_os = "linux")] + fn frozen_batch(len: usize, seed: usize) -> Frozen<4096> { + let bytes: Vec<u8> = (1u8..=251).cycle().skip(seed).take(len).collect(); + Frozen::from(Owned::<4096>::copy_from_slice(&bytes)) + } + + #[cfg(target_os = "linux")] + async fn open_writer( + directory: &tempfile::TempDir, + name: &str, + write_io: SegmentIoMode, + size: &Rc<AtomicU64>, + ) -> MessagesWriter { + let path = directory.path().join(name); + MessagesWriter::new( + path.to_str().expect("utf-8 path"), + Rc::clone(size), + false, + write_io, + false, + None, + ) + .await + .expect("open messages writer") + } + + #[cfg(target_os = "linux")] + #[compio::test] + async fn given_uncached_mode_when_saving_batches_should_persist_bytes_and_advance_cursor() { + let Some(directory) = uncached_scratch_dir().await else { + return; + }; + let size = Rc::new(AtomicU64::new(0)); + let writer = open_writer(&directory, "segment.log", SegmentIoMode::Uncached, &size).await; + + let batches: Vec<Frozen<4096>> = [300usize, 4113, 70_000] + .iter() + .enumerate() + .map(|(seed, len)| frozen_batch(*len, seed)) + .collect(); + let expected: Vec<u8> = batches + .iter() + .flat_map(|batch| batch.as_slice().to_vec()) + .collect(); + + let saved = writer.save_frozen_batches(&batches).await.unwrap(); + writer.advance(saved.as_bytes_u64()); + + assert_eq!(saved.as_bytes_u64(), expected.len() as u64); + assert_eq!(size.load(Ordering::Relaxed), expected.len() as u64); + assert_eq!( + std::fs::read(directory.path().join("segment.log")).unwrap(), + expected + ); + } + + /// The negative control: tmpfs refuses `RWF_DONTCACHE` while taking the + /// identical buffered write, so only a submission that really carries the + /// flag can fail here. Without this, deleting the uncached branch of + /// `write_vectored_all` leaves every other test in this file green. + #[cfg(target_os = "linux")] + #[compio::test] + async fn given_tmpfs_when_saving_batches_should_fail_uncached_but_succeed_buffered() { + let Some(directory) = tmpfs_scratch_dir() else { + return; + }; + let batches = vec![frozen_batch(300, 0), frozen_batch(4113, 1)]; + let total: u64 = batches.iter().map(|batch| batch.len() as u64).sum(); + + let uncached_size = Rc::new(AtomicU64::new(0)); + let uncached = open_writer( + &directory, + "uncached.log", + SegmentIoMode::Uncached, + &uncached_size, + ) + .await; + let error = uncached.save_frozen_batches(&batches).await.expect_err( + "tmpfs must reject RWF_DONTCACHE; a write that succeeds here never carried the flag", + ); + assert!(matches!(error, IggyError::CannotWriteToFile), "{error}"); + + let buffered_size = Rc::new(AtomicU64::new(0)); + let buffered = open_writer( + &directory, + "buffered.log", + SegmentIoMode::Buffered, + &buffered_size, + ) + .await; + let saved = buffered + .save_frozen_batches(&batches) + .await + .expect("the same directory takes buffered writes"); + assert_eq!(saved.as_bytes_u64(), total); + } } diff --git a/core/partitions/src/state_transfer.rs b/core/partitions/src/state_transfer.rs index 7444b9d76..b64d28743 100644 --- a/core/partitions/src/state_transfer.rs +++ b/core/partitions/src/state_transfer.rs @@ -2579,6 +2579,7 @@ where &messages_reader.path(), messages_w.size_counter(), enforce_fsync, + config.write_io, true, preallocate_segments.then_some(segment_size), ) @@ -2591,6 +2592,7 @@ where &index_reader.path(), index_w.size_counter(), enforce_fsync, + config.write_io, true, ) .await diff --git a/core/partitions/src/types.rs b/core/partitions/src/types.rs index 4762b772a..d6fd5f7cc 100644 --- a/core/partitions/src/types.rs +++ b/core/partitions/src/types.rs @@ -18,6 +18,7 @@ use iggy_binary_protocol::Operation; use iggy_common::{EncryptorKind, IggyByteSize, PollingStrategy}; use server_common::iobuf::Frozen; +use server_common::segment_io::SegmentIoMode; use smallvec::SmallVec; use std::sync::Arc; @@ -356,6 +357,9 @@ pub struct PartitionsConfig { pub segment_size: IggyByteSize, /// Whether local message files reserve the configured segment size on open. pub preallocate_segments: bool, + /// Page-cache policy for segment `.log` and `.index` writes; host-wide + /// because it depends on the kernel and filesystem under `system.path`. + pub write_io: SegmentIoMode, /// Server-side at-rest encryption. Applied ONCE, on the primary at /// ingestion, so the ciphertext replicates verbatim: every replica /// journals, acks, and persists identical bytes (checksums and the diff --git a/core/server/config.toml b/core/server/config.toml index 5584b0378..15879a3ac 100644 --- a/core/server/config.toml +++ b/core/server/config.toml @@ -478,6 +478,35 @@ validate_checksum = true # Unsupported: setting this to `true` aborts boot. archive_expired = false +# How segment `.log` and `.index` writes go through the kernel (string). +# `buffered` keeps the kernel's page-cache behaviour: writes land in the page +# cache and writeback runs on the kernel's schedule. +# `uncached` sets RWF_DONTCACHE on every segment write: writeback is kicked at +# once, and a page leaves the page cache when that writeback COMPLETES, only +# where the flush wrote that whole 4 KiB block (a 1 MiB flush leaves nothing +# resident). With `enforce_fsync = false` nothing waits on that completion, so +# the pages linger until the bdi flusher reaches them - 15-30 s under stock +# `dirty_writeback_centisecs` / `dirty_expire_centisecs`, longer on a loaded +# box. Small appends stay fully resident just as with `buffered`, so the +# default flush sizes get no page-cache saving at all. Dirty pages still peak +# during a burst either way: the flag starts writeback, it does not make the +# device keep up with the writer. +# Cost, for the flushes whose pages do leave: a consumer reading more than one +# flush behind the producer is served from the device instead of the page +# cache, and each newly allocated 4 KiB block costs one synchronous read-back +# on the append that follows its first writeback. +# Measured on ext4 (XFS not yet measured). With `enforce_fsync = true`, +# `uncached` was never slower (p50 neutral to 9% faster in a micro-bench). +# Without fsync, tiny flushes (small `messages_required_to_save`) are a clear +# loss: about 9x the device writes for 300 B appends and about 190 us more at +# p99, because every append re-dirties a page the flusher just cleaned. Keep +# the default flush thresholds or enable fsync. +# `uncached` needs Linux >= 6.14 (>= 7.2 recommended: older kernels submit the +# writeback inline in the writer) and a filesystem that supports it (ext4, +# XFS; not btrfs, tmpfs, overlayfs, FUSE). The server refuses to start when +# the mode is requested but unsupported. +write_io = "buffered" + # Recovery configuration in case of lost data [system.recovery] # Controls whether streams/topics/partitions should be recreated if the expected data for existing state is missing (boolean). diff --git a/core/server/src/boot/mod.rs b/core/server/src/boot/mod.rs index 841b3e3ab..e875687d6 100644 --- a/core/server/src/boot/mod.rs +++ b/core/server/src/boot/mod.rs @@ -83,6 +83,10 @@ use server_common::Message; use server_common::bootstrap::create_directories; use server_common::fs_utils::remove_dir_all; use server_common::log::{Logging, LoggingSettings, TelemetrySettings}; +#[cfg(target_os = "linux")] +use server_common::segment_io::SegmentIoMode; +#[cfg(target_os = "linux")] +use server_common::uncached_io::{probe_uncached_write, uncached_write_kernel_warning}; use shard::metrics::{ShardMetrics, frame_drop_reason, frame_drop_variant}; use shard::{ LifecycleFrame, Receiver as ShardReceiver, ShardFrame, TaggedSender, channel, @@ -150,13 +154,17 @@ pub async fn load_config() -> Result<ServerConfig, ServerError> { /// /// # Errors /// -/// Returns an error if the wipe, directory preparation, or logging setup -/// fails. +/// Returns an error if the uncached write probe, the wipe, directory +/// preparation, or logging setup fails. pub async fn prepare_runtime_dirs( config: &ServerConfig, logging: &mut Logging, fresh: bool, ) -> Result<(), ServerError> { + // Ahead of the wipe: a write mode this box cannot honour must cost the + // operator a refused boot, never their data. + #[cfg(target_os = "linux")] + ensure_uncached_write_support(config).await?; if fresh { wipe_system_path(config).await?; } @@ -175,10 +183,59 @@ pub async fn prepare_runtime_dirs( &TelemetrySettings::from(&config.telemetry), ) .map_err(ServerError::Logging)?; + #[cfg(target_os = "linux")] + warn_on_inline_writeback_kick(config); Ok(()) } +/// Refuses boot when `system.segment.write_io = "uncached"` cannot work where +/// the segments will live: one probe write here beats the first segment flush +/// failing after clients are connected. +#[cfg(target_os = "linux")] +async fn ensure_uncached_write_support(config: &ServerConfig) -> Result<(), ServerError> { + if config.system.segment.write_io != SegmentIoMode::Uncached { + return Ok(()); + } + let path = segment_probe_dir(config); + probe_uncached_write(&path).await.map_err(|source| { + // The only channel the operator sees: this runs before `late_init`, + // where tracing still writes to a null stdout and to an in-memory + // buffer a refused boot never flushes, and `main` prints the returned + // error through `Debug`, which drops every `Display` message. + eprintln!( + "system.segment.write_io = \"uncached\" is unusable under {}: {source}", + path.display() + ); + ServerError::UncachedWriteUnsupported { path, source } + }) +} + +/// Deepest existing ancestor of the streams path, which is where the segment +/// files land. Boot has created nothing yet, and a submount can only attach to +/// a directory that already exists, so no unprobed filesystem can slip in +/// between this directory and the segments created under it later. +#[cfg(target_os = "linux")] +fn segment_probe_dir(config: &ServerConfig) -> PathBuf { + PathBuf::from(config.system.get_streams_path()) + .ancestors() + .find(|ancestor| ancestor.is_dir()) + .map_or_else(|| PathBuf::from("."), Path::to_path_buf) +} + +/// Kernels between 6.14 and 7.2 take `RWF_DONTCACHE` but kick writeback inline +/// on the writing shard. Reported after `late_init` so the warning reaches the +/// operator's stdout instead of the pre-boot buffer. +#[cfg(target_os = "linux")] +fn warn_on_inline_writeback_kick(config: &ServerConfig) { + if config.system.segment.write_io != SegmentIoMode::Uncached { + return; + } + if let Some(warning) = uncached_write_kernel_warning() { + warn!("{warning}"); + } +} + /// Delete the configured system path so the server boots on empty state. async fn wipe_system_path(config: &ServerConfig) -> Result<(), ServerError> { let path = config.system.get_system_path(); diff --git a/core/server/src/boot/recovery.rs b/core/server/src/boot/recovery.rs index e614e688f..b2804dad2 100644 --- a/core/server/src/boot/recovery.rs +++ b/core/server/src/boot/recovery.rs @@ -111,6 +111,7 @@ pub(in crate::boot) async fn build_shard_for_thread( validate_checksum: config.system.partition.validate_checksum, segment_size: IggyByteSize::from(iggy_common::DEFAULT_SEGMENT_SIZE), preallocate_segments: iggy_common::DEFAULT_PREALLOCATE_SEGMENTS, + write_io: config.system.segment.write_io, encryptor, path_layout: partitions::PartitionPathLayout { streams_root: config.system.get_streams_path(), diff --git a/core/server/src/dispatch/mod.rs b/core/server/src/dispatch/mod.rs index 8f69e7886..62145e4b0 100644 --- a/core/server/src/dispatch/mod.rs +++ b/core/server/src/dispatch/mod.rs @@ -1099,6 +1099,7 @@ mod tests { validate_checksum: true, segment_size: iggy_common::IggyByteSize::from(1_048_576_u64), preallocate_segments: false, + write_io: server_common::segment_io::SegmentIoMode::Buffered, encryptor: None, path_layout: PartitionPathLayout::default(), }, diff --git a/core/server/src/dispatch/partition.rs b/core/server/src/dispatch/partition.rs index b2e0789bf..5850cdff5 100644 --- a/core/server/src/dispatch/partition.rs +++ b/core/server/src/dispatch/partition.rs @@ -1326,6 +1326,7 @@ mod tests { validate_checksum: true, segment_size: iggy_common::IggyByteSize::from(1_048_576_u64), preallocate_segments: false, + write_io: server_common::segment_io::SegmentIoMode::Buffered, encryptor: None, path_layout: PartitionPathLayout::default(), }, @@ -1452,6 +1453,7 @@ mod tests { validate_checksum: true, segment_size: iggy_common::IggyByteSize::from(1_048_576_u64), preallocate_segments: false, + write_io: server_common::segment_io::SegmentIoMode::Buffered, encryptor: None, path_layout: PartitionPathLayout::default(), }, @@ -1517,6 +1519,7 @@ mod tests { validate_checksum: true, segment_size: iggy_common::IggyByteSize::from(1_048_576_u64), preallocate_segments: false, + write_io: server_common::segment_io::SegmentIoMode::Buffered, encryptor: None, path_layout: PartitionPathLayout::default(), }, diff --git a/core/server/src/dispatch/session_ops.rs b/core/server/src/dispatch/session_ops.rs index d43bcc17b..5f39c28c4 100644 --- a/core/server/src/dispatch/session_ops.rs +++ b/core/server/src/dispatch/session_ops.rs @@ -1617,6 +1617,7 @@ mod tests { validate_checksum: true, segment_size: iggy_common::IggyByteSize::from(1_048_576_u64), preallocate_segments: false, + write_io: server_common::segment_io::SegmentIoMode::Buffered, encryptor: None, path_layout: PartitionPathLayout::default(), }, diff --git a/core/server/src/dispatch/test_support.rs b/core/server/src/dispatch/test_support.rs index e52578c64..5914b8055 100644 --- a/core/server/src/dispatch/test_support.rs +++ b/core/server/src/dispatch/test_support.rs @@ -180,6 +180,7 @@ pub fn test_shard(bus: &SpyBus, replica: u8, replica_count: u8, incarnation: u12 validate_checksum: true, segment_size: iggy_common::IggyByteSize::from(1_048_576_u64), preallocate_segments: false, + write_io: server_common::segment_io::SegmentIoMode::Buffered, encryptor: None, path_layout: PartitionPathLayout::default(), }, diff --git a/core/server/src/partition_helpers.rs b/core/server/src/partition_helpers.rs index ec39868e5..04be404d4 100644 --- a/core/server/src/partition_helpers.rs +++ b/core/server/src/partition_helpers.rs @@ -49,6 +49,7 @@ use metadata::{IdentityField, ReplicaIdentity}; use partitions::{IggyIndexWriter, IggyPartition, IggyPartitions, MessagesWriter, Segment}; use server_common::SegmentStorage; use server_common::fs_utils::remove_dir_all; +use server_common::segment_io::SegmentIoMode; use server_common::sharding::IggyNamespace; use std::path::{Path, PathBuf}; use std::rc::Rc; @@ -384,6 +385,7 @@ pub async fn ensure_initial_segment( &messages_path, messages_size_counter, enforce_fsync, + config.system.segment.write_io, false, preallocate_segments.then_some(segment_size), ) @@ -401,19 +403,25 @@ pub async fn ensure_initial_segment( })?, )), Some(Rc::new( - IggyIndexWriter::new(&index_path, index_size_counter, enforce_fsync, false) - .await - .map_err(|source| { - error!( - stream_id, - topic_id, - partition_id, - path = %index_path, - error = %source, - "failed to initialize initial sparse index writer" - ); - source - })?, + IggyIndexWriter::new( + &index_path, + index_size_counter, + enforce_fsync, + config.system.segment.write_io, + false, + ) + .await + .map_err(|source| { + error!( + stream_id, + topic_id, + partition_id, + path = %index_path, + error = %source, + "failed to initialize initial sparse index writer" + ); + source + })?, )), ); partition.stats.increment_segments_count(1); @@ -810,6 +818,7 @@ async fn load_partition( topic_id, partition_id, recovered_segments, + config.system.segment.write_io, ) .await?; @@ -913,6 +922,7 @@ async fn recover_partition_segments( /// /// Takes no `&ServerConfig`: every knob it needs is the partition's own /// resolved topic option now, which is the whole point of the per-topic move. +/// `write_io` is the one exception, a host-wide filesystem property. async fn hydrate_partition_log( partition: &mut IggyPartition<Rc<IggyMessageBus>>, partition_dir: &str, @@ -920,6 +930,7 @@ async fn hydrate_partition_log( topic_id: usize, partition_id: usize, recovered_segments: Vec<RecoveredSegment>, + write_io: SegmentIoMode, ) -> Result<(), ServerError> { // The partition's own resolved knobs, not the shard-wide config: a topic // created with `enforce_fsync` or a per-topic `segment_size` must get them @@ -966,6 +977,7 @@ async fn hydrate_partition_log( &messages_reader.path(), messages_size_counter, enforce_fsync, + write_io, true, preallocate_segments.then_some(segment_size), ) @@ -990,26 +1002,32 @@ async fn hydrate_partition_log( })?, )); partition.log.index_writers_mut()[active_index] = Some(Rc::new( - IggyIndexWriter::new(&index_path, index_size_counter, enforce_fsync, true) - .await - .map_err(|source| { - error!( - stream_id, - topic_id, - partition_id, - path = %index_path, - error = %source, - "failed to initialize persisted sparse index writer" - ); - hydrate_reopen_error( - source, - partition_dir, - stream_id, - topic_id, - partition_id, - start_offset, - ) - })?, + IggyIndexWriter::new( + &index_path, + index_size_counter, + enforce_fsync, + write_io, + true, + ) + .await + .map_err(|source| { + error!( + stream_id, + topic_id, + partition_id, + path = %index_path, + error = %source, + "failed to initialize persisted sparse index writer" + ); + hydrate_reopen_error( + source, + partition_dir, + stream_id, + topic_id, + partition_id, + start_offset, + ) + })?, )); } } diff --git a/core/server/src/partition_reconciler.rs b/core/server/src/partition_reconciler.rs index 4db62a2b7..ad25b227e 100644 --- a/core/server/src/partition_reconciler.rs +++ b/core/server/src/partition_reconciler.rs @@ -1608,6 +1608,7 @@ mod tests { validate_checksum: true, segment_size: iggy_common::IggyByteSize::from(iggy_common::DEFAULT_SEGMENT_SIZE), preallocate_segments: false, + write_io: server_common::segment_io::SegmentIoMode::Buffered, encryptor: None, path_layout: PartitionPathLayout::default(), }, diff --git a/core/server/src/server_error.rs b/core/server/src/server_error.rs index 7f0f86c5e..49d2464cd 100644 --- a/core/server/src/server_error.rs +++ b/core/server/src/server_error.rs @@ -132,6 +132,16 @@ pub enum ServerError { }, #[error("failed to initialize server logging")] Logging(#[source] LogError), + // `main` returns this enum, so boot prints it through `Debug` and no + // `Display` text here reaches the operator: `bootstrap` writes the failed + // requirement and the remedy to stderr itself. + #[cfg(target_os = "linux")] + #[error("system.segment.write_io = \"uncached\" is unusable under {path}")] + UncachedWriteUnsupported { + path: PathBuf, + #[source] + source: server_common::uncached_io::UncachedIoError, + }, #[error("failed to recover metadata snapshot and journal")] MetadataRecovery(#[source] RecoveryError), #[error("failed to open partition superblock at {dir}")] diff --git a/core/server_common/Cargo.toml b/core/server_common/Cargo.toml index ce10ca165..f28bc753c 100644 --- a/core/server_common/Cargo.toml +++ b/core/server_common/Cargo.toml @@ -65,6 +65,10 @@ tracing-opentelemetry = { workspace = true } tracing-subscriber = { workspace = true } twox-hash = { workspace = true } +[target.'cfg(target_os = "linux")'.dependencies] +io-uring = { workspace = true } +libc = { workspace = true } + [target.'cfg(unix)'.dependencies] nix = { workspace = true } diff --git a/core/server_common/src/diagnostics.rs b/core/server_common/src/diagnostics.rs index 2b745d7e4..a9da6ce39 100644 --- a/core/server_common/src/diagnostics.rs +++ b/core/server_common/src/diagnostics.rs @@ -409,7 +409,7 @@ fn report_io_uring_environment() { /// Parses "major.minor[.patch...][-suffix]" from a kernel release string. #[cfg(target_os = "linux")] -fn parse_kernel_version(release: &str) -> Option<(u32, u32)> { +pub(crate) fn parse_kernel_version(release: &str) -> Option<(u32, u32)> { let mut parts = release .split(|c: char| !c.is_ascii_digit()) .filter(|s| !s.is_empty()); diff --git a/core/server_common/src/lib.rs b/core/server_common/src/lib.rs index 9c7c1a55b..00c416186 100644 --- a/core/server_common/src/lib.rs +++ b/core/server_common/src/lib.rs @@ -27,10 +27,13 @@ pub mod iobuf; pub mod log; mod memory_pool; mod reactor_yield; +pub mod segment_io; mod segment_storage; pub mod send_messages; pub mod sharding; mod storage; +#[cfg(target_os = "linux")] +pub mod uncached_io; pub use bootstrap::create_directories; pub use buffer::PooledBuffer; diff --git a/core/server_common/src/segment_io.rs b/core/server_common/src/segment_io.rs new file mode 100644 index 000000000..6d69d024e --- /dev/null +++ b/core/server_common/src/segment_io.rs @@ -0,0 +1,119 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! The `[system.segment] write_io` knob: how segment `.log` and `.index` +//! writes go through the kernel. +//! +//! `buffered` is the kernel default: writes land in the page cache and +//! writeback runs on the kernel's schedule. `uncached` carries +//! `RWF_DONTCACHE` on every segment write, so writeback starts at once and +//! the written pages are dropped from the page cache when it completes. + +use serde::{Deserialize, Serialize}; +use std::fmt::{Display, Formatter}; +use std::str::FromStr; + +const BUFFERED: &str = "buffered"; +const UNCACHED: &str = "uncached"; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum SegmentIoMode { + #[default] + Buffered, + Uncached, +} + +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +#[error("invalid segment write_io mode '{0}', expected '{BUFFERED}' or '{UNCACHED}'")] +pub struct InvalidSegmentIoMode(pub String); + +impl SegmentIoMode { + pub const fn as_str(self) -> &'static str { + match self { + Self::Buffered => BUFFERED, + Self::Uncached => UNCACHED, + } + } +} + +impl Display for SegmentIoMode { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + f.write_str(self.as_str()) + } +} + +impl FromStr for SegmentIoMode { + type Err = InvalidSegmentIoMode; + + fn from_str(s: &str) -> Result<Self, Self::Err> { + match s { + BUFFERED => Ok(Self::Buffered), + UNCACHED => Ok(Self::Uncached), + other => Err(InvalidSegmentIoMode(other.to_owned())), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde::de::IntoDeserializer; + use serde::de::value::{Error as DeError, StrDeserializer}; + + const ALL_MODES: [SegmentIoMode; 2] = [SegmentIoMode::Buffered, SegmentIoMode::Uncached]; + + fn deserialize_str(value: &str) -> Result<SegmentIoMode, DeError> { + let deserializer: StrDeserializer<'_, DeError> = value.into_deserializer(); + SegmentIoMode::deserialize(deserializer) + } + + #[test] + fn display_and_from_str_round_trip_every_mode() { + for mode in ALL_MODES { + let parsed: SegmentIoMode = mode.to_string().parse().expect("round trip"); + assert_eq!(parsed, mode); + } + assert_eq!(SegmentIoMode::Buffered.to_string(), "buffered"); + assert_eq!(SegmentIoMode::Uncached.to_string(), "uncached"); + } + + #[test] + fn from_str_rejects_unknown_and_wrong_case_values() { + for value in ["direct", "", "Buffered", "UNCACHED", " uncached"] { + let error = SegmentIoMode::from_str(value).expect_err(value); + assert_eq!(error, InvalidSegmentIoMode(value.to_owned())); + assert!(error.to_string().contains(value), "{error}"); + } + } + + #[test] + fn default_is_buffered() { + assert_eq!(SegmentIoMode::default(), SegmentIoMode::Buffered); + } + + // The config file and env override reach the enum through serde, the + // embedded defaults through FromStr: both spellings must agree. + #[test] + fn serde_deserializes_the_same_names_as_from_str() { + for mode in ALL_MODES { + assert_eq!(deserialize_str(mode.as_str()).expect("serde"), mode); + } + assert!(deserialize_str("Uncached").is_err()); + assert!(deserialize_str("direct").is_err()); + } +} diff --git a/core/server_common/src/uncached_io.rs b/core/server_common/src/uncached_io.rs new file mode 100644 index 000000000..7e8579a9b --- /dev/null +++ b/core/server_common/src/uncached_io.rs @@ -0,0 +1,688 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Segment writes that carry `RWF_DONTCACHE` (Linux >= 6.14): the kernel +//! kicks writeback as soon as the write lands and drops the pages from the +//! page cache once it completes, so a busy producer stops evicting the data +//! readers still want. compio does not expose `rw_flags`, hence our own +//! io_uring ops. Only ext4 and XFS honour the flag; every other filesystem +//! answers `EOPNOTSUPP`, which [`probe_uncached_write`] turns into a +//! boot-time error instead of a per-write surprise. + +use std::env; +use std::fs::OpenOptions; +use std::io; +use std::os::fd::{AsFd, AsRawFd, OwnedFd}; +use std::os::unix::fs::OpenOptionsExt; +use std::path::Path; + +use compio::driver::{OpCode, OpEntry, SharedFd}; +use compio_buf::{BufResult, IntoInner, IoBuf, IoVectoredBuf}; +use io_uring::opcode; +use io_uring::types::Fd; +use nix::sys::statfs::{self, FsType}; +use nix::sys::utsname::uname; + +use crate::diagnostics::parse_kernel_version; + +pub const RWF_DONTCACHE: i32 = libc::RWF_DONTCACHE; + +/// First kernel that accepts `RWF_DONTCACHE` at all. +const RWF_DONTCACHE_MIN_KERNEL: (u32, u32) = (6, 14); +/// First kernel that runs the post-write writeback kick off the writer's +/// thread; before it the kick runs inline in the submitting shard. +const WRITEBACK_KICK_OFF_WRITER_KERNEL: (u32, u32) = (7, 2); +const PROBE_FILE_PREFIX: &str = ".iggy_uncached_probe."; +const PROBE_LEN: usize = 4096; +const UNKNOWN: &str = "unknown"; +const REQUIRE_UNCACHED_IO_TESTS_ENV: &str = "IGGY_REQUIRE_UNCACHED_IO"; +const FILESYSTEM_NAMES: [(FsType, &str); 7] = [ + (statfs::EXT4_SUPER_MAGIC, "ext4"), + (statfs::XFS_SUPER_MAGIC, "xfs"), + (statfs::BTRFS_SUPER_MAGIC, "btrfs"), + (statfs::TMPFS_MAGIC, "tmpfs"), + (statfs::OVERLAYFS_SUPER_MAGIC, "overlayfs"), + (statfs::FUSE_SUPER_MAGIC, "fuse"), + (statfs::NFS_SUPER_MAGIC, "nfs"), +]; + +#[derive(Debug, thiserror::Error)] +pub enum UncachedIoError { + #[error("{}", unsupported_reason(.kernel_release, .filesystem))] + Unsupported { + kernel_release: String, + filesystem: String, + }, + #[error("uncached write probe failed: {0}")] + Io(#[from] io::Error), +} + +/// compio's `WriteAt` plus caller-chosen `rw_flags`. +pub struct WriteAtFlags<T: IoBuf, S> { + fd: S, + offset: u64, + flags: i32, + buffer: T, +} + +impl<T: IoBuf, S> WriteAtFlags<T, S> { + pub fn new(fd: S, offset: u64, flags: i32, buffer: T) -> Self { + Self { + fd, + offset, + flags, + buffer, + } + } +} + +impl<T: IoBuf, S> IntoInner for WriteAtFlags<T, S> { + type Inner = T; + + fn into_inner(self) -> Self::Inner { + self.buffer + } +} + +unsafe impl<T: IoBuf, S: AsFd> OpCode for WriteAtFlags<T, S> { + type Control = (); + + fn create_entry(&mut self, _: &mut Self::Control) -> OpEntry { + let slice = self.buffer.as_init(); + opcode::Write::new( + Fd(self.fd.as_fd().as_raw_fd()), + slice.as_ptr(), + slice.len().try_into().unwrap_or(u32::MAX), + ) + .offset(self.offset) + .rw_flags(self.flags) + .build() + .into() + } +} + +/// compio's `WriteVectoredAt` plus caller-chosen `rw_flags`. +pub struct WriteVectoredAtFlags<T: IoVectoredBuf, S> { + fd: S, + offset: u64, + flags: i32, + buffer: T, +} + +impl<T: IoVectoredBuf, S> WriteVectoredAtFlags<T, S> { + pub fn new(fd: S, offset: u64, flags: i32, buffer: T) -> Self { + Self { + fd, + offset, + flags, + buffer, + } + } +} + +impl<T: IoVectoredBuf, S> IntoInner for WriteVectoredAtFlags<T, S> { + type Inner = T; + + fn into_inner(self) -> Self::Inner { + self.buffer + } +} + +/// The iovec array a `Writev` submission points at. +#[derive(Default)] +pub struct IovecControl { + slices: Vec<libc::iovec>, +} + +unsafe impl<T: IoVectoredBuf, S: AsFd> OpCode for WriteVectoredAtFlags<T, S> { + type Control = IovecControl; + + // The driver calls `init` only after boxing the op, so the iovecs point + // at memory that no longer moves. They stay valid until the completion + // arrives even if the submitting future is dropped: the driver keeps the + // boxed op (buffer, iovecs and fd clone) alive until then. + unsafe fn init(&mut self, control: &mut Self::Control) { + control.slices = self.buffer.iter_slice().map(iovec_from_slice).collect(); + } + + fn create_entry(&mut self, control: &mut Self::Control) -> OpEntry { + opcode::Writev::new( + Fd(self.fd.as_fd().as_raw_fd()), + control.slices.as_ptr(), + control.slices.len().try_into().unwrap_or(u32::MAX), + ) + .offset(self.offset) + .rw_flags(self.flags) + .build() + .into() + } +} + +/// Writes the whole buffer at `pos` with `RWF_DONTCACHE`, resubmitting the +/// tail after short writes like compio-io's `write_all_at`. +pub async fn write_all_at_uncached<T: IoBuf, S: AsFd + Clone + 'static>( + fd: &S, + buf: T, + pos: u64, +) -> BufResult<(), T> { + let len = buf.buf_len(); + write_all_with(buf, len, |buf, written| { + let op = WriteAtFlags::new( + fd.clone(), + pos + written as u64, + RWF_DONTCACHE, + buf.slice(written..), + ); + async move { compio::runtime::submit(op).await.into_inner().into_inner() } + }) + .await +} + +/// Vectored twin of [`write_all_at_uncached`]. +pub async fn write_vectored_all_at_uncached<T: IoVectoredBuf, S: AsFd + Clone + 'static>( + fd: &S, + buf: T, + pos: u64, +) -> BufResult<(), T> { + let len = buf.total_len(); + write_all_with(buf, len, |buf, written| { + let op = WriteVectoredAtFlags::new( + fd.clone(), + pos + written as u64, + RWF_DONTCACHE, + buf.slice(written), + ); + async move { compio::runtime::submit(op).await.into_inner().into_inner() } + }) + .await +} + +/// Boot-time check that `dir` takes `RWF_DONTCACHE` writes: one page through +/// the very io_uring op the segment writers submit, on a private file that is +/// always removed again. A kernel and filesystem pair can accept the flag on +/// the `pwritev2` path and still refuse it on the io_uring path, so nothing +/// short of the real op proves the mode is usable. +/// +/// Must be awaited inside a compio runtime; boot does that before any shard +/// exists. +pub async fn probe_uncached_write(dir: &Path) -> Result<(), UncachedIoError> { + let path = dir.join(probe_file_name()); + let file = OpenOptions::new() + .write(true) + .create_new(true) + .mode(0o600) + .open(&path)?; + let _unlink = UnlinkOnDrop(&path); + let fd = SharedFd::new(OwnedFd::from(file)); + let BufResult(result, _) = write_all_at_uncached(&fd, vec![0u8; PROBE_LEN], 0).await; + result.map_err(|error| classify_probe_failure(error, dir)) +} + +/// Warns on kernels that accept `RWF_DONTCACHE` but still run the writeback +/// kick inline in the writer. `None` on 7.2+ or when the release string is +/// unparsable. +pub fn uncached_write_kernel_warning() -> Option<String> { + kernel_warning_for(&kernel_release()?) +} + +/// Test support: with `IGGY_REQUIRE_UNCACHED_IO=1` a test that would skip +/// itself for want of uncached I/O must fail instead, so CI cannot go green +/// on a box where the flag never reached a write. +pub fn require_uncached_io_tests() -> bool { + env::var(REQUIRE_UNCACHED_IO_TESTS_ENV).is_ok_and(|value| value == "1") +} + +/// Mirrors compio-io's `loop_write_all!`: resubmits the unwritten tail after +/// a short write, retries `Interrupted`, and reports a zero-length completion +/// as `WriteZero`. +async fn write_all_with<B, F, Fut>(mut buf: B, len: usize, mut write: F) -> BufResult<(), B> +where + F: FnMut(B, usize) -> Fut, + Fut: Future<Output = BufResult<usize, B>>, +{ + let mut written = 0; + while written < len { + match write(buf, written).await { + BufResult(Ok(0), whole) => { + return BufResult( + Err(io::Error::new( + io::ErrorKind::WriteZero, + "failed to write whole buffer", + )), + whole, + ); + } + BufResult(Ok(n), whole) => { + written += n; + buf = whole; + } + BufResult(Err(ref error), whole) if error.kind() == io::ErrorKind::Interrupted => { + buf = whole; + } + BufResult(Err(error), whole) => return BufResult(Err(error), whole), + } + } + BufResult(Ok(()), buf) +} + +fn iovec_from_slice(slice: &[u8]) -> libc::iovec { + libc::iovec { + iov_base: slice.as_ptr().cast_mut().cast(), + iov_len: slice.len(), + } +} + +/// The pid alone repeats across PID namespaces, so two servers sharing one +/// data directory would fight over a single probe file and misdiagnose each +/// other's writes. +fn probe_file_name() -> String { + format!( + "{PROBE_FILE_PREFIX}{}.{:016x}", + std::process::id(), + rand::random::<u64>() + ) +} + +fn classify_probe_failure(error: io::Error, dir: &Path) -> UncachedIoError { + if error.kind() == io::ErrorKind::WriteZero { + return UncachedIoError::Io(io::Error::other(format!( + "uncached probe write stopped short of {PROBE_LEN} bytes under {}; \ + RWF_DONTCACHE itself was accepted, so check free space and quotas", + dir.display() + ))); + } + match error.raw_os_error() { + Some(libc::EOPNOTSUPP | libc::EINVAL | libc::ENOSYS) => UncachedIoError::Unsupported { + kernel_release: kernel_release().unwrap_or_else(|| UNKNOWN.to_owned()), + filesystem: filesystem_name(dir), + }, + _ => UncachedIoError::Io(error), + } +} + +fn unsupported_reason(kernel_release: &str, filesystem: &str) -> String { + let failed = match parse_kernel_version(kernel_release) { + Some(version) if version < RWF_DONTCACHE_MIN_KERNEL => { + format!("kernel {kernel_release} predates RWF_DONTCACHE") + } + Some(_) => { + format!("filesystem {filesystem} rejects RWF_DONTCACHE (kernel {kernel_release})") + } + None => format!( + "the kernel version could not be determined from release {kernel_release}, and \ + filesystem {filesystem} rejected RWF_DONTCACHE" + ), + }; + format!( + "{failed}: set write_io = \"buffered\", or keep segments on ext4/XFS under Linux >= 6.14" + ) +} + +fn kernel_warning_for(release: &str) -> Option<String> { + let version = parse_kernel_version(release)?; + (RWF_DONTCACHE_MIN_KERNEL..WRITEBACK_KICK_OFF_WRITER_KERNEL) + .contains(&version) + .then(|| { + format!( + "Linux {release} runs the RWF_DONTCACHE writeback kick inline on the writing \ + shard thread; Linux >= 7.2 moves it off the writer. Upgrade the kernel to keep \ + uncached segment writes off the hot path." + ) + }) +} + +fn kernel_release() -> Option<String> { + uname() + .ok() + .map(|info| info.release().to_string_lossy().into_owned()) +} + +fn filesystem_name(dir: &Path) -> String { + let Ok(stat) = statfs::statfs(dir) else { + return UNKNOWN.to_owned(); + }; + let fs_type = stat.filesystem_type(); + FILESYSTEM_NAMES + .iter() + .find(|(magic, _)| *magic == fs_type) + .map_or_else( + || format!("{:#x}", fs_type.0), + |(_, name)| (*name).to_owned(), + ) +} + +struct UnlinkOnDrop<'a>(&'a Path); + +impl Drop for UnlinkOnDrop<'_> { + fn drop(&mut self) { + // Best effort: whatever brought us here is the error worth reporting. + let _ = std::fs::remove_file(self.0); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::HashSet; + use std::fs::File; + use std::future::ready; + use tempfile::{TempDir, tempdir_in}; + + const SHM_DIR: &str = "/dev/shm"; + + /// `/tmp` is tmpfs on many boxes and tmpfs rejects `RWF_DONTCACHE`, so + /// fixtures live next to the test binary, on the build tree's filesystem. + fn target_scratch_dir() -> TempDir { + let exe = std::env::current_exe().expect("current_exe"); + let base = exe.parent().expect("test binary has a parent dir"); + tempdir_in(base).expect("scratch dir under target/") + } + + /// Loud enough to spot in a scrolling CI log, and a hard failure wherever + /// the uncached path is required to have run. + fn skip_or_fail(reason: &str) { + assert!( + !require_uncached_io_tests(), + "### {REQUIRE_UNCACHED_IO_TESTS_ENV}=1 forbids skipping: {reason} ###" + ); + eprintln!("######## SKIPPING UNCACHED TEST: {reason} ########"); + } + + /// `None`, with the reason reported, when this box cannot do uncached + /// writes at all (old kernel or unsupported build-tree filesystem). + async fn supported_scratch_dir() -> Option<TempDir> { + let dir = target_scratch_dir(); + match probe_uncached_write(dir.path()).await { + Ok(()) => Some(dir), + Err(error @ UncachedIoError::Unsupported { .. }) => { + skip_or_fail(&error.to_string()); + None + } + Err(error) => panic!("probe hit an I/O error: {error}"), + } + } + + fn shm_tmpfs_dir() -> Option<TempDir> { + let shm = Path::new(SHM_DIR); + if filesystem_name(shm) != "tmpfs" { + skip_or_fail(&format!("{SHM_DIR} is not tmpfs")); + return None; + } + match tempdir_in(shm) { + Ok(dir) => Some(dir), + Err(error) => { + skip_or_fail(&format!( + "cannot create a directory under {SHM_DIR}: {error}" + )); + None + } + } + } + + fn open_new(dir: &TempDir) -> (SharedFd<OwnedFd>, std::path::PathBuf) { + let path = dir.path().join("segment.log"); + let file = File::options() + .read(true) + .write(true) + .create_new(true) + .open(&path) + .expect("create segment file"); + (SharedFd::new(OwnedFd::from(file)), path) + } + + /// Non-zero bytes that differ per offset, so a misplaced or repeated + /// write cannot pass the byte-exact comparison. + fn pattern(len: usize, seed: usize) -> Vec<u8> { + (0..len).map(|i| ((i + seed) % 251 + 1) as u8).collect() + } + + fn assert_no_probe_leftover(dir: &TempDir) { + let leftovers: Vec<_> = std::fs::read_dir(dir.path()) + .expect("read scratch dir") + .map(|entry| entry.expect("dir entry").file_name()) + .collect(); + assert!(leftovers.is_empty(), "probe left {leftovers:?} behind"); + } + + #[compio::test] + async fn probe_accepts_the_build_tree_filesystem() { + let dir = target_scratch_dir(); + match probe_uncached_write(dir.path()).await { + Ok(()) => {} + Err(error @ UncachedIoError::Unsupported { .. }) => skip_or_fail(&error.to_string()), + Err(error) => panic!("probe hit an I/O error: {error}"), + } + assert_no_probe_leftover(&dir); + } + + #[compio::test] + async fn probe_reports_tmpfs_as_unsupported() { + let Some(dir) = tmpfs_dir_that_rejects_the_flag() else { + return; + }; + let error = probe_uncached_write(dir.path()) + .await + .expect_err("tmpfs never takes RWF_DONTCACHE"); + match &error { + UncachedIoError::Unsupported { filesystem, .. } => assert_eq!(filesystem, "tmpfs"), + other => panic!("expected Unsupported, got {other}"), + } + assert!(error.to_string().contains("tmpfs"), "{error}"); + assert_no_probe_leftover(&dir); + } + + #[compio::test] + async fn probe_reports_a_missing_directory_as_io_error() { + let dir = target_scratch_dir(); + let error = probe_uncached_write(&dir.path().join("missing")) + .await + .expect_err("no such dir"); + match error { + UncachedIoError::Io(error) => assert_eq!(error.kind(), io::ErrorKind::NotFound), + other => panic!("expected Io, got {other}"), + } + } + + #[test] + fn probe_file_names_never_repeat() { + let names: HashSet<String> = (0..64).map(|_| probe_file_name()).collect(); + assert_eq!(names.len(), 64, "probe file name is not unique enough"); + } + + #[compio::test] + async fn uncached_writes_round_trip_unaligned_buffers() { + let Some(dir) = supported_scratch_dir().await else { + return; + }; + let (fd, path) = open_new(&dir); + let mut expected = Vec::new(); + + for len in [300, 4113, 1024 * 1024] { + let buf = pattern(len, expected.len()); + let BufResult(result, buf) = + write_all_at_uncached(&fd, buf, expected.len() as u64).await; + result.expect("uncached write"); + expected.extend_from_slice(&buf); + } + + let slices: Vec<Vec<u8>> = [7, 4096, 1, 9000, 33] + .iter() + .enumerate() + .map(|(index, len)| pattern(*len, expected.len() + index)) + .collect(); + let BufResult(result, slices) = + write_vectored_all_at_uncached(&fd, slices, expected.len() as u64).await; + result.expect("uncached vectored write"); + for slice in &slices { + expected.extend_from_slice(slice); + } + + let on_disk = std::fs::read(&path).expect("read back"); + assert_eq!(on_disk.len(), expected.len()); + assert!( + on_disk == expected, + "file content differs from what was written" + ); + } + + /// tmpfs rejecting the write is the only proof that `RWF_DONTCACHE` + /// really rides on the submission, so the gate must not consult the ops + /// under test: doing so would let a dropped `rw_flags` disable its own + /// test. Kernel version plus filesystem decide it instead. + fn tmpfs_dir_that_rejects_the_flag() -> Option<TempDir> { + let release = kernel_release().unwrap_or_else(|| UNKNOWN.to_owned()); + match parse_kernel_version(&release) { + Some(version) if version >= RWF_DONTCACHE_MIN_KERNEL => shm_tmpfs_dir(), + _ => { + skip_or_fail(&format!("kernel {release} predates RWF_DONTCACHE")); + None + } + } + } + + #[compio::test] + async fn scalar_write_on_tmpfs_fails_with_eopnotsupp() { + let Some(dir) = tmpfs_dir_that_rejects_the_flag() else { + return; + }; + let (fd, _path) = open_new(&dir); + let BufResult(result, _) = write_all_at_uncached(&fd, pattern(300, 0), 0).await; + let error = result.expect_err("tmpfs must reject RWF_DONTCACHE"); + assert_eq!(error.raw_os_error(), Some(libc::EOPNOTSUPP), "{error}"); + } + + #[compio::test] + async fn vectored_write_on_tmpfs_fails_with_eopnotsupp() { + let Some(dir) = tmpfs_dir_that_rejects_the_flag() else { + return; + }; + let (fd, _path) = open_new(&dir); + let slices = vec![pattern(100, 0), pattern(200, 100)]; + let BufResult(result, _) = write_vectored_all_at_uncached(&fd, slices, 0).await; + let error = result.expect_err("tmpfs must reject RWF_DONTCACHE"); + assert_eq!(error.raw_os_error(), Some(libc::EOPNOTSUPP), "{error}"); + } + + #[compio::test] + async fn write_all_with_resubmits_the_tail_and_retries_interrupted() { + let buf = pattern(10, 0); + let mut calls = Vec::new(); + let mut outcomes = vec![ + Ok(3), + Err(io::Error::from(io::ErrorKind::Interrupted)), + Ok(7), + ] + .into_iter(); + let BufResult(result, buf) = write_all_with(buf, 10, |buf: Vec<u8>, written| { + calls.push(written); + ready(BufResult(outcomes.next().expect("scripted outcome"), buf)) + }) + .await; + result.expect("all bytes written"); + assert_eq!(calls, [0, 3, 3]); + assert_eq!(buf, pattern(10, 0)); + } + + /// A short vectored write must resume mid-iovec. Restarting the whole + /// iovec array instead would duplicate bytes on disk and still report + /// success. + #[compio::test] + async fn write_all_with_resumes_a_vectored_buffer_mid_iovec() { + let slices = vec![pattern(7, 0), pattern(4096, 7), pattern(11, 4103)]; + let total = slices.iter().map(Vec::len).sum::<usize>(); + let mut handed = Vec::new(); + let mut outcomes = vec![Ok(7), Ok(4090), Ok(total - 7 - 4090)].into_iter(); + + let BufResult(result, slices) = + write_all_with(slices, total, |buf: Vec<Vec<u8>>, written| { + let view = buf.slice(written); + handed.push(view.iter_slice().map(<[u8]>::to_vec).collect::<Vec<_>>()); + let outcome = outcomes.next().expect("scripted outcome"); + ready(BufResult(outcome, view.into_inner())) + }) + .await; + result.expect("all bytes written"); + + let expected = [ + vec![pattern(7, 0), pattern(4096, 7), pattern(11, 4103)], + vec![pattern(4096, 7), pattern(11, 4103)], + vec![pattern(4096, 7)[4090..].to_vec(), pattern(11, 4103)], + ]; + assert_eq!(handed, expected); + assert_eq!( + slices, + vec![pattern(7, 0), pattern(4096, 7), pattern(11, 4103)] + ); + } + + #[compio::test] + async fn write_all_with_reports_zero_length_completion_as_write_zero() { + let BufResult(result, _) = write_all_with(pattern(4, 0), 4, |buf: Vec<u8>, _| { + ready(BufResult(Ok(0), buf)) + }) + .await; + let error = result.expect_err("zero-length completion"); + assert_eq!(error.kind(), io::ErrorKind::WriteZero); + } + + #[test] + fn kernel_warning_covers_only_the_inline_kick_range() { + assert!(kernel_warning_for("6.13.9-arch1").is_none()); + assert!(kernel_warning_for("6.14.0-1-generic").is_some()); + assert!(kernel_warning_for("7.1.5").is_some()); + assert!(kernel_warning_for("7.2.0-1-cachyos").is_none()); + assert!(kernel_warning_for("not-a-version").is_none()); + } + + #[test] + fn unsupported_message_names_the_failed_requirement_and_the_remedy() { + let old_kernel = UncachedIoError::Unsupported { + kernel_release: "6.8.0-45-generic".to_owned(), + filesystem: "ext4".to_owned(), + } + .to_string(); + assert!( + old_kernel.contains("kernel 6.8.0-45-generic predates"), + "{old_kernel}" + ); + + let bad_filesystem = UncachedIoError::Unsupported { + kernel_release: "7.2.0-1-cachyos".to_owned(), + filesystem: "tmpfs".to_owned(), + } + .to_string(); + assert!( + bad_filesystem.contains("filesystem tmpfs rejects"), + "{bad_filesystem}" + ); + + let unknown_kernel = UncachedIoError::Unsupported { + kernel_release: UNKNOWN.to_owned(), + filesystem: "ext4".to_owned(), + } + .to_string(); + assert!( + unknown_kernel.contains("kernel version could not be determined"), + "{unknown_kernel}" + ); + + for message in [old_kernel, bad_filesystem, unknown_kernel] { + assert!(message.contains("write_io = \"buffered\""), "{message}"); + assert!(message.contains("Linux >= 6.14"), "{message}"); + } + } +} diff --git a/core/simulator/src/replica.rs b/core/simulator/src/replica.rs index 869dc6185..661af9ce5 100644 --- a/core/simulator/src/replica.rs +++ b/core/simulator/src/replica.rs @@ -379,6 +379,7 @@ pub fn new_shard( validate_checksum: true, segment_size: IggyByteSize::from(1024 * 1024 * 1024), preallocate_segments: false, + write_io: server_common::segment_io::SegmentIoMode::Buffered, encryptor: None, path_layout: PartitionPathLayout::default(), };
