This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 420047743 chore(server-ng): extract compio executor and io_uring
diagnostics into shared crate (#3331)
420047743 is described below
commit 420047743c041171092bbcd00c1b7bd53760a136
Author: Jaya Kasa <[email protected]>
AuthorDate: Thu May 28 18:19:10 2026 -0400
chore(server-ng): extract compio executor and io_uring diagnostics into
shared crate (#3331)
---
.../configs/src/configs_impl/typed_env_provider.rs | 2 +-
core/server-ng/src/main.rs | 11 +-
core/server/src/bootstrap.rs | 40 +--
core/server/src/diagnostics.rs | 321 +--------------------
core/{server => server_common}/src/diagnostics.rs | 85 ++++--
core/server_common/src/executor.rs | 132 +++++++++
core/server_common/src/lib.rs | 3 +
7 files changed, 199 insertions(+), 395 deletions(-)
diff --git a/core/configs/src/configs_impl/typed_env_provider.rs
b/core/configs/src/configs_impl/typed_env_provider.rs
index 2dff687e6..772fb217a 100644
--- a/core/configs/src/configs_impl/typed_env_provider.rs
+++ b/core/configs/src/configs_impl/typed_env_provider.rs
@@ -55,7 +55,7 @@ const IGNORED_ENV_VARS: &[&str] = &[
"IGGY_ROOT_PASSWORD",
"IGGY_ROOT_USERNAME",
// Tunes per-shard io_uring SQ/CQ capacity; read directly by
- // `server::bootstrap::create_shard_executor` (see that fn for rationale).
+ // `server_common::executor::create_shard_executor` (see that fn for
rationale).
"IGGY_SHARD_RUNTIME_CAPACITY",
"IGGY_TEST_CLEANUP_DISABLED",
"IGGY_TEST_VERBOSE",
diff --git a/core/server-ng/src/main.rs b/core/server-ng/src/main.rs
index 6ac5ca2b1..b9a7c7da3 100644
--- a/core/server-ng/src/main.rs
+++ b/core/server-ng/src/main.rs
@@ -28,21 +28,18 @@ use server_ng::server_error::ServerNgError;
use tracing::{error, info};
fn main() -> Result<(), ServerNgError> {
- // TODO(hubcio): decouple runtime creation from the `server` crate and
- // move the shared compio executor setup into a lower-level crate/module
- // used by both binaries.
- let bootstrap_runtime = match server::bootstrap::create_shard_executor() {
+ let bootstrap_runtime = match server_common::create_shard_executor() {
Ok(rt) => rt,
Err(e) => {
match e.kind() {
std::io::ErrorKind::InvalidInput => {
- server::diagnostics::print_invalid_io_uring_args_info();
+
server_common::diagnostics::print_invalid_io_uring_args_info();
}
std::io::ErrorKind::OutOfMemory => {
- server::diagnostics::print_locked_memory_limit_info();
+
server_common::diagnostics::print_locked_memory_limit_info();
}
std::io::ErrorKind::PermissionDenied => {
- server::diagnostics::print_io_uring_permission_info();
+
server_common::diagnostics::print_io_uring_permission_info();
}
_ => {}
}
diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs
index d00562aa7..fac9bed5e 100644
--- a/core/server/src/bootstrap.rs
+++ b/core/server/src/bootstrap.rs
@@ -50,7 +50,7 @@ use crate::{
utils::{crypto, file::overwrite},
},
};
-use compio::{fs::create_dir_all, runtime::Runtime};
+use compio::fs::create_dir_all;
use err_trail::ErrContext;
use iggy_common::SemanticVersion;
use iggy_common::{
@@ -170,43 +170,7 @@ pub fn create_root_user() -> User {
User::root(&username, &password)
}
-// Shard executors require IORING_SETUP_COOP_TASKRUN for predictable latency.
-// Falling back to default flags would silently degrade shard performance -
-// do not add a retry with reduced flags here.
-pub fn create_shard_executor() -> Result<Runtime, std::io::Error> {
- // TODO: The event interval tick, could be configured based on the fact
- // How many clients we expect to have connected.
- // This roughly estimates the number of tasks we will create.
- let mut proactor = compio::driver::ProactorBuilder::new();
-
- // Each shard reserves io_uring SQ + CQ entries against `RLIMIT_MEMLOCK`.
- // The multi-node integration tests spawn N nodes * M shards in a single
- // process tree under a memlock budget that's often capped to 8 MiB on
- // dev machines; the 4096-entry default blows past that and the OS
- // returns `Cannot allocate memory` (EAGAIN/ENOMEM) during ring setup.
- // The env knob lets the test harness shrink the per-ring footprint
- // without compiling a separate test build. Production keeps the
- // higher default for throughput.
- let capacity = std::env::var("IGGY_SHARD_RUNTIME_CAPACITY")
- .ok()
- .and_then(|v| v.parse::<u32>().ok())
- .unwrap_or(4096);
- proactor
- .capacity(capacity)
- .coop_taskrun(true)
- .taskrun_flag(true);
-
- // FIXME(hubcio): Only set thread_pool_limit(0) on non-macOS platforms
- // This causes a freeze on macOS with compio fs operations
- // see https://github.com/compio-rs/compio/issues/446
- #[cfg(not(all(target_os = "macos", target_arch = "aarch64")))]
- proactor.thread_pool_limit(0);
-
- compio::runtime::RuntimeBuilder::new()
- .with_proactor(proactor.to_owned())
- .event_interval(128)
- .build()
-}
+pub use server_common::create_shard_executor;
pub fn resolve_persister(enforce_fsync: bool) -> Arc<PersisterKind> {
match enforce_fsync {
diff --git a/core/server/src/diagnostics.rs b/core/server/src/diagnostics.rs
index 32a19ea3b..b2ed5ccdc 100644
--- a/core/server/src/diagnostics.rs
+++ b/core/server/src/diagnostics.rs
@@ -16,321 +16,6 @@
* under the License.
*/
-#[cfg(target_os = "linux")]
-const DISCORD_SUPPORT_URL: &str = "https://discord.gg/apache-iggy";;
-
-#[cfg(target_os = "linux")]
-fn print_discord_link() {
- eprintln!(" Need help? Join our Discord: {DISCORD_SUPPORT_URL}");
- eprintln!();
-}
-
-/// Prints information about locked memory limits when runtime creation fails.
-/// This is typically needed when io_uring cannot allocate memory due to
RLIMIT_MEMLOCK.
-#[cfg(target_os = "linux")]
-pub fn print_locked_memory_limit_info() {
- use nix::sys::resource::{Resource, getrlimit};
-
- let (soft, hard) = match getrlimit(Resource::RLIMIT_MEMLOCK) {
- Ok(limits) => limits,
- Err(_) => {
- eprintln!("Failed to retrieve locked memory limits");
- return;
- }
- };
-
- let format_limit = |limit: u64| -> String {
- if limit == u64::MAX {
- "unlimited".to_string()
- } else {
- let kb = limit / 1024;
- let mb = kb / 1024;
- if mb > 0 {
- format!("{} bytes ({} MB)", limit, mb)
- } else {
- format!("{} bytes ({} KB)", limit, kb)
- }
- }
- };
-
- eprintln!();
- eprintln!("=== Locked Memory Limit Information ===");
- eprintln!("Current soft limit: {}", format_limit(soft));
- eprintln!("Current hard limit: {}", format_limit(hard));
- eprintln!();
- eprintln!("The io_uring runtime requires sufficient locked memory to
operate.");
- eprintln!("To increase the limit, you can:");
- eprintln!();
- eprintln!(" 1. Temporarily (current session only):");
- eprintln!(" ulimit -l unlimited");
- eprintln!();
- eprintln!(" 2. Docker run:");
- eprintln!(" docker run --ulimit memlock=-1:-1 ...");
- eprintln!();
- eprintln!(" 3. Docker Compose (add to service):");
- eprintln!(" ulimits:");
- eprintln!(" memlock:");
- eprintln!(" soft: -1");
- eprintln!(" hard: -1");
- eprintln!();
- eprintln!(" 4. Persistently (add to /etc/security/limits.conf):");
- eprintln!(" * soft memlock unlimited");
- eprintln!(" * hard memlock unlimited");
- eprintln!();
- eprintln!(" 5. For systemd services (add to service file):");
- eprintln!(" LimitMEMLOCK=infinity");
- eprintln!();
- print_discord_link();
-}
-
-/// Prints information about io_uring permission issues in containerized
environments.
-/// This occurs when seccomp blocks io_uring syscalls.
-#[cfg(target_os = "linux")]
-pub fn print_io_uring_permission_info() {
- eprintln!();
- eprintln!("=== io_uring Permission Denied ===");
- eprintln!();
- eprintln!("The io_uring runtime requires specific syscalls that are
blocked by default");
- eprintln!("in containerized environments (Docker, Podman, etc.).");
- eprintln!();
- eprintln!("To resolve this issue:");
- eprintln!();
- eprintln!(" 1. Docker Compose (add to service):");
- eprintln!(" security_opt:");
- eprintln!(" - seccomp:unconfined");
- eprintln!();
- eprintln!(" 2. Docker run:");
- eprintln!(" docker run --security-opt seccomp=unconfined ...");
- eprintln!();
- eprintln!(" 3. Custom seccomp profile (more secure):");
- eprintln!(" Create a profile allowing io_uring_setup,
io_uring_enter,");
- eprintln!(" and io_uring_register syscalls.");
- eprintln!();
- eprintln!(" 4. Kubernetes (add to pod spec):");
- eprintln!(" securityContext:");
- eprintln!(" seccompProfile:");
- eprintln!(" type: Unconfined");
- eprintln!();
- print_discord_link();
-}
-
-/// Minimum kernel version for IORING_SETUP_COOP_TASKRUN and
IORING_SETUP_TASKRUN_FLAG.
-#[cfg(target_os = "linux")]
-const MIN_KERNEL_MAJOR: u32 = 5;
-#[cfg(target_os = "linux")]
-const MIN_KERNEL_MINOR: u32 = 19;
-
-/// Minimum kernel version for kernel.io_uring_disabled sysctl.
-#[cfg(target_os = "linux")]
-const SYSCTL_IO_URING_DISABLED_KERNEL_MAJOR: u32 = 6;
-#[cfg(target_os = "linux")]
-const SYSCTL_IO_URING_DISABLED_KERNEL_MINOR: u32 = 1;
-
-/// Prints diagnostic information when io_uring setup fails with EINVAL.
-///
-/// This typically occurs when the kernel does not support the io_uring flags
-/// required by shard executors (IORING_SETUP_COOP_TASKRUN,
IORING_SETUP_TASKRUN_FLAG).
-/// The caller is responsible for deduplication (e.g., via `std::sync::Once`).
-#[cfg(target_os = "linux")]
-pub fn print_invalid_io_uring_args_info() {
- use nix::sys::utsname::uname;
- use std::fs;
-
- eprintln!();
- eprintln!("=== io_uring Invalid Argument (EINVAL) ===");
- eprintln!();
- eprintln!("The shard executor failed to initialize because the kernel
rejected");
- eprintln!("io_uring setup flags required for shard operation.");
- eprintln!();
- eprintln!(" The main thread's io_uring runtime uses default settings and
initialized");
- eprintln!(" successfully. Shard executors require additional flags:");
- eprintln!(" - IORING_SETUP_COOP_TASKRUN (cooperative task running)");
- eprintln!(" - IORING_SETUP_TASKRUN_FLAG (task runner flag
notification)");
- eprintln!(
- " These flags require Linux kernel >=
{MIN_KERNEL_MAJOR}.{MIN_KERNEL_MINOR} with full io_uring support."
- );
- eprintln!();
-
- let mut detected_issues: Vec<String> = Vec::new();
-
- // 1. Kernel version check
- let uname_info = match uname() {
- Ok(info) => Some(info),
- Err(_) => {
- eprintln!(" [!] Could not retrieve kernel information via
uname(2).");
- None
- }
- };
-
- let mut kernel_version: Option<(u32, u32)> = None;
-
- if let Some(ref info) = uname_info {
- let release = info.release().to_string_lossy();
- eprintln!(" Kernel release: {release}");
-
- if let Some((major, minor)) = parse_kernel_version(&release) {
- kernel_version = Some((major, minor));
- if (major, minor) < (MIN_KERNEL_MAJOR, MIN_KERNEL_MINOR) {
- detected_issues.push(format!(
- "Kernel {major}.{minor} is too old (need >=
{MIN_KERNEL_MAJOR}.{MIN_KERNEL_MINOR})"
- ));
- }
- } else {
- eprintln!(" [!] Could not parse kernel version from release
string.");
- }
-
- // 2. WSL2 detection
- let release_is_wsl = release.contains("microsoft") ||
release.contains("Microsoft");
- let proc_version_is_wsl = fs::read_to_string("/proc/version")
- .map(|v| v.contains("Microsoft") || v.contains("microsoft"))
- .unwrap_or(false);
-
- if release_is_wsl || proc_version_is_wsl {
- eprintln!(" Environment: WSL2 (Microsoft kernel fork detected)");
- detected_issues.push(
- "WSL2 kernel may not support IORING_SETUP_COOP_TASKRUN even if
version >= 5.19"
- .to_string(),
- );
- }
- }
-
- // 3. kernel.io_uring_disabled sysctl (available since kernel 6.1)
- match fs::read_to_string("/proc/sys/kernel/io_uring_disabled") {
- Ok(value) => {
- let value = value.trim();
- eprintln!(" kernel.io_uring_disabled = {value}");
- match value {
- "1" => detected_issues
- .push("io_uring is disabled for unprivileged users (sysctl
= 1)".to_string()),
- "2" => detected_issues
- .push("io_uring is fully disabled by sysctl (sysctl =
2)".to_string()),
- _ => {}
- }
- }
- Err(_) => {
- // The sysctl was introduced in kernel 6.1. If the file is absent
on a kernel >= 6.1,
- // io_uring is likely not compiled in (CONFIG_IO_URING=n).
- if let Some((major, minor)) = kernel_version
- && (major, minor)
- >= (
- SYSCTL_IO_URING_DISABLED_KERNEL_MAJOR,
- SYSCTL_IO_URING_DISABLED_KERNEL_MINOR,
- )
- {
- detected_issues.push(format!(
- "kernel.io_uring_disabled sysctl not found on kernel >= \
-
{SYSCTL_IO_URING_DISABLED_KERNEL_MAJOR}.{SYSCTL_IO_URING_DISABLED_KERNEL_MINOR}
\
- - io_uring may not be compiled in (CONFIG_IO_URING=n)"
- ));
- }
- }
- }
-
- // 4. AppArmor - informational only, not added to detected_issues
- let apparmor_profile =
fs::read_to_string("/proc/self/attr/apparmor/current")
- .ok()
- .map(|s| s.trim().to_string());
-
- if let Some(ref profile) = apparmor_profile
- && profile != "unconfined"
- && !profile.is_empty()
- {
- eprintln!(" AppArmor profile: {profile}");
- }
-
- // Print detected issues
- if detected_issues.is_empty() {
- eprintln!();
- eprintln!(" No specific issue was detected. The kernel may lack
io_uring support");
- eprintln!(" for the flags used by Iggy's shard executors.");
- } else {
- eprintln!();
- eprintln!(" Detected issues:");
- for (i, issue) in detected_issues.iter().enumerate() {
- eprintln!(" {}. {issue}", i + 1);
- }
- }
-
- eprintln!();
- eprintln!(" To resolve this:");
- eprintln!();
- eprintln!(
- " 1. Upgrade to Linux kernel >= {MIN_KERNEL_MAJOR}.{MIN_KERNEL_MINOR}
(>=
{SYSCTL_IO_URING_DISABLED_KERNEL_MAJOR}.{SYSCTL_IO_URING_DISABLED_KERNEL_MINOR}
recommended)"
- );
- eprintln!();
- eprintln!(" 2. If running under WSL2:");
- eprintln!(" - Update WSL: wsl --update (from PowerShell)");
- eprintln!(" - Or build a custom kernel with full io_uring support:");
- eprintln!("
https://learn.microsoft.com/en-us/windows/wsl/wsl-config#wsl-2-settings";);
- eprintln!(" - Or use Docker Desktop / a native Linux VM instead of
WSL2");
- eprintln!();
- eprintln!(" 3. If io_uring is disabled via sysctl:");
- eprintln!(" sudo sysctl -w kernel.io_uring_disabled=0");
- eprintln!();
- eprintln!(" 4. If AppArmor is restricting io_uring:");
- eprintln!(" sudo aa-complain <profile-name>");
- eprintln!();
- eprintln!(" 5. Check kernel logs for more details:");
- eprintln!(" dmesg | grep -i io_uring");
- eprintln!();
- print_discord_link();
-}
-
-/// Parses "major.minor[.patch...][-suffix]" from a kernel release string.
-#[cfg(target_os = "linux")]
-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());
- let major = parts.next()?.parse::<u32>().ok()?;
- let minor = parts.next()?.parse::<u32>().ok()?;
- Some((major, minor))
-}
-
-#[cfg(not(target_os = "linux"))]
-pub fn print_locked_memory_limit_info() {}
-
-#[cfg(not(target_os = "linux"))]
-pub fn print_io_uring_permission_info() {}
-
-#[cfg(not(target_os = "linux"))]
-pub fn print_invalid_io_uring_args_info() {}
-
-#[cfg(test)]
-mod tests {
- #[cfg(target_os = "linux")]
- use super::parse_kernel_version;
-
- #[test]
- #[cfg(target_os = "linux")]
- fn test_parse_standard_kernel_version() {
- assert_eq!(parse_kernel_version("6.8.0-45-generic"), Some((6, 8)));
- }
-
- #[test]
- #[cfg(target_os = "linux")]
- fn test_parse_wsl2_kernel_version() {
- assert_eq!(
- parse_kernel_version("5.15.153.1-microsoft-standard-WSL2"),
- Some((5, 15))
- );
- }
-
- #[test]
- #[cfg(target_os = "linux")]
- fn test_parse_minimal_version() {
- assert_eq!(parse_kernel_version("5.19"), Some((5, 19)));
- }
-
- #[test]
- #[cfg(target_os = "linux")]
- fn test_parse_garbage_returns_none() {
- assert_eq!(parse_kernel_version("not-a-version"), None);
- }
-
- #[test]
- #[cfg(target_os = "linux")]
- fn test_parse_empty_returns_none() {
- assert_eq!(parse_kernel_version(""), None);
- }
-}
+pub use server_common::diagnostics::print_invalid_io_uring_args_info;
+pub use server_common::diagnostics::print_io_uring_permission_info;
+pub use server_common::diagnostics::print_locked_memory_limit_info;
diff --git a/core/server/src/diagnostics.rs
b/core/server_common/src/diagnostics.rs
similarity index 86%
copy from core/server/src/diagnostics.rs
copy to core/server_common/src/diagnostics.rs
index 32a19ea3b..568d4e44d 100644
--- a/core/server/src/diagnostics.rs
+++ b/core/server_common/src/diagnostics.rs
@@ -1,4 +1,5 @@
-/* Licensed to the Apache Software Foundation (ASF) under one
+/*
+ * 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
@@ -25,6 +26,27 @@ fn print_discord_link() {
eprintln!();
}
+/// Formats an `RLIMIT_MEMLOCK` value for diagnostic output.
+///
+/// `u64::MAX` is rendered as `unlimited` (the value getrlimit returns
+/// for an uncapped limit); other values are rendered as raw bytes plus
+/// a coarse MB-or-KB suffix so operators can eyeball whether the limit
+/// is in the ballpark of the 4096-entry io_uring ring footprint.
+#[cfg(target_os = "linux")]
+fn format_limit(limit: u64) -> String {
+ if limit == u64::MAX {
+ "unlimited".to_string()
+ } else {
+ let kb = limit / 1024;
+ let mb = kb / 1024;
+ if mb > 0 {
+ format!("{limit} bytes ({mb} MB)")
+ } else {
+ format!("{limit} bytes ({kb} KB)")
+ }
+ }
+}
+
/// Prints information about locked memory limits when runtime creation fails.
/// This is typically needed when io_uring cannot allocate memory due to
RLIMIT_MEMLOCK.
#[cfg(target_os = "linux")]
@@ -39,20 +61,6 @@ pub fn print_locked_memory_limit_info() {
}
};
- let format_limit = |limit: u64| -> String {
- if limit == u64::MAX {
- "unlimited".to_string()
- } else {
- let kb = limit / 1024;
- let mb = kb / 1024;
- if mb > 0 {
- format!("{} bytes ({} MB)", limit, mb)
- } else {
- format!("{} bytes ({} KB)", limit, kb)
- }
- }
- };
-
eprintln!();
eprintln!("=== Locked Memory Limit Information ===");
eprintln!("Current soft limit: {}", format_limit(soft));
@@ -288,28 +296,25 @@ fn parse_kernel_version(release: &str) -> Option<(u32,
u32)> {
}
#[cfg(not(target_os = "linux"))]
-pub fn print_locked_memory_limit_info() {}
+pub const fn print_locked_memory_limit_info() {}
#[cfg(not(target_os = "linux"))]
-pub fn print_io_uring_permission_info() {}
+pub const fn print_io_uring_permission_info() {}
#[cfg(not(target_os = "linux"))]
-pub fn print_invalid_io_uring_args_info() {}
+pub const fn print_invalid_io_uring_args_info() {}
-#[cfg(test)]
+#[cfg(all(test, target_os = "linux"))]
mod tests {
- #[cfg(target_os = "linux")]
- use super::parse_kernel_version;
+ use super::{format_limit, parse_kernel_version};
#[test]
- #[cfg(target_os = "linux")]
- fn test_parse_standard_kernel_version() {
+ fn parse_kernel_version_standard() {
assert_eq!(parse_kernel_version("6.8.0-45-generic"), Some((6, 8)));
}
#[test]
- #[cfg(target_os = "linux")]
- fn test_parse_wsl2_kernel_version() {
+ fn parse_kernel_version_wsl2() {
assert_eq!(
parse_kernel_version("5.15.153.1-microsoft-standard-WSL2"),
Some((5, 15))
@@ -317,20 +322,38 @@ mod tests {
}
#[test]
- #[cfg(target_os = "linux")]
- fn test_parse_minimal_version() {
+ fn parse_kernel_version_minimal() {
assert_eq!(parse_kernel_version("5.19"), Some((5, 19)));
}
#[test]
- #[cfg(target_os = "linux")]
- fn test_parse_garbage_returns_none() {
+ fn parse_kernel_version_garbage_returns_none() {
assert_eq!(parse_kernel_version("not-a-version"), None);
}
#[test]
- #[cfg(target_os = "linux")]
- fn test_parse_empty_returns_none() {
+ fn parse_kernel_version_empty_returns_none() {
assert_eq!(parse_kernel_version(""), None);
}
+
+ #[test]
+ fn parse_kernel_version_overflow_returns_none() {
+ // u32::MAX + 1 in the major slot must not silently wrap.
+ assert_eq!(parse_kernel_version("4294967296.0"), None);
+ }
+
+ #[test]
+ fn format_limit_unlimited() {
+ assert_eq!(format_limit(u64::MAX), "unlimited");
+ }
+
+ #[test]
+ fn format_limit_sub_mb_uses_kb_suffix() {
+ assert_eq!(format_limit(64 * 1024), "65536 bytes (64 KB)");
+ }
+
+ #[test]
+ fn format_limit_mb_range_uses_mb_suffix() {
+ assert_eq!(format_limit(8 * 1024 * 1024), "8388608 bytes (8 MB)");
+ }
}
diff --git a/core/server_common/src/executor.rs
b/core/server_common/src/executor.rs
new file mode 100644
index 000000000..c6e20e05e
--- /dev/null
+++ b/core/server_common/src/executor.rs
@@ -0,0 +1,132 @@
+/*
+ * 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.
+ */
+
+use compio::runtime::Runtime;
+
+const DEFAULT_SHARD_RUNTIME_CAPACITY: u32 = 4096;
+const SHARD_RUNTIME_CAPACITY_ENV: &str = "IGGY_SHARD_RUNTIME_CAPACITY";
+
+/// Resolves the per-shard io_uring SQ/CQ capacity from
`IGGY_SHARD_RUNTIME_CAPACITY`,
+/// falling back to [`DEFAULT_SHARD_RUNTIME_CAPACITY`] when the var is missing
or
+/// fails to parse as `u32`.
+fn shard_capacity_from_env() -> u32 {
+ std::env::var(SHARD_RUNTIME_CAPACITY_ENV)
+ .ok()
+ .and_then(|v| v.parse::<u32>().ok())
+ .unwrap_or(DEFAULT_SHARD_RUNTIME_CAPACITY)
+}
+
+/// Creates a compio runtime for a shard thread, with shard-specific
`io_uring` flags.
+///
+/// The per-ring SQ/CQ capacity defaults to `4096` and can be overridden via
the
+/// `IGGY_SHARD_RUNTIME_CAPACITY` env var, which the multi-node integration
+/// harness sets to `256` so N nodes * M shards fit under an 8 MiB
+/// `RLIMIT_MEMLOCK` budget without `ENOMEM` at ring setup.
+///
+/// # Errors
+///
+/// Returns an `std::io::Error` if the underlying `io_uring` proactor cannot
be initialised.
+/// On `InvalidInput` the kernel rejected the required flags; on `OutOfMemory`
or
+/// `PermissionDenied` the caller should print the appropriate diagnostic
before panicking.
+///
+/// Shard executors require `IORING_SETUP_COOP_TASKRUN` for predictable
latency.
+/// Falling back to default flags would silently degrade shard performance -
+/// do not add a retry with reduced flags here.
+pub fn create_shard_executor() -> Result<Runtime, std::io::Error> {
+ // TODO: The event interval tick, could be configured based on the fact
+ // How many clients we expect to have connected.
+ // This roughly estimates the number of tasks we will create.
+ let mut proactor = compio::driver::ProactorBuilder::new();
+
+ proactor
+ .capacity(shard_capacity_from_env())
+ .coop_taskrun(true)
+ .taskrun_flag(true);
+
+ // FIXME(hubcio): Only set thread_pool_limit(0) on non-macOS platforms
+ // This causes a freeze on macOS with compio fs operations
+ // see https://github.com/compio-rs/compio/issues/446
+ #[cfg(not(all(target_os = "macos", target_arch = "aarch64")))]
+ proactor.thread_pool_limit(0);
+
+ compio::runtime::RuntimeBuilder::new()
+ .with_proactor(proactor.to_owned())
+ .event_interval(128)
+ .build()
+}
+
+#[cfg(test)]
+mod tests {
+ use super::{
+ DEFAULT_SHARD_RUNTIME_CAPACITY, SHARD_RUNTIME_CAPACITY_ENV,
shard_capacity_from_env,
+ };
+ use serial_test::serial;
+
+ fn with_capacity_env<R>(value: Option<&str>, f: impl FnOnce() -> R) -> R {
+ // SAFETY: tests in this module are #[serial], so no other thread races
+ // on the process-wide environment while the guard is active.
+ let prev = std::env::var(SHARD_RUNTIME_CAPACITY_ENV).ok();
+ unsafe {
+ match value {
+ Some(v) => std::env::set_var(SHARD_RUNTIME_CAPACITY_ENV, v),
+ None => std::env::remove_var(SHARD_RUNTIME_CAPACITY_ENV),
+ }
+ }
+ let out = f();
+ unsafe {
+ match prev {
+ Some(v) => std::env::set_var(SHARD_RUNTIME_CAPACITY_ENV, v),
+ None => std::env::remove_var(SHARD_RUNTIME_CAPACITY_ENV),
+ }
+ }
+ out
+ }
+
+ #[test]
+ #[serial]
+ fn shard_capacity_from_env_uses_parsed_value() {
+ with_capacity_env(Some("256"), || {
+ assert_eq!(shard_capacity_from_env(), 256);
+ });
+ }
+
+ #[test]
+ #[serial]
+ fn shard_capacity_from_env_falls_back_when_unset() {
+ with_capacity_env(None, || {
+ assert_eq!(shard_capacity_from_env(),
DEFAULT_SHARD_RUNTIME_CAPACITY);
+ });
+ }
+
+ #[test]
+ #[serial]
+ fn shard_capacity_from_env_falls_back_on_unparsable() {
+ with_capacity_env(Some("not-a-number"), || {
+ assert_eq!(shard_capacity_from_env(),
DEFAULT_SHARD_RUNTIME_CAPACITY);
+ });
+ }
+
+ #[test]
+ #[serial]
+ fn shard_capacity_from_env_falls_back_on_negative() {
+ with_capacity_env(Some("-1"), || {
+ assert_eq!(shard_capacity_from_env(),
DEFAULT_SHARD_RUNTIME_CAPACITY);
+ });
+ }
+}
diff --git a/core/server_common/src/lib.rs b/core/server_common/src/lib.rs
index b11447e59..75be8cb01 100644
--- a/core/server_common/src/lib.rs
+++ b/core/server_common/src/lib.rs
@@ -20,6 +20,8 @@ mod buffer;
mod certificates;
mod consensus_message;
mod deduplication;
+pub mod diagnostics;
+mod executor;
mod in_flight;
mod indexes_mut;
// TODO(hubcio): iobuf was relocated verbatim from
`core/binary_protocol/src/consensus/iobuf.rs`
@@ -50,6 +52,7 @@ pub use consensus_message::{
MutableBacking, RequestBacking, RequestBackingKind, ResponseBacking,
ResponseBackingKind,
};
pub use deduplication::MessageDeduplicator;
+pub use executor::create_shard_executor;
pub use in_flight::IggyMessagesBatchSetInFlight;
pub use indexes_mut::IggyIndexesMut;
pub use memory_pool::{MEMORY_POOL, MemoryPool, MemoryPoolConfigOther,
memory_pool};
