This is an automated email from the ASF dual-hosted git repository.
mmodzelewski 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 ebaffde44 feat(server): respect CPU affinity and cgroup limits in
shards and stats (#3615)
ebaffde44 is described below
commit ebaffde44afb46f792a512a044ec7553c922faf2
Author: Maciej Modzelewski <[email protected]>
AuthorDate: Wed Jul 8 11:43:54 2026 +0200
feat(server): respect CPU affinity and cgroup limits in shards and stats
(#3615)
The server assumed it owned the whole host. Shards pinned to
absolute core ids 0..n, which fails with EINVAL (or lands on
forbidden cores) under systemd AllowedCPUs=, container cpusets,
or taskset. Stats reported host-wide CPU and memory, so a
confined instance showed its neighbors' load and a meaningless
host total.
Draw pinned cores from the process's allowed CPU set
(sched_getaffinity) instead, and validate configured ranges
against it at boot. A new pin_cores knob (default true) turns
pinning off for shared-core hosts (cgroup CPU quotas), where
every tenant pinning to the same low-numbered cores would pile
onto one core; unpinned shards let the kernel scheduler place
threads freely.
Stats now scope total_cpu_usage to the allowed cores and report
the effective cgroup memory limit as total_memory. Available
memory adds reclaimable file cache back
(limit - (current - inactive_file - active_file), minimum over
capped ancestors, v1 and v2), since memory.current charges page
cache as used and the naive limit - current trends to zero on a
cache-heavy server long before real OOM pressure.
---
Cargo.lock | 2 +
core/common/src/types/stats/mod.rs | 8 +-
core/configs/src/server_config/sharding.rs | 17 ++
core/configs/src/server_config/validators.rs | 84 +++++++-
core/cpu_allocation/Cargo.toml | 5 +-
core/cpu_allocation/src/allowed_cpus.rs | 59 ++++++
core/cpu_allocation/src/lib.rs | 8 +-
core/server-ng/config.toml | 9 +
core/server-ng/src/bootstrap.rs | 53 +++--
core/server/Cargo.toml | 1 +
core/server/config.toml | 8 +
core/server/src/main.rs | 5 +-
core/server/src/shard/system/cgroup_memory.rs | 293 ++++++++++++++++++++++++++
core/server/src/shard/system/mod.rs | 1 +
core/server/src/shard/system/stats.rs | 38 +++-
core/shard_allocator/src/lib.rs | 212 +++++++++++++++++--
16 files changed, 753 insertions(+), 50 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 4e7a3fce9..ed13f356d 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3349,6 +3349,7 @@ dependencies = [
name = "cpu_allocation"
version = "0.1.0"
dependencies = [
+ "nix",
"serde",
"serde_json",
]
@@ -11679,6 +11680,7 @@ dependencies = [
"clap",
"compio",
"configs",
+ "cpu_allocation",
"ctrlc",
"cyper",
"cyper-axum",
diff --git a/core/common/src/types/stats/mod.rs
b/core/common/src/types/stats/mod.rs
index fa848858d..5de7f03b4 100644
--- a/core/common/src/types/stats/mod.rs
+++ b/core/common/src/types/stats/mod.rs
@@ -26,13 +26,15 @@ pub struct Stats {
pub process_id: u32,
/// The CPU usage of the process.
pub cpu_usage: f32,
- /// the total CPU usage of the system.
+ /// The total CPU usage of the system, scoped to the cores this process
may run on
+ /// when confined by an affinity/cpuset mask.
pub total_cpu_usage: f32,
/// The memory usage of the process.
pub memory_usage: IggyByteSize,
- /// The total memory of the system.
+ /// The total memory of the system, or the effective cgroup memory limit
when the
+ /// server runs inside a memory-capped cgroup (container, systemd slice).
pub total_memory: IggyByteSize,
- /// The available memory of the system.
+ /// The available memory of the system, scoped to the cgroup limit when
one applies.
pub available_memory: IggyByteSize,
/// The run time of the process.
pub run_time: IggyDuration,
diff --git a/core/configs/src/server_config/sharding.rs
b/core/configs/src/server_config/sharding.rs
index 0d0c1fc6e..2f23cfdb7 100644
--- a/core/configs/src/server_config/sharding.rs
+++ b/core/configs/src/server_config/sharding.rs
@@ -101,6 +101,10 @@ const fn default_inbox_capacity() -> usize {
DEFAULT_INBOX_CAPACITY
}
+const fn default_pin_cores() -> bool {
+ true
+}
+
fn default_shutdown_drain_timeout() -> IggyDuration {
IggyDuration::new(DEFAULT_SHUTDOWN_DRAIN_TIMEOUT)
}
@@ -119,6 +123,18 @@ pub struct ShardingConfig {
#[serde(default)]
#[config_env(leaf)]
pub cpu_allocation: CpuAllocation,
+ /// Whether shard threads are pinned to dedicated CPU cores
+ /// (`sched_setaffinity`). Pinning maximizes cache locality when this
+ /// server owns its cores (dedicated host, `numa:` allocations). Set to
+ /// `false` when the server shares cores with other workloads — e.g. a
+ /// multi-tenant host slicing CPU via cgroup quotas — where every process
+ /// pinning to the same low-numbered cores would pile onto one core while
+ /// the rest sit idle; unpinned shards let the kernel scheduler place
+ /// threads freely within the allowed set. With a NUMA-aware allocation,
+ /// `false` drops both the CPU and memory-node bindings (and logs a
+ /// warning, since NUMA placement without pinning is meaningless).
+ #[serde(default = "default_pin_cores")]
+ pub pin_cores: bool,
/// Per-shard inter-shard inbox channel capacity. Bounded by design.
/// Drops on full inbox of consensus frames are recovered by VSR
/// retransmit. Drops of cross-shard client Reply frames are terminal:
@@ -176,6 +192,7 @@ impl Default for ShardingConfig {
fn default() -> Self {
Self {
cpu_allocation: CpuAllocation::default(),
+ pin_cores: default_pin_cores(),
inbox_capacity: DEFAULT_INBOX_CAPACITY,
shutdown_drain_timeout: default_shutdown_drain_timeout(),
shutdown_poll_interval: default_shutdown_poll_interval(),
diff --git a/core/configs/src/server_config/validators.rs
b/core/configs/src/server_config/validators.rs
index 0aafe1ef5..ca8410e28 100644
--- a/core/configs/src/server_config/validators.rs
+++ b/core/configs/src/server_config/validators.rs
@@ -28,6 +28,7 @@ use super::sharding::{
use super::system::SegmentConfig;
use super::system::{CompressionConfig, LoggingConfig, PartitionConfig};
use crate::ConfigurationError;
+use cpu_allocation::allowed_cpus;
use err_trail::ErrContext;
use iggy_common::CompressionAlgorithm;
use iggy_common::IggyExpiry;
@@ -488,9 +489,20 @@ impl Validatable<ConfigurationError> for ShardingConfig {
);
return Err(ConfigurationError::InvalidConfigurationValue);
}
- if *end > available_cpus {
+ if *end - *start > available_cpus {
eprintln!(
- "Invalid sharding configuration: cpu_allocation range
{start}..{end} exceeds available CPU cores (max: {available_cpus})"
+ "Invalid sharding configuration: cpu_allocation range
{start}..{end} yields {} shards, exceeding available CPU cores
{available_cpus}",
+ *end - *start
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+ if !self.pin_cores {
+ return Ok(());
+ }
+ let allowed = allowed_cpus();
+ if let Some(cpu) = (*start..*end).find(|cpu|
!allowed.contains(cpu)) {
+ eprintln!(
+ "Invalid sharding configuration: cpu_allocation range
{start}..{end} includes CPU {cpu}, which is outside the set of cores allowed
for this process (affinity/cpuset mask)"
);
return Err(ConfigurationError::InvalidConfigurationValue);
}
@@ -1008,3 +1020,71 @@ mod sharding_shutdown_knob_tests {
assert!(cfg.validate().is_err());
}
}
+
+#[cfg(test)]
+mod sharding_cpu_range_tests {
+ use super::*;
+
+ #[test]
+ fn inverted_range_is_rejected() {
+ let cfg = ShardingConfig {
+ cpu_allocation: CpuAllocation::Range(2, 2),
+ ..ShardingConfig::default()
+ };
+ assert!(cfg.validate().is_err());
+ }
+
+ #[test]
+ fn pinned_range_within_allowed_set_is_accepted() {
+ let first = allowed_cpus()[0];
+ let cfg = ShardingConfig {
+ cpu_allocation: CpuAllocation::Range(first, first + 1),
+ ..ShardingConfig::default()
+ };
+ assert!(cfg.validate().is_ok());
+ }
+
+ #[test]
+ fn pinned_range_outside_allowed_set_is_rejected() {
+ let past_last = allowed_cpus().last().copied().unwrap() + 1;
+ let cfg = ShardingConfig {
+ cpu_allocation: CpuAllocation::Range(past_last, past_last + 1),
+ ..ShardingConfig::default()
+ };
+ assert!(cfg.validate().is_err());
+ }
+
+ #[test]
+ fn pinned_range_wider_than_parallelism_is_rejected() {
+ // Under a cgroup CPU quota the affinity mask stays full while
+ // `available_parallelism` shrinks, so membership alone would
+ // accept this; the shard-count cap must reject it.
+ let first = allowed_cpus()[0];
+ let available = available_parallelism().unwrap().get();
+ let cfg = ShardingConfig {
+ cpu_allocation: CpuAllocation::Range(first, first + available + 1),
+ ..ShardingConfig::default()
+ };
+ assert!(cfg.validate().is_err());
+ }
+
+ #[test]
+ fn unpinned_range_is_capped_by_shard_count_not_core_ids() {
+ // Core ids outside the machine are fine unpinned; only the
+ // resulting shard count matters.
+ let cfg = ShardingConfig {
+ cpu_allocation: CpuAllocation::Range(1 << 20, (1 << 20) + 1),
+ pin_cores: false,
+ ..ShardingConfig::default()
+ };
+ assert!(cfg.validate().is_ok());
+
+ let available = available_parallelism().unwrap().get();
+ let cfg = ShardingConfig {
+ cpu_allocation: CpuAllocation::Range(0, available + 1),
+ pin_cores: false,
+ ..ShardingConfig::default()
+ };
+ assert!(cfg.validate().is_err());
+ }
+}
diff --git a/core/cpu_allocation/Cargo.toml b/core/cpu_allocation/Cargo.toml
index 74eea3230..24cfb9b17 100644
--- a/core/cpu_allocation/Cargo.toml
+++ b/core/cpu_allocation/Cargo.toml
@@ -18,7 +18,7 @@
[package]
name = "cpu_allocation"
version = "0.1.0"
-description = "Shard CPU/NUMA allocation config types (CpuAllocation,
NumaConfig) parsed from the iggy server config."
+description = "Shard CPU/NUMA allocation config types (CpuAllocation,
NumaConfig) parsed from the iggy server config, plus the allowed-CPU-set probe."
edition = "2024"
license = "Apache-2.0"
publish = false
@@ -26,5 +26,8 @@ publish = false
[dependencies]
serde = { workspace = true }
+[target.'cfg(target_os = "linux")'.dependencies]
+nix = { workspace = true }
+
[dev-dependencies]
serde_json = { workspace = true }
diff --git a/core/cpu_allocation/src/allowed_cpus.rs
b/core/cpu_allocation/src/allowed_cpus.rs
new file mode 100644
index 000000000..df7fe5490
--- /dev/null
+++ b/core/cpu_allocation/src/allowed_cpus.rs
@@ -0,0 +1,59 @@
+// 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.
+
+#[cfg(target_os = "linux")]
+use nix::{
+ sched::{CpuSet, sched_getaffinity},
+ unistd::Pid,
+};
+use std::thread::available_parallelism;
+
+/// CPUs the calling process is currently allowed to run on, ascending.
+///
+/// Respects restrictions imposed by the parent environment (systemd
+/// `AllowedCPUs=`, container cpusets, `taskset`), which absolute core ids
+/// `0..n` would silently violate: `sched_setaffinity` to a core outside the
+/// allowed set fails with `EINVAL`. Falls back to `0..available_parallelism()`
+/// where the affinity mask is unavailable (non-Linux).
+pub fn allowed_cpus() -> Vec<usize> {
+ #[cfg(target_os = "linux")]
+ {
+ if let Ok(mask) = sched_getaffinity(Pid::from_raw(0)) {
+ let cpus: Vec<usize> = (0..CpuSet::count())
+ .filter(|&cpu| mask.is_set(cpu).unwrap_or(false))
+ .collect();
+ if !cpus.is_empty() {
+ return cpus;
+ }
+ }
+ }
+
+ let fallback = available_parallelism().map(|n| n.get()).unwrap_or(1);
+ (0..fallback).collect()
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn allowed_cpus_is_non_empty_and_ascending() {
+ let allowed = allowed_cpus();
+ assert!(!allowed.is_empty());
+ assert!(allowed.windows(2).all(|pair| pair[0] < pair[1]));
+ }
+}
diff --git a/core/cpu_allocation/src/lib.rs b/core/cpu_allocation/src/lib.rs
index 5391e73b1..48719f889 100644
--- a/core/cpu_allocation/src/lib.rs
+++ b/core/cpu_allocation/src/lib.rs
@@ -22,11 +22,17 @@
//! the server config (TOML). The config crate re-exports them, and the
//! `shard_allocator` crate turns them into a real plan. Kept in their
//! own little crate so neither side has to pull in the other's heavy
-//! dependencies just to share two small enums.
+//! dependencies just to share two small enums. Also home to
+//! [`allowed_cpus`], the probe for the process's allowed CPU set, which
+//! both sides consult when validating and pinning cores.
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use std::str::FromStr;
+mod allowed_cpus;
+
+pub use allowed_cpus::allowed_cpus;
+
/// Tell server how many CPU cores to grab for shards, and how.
///
/// Server make one shard per core. This say which cores. Pick one:
diff --git a/core/server-ng/config.toml b/core/server-ng/config.toml
index 11de70d34..70af981f5 100644
--- a/core/server-ng/config.toml
+++ b/core/server-ng/config.toml
@@ -630,6 +630,15 @@ ports = { tcp = 8091, quic = 8081, http = 3001, websocket
= 8093, tcp_replica =
# TODO(hubcio): revert to "numa:auto" once multi-shard server-ng is stable.
cpu_allocation = 1
+# Whether shard threads are pinned to dedicated CPU cores (default: true).
+# Pinned cores are drawn from the process's allowed CPU set (affinity/cpuset
+# mask), so the server cooperates with systemd `AllowedCPUs=` and container
+# cpusets. Set to false when the server shares cores with other workloads
+# (e.g. a multi-tenant host slicing CPU via cgroup quotas): unpinned shards
+# let the kernel scheduler place threads freely instead of piling every
+# process onto the same low-numbered cores.
+# pin_cores = true
+
# Wall-clock budget for a single shard's bus drain on shutdown. Drives
# the per-shard watchdog and the parallel-join survivor path; sized
# larger than typical TCP RTT times in-flight write-batch so writers
diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs
index ca7710148..1c3a6799a 100644
--- a/core/server-ng/src/bootstrap.rs
+++ b/core/server-ng/src/bootstrap.rs
@@ -451,6 +451,35 @@ pub async fn load_config(logging: &mut Logging) ->
Result<ServerNgConfig, Server
Ok(config)
}
+/// Resolve the operator's `cpu_allocation` into concrete shard
+/// assignments plus the checked `u16` shard count.
+///
+/// Shard ids index `ReplicaOwnerTable` slots as `u16`. `OWNER_NONE`
+/// (`u16::MAX`) is reserved as the empty-slot sentinel, so a server
+/// configured with `u16::MAX` shards would mint a shard id that
+/// collides with the sentinel and an owner-table lookup could never
+/// tell that shard apart from an unowned slot. Reject at boot so the
+/// invariant is held by the type system, not by hoping the operator
+/// never configures 65535 cores worth of shards.
+fn resolve_shard_assignments(
+ sharding: &configs::sharding::ShardingConfig,
+) -> Result<(Vec<ShardInfo>, u16), ServerNgError> {
+ let allocator = ShardAllocator::new(&sharding.cpu_allocation,
sharding.pin_cores)
+ .map_err(ServerNgError::ShardAllocator)?;
+ let assignments = allocator
+ .to_shard_assignments()
+ .map_err(ServerNgError::ShardAllocator)?;
+ if assignments.is_empty() {
+ return Err(ServerNgError::ShardsCountZero);
+ }
+ match u16::try_from(assignments.len()) {
+ Ok(count) if count < message_bus::OWNER_NONE => Ok((assignments,
count)),
+ _ => Err(ServerNgError::ShardsCountOverflow {
+ count: assignments.len(),
+ }),
+ }
+}
+
/// Re-validate the runtime sharding knobs that the per-shard runtime
/// consumes directly. Mirrors `ShardingConfig::validate` so a caller
/// that built the config without running it (e.g. tests, embedded
@@ -523,30 +552,8 @@ pub fn bootstrap(
current_replica_id: Option<u8>,
) -> Result<ShardHandles, ServerNgError> {
warm_dummy_password_hash();
- let allocator = ShardAllocator::new(&config.system.sharding.cpu_allocation)
- .map_err(ServerNgError::ShardAllocator)?;
- let assignments = allocator
- .to_shard_assignments()
- .map_err(ServerNgError::ShardAllocator)?;
+ let (assignments, total_shards) =
resolve_shard_assignments(&config.system.sharding)?;
let shards_count = assignments.len();
- if shards_count == 0 {
- return Err(ServerNgError::ShardsCountZero);
- }
- // Shard ids index `ReplicaOwnerTable` slots as `u16`. `OWNER_NONE`
- // (`u16::MAX`) is reserved as the empty-slot sentinel, so a server
- // configured with `u16::MAX` shards would mint a shard id that
- // collides with the sentinel and an owner-table lookup could never
- // tell that shard apart from an unowned slot. Reject at boot so the
- // invariant is held by the type system above this line, not by hoping
- // the operator never configures 65535 cores worth of shards.
- let total_shards = match u16::try_from(shards_count) {
- Ok(count) if count < message_bus::OWNER_NONE => count,
- _ => {
- return Err(ServerNgError::ShardsCountOverflow {
- count: shards_count,
- });
- }
- };
// Re-check the full valid range, not just the zero floor: a caller
// that built the config without running `ShardingConfig::validate`
diff --git a/core/server/Cargo.toml b/core/server/Cargo.toml
index c2f11503c..cb8f78640 100644
--- a/core/server/Cargo.toml
+++ b/core/server/Cargo.toml
@@ -48,6 +48,7 @@ chrono = { workspace = true }
clap = { workspace = true }
compio = { workspace = true }
configs = { workspace = true }
+cpu_allocation = { workspace = true }
ctrlc = { workspace = true }
cyper = { workspace = true }
cyper-axum = { workspace = true }
diff --git a/core/server/config.toml b/core/server/config.toml
index d6d0bdf24..b31bf8a1c 100644
--- a/core/server/config.toml
+++ b/core/server/config.toml
@@ -592,6 +592,14 @@ ports = { tcp = 8091, quic = 8081, http = 3001, websocket
= 8093, tcp_replica =
# + "numa:auto": Use all available numa node, cores
# + "numa:nodes=0,1;cores=4;no_ht=true": Use NUMA node 0 and 1, each nodes
use 4 cores, and no hyperthreads
cpu_allocation = "numa:auto"
+# Whether shard threads are pinned to dedicated CPU cores (default: true).
+# Pinned cores are drawn from the process's allowed CPU set (affinity/cpuset
+# mask), so the server cooperates with systemd `AllowedCPUs=` and container
+# cpusets. Set to false when the server shares cores with other workloads
+# (e.g. a multi-tenant host slicing CPU via cgroup quotas): unpinned shards
+# let the kernel scheduler place threads freely instead of piling every
+# process onto the same low-numbered cores.
+# pin_cores = true
[websocket]
enabled = true
diff --git a/core/server/src/main.rs b/core/server/src/main.rs
index 542ec354a..2f598c75c 100644
--- a/core/server/src/main.rs
+++ b/core/server/src/main.rs
@@ -303,7 +303,10 @@ fn main() -> Result<(), ServerError> {
);
// ELEVENTH DISCRETE LOADING STEP.
- let shard_allocator =
ShardAllocator::new(&config.system.sharding.cpu_allocation)?;
+ let shard_allocator = ShardAllocator::new(
+ &config.system.sharding.cpu_allocation,
+ config.system.sharding.pin_cores,
+ )?;
let shard_assignment = shard_allocator.to_shard_assignments()?;
#[cfg(feature = "disable-mimalloc")]
diff --git a/core/server/src/shard/system/cgroup_memory.rs
b/core/server/src/shard/system/cgroup_memory.rs
new file mode 100644
index 000000000..35a6d9df8
--- /dev/null
+++ b/core/server/src/shard/system/cgroup_memory.rs
@@ -0,0 +1,293 @@
+// 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.
+
+//! MemAvailable-equivalent headroom for a memory-capped cgroup.
+//!
+//! `memory.current` charges reclaimable page cache as used, so
+//! `limit - current` trends toward zero on a cache-heavy server even
+//! though the kernel reclaims that cache long before OOM. The honest
+//! number adds the reclaimable file cache back:
+//! `limit - (current - inactive_file - active_file)`, evaluated at
+//! every capped ancestor with the minimum kept, since a parent's cap
+//! binds descendants even when the leaf itself is unlimited.
+
+use std::fs;
+use std::path::{Path, PathBuf};
+
+/// Reclaimable-aware available memory for the calling process's memory
+/// cgroup. `None` when no ancestor caps memory below the host total or
+/// the cgroup fs is unreadable; callers keep their fallback then.
+pub fn cgroup_available_memory(host_total_memory: u64) -> Option<u64> {
+ let cgroup = fs::read_to_string("/proc/self/cgroup").ok()?;
+
+ let from_v2 = v2_cgroup_path(&cgroup).and_then(|path| {
+ let root = Path::new("/sys/fs/cgroup");
+ available_memory_within(&root.join(path), root, &CGROUP_V2,
host_total_memory)
+ });
+ if from_v2.is_some() {
+ return from_v2;
+ }
+
+ v1_cgroup_path(&cgroup).and_then(|path| {
+ let root = Path::new("/sys/fs/cgroup/memory");
+ available_memory_within(&root.join(path), root, &CGROUP_V1,
host_total_memory)
+ })
+}
+
+/// v1 and v2 name the limit/usage files differently, and v1 needs the
+/// `total_`-prefixed keys for hierarchical (descendants included) stats,
+/// which v2 reports by default.
+struct CgroupMemoryFiles {
+ limit: &'static str,
+ usage: &'static str,
+ reclaimable_keys: [&'static str; 2],
+}
+
+const CGROUP_V2: CgroupMemoryFiles = CgroupMemoryFiles {
+ limit: "memory.max",
+ usage: "memory.current",
+ reclaimable_keys: ["inactive_file", "active_file"],
+};
+
+const CGROUP_V1: CgroupMemoryFiles = CgroupMemoryFiles {
+ limit: "memory.limit_in_bytes",
+ usage: "memory.usage_in_bytes",
+ reclaimable_keys: ["total_inactive_file", "total_active_file"],
+};
+
+fn available_memory_within(
+ base: &Path,
+ root: &Path,
+ files: &CgroupMemoryFiles,
+ host_total_memory: u64,
+) -> Option<u64> {
+ let mut available: Option<u64> = None;
+
+ for level in base.ancestors() {
+ // An unlimited level reads as "max" (v2, unparsable) or a value
+ // beyond the host total (v1); neither constrains anything.
+ if let Some(limit) =
+ read_u64(&level.join(files.limit)).filter(|limit| *limit <=
host_total_memory)
+ {
+ let usage = read_u64(&level.join(files.usage))?;
+ let reclaimable =
+ read_reclaimable(&level.join("memory.stat"),
&files.reclaimable_keys)?;
+ let level_available =
limit.saturating_sub(usage.saturating_sub(reclaimable));
+ available =
+ Some(available.map_or(level_available, |tightest|
tightest.min(level_available)));
+ }
+ if level == root {
+ break;
+ }
+ }
+
+ available
+}
+
+fn v2_cgroup_path(cgroup: &str) -> Option<PathBuf> {
+ cgroup_relative_path(cgroup, |hierarchy_id, controllers| {
+ hierarchy_id == "0" && controllers.is_empty()
+ })
+}
+
+fn v1_cgroup_path(cgroup: &str) -> Option<PathBuf> {
+ cgroup_relative_path(cgroup, |_, controllers| {
+ controllers
+ .split(',')
+ .any(|controller| controller == "memory")
+ })
+}
+
+fn cgroup_relative_path(
+ cgroup: &str,
+ line_matches: impl Fn(&str, &str) -> bool,
+) -> Option<PathBuf> {
+ cgroup.lines().find_map(|line| {
+ let mut fields = line.splitn(3, ':');
+ let hierarchy_id = fields.next()?;
+ let controllers = fields.next()?;
+ let path = fields.next()?;
+
+ if !line_matches(hierarchy_id, controllers) {
+ return None;
+ }
+
+ Some(Path::new(path).strip_prefix("/").ok()?.to_path_buf())
+ })
+}
+
+fn read_u64(path: &Path) -> Option<u64> {
+ fs::read_to_string(path).ok()?.trim().parse().ok()
+}
+
+/// A key absent from `memory.stat` counts as zero (kernels omit
+/// feature-gated counters); only an unreadable file yields `None`.
+fn read_reclaimable(stat_path: &Path, keys: &[&'static str; 2]) -> Option<u64>
{
+ let stat = fs::read_to_string(stat_path).ok()?;
+ let mut reclaimable = 0u64;
+
+ for key in keys {
+ let value = stat.lines().find_map(|line| {
+ let (name, value) = line.split_once(' ')?;
+ if name != *key {
+ return None;
+ }
+ value.trim().parse::<u64>().ok()
+ });
+ reclaimable = reclaimable.saturating_add(value.unwrap_or(0));
+ }
+
+ Some(reclaimable)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use std::fs::{create_dir_all, write};
+ use tempfile::tempdir;
+
+ const HOST_TOTAL: u64 = 10_000;
+
+ #[test]
+ fn v2_available_memory_adds_reclaimable_file_cache_back() {
+ let root = tempdir().unwrap();
+ let leaf = root.path().join("iggy.slice");
+ create_dir_all(&leaf).unwrap();
+ write(root.path().join("memory.max"), "max").unwrap();
+ write(leaf.join("memory.max"), "1000").unwrap();
+ write(leaf.join("memory.current"), "900").unwrap();
+ write(
+ leaf.join("memory.stat"),
+ "anon 300\ninactive_file 400\nactive_file 100\n",
+ )
+ .unwrap();
+
+ let available = available_memory_within(&leaf, root.path(),
&CGROUP_V2, HOST_TOTAL);
+
+ // limit - current would report 100; 500 of the 900 is reclaimable
cache.
+ assert_eq!(available, Some(600));
+ }
+
+ #[test]
+ fn v2_tighter_parent_cap_wins_over_unlimited_leaf() {
+ let root = tempdir().unwrap();
+ let parent = root.path().join("parent");
+ let leaf = parent.join("leaf");
+ create_dir_all(&leaf).unwrap();
+ write(root.path().join("memory.max"), "max").unwrap();
+ write(parent.join("memory.max"), "500").unwrap();
+ write(parent.join("memory.current"), "450").unwrap();
+ write(
+ parent.join("memory.stat"),
+ "inactive_file 50\nactive_file 0\n",
+ )
+ .unwrap();
+ write(leaf.join("memory.max"), "max").unwrap();
+ write(leaf.join("memory.current"), "100").unwrap();
+ write(
+ leaf.join("memory.stat"),
+ "inactive_file 10\nactive_file 0\n",
+ )
+ .unwrap();
+
+ let available = available_memory_within(&leaf, root.path(),
&CGROUP_V2, HOST_TOTAL);
+
+ assert_eq!(available, Some(100));
+ }
+
+ #[test]
+ fn v2_missing_reclaimable_key_counts_as_zero() {
+ let root = tempdir().unwrap();
+ let leaf = root.path().join("leaf");
+ create_dir_all(&leaf).unwrap();
+ write(root.path().join("memory.max"), "max").unwrap();
+ write(leaf.join("memory.max"), "1000").unwrap();
+ write(leaf.join("memory.current"), "900").unwrap();
+ write(leaf.join("memory.stat"), "anon 300\ninactive_file
400\n").unwrap();
+
+ let available = available_memory_within(&leaf, root.path(),
&CGROUP_V2, HOST_TOTAL);
+
+ assert_eq!(available, Some(500));
+ }
+
+ #[test]
+ fn v2_no_capped_ancestor_yields_none() {
+ let root = tempdir().unwrap();
+ let leaf = root.path().join("leaf");
+ create_dir_all(&leaf).unwrap();
+ write(root.path().join("memory.max"), "max").unwrap();
+ write(leaf.join("memory.max"), "max").unwrap();
+ write(leaf.join("memory.current"), "100").unwrap();
+ write(
+ leaf.join("memory.stat"),
+ "inactive_file 10\nactive_file 0\n",
+ )
+ .unwrap();
+
+ let available = available_memory_within(&leaf, root.path(),
&CGROUP_V2, HOST_TOTAL);
+
+ assert_eq!(available, None);
+ }
+
+ #[test]
+ fn v2_unreadable_usage_at_capped_level_yields_none() {
+ let root = tempdir().unwrap();
+ let leaf = root.path().join("leaf");
+ create_dir_all(&leaf).unwrap();
+ write(root.path().join("memory.max"), "max").unwrap();
+ write(leaf.join("memory.max"), "1000").unwrap();
+
+ let available = available_memory_within(&leaf, root.path(),
&CGROUP_V2, HOST_TOTAL);
+
+ assert_eq!(available, None);
+ }
+
+ #[test]
+ fn v1_available_memory_uses_hierarchical_stat_keys() {
+ let root = tempdir().unwrap();
+ let leaf = root.path().join("iggy");
+ create_dir_all(&leaf).unwrap();
+ write(
+ root.path().join("memory.limit_in_bytes"),
+ u64::MAX.to_string(),
+ )
+ .unwrap();
+ write(leaf.join("memory.limit_in_bytes"), "1000").unwrap();
+ write(leaf.join("memory.usage_in_bytes"), "800").unwrap();
+ write(
+ leaf.join("memory.stat"),
+ "inactive_file 999\ntotal_inactive_file 250\ntotal_active_file
50\n",
+ )
+ .unwrap();
+
+ let available = available_memory_within(&leaf, root.path(),
&CGROUP_V1, HOST_TOTAL);
+
+ assert_eq!(available, Some(500));
+ }
+
+ #[test]
+ fn cgroup_paths_parse_v2_and_v1_lines() {
+ let hybrid =
"12:cpuset:/\n11:memory,cpuacct:/system.slice/iggy.service\n0::/user.slice";
+
+ assert_eq!(v2_cgroup_path(hybrid), Some(PathBuf::from("user.slice")));
+ assert_eq!(
+ v1_cgroup_path(hybrid),
+ Some(PathBuf::from("system.slice/iggy.service"))
+ );
+ assert_eq!(v1_cgroup_path("12:cpuset:/\n10:cpu:/"), None);
+ }
+}
diff --git a/core/server/src/shard/system/mod.rs
b/core/server/src/shard/system/mod.rs
index f326ea08e..907d2c8ab 100644
--- a/core/server/src/shard/system/mod.rs
+++ b/core/server/src/shard/system/mod.rs
@@ -15,6 +15,7 @@
// specific language governing permissions and limitations
// under the License.
+mod cgroup_memory;
pub mod clients;
pub mod cluster;
pub mod consumer_groups;
diff --git a/core/server/src/shard/system/stats.rs
b/core/server/src/shard/system/stats.rs
index dc1e92e07..6da51fce0 100644
--- a/core/server/src/shard/system/stats.rs
+++ b/core/server/src/shard/system/stats.rs
@@ -16,7 +16,9 @@
// under the License.
use crate::shard::IggyShard;
+use crate::shard::system::cgroup_memory::cgroup_available_memory;
use crate::{SEMANTIC_VERSION, VERSION};
+use cpu_allocation::allowed_cpus;
use iggy_common::{IggyDuration, IggyError, Stats};
use std::cell::RefCell;
use sysinfo::{Pid, ProcessesToUpdate, System as SysinfoSystem};
@@ -44,7 +46,8 @@ impl IggyShard {
sys.refresh_memory();
sys.refresh_processes(ProcessesToUpdate::Some(&[Pid::from_u32(process_id)]),
true);
- let total_cpu_usage = sys.global_cpu_usage();
+ let total_cpu_usage =
+ allowed_cores_cpu_usage(sys).unwrap_or_else(||
sys.global_cpu_usage());
let total_memory = sys.total_memory().into();
let available_memory = sys.available_memory().into();
let clients_count = self.client_manager.get_clients().len() as u32;
@@ -88,6 +91,17 @@ impl IggyShard {
stats.written_bytes = disk_usage.total_written_bytes.into();
stats.threads_count = process.tasks().map(|t| t.len() as
u32).unwrap_or(0);
+
+ if let Some(limits) = process
+ .cgroup_limits()
+ .filter(|limits| limits.total_memory < sys.total_memory())
+ {
+ stats.total_memory = limits.total_memory.into();
+ stats.available_memory =
cgroup_available_memory(sys.total_memory())
+ .unwrap_or(limits.free_memory)
+ .min(limits.total_memory)
+ .into();
+ }
}
let (streams_count, topics_count, partitions_count,
consumer_groups_count, stream_ids) =
@@ -142,3 +156,25 @@ impl IggyShard {
})
}
}
+
+/// Average usage over the cores this process may run on.
+///
+/// `global_cpu_usage` averages every host core, so a cpuset-confined
+/// instance would report its neighbors' load. `None` when the process
+/// is unrestricted (or an allowed core is missing from `sys.cpus()`),
+/// where the host-global number is already the right one.
+fn allowed_cores_cpu_usage(sys: &SysinfoSystem) -> Option<f32> {
+ let cpus = sys.cpus();
+ let allowed = allowed_cpus();
+ if allowed.is_empty() || allowed.len() >= cpus.len() {
+ return None;
+ }
+
+ let mut total_usage = 0.0f32;
+ for cpu_id in &allowed {
+ let name = format!("cpu{cpu_id}");
+ total_usage += cpus.iter().find(|cpu| cpu.name() == name)?.cpu_usage();
+ }
+
+ Some(total_usage / allowed.len() as f32)
+}
diff --git a/core/shard_allocator/src/lib.rs b/core/shard_allocator/src/lib.rs
index b02343a53..c1cf2395a 100644
--- a/core/shard_allocator/src/lib.rs
+++ b/core/shard_allocator/src/lib.rs
@@ -24,7 +24,7 @@
//! shard. On Linux it also pins each shard's thread to its core and
//! pins memory to the right NUMA node, so memory stays close and fast.
-use cpu_allocation::{CpuAllocation, NumaConfig};
+use cpu_allocation::{CpuAllocation, NumaConfig, allowed_cpus};
use hwlocality::Topology;
use hwlocality::bitmap::SpecializedBitmapRef;
use hwlocality::cpu::cpuset::CpuSet;
@@ -61,6 +61,19 @@ pub enum ShardingError {
node: usize,
},
+ #[error(
+ "Requested {requested} shard(s) but only {available} CPU core(s) are
allowed for this process (affinity/cpuset mask)"
+ )]
+ AllowedCpusExceeded { requested: usize, available: usize },
+
+ #[error(
+ "Configured CPU {cpu} is outside the set of cores allowed for this
process (affinity/cpuset mask)"
+ )]
+ CpuNotAllowed { cpu: usize },
+
+ #[error("Invalid CPU range {start}..{end}: start must be less than end")]
+ InvalidRange { start: usize, end: usize },
+
#[error("Invalid NUMA node: requested {requested}, only available
{available} node")]
InvalidNode { requested: usize, available: usize },
@@ -257,17 +270,54 @@ impl ShardInfo {
}
}
+/// One shard per core drawn from `allowed`, each pinned to its own core.
+fn pinned_from_allowed(allowed: &[usize], count: usize) ->
Result<Vec<ShardInfo>, ShardingError> {
+ if count > allowed.len() {
+ return Err(ShardingError::AllowedCpusExceeded {
+ requested: count,
+ available: allowed.len(),
+ });
+ }
+
+ Ok(allowed
+ .iter()
+ .take(count)
+ .map(|&cpu_id| ShardInfo {
+ cpu_set: HashSet::from([cpu_id]),
+ numa_node: None,
+ })
+ .collect())
+}
+
+/// `count` shards with no CPU affinity: `bind_cpu` becomes a no-op and the
+/// kernel scheduler places shard threads freely. This is the right mode when
+/// the process shares its cores with other workloads (e.g. a multi-tenant
+/// host slicing CPU via cgroup quotas), where per-process pinning to the same
+/// low-numbered cores would pile every tenant onto one core.
+fn unpinned(count: usize) -> Vec<ShardInfo> {
+ (0..count)
+ .map(|_| ShardInfo {
+ cpu_set: HashSet::new(),
+ numa_node: None,
+ })
+ .collect()
+}
+
/// Turns the operator's [`CpuAllocation`] choice into a concrete plan
/// of shards. Reads the NUMA topology only when the choice needs it.
pub struct ShardAllocator {
allocation: CpuAllocation,
+ pin_cores: bool,
topology: Option<Arc<NumaTopology>>,
}
impl ShardAllocator {
/// Build an allocator for the given choice. Only `NumaAware` reads
/// the machine topology up front; the simpler modes do not.
- pub fn new(allocation: &CpuAllocation) -> Result<ShardAllocator,
ShardingError> {
+ pub fn new(
+ allocation: &CpuAllocation,
+ pin_cores: bool,
+ ) -> Result<ShardAllocator, ShardingError> {
let topology = if matches!(allocation, CpuAllocation::NumaAware(_)) {
let numa_topology = NumaTopology::detect()?;
@@ -278,6 +328,7 @@ impl ShardAllocator {
Ok(Self {
allocation: allocation.clone(),
+ pin_cores,
topology,
})
}
@@ -287,39 +338,73 @@ impl ShardAllocator {
pub fn to_shard_assignments(&self) -> Result<Vec<ShardInfo>,
ShardingError> {
match &self.allocation {
CpuAllocation::All => {
+ // `available_parallelism` already accounts for both the
+ // affinity mask and any cgroup CPU quota, so a
+ // quota-restricted process gets proportionally fewer shards.
let available_cpus = available_parallelism()
.map_err(|err| ShardingError::Other {
msg: format!("Failed to get available_parallelism:
{:?}", err),
})?
.get();
- let shard_assignments: Vec<_> = (0..available_cpus)
- .map(|cpu_id| ShardInfo {
- cpu_set: HashSet::from([cpu_id]),
- numa_node: None,
- })
- .collect();
+ if !self.pin_cores {
+ info!("Using all available CPU cores ({available_cpus}
shards, unpinned)");
+ return Ok(unpinned(available_cpus));
+ }
+
+ let allowed = allowed_cpus();
+ let shard_assignments =
+ pinned_from_allowed(&allowed,
available_cpus.min(allowed.len()))?;
info!(
- "Using all available CPU cores ({} shards with affinity)",
- shard_assignments.len()
+ "Using all available CPU cores ({} shards pinned within
allowed set {:?})",
+ shard_assignments.len(),
+ allowed
);
Ok(shard_assignments)
}
CpuAllocation::Count(count) => {
- let shard_assignments = (0..*count)
- .map(|cpu_id| ShardInfo {
- cpu_set: HashSet::from([cpu_id]),
- numa_node: None,
- })
- .collect();
+ if !self.pin_cores {
+ info!("Using {count} shard(s), unpinned");
+ return Ok(unpinned(*count));
+ }
- info!("Using {count} shards with affinity to cores
0..{count}");
+ let allowed = allowed_cpus();
+ let shard_assignments = pinned_from_allowed(&allowed, *count)?;
+
+ info!(
+ "Using {count} shard(s) with affinity to cores {:?}",
+ &allowed[..*count]
+ );
Ok(shard_assignments)
}
CpuAllocation::Range(start, end) => {
+ if start >= end {
+ return Err(ShardingError::InvalidRange {
+ start: *start,
+ end: *end,
+ });
+ }
+
+ if !self.pin_cores {
+ info!(
+ "Using {} shard(s) for range {start}..{end}, unpinned",
+ end - start
+ );
+ return Ok(unpinned(end - start));
+ }
+
+ // An explicit range names exact cores and is never remapped
+ // into the allowed set; each core must be a member of it, or
+ // we fail fast instead of letting `sched_setaffinity` EINVAL
+ // later.
+ let allowed = allowed_cpus();
+ if let Some(cpu) = (*start..*end).find(|cpu|
!allowed.contains(cpu)) {
+ return Err(ShardingError::CpuNotAllowed { cpu });
+ }
+
let shard_assignments = (*start..*end)
.map(|cpu_id| ShardInfo {
cpu_set: HashSet::from([cpu_id]),
@@ -336,7 +421,16 @@ impl ShardAllocator {
}
CpuAllocation::NumaAware(numa_config) => {
let topology =
self.topology.as_ref().ok_or(ShardingError::NoTopology)?;
- self.compute_numa_assignments(topology, numa_config)
+ let assignments = self.compute_numa_assignments(topology,
numa_config)?;
+
+ if !self.pin_cores {
+ tracing::warn!(
+ "pin_cores = false with a NUMA-aware cpu_allocation:
NUMA bindings are ignored"
+ );
+ return Ok(unpinned(assignments.len()));
+ }
+
+ Ok(assignments)
}
}
}
@@ -408,3 +502,85 @@ impl ShardAllocator {
Ok(shard_infos)
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn pinned_from_allowed_draws_cores_from_the_allowed_set() {
+ let allowed = vec![2, 3, 6, 7];
+ let shards = pinned_from_allowed(&allowed, 2).unwrap();
+ assert_eq!(shards.len(), 2);
+ assert_eq!(shards[0].cpu_set, HashSet::from([2]));
+ assert_eq!(shards[1].cpu_set, HashSet::from([3]));
+ assert!(shards.iter().all(|shard| shard.numa_node.is_none()));
+ }
+
+ #[test]
+ fn pinned_from_allowed_rejects_more_shards_than_allowed_cores() {
+ let allowed = vec![0, 1];
+ let err = pinned_from_allowed(&allowed, 3).unwrap_err();
+ assert!(matches!(
+ err,
+ ShardingError::AllowedCpusExceeded {
+ requested: 3,
+ available: 2,
+ }
+ ));
+ }
+
+ #[test]
+ fn unpinned_shards_have_empty_cpu_sets() {
+ let shards = unpinned(4);
+ assert_eq!(shards.len(), 4);
+ assert!(shards.iter().all(|shard| shard.cpu_set.is_empty()));
+ assert!(shards.iter().all(|shard| shard.numa_node.is_none()));
+ }
+
+ #[test]
+ fn count_without_pinning_yields_unpinned_shards() {
+ let allocator = ShardAllocator::new(&CpuAllocation::Count(1),
false).unwrap();
+ let shards = allocator.to_shard_assignments().unwrap();
+ assert_eq!(shards.len(), 1);
+ assert!(shards[0].cpu_set.is_empty());
+ }
+
+ #[test]
+ fn count_with_pinning_stays_within_the_allowed_set() {
+ let allocator = ShardAllocator::new(&CpuAllocation::Count(1),
true).unwrap();
+ let shards = allocator.to_shard_assignments().unwrap();
+ assert_eq!(shards.len(), 1);
+ let allowed = allowed_cpus();
+ assert!(shards[0].cpu_set.iter().all(|cpu| allowed.contains(cpu)));
+ }
+
+ #[test]
+ fn range_without_pinning_keeps_shard_count() {
+ let allocator = ShardAllocator::new(&CpuAllocation::Range(0, 1),
false).unwrap();
+ let shards = allocator.to_shard_assignments().unwrap();
+ assert_eq!(shards.len(), 1);
+ assert!(shards[0].cpu_set.is_empty());
+ }
+
+ #[test]
+ fn inverted_range_is_rejected_regardless_of_pinning() {
+ for pin_cores in [false, true] {
+ let allocator = ShardAllocator::new(&CpuAllocation::Range(2, 1),
pin_cores).unwrap();
+ let err = allocator.to_shard_assignments().unwrap_err();
+ assert!(matches!(
+ err,
+ ShardingError::InvalidRange { start: 2, end: 1 }
+ ));
+ }
+ }
+
+ #[test]
+ fn bind_cpu_with_empty_set_is_a_no_op() {
+ let shard = ShardInfo {
+ cpu_set: HashSet::new(),
+ numa_node: None,
+ };
+ assert!(shard.bind_cpu().is_ok());
+ }
+}
