This is an automated email from the ASF dual-hosted git repository.
numinnex 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 15044ea44 refactor: extract shard allocator into a dedicated crate
(#3578)
15044ea44 is described below
commit 15044ea44972124b57226983fdf09542af28b978
Author: Jaya Kasa <[email protected]>
AuthorDate: Mon Jul 6 06:34:21 2026 -0400
refactor: extract shard allocator into a dedicated crate (#3578)
---
Cargo.lock | 23 +-
Cargo.toml | 4 +
core/configs/Cargo.toml | 1 +
core/configs/src/server_config/sharding.rs | 241 +------------
core/configs/src/server_config/validators.rs | 2 +-
core/{configs => cpu_allocation}/Cargo.toml | 17 +-
core/cpu_allocation/src/lib.rs | 374 +++++++++++++++++++++
core/server-ng/Cargo.toml | 1 +
core/server-ng/src/bootstrap.rs | 2 +-
core/server-ng/src/server_error.rs | 2 +-
core/server/Cargo.toml | 7 +-
core/server/src/bootstrap.rs | 2 +-
core/server/src/lib.rs | 1 -
core/server/src/main.rs | 2 +-
core/server/src/server_error.rs | 2 +-
core/{configs => shard_allocator}/Cargo.toml | 27 +-
core/shard_allocator/build.rs | 43 +++
.../src/lib.rs} | 35 +-
18 files changed, 510 insertions(+), 276 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 106aa2445..45f9b0a2d 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3152,6 +3152,7 @@ name = "configs"
version = "0.1.0"
dependencies = [
"configs_derive",
+ "cpu_allocation",
"derive_more",
"err_trail",
"figment",
@@ -3345,6 +3346,14 @@ dependencies = [
"libm",
]
+[[package]]
+name = "cpu_allocation"
+version = "0.1.0"
+dependencies = [
+ "serde",
+ "serde_json",
+]
+
[[package]]
name = "cpufeatures"
version = "0.2.17"
@@ -11684,7 +11693,6 @@ dependencies = [
"futures",
"hash32 1.0.0",
"human-repr",
- "hwlocality",
"iggy_binary_protocol",
"iggy_common",
"jsonwebtoken",
@@ -11711,6 +11719,7 @@ dependencies = [
"serde",
"serde_json",
"server_common",
+ "shard_allocator",
"slab",
"socket2 0.6.4",
"strum 0.28.0",
@@ -11792,6 +11801,7 @@ dependencies = [
"server",
"server_common",
"shard",
+ "shard_allocator",
"slab",
"socket2 0.6.4",
"strum 0.28.0",
@@ -11915,6 +11925,17 @@ dependencies = [
"tracing",
]
+[[package]]
+name = "shard_allocator"
+version = "0.1.0"
+dependencies = [
+ "cpu_allocation",
+ "hwlocality",
+ "nix",
+ "thiserror 2.0.18",
+ "tracing",
+]
+
[[package]]
name = "sharded-slab"
version = "0.1.7"
diff --git a/Cargo.toml b/Cargo.toml
index 027209cea..5ef321639 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -49,6 +49,7 @@ members = [
"core/connectors/sources/postgres_source",
"core/connectors/sources/random_source",
"core/consensus",
+ "core/cpu_allocation",
"core/harness_derive",
"core/integration",
"core/journal",
@@ -60,6 +61,7 @@ members = [
"core/server-ng",
"core/server_common",
"core/shard",
+ "core/shard_allocator",
"core/simulator",
"core/tools",
"examples/rust",
@@ -134,6 +136,7 @@ configs = { path = "core/configs", version = "0.1.0" }
configs_derive = { path = "core/configs_derive", version = "0.1.0" }
consensus = { path = "core/consensus" }
console-subscriber = "0.5.0"
+cpu_allocation = { path = "core/cpu_allocation" }
crossbeam = "0.8.4"
crossfire = "3.1.16"
csv = "1.4.0"
@@ -279,6 +282,7 @@ server = { path = "core/server" }
server-ng = { path = "core/server-ng" }
server_common = { path = "core/server_common" }
shard = { path = "core/shard" }
+shard_allocator = { path = "core/shard_allocator" }
simd-json = { version = "0.17.0", features = ["serde_impl"] }
slab = "0.4.12"
smallvec = "1.15"
diff --git a/core/configs/Cargo.toml b/core/configs/Cargo.toml
index c0f606b8a..7d5d7b104 100644
--- a/core/configs/Cargo.toml
+++ b/core/configs/Cargo.toml
@@ -24,6 +24,7 @@ publish = false
[dependencies]
configs_derive = { workspace = true }
+cpu_allocation = { workspace = true }
derive_more = { workspace = true }
err_trail = { workspace = true }
figment = { workspace = true }
diff --git a/core/configs/src/server_config/sharding.rs
b/core/configs/src/server_config/sharding.rs
index ec332462b..0d0c1fc6e 100644
--- a/core/configs/src/server_config/sharding.rs
+++ b/core/configs/src/server_config/sharding.rs
@@ -16,13 +16,18 @@
// under the License.
use iggy_common::IggyDuration;
-use serde::{Deserialize, Deserializer, Serialize, Serializer};
+use serde::{Deserialize, Serialize};
use serde_with::{DisplayFromStr, serde_as};
-use std::str::FromStr;
use std::time::Duration;
use configs::ConfigEnv;
+// `CpuAllocation`/`NumaConfig` are pure config types and live in their own
+// leaf crate so both `configs` and `shard_allocator` can share them without
+// pulling each other's heavier dependency trees. Re-exported here to keep the
+// `configs::sharding::*` path stable for existing callers.
+pub use cpu_allocation::{CpuAllocation, NumaConfig};
+
/// Default capacity of the per-shard inter-shard inbox channel. Sized
/// comfortably above the consensus working set, which is roughly
/// `PIPELINE_PREPARE_QUEUE_MAX (= 32) * replica_count * directions`
@@ -178,235 +183,3 @@ impl Default for ShardingConfig {
}
}
}
-
-#[derive(Debug, Clone, PartialEq, Default)]
-pub enum CpuAllocation {
- #[default]
- All,
- Count(usize),
- Range(usize, usize),
- NumaAware(NumaConfig),
-}
-
-/// NUMA specific configuration
-#[derive(Debug, Clone, PartialEq, Default)]
-pub struct NumaConfig {
- /// Which NUMA nodes to use (empty = auto-detect all)
- pub nodes: Vec<usize>,
- /// Cores per node to use (0 = use all available)
- pub cores_per_node: usize,
- /// skip hyperthread sibling
- pub avoid_hyperthread: bool,
-}
-
-impl CpuAllocation {
- fn parse_numa(s: &str) -> Result<CpuAllocation, String> {
- let params = s
- .strip_prefix("numa:")
- .ok_or_else(|| "Numa config must start with 'numa:'".to_string())?;
-
- if params == "auto" {
- return Ok(CpuAllocation::NumaAware(NumaConfig {
- nodes: vec![],
- cores_per_node: 0,
- avoid_hyperthread: true,
- }));
- }
-
- let mut nodes = Vec::new();
- let mut cores_per_node = 0;
- let mut avoid_hyperthread = true;
-
- for param in params.split(';') {
- let kv: Vec<&str> = param.split('=').collect();
- if kv.len() != 2 {
- return Err(format!(
- "Invalid NUMA parameter: '{param}', only available: 'auto'"
- ));
- }
-
- match kv[0] {
- "nodes" => {
- nodes = kv[1]
- .split(',')
- .map(|n| {
- n.parse::<usize>()
- .map_err(|_| format!("Invalid node number:
{n}"))
- })
- .collect::<Result<Vec<_>, _>>()?;
- }
- "cores" => {
- cores_per_node = kv[1]
- .parse::<usize>()
- .map_err(|_| format!("Invalid cores value: {}",
kv[1]))?;
- }
- "no_ht" => {
- avoid_hyperthread = kv[1]
- .parse::<bool>()
- .map_err(|_| format!("Invalid no ht value: {}",
kv[1]))?;
- }
- _ => {
- return Err(format!(
- "Unknown NUMA parameter: {}, example:
numa:nodes=0;cores=4;no_ht=true",
- kv[0]
- ));
- }
- }
- }
-
- Ok(CpuAllocation::NumaAware(NumaConfig {
- nodes,
- cores_per_node,
- avoid_hyperthread,
- }))
- }
-}
-
-impl FromStr for CpuAllocation {
- type Err = String;
-
- fn from_str(s: &str) -> Result<Self, Self::Err> {
- match s {
- "all" => Ok(CpuAllocation::All),
- s if s.starts_with("numa:") => Self::parse_numa(s),
- s if s.contains("..") => {
- let parts: Vec<&str> = s.split("..").collect();
- if parts.len() != 2 {
- return Err(format!("Invalid range format: {s}. Expected
'start..end'"));
- }
- let start = parts[0]
- .parse::<usize>()
- .map_err(|_| format!("Invalid start value: {}",
parts[0]))?;
- let end = parts[1]
- .parse::<usize>()
- .map_err(|_| format!("Invalid end value: {}", parts[1]))?;
- Ok(CpuAllocation::Range(start, end))
- }
- s => {
- let count = s
- .parse::<usize>()
- .map_err(|_| format!("Invalid shard count: {s}"))?;
- Ok(CpuAllocation::Count(count))
- }
- }
- }
-}
-
-impl Serialize for CpuAllocation {
- fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
- where
- S: Serializer,
- {
- match self {
- CpuAllocation::All => serializer.serialize_str("all"),
- CpuAllocation::Count(n) => serializer.serialize_u64(*n as u64),
- CpuAllocation::Range(start, end) => {
- serializer.serialize_str(&format!("{start}..{end}"))
- }
- CpuAllocation::NumaAware(numa) => {
- if numa.nodes.is_empty() && numa.cores_per_node == 0 {
- serializer.serialize_str("numa:auto")
- } else {
- let nodes_str = numa
- .nodes
- .iter()
- .map(|n| n.to_string())
- .collect::<Vec<_>>()
- .join(",");
-
- let full_str = format!(
- "numa:nodes={};cores={};no_ht={}",
- nodes_str, numa.cores_per_node, numa.avoid_hyperthread
- );
-
- serializer.serialize_str(&full_str)
- }
- }
- }
- }
-}
-
-impl<'de> Deserialize<'de> for CpuAllocation {
- fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
- where
- D: Deserializer<'de>,
- {
- #[derive(Deserialize)]
- #[serde(untagged)]
- enum CpuAllocationHelper {
- String(String),
- Number(usize),
- }
-
- match CpuAllocationHelper::deserialize(deserializer)? {
- CpuAllocationHelper::String(s) => {
- CpuAllocation::from_str(&s).map_err(serde::de::Error::custom)
- }
- CpuAllocationHelper::Number(n) => Ok(CpuAllocation::Count(n)),
- }
- }
-}
-
-#[cfg(test)]
-mod tests {
- use super::*;
-
- #[test]
- fn test_parse_all() {
- assert_eq!(CpuAllocation::from_str("all").unwrap(),
CpuAllocation::All);
- }
-
- #[test]
- fn test_parse_count() {
- assert_eq!(
- CpuAllocation::from_str("4").unwrap(),
- CpuAllocation::Count(4)
- );
- }
-
- #[test]
- fn test_parse_range() {
- assert_eq!(
- CpuAllocation::from_str("2..8").unwrap(),
- CpuAllocation::Range(2, 8)
- );
- }
-
- #[test]
- fn test_parse_numa_auto() {
- let result = CpuAllocation::from_str("numa:auto").unwrap();
- match result {
- CpuAllocation::NumaAware(numa) => {
- assert!(numa.nodes.is_empty());
- assert_eq!(numa.cores_per_node, 0);
- assert!(numa.avoid_hyperthread);
- }
- _ => panic!("Expected NumaAware"),
- }
- }
-
- #[test]
- fn test_parse_numa_explicit() {
- let result =
CpuAllocation::from_str("numa:nodes=0,1;cores=4;no_ht=true").unwrap();
- match result {
- CpuAllocation::NumaAware(numa) => {
- assert_eq!(numa.nodes, vec![0, 1]);
- assert_eq!(numa.cores_per_node, 4);
- assert!(numa.avoid_hyperthread);
- }
- _ => panic!("Expected NumaAware"),
- }
- }
-
- #[test]
- fn test_numa_explicit_serde_roundtrip() {
- let original = CpuAllocation::NumaAware(NumaConfig {
- nodes: vec![0, 1],
- cores_per_node: 4,
- avoid_hyperthread: true,
- });
- let serialized = serde_json::to_string(&original).unwrap();
- let deserialized: CpuAllocation =
serde_json::from_str(&serialized).unwrap();
- assert_eq!(original, deserialized);
- }
-}
diff --git a/core/configs/src/server_config/validators.rs
b/core/configs/src/server_config/validators.rs
index 93c61fb9b..0aafe1ef5 100644
--- a/core/configs/src/server_config/validators.rs
+++ b/core/configs/src/server_config/validators.rs
@@ -497,7 +497,7 @@ impl Validatable<ConfigurationError> for ShardingConfig {
Ok(())
}
// NUMA topology validation requires hwlocality (runtime dep).
- // Full NUMA validation happens in server::shard_allocator at
startup.
+ // Full NUMA validation happens in shard_allocator at startup.
CpuAllocation::NumaAware(_) => Ok(()),
}
}
diff --git a/core/configs/Cargo.toml b/core/cpu_allocation/Cargo.toml
similarity index 69%
copy from core/configs/Cargo.toml
copy to core/cpu_allocation/Cargo.toml
index c0f606b8a..74eea3230 100644
--- a/core/configs/Cargo.toml
+++ b/core/cpu_allocation/Cargo.toml
@@ -16,24 +16,15 @@
# under the License.
[package]
-name = "configs"
+name = "cpu_allocation"
version = "0.1.0"
+description = "Shard CPU/NUMA allocation config types (CpuAllocation,
NumaConfig) parsed from the iggy server config."
edition = "2024"
license = "Apache-2.0"
publish = false
[dependencies]
-configs_derive = { workspace = true }
-derive_more = { workspace = true }
-err_trail = { workspace = true }
-figment = { workspace = true }
-iggy_common = { workspace = true }
-jsonwebtoken = { workspace = true }
serde = { workspace = true }
+
+[dev-dependencies]
serde_json = { workspace = true }
-serde_with = { workspace = true }
-server_common = { workspace = true }
-static-toml = { workspace = true }
-strum = { workspace = true }
-tracing = { workspace = true }
-tungstenite = { workspace = true }
diff --git a/core/cpu_allocation/src/lib.rs b/core/cpu_allocation/src/lib.rs
new file mode 100644
index 000000000..5391e73b1
--- /dev/null
+++ b/core/cpu_allocation/src/lib.rs
@@ -0,0 +1,374 @@
+// 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.
+
+//! `cpu_allocation`: tiny config types that say how many CPU cores the
+//! server should grab for its shards, and how.
+//!
+//! These two types ([`CpuAllocation`] and [`NumaConfig`]) are read from
+//! 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.
+
+use serde::{Deserialize, Deserializer, Serialize, Serializer};
+use std::str::FromStr;
+
+/// Tell server how many CPU cores to grab for shards, and how.
+///
+/// Server make one shard per core. This say which cores. Pick one:
+/// - `All`: take every core machine have.
+/// - `Count(n)`: take first `n` cores.
+/// - `Range(a, b)`: take cores `a` up to (not including) `b`.
+/// - `NumaAware(..)`: smart pick by NUMA node, keep memory close to core.
+///
+/// Parse from a string in TOML, e.g. `"all"`, `4`, `"2..8"`,
+/// `"numa:auto"`, or `"numa:nodes=0,1;cores=4;no_ht=true"`.
+#[derive(Debug, Clone, PartialEq, Default)]
+pub enum CpuAllocation {
+ #[default]
+ All,
+ Count(usize),
+ Range(usize, usize),
+ NumaAware(NumaConfig),
+}
+
+/// Knobs for NUMA-aware core picking.
+///
+/// NUMA = machine split into groups (nodes). Each node have own cores
+/// and own memory. Memory of same node is fast; far node is slow. This
+/// struct say which nodes to use and how many cores from each.
+#[derive(Debug, Clone, PartialEq, Default)]
+pub struct NumaConfig {
+ /// Which NUMA nodes to use. Empty means: use all of them.
+ pub nodes: Vec<usize>,
+ /// How many cores to take from each node. `0` means: take all.
+ pub cores_per_node: usize,
+ /// `true` means skip hyperthread twins, use only one thread per core.
+ pub avoid_hyperthread: bool,
+}
+
+impl CpuAllocation {
+ fn parse_numa(s: &str) -> Result<CpuAllocation, String> {
+ let params = s
+ .strip_prefix("numa:")
+ .ok_or_else(|| "Numa config must start with 'numa:'".to_string())?;
+
+ if params == "auto" {
+ return Ok(CpuAllocation::NumaAware(NumaConfig {
+ nodes: vec![],
+ cores_per_node: 0,
+ avoid_hyperthread: true,
+ }));
+ }
+
+ let mut nodes = Vec::new();
+ let mut cores_per_node = 0;
+ let mut avoid_hyperthread = true;
+
+ for param in params.split(';') {
+ let kv: Vec<&str> = param.split('=').collect();
+ if kv.len() != 2 {
+ return Err(format!(
+ "Invalid NUMA parameter: '{param}', only available: 'auto'"
+ ));
+ }
+
+ match kv[0] {
+ "nodes" => {
+ nodes = kv[1]
+ .split(',')
+ .map(|n| {
+ n.parse::<usize>()
+ .map_err(|_| format!("Invalid node number:
{n}"))
+ })
+ .collect::<Result<Vec<_>, _>>()?;
+ }
+ "cores" => {
+ cores_per_node = kv[1]
+ .parse::<usize>()
+ .map_err(|_| format!("Invalid cores value: {}",
kv[1]))?;
+ }
+ "no_ht" => {
+ avoid_hyperthread = kv[1]
+ .parse::<bool>()
+ .map_err(|_| format!("Invalid no ht value: {}",
kv[1]))?;
+ }
+ _ => {
+ return Err(format!(
+ "Unknown NUMA parameter: {}, example:
numa:nodes=0;cores=4;no_ht=true",
+ kv[0]
+ ));
+ }
+ }
+ }
+
+ Ok(CpuAllocation::NumaAware(NumaConfig {
+ nodes,
+ cores_per_node,
+ avoid_hyperthread,
+ }))
+ }
+}
+
+impl FromStr for CpuAllocation {
+ type Err = String;
+
+ fn from_str(s: &str) -> Result<Self, Self::Err> {
+ match s {
+ "all" => Ok(CpuAllocation::All),
+ s if s.starts_with("numa:") => Self::parse_numa(s),
+ s if s.contains("..") => {
+ let parts: Vec<&str> = s.split("..").collect();
+ if parts.len() != 2 {
+ return Err(format!("Invalid range format: {s}. Expected
'start..end'"));
+ }
+ let start = parts[0]
+ .parse::<usize>()
+ .map_err(|_| format!("Invalid start value: {}",
parts[0]))?;
+ let end = parts[1]
+ .parse::<usize>()
+ .map_err(|_| format!("Invalid end value: {}", parts[1]))?;
+ Ok(CpuAllocation::Range(start, end))
+ }
+ s => {
+ let count = s
+ .parse::<usize>()
+ .map_err(|_| format!("Invalid shard count: {s}"))?;
+ Ok(CpuAllocation::Count(count))
+ }
+ }
+ }
+}
+
+impl Serialize for CpuAllocation {
+ fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+ where
+ S: Serializer,
+ {
+ match self {
+ CpuAllocation::All => serializer.serialize_str("all"),
+ CpuAllocation::Count(n) => serializer.serialize_u64(*n as u64),
+ CpuAllocation::Range(start, end) => {
+ serializer.serialize_str(&format!("{start}..{end}"))
+ }
+ CpuAllocation::NumaAware(numa) => {
+ if numa.nodes.is_empty() && numa.cores_per_node == 0 {
+ serializer.serialize_str("numa:auto")
+ } else {
+ let nodes_str = numa
+ .nodes
+ .iter()
+ .map(|n| n.to_string())
+ .collect::<Vec<_>>()
+ .join(",");
+
+ let full_str = format!(
+ "numa:nodes={};cores={};no_ht={}",
+ nodes_str, numa.cores_per_node, numa.avoid_hyperthread
+ );
+
+ serializer.serialize_str(&full_str)
+ }
+ }
+ }
+ }
+}
+
+impl<'de> Deserialize<'de> for CpuAllocation {
+ fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+ where
+ D: Deserializer<'de>,
+ {
+ #[derive(Deserialize)]
+ #[serde(untagged)]
+ enum CpuAllocationHelper {
+ String(String),
+ Number(usize),
+ }
+
+ match CpuAllocationHelper::deserialize(deserializer)? {
+ CpuAllocationHelper::String(s) => {
+ CpuAllocation::from_str(&s).map_err(serde::de::Error::custom)
+ }
+ CpuAllocationHelper::Number(n) => Ok(CpuAllocation::Count(n)),
+ }
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn test_parse_all() {
+ assert_eq!(CpuAllocation::from_str("all").unwrap(),
CpuAllocation::All);
+ }
+
+ #[test]
+ fn test_parse_count() {
+ assert_eq!(
+ CpuAllocation::from_str("4").unwrap(),
+ CpuAllocation::Count(4)
+ );
+ }
+
+ #[test]
+ fn test_parse_range() {
+ assert_eq!(
+ CpuAllocation::from_str("2..8").unwrap(),
+ CpuAllocation::Range(2, 8)
+ );
+ }
+
+ #[test]
+ fn test_parse_numa_auto() {
+ let result = CpuAllocation::from_str("numa:auto").unwrap();
+ match result {
+ CpuAllocation::NumaAware(numa) => {
+ assert!(numa.nodes.is_empty());
+ assert_eq!(numa.cores_per_node, 0);
+ assert!(numa.avoid_hyperthread);
+ }
+ _ => panic!("Expected NumaAware"),
+ }
+ }
+
+ #[test]
+ fn test_parse_numa_explicit() {
+ let result =
CpuAllocation::from_str("numa:nodes=0,1;cores=4;no_ht=true").unwrap();
+ match result {
+ CpuAllocation::NumaAware(numa) => {
+ assert_eq!(numa.nodes, vec![0, 1]);
+ assert_eq!(numa.cores_per_node, 4);
+ assert!(numa.avoid_hyperthread);
+ }
+ _ => panic!("Expected NumaAware"),
+ }
+ }
+
+ #[test]
+ fn test_numa_explicit_serde_roundtrip() {
+ let original = CpuAllocation::NumaAware(NumaConfig {
+ nodes: vec![0, 1],
+ cores_per_node: 4,
+ avoid_hyperthread: true,
+ });
+ let serialized = serde_json::to_string(&original).unwrap();
+ let deserialized: CpuAllocation =
serde_json::from_str(&serialized).unwrap();
+ assert_eq!(original, deserialized);
+ }
+
+ #[test]
+ fn test_parse_invalid_range_too_many_parts() {
+ assert!(CpuAllocation::from_str("1..2..3").is_err());
+ }
+
+ #[test]
+ fn test_parse_invalid_range_start() {
+ assert!(CpuAllocation::from_str("x..8").is_err());
+ }
+
+ #[test]
+ fn test_parse_invalid_range_end() {
+ assert!(CpuAllocation::from_str("2..y").is_err());
+ }
+
+ #[test]
+ fn test_parse_invalid_count() {
+ assert!(CpuAllocation::from_str("abc").is_err());
+ }
+
+ #[test]
+ fn test_parse_numa_missing_prefix() {
+ assert!(CpuAllocation::parse_numa("nodes=0").is_err());
+ }
+
+ #[test]
+ fn test_parse_numa_param_without_equals() {
+ assert!(CpuAllocation::from_str("numa:nodes").is_err());
+ }
+
+ #[test]
+ fn test_parse_numa_invalid_node_number() {
+ assert!(CpuAllocation::from_str("numa:nodes=a").is_err());
+ }
+
+ #[test]
+ fn test_parse_numa_invalid_cores() {
+ assert!(CpuAllocation::from_str("numa:cores=x").is_err());
+ }
+
+ #[test]
+ fn test_parse_numa_invalid_no_ht() {
+ assert!(CpuAllocation::from_str("numa:no_ht=maybe").is_err());
+ }
+
+ #[test]
+ fn test_parse_numa_unknown_param() {
+ assert!(CpuAllocation::from_str("numa:foo=1").is_err());
+ }
+
+ #[test]
+ fn test_serialize_all() {
+ assert_eq!(
+ serde_json::to_string(&CpuAllocation::All).unwrap(),
+ "\"all\""
+ );
+ }
+
+ #[test]
+ fn test_serialize_count() {
+ assert_eq!(
+ serde_json::to_string(&CpuAllocation::Count(4)).unwrap(),
+ "4"
+ );
+ }
+
+ #[test]
+ fn test_serialize_range() {
+ assert_eq!(
+ serde_json::to_string(&CpuAllocation::Range(2, 8)).unwrap(),
+ "\"2..8\""
+ );
+ }
+
+ #[test]
+ fn test_serialize_numa_auto() {
+ let auto = CpuAllocation::NumaAware(NumaConfig {
+ nodes: vec![],
+ cores_per_node: 0,
+ avoid_hyperthread: true,
+ });
+ assert_eq!(serde_json::to_string(&auto).unwrap(), "\"numa:auto\"");
+ }
+
+ #[test]
+ fn test_deserialize_number() {
+ assert_eq!(
+ serde_json::from_str::<CpuAllocation>("4").unwrap(),
+ CpuAllocation::Count(4)
+ );
+ }
+
+ #[test]
+ fn test_deserialize_string() {
+ assert_eq!(
+ serde_json::from_str::<CpuAllocation>("\"all\"").unwrap(),
+ CpuAllocation::All
+ );
+ }
+}
diff --git a/core/server-ng/Cargo.toml b/core/server-ng/Cargo.toml
index bf42650ed..3d2a2ab5a 100644
--- a/core/server-ng/Cargo.toml
+++ b/core/server-ng/Cargo.toml
@@ -149,6 +149,7 @@ serde = { workspace = true }
server = { workspace = true }
server_common = { workspace = true }
shard = { workspace = true }
+shard_allocator = { workspace = true }
slab = { workspace = true }
socket2 = { workspace = true }
strum = { workspace = true }
diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs
index 46bd45c29..e430fed3e 100644
--- a/core/server-ng/src/bootstrap.rs
+++ b/core/server-ng/src/bootstrap.rs
@@ -72,9 +72,9 @@ use rustls::pki_types::ServerName;
use server_common::bootstrap::create_directories;
use server_common::executor::create_shard_executor;
use server_common::sharding::{IggyNamespace, PartitionLocation, ShardId};
+use shard_allocator::{ShardAllocator, ShardInfo};
// TODO: decouple bootstrap/storage helpers and logging from the `server`
crate.
use server::log::logger::Logging;
-use server::shard_allocator::{ShardAllocator, ShardInfo};
use server::streaming::users::user::User as LegacyUser;
use server::{IGGY_ROOT_PASSWORD_ENV, IGGY_ROOT_USERNAME_ENV};
use shard::builder::IggyShardBuilder;
diff --git a/core/server-ng/src/server_error.rs
b/core/server-ng/src/server_error.rs
index a3070dedd..a25ebfec7 100644
--- a/core/server-ng/src/server_error.rs
+++ b/core/server-ng/src/server_error.rs
@@ -18,8 +18,8 @@
use metadata::impls::recovery::RecoveryError;
// TODO: decouple logging errors from the `server` crate.
use server::server_error::LogError;
-use server::shard_allocator::ShardingError;
use shard::ShardCtorError;
+use shard_allocator::ShardingError;
use thiserror::Error;
#[derive(Debug, Error)]
diff --git a/core/server/Cargo.toml b/core/server/Cargo.toml
index c48abee12..09c352bae 100644
--- a/core/server/Cargo.toml
+++ b/core/server/Cargo.toml
@@ -90,6 +90,7 @@ send_wrapper = { workspace = true }
serde = { workspace = true }
serde_json = { workspace = true }
server_common = { workspace = true }
+shard_allocator = { workspace = true }
slab = { workspace = true }
socket2 = { workspace = true }
strum = { workspace = true }
@@ -105,11 +106,5 @@ tracing-subscriber = { workspace = true }
ulid = { workspace = true }
uuid = { workspace = true }
-[target.'cfg(not(target_env = "musl"))'.dependencies]
-hwlocality = { workspace = true }
-
-[target.'cfg(target_env = "musl")'.dependencies]
-hwlocality = { workspace = true, features = ["vendored"] }
-
[build-dependencies]
vergen-git2 = { workspace = true }
diff --git a/core/server/src/bootstrap.rs b/core/server/src/bootstrap.rs
index f35a43a2c..eb42c0539 100644
--- a/core/server/src/bootstrap.rs
+++ b/core/server/src/bootstrap.rs
@@ -33,7 +33,6 @@ use crate::{
frame::ShardFrame,
},
},
- shard_allocator::ShardInfo,
state::system::{StreamState, TopicState, UserState},
streaming::{
partitions::{
@@ -57,6 +56,7 @@ use iggy_common::{
MIN_USERNAME_LENGTH,
},
};
+use shard_allocator::ShardInfo;
use slab::Slab;
use std::{env, sync::Arc};
use tracing::{info, warn};
diff --git a/core/server/src/lib.rs b/core/server/src/lib.rs
index 6807fb6ba..70a3e6c4d 100644
--- a/core/server/src/lib.rs
+++ b/core/server/src/lib.rs
@@ -41,7 +41,6 @@ pub mod quic;
pub mod sender;
pub mod server_error;
pub mod shard;
-pub mod shard_allocator;
pub mod state;
pub mod streaming;
pub mod tcp;
diff --git a/core/server/src/main.rs b/core/server/src/main.rs
index d9baaac85..fd1908260 100644
--- a/core/server/src/main.rs
+++ b/core/server/src/main.rs
@@ -40,7 +40,6 @@ use server::metadata::{Metadata, create_metadata_handles};
use server::server_error::ServerError;
use server::shard::system::info::SystemInfo;
use server::shard::{IggyShard, calculate_shard_assignment};
-use server::shard_allocator::ShardAllocator;
use server::state::file::FileState;
use server::state::system::SystemState;
use server::streaming::clients::client_manager::{Client, ClientManager};
@@ -49,6 +48,7 @@ use server::streaming::storage::SystemStorage;
use server::streaming::utils::ptr::EternalPtr;
use server_common::MemoryPool;
use server_common::sharding::{IggyNamespace, PartitionLocation, ShardId};
+use shard_allocator::ShardAllocator;
use std::panic::AssertUnwindSafe;
use std::rc::Rc;
use std::str::FromStr;
diff --git a/core/server/src/server_error.rs b/core/server/src/server_error.rs
index 6a63fac8f..5dde456e4 100644
--- a/core/server/src/server_error.rs
+++ b/core/server/src/server_error.rs
@@ -36,7 +36,7 @@ error_set!(
NumaError := {
#[display("{0}")]
- Sharding(crate::shard_allocator::ShardingError),
+ Sharding(shard_allocator::ShardingError),
}
ConfigurationError := {
diff --git a/core/configs/Cargo.toml b/core/shard_allocator/Cargo.toml
similarity index 66%
copy from core/configs/Cargo.toml
copy to core/shard_allocator/Cargo.toml
index c0f606b8a..d44574376 100644
--- a/core/configs/Cargo.toml
+++ b/core/shard_allocator/Cargo.toml
@@ -16,24 +16,23 @@
# under the License.
[package]
-name = "configs"
+name = "shard_allocator"
version = "0.1.0"
+description = "CPU and NUMA shard allocation for the iggy server, backed by
hwloc."
edition = "2024"
license = "Apache-2.0"
publish = false
[dependencies]
-configs_derive = { workspace = true }
-derive_more = { workspace = true }
-err_trail = { workspace = true }
-figment = { workspace = true }
-iggy_common = { workspace = true }
-jsonwebtoken = { workspace = true }
-serde = { workspace = true }
-serde_json = { workspace = true }
-serde_with = { workspace = true }
-server_common = { workspace = true }
-static-toml = { workspace = true }
-strum = { workspace = true }
+cpu_allocation = { workspace = true }
+thiserror = { workspace = true }
tracing = { workspace = true }
-tungstenite = { workspace = true }
+
+[target.'cfg(not(target_env = "musl"))'.dependencies]
+hwlocality = { workspace = true }
+
+[target.'cfg(target_env = "musl")'.dependencies]
+hwlocality = { workspace = true, features = ["vendored"] }
+
+[target.'cfg(target_os = "linux")'.dependencies]
+nix = { workspace = true }
diff --git a/core/shard_allocator/build.rs b/core/shard_allocator/build.rs
new file mode 100644
index 000000000..d66e93f07
--- /dev/null
+++ b/core/shard_allocator/build.rs
@@ -0,0 +1,43 @@
+// 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.
+
+// Vendored `hwloc` references `cbrt`, which makes the linker pull in a
+// `libm`. musl folds the math functions into its `libc`, so the Rust
+// musl sysroot ships no `libm.a`. Without one, `-lm` falls through to
+// the host glibc's `libm.a`, whose `cbrt` needs glibc-internal
+// `__frexp`/`__ldexp` symbols that do not exist on musl, and the static
+// link fails. Drop an empty `libm.a` stub on the search path so `-lm`
+// resolves to nothing and `cbrt` is satisfied later by musl's own
+// `libc`. No effect on non-musl targets.
+
+use std::env;
+use std::fs;
+use std::path::Path;
+
+fn main() {
+ if env::var("CARGO_CFG_TARGET_ENV").as_deref() != Ok("musl") {
+ return;
+ }
+
+ let out_dir = env::var("OUT_DIR").expect("OUT_DIR is set by cargo for
build scripts");
+ let stub = Path::new(&out_dir).join("libm.a");
+
+ // `!<arch>\n` is the canonical header of an empty `ar` archive.
+ fs::write(&stub, b"!<arch>\n").expect("write empty libm.a stub");
+
+ println!("cargo:rustc-link-search=native={out_dir}");
+}
diff --git a/core/server/src/shard_allocator.rs
b/core/shard_allocator/src/lib.rs
similarity index 85%
rename from core/server/src/shard_allocator.rs
rename to core/shard_allocator/src/lib.rs
index 26a1f0e38..b02343a53 100644
--- a/core/server/src/shard_allocator.rs
+++ b/core/shard_allocator/src/lib.rs
@@ -15,7 +15,16 @@
// specific language governing permissions and limitations
// under the License.
-use configs::sharding::{CpuAllocation, NumaConfig};
+//! `shard_allocator`: decide which CPU cores each shard lives on.
+//!
+//! The server makes many shards and wants each one to run on its own
+//! core so they do not fight over CPU time. This crate reads the
+//! operator's choice ([`CpuAllocation`] from the config), looks at the
+//! real machine with `hwloc`, and hands back one [`ShardInfo`] per
+//! 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 hwlocality::Topology;
use hwlocality::bitmap::SpecializedBitmapRef;
use hwlocality::cpu::cpuset::CpuSet;
@@ -28,6 +37,9 @@ use std::sync::Arc;
use std::thread::available_parallelism;
use tracing::info;
+/// All the ways shard allocation can go wrong: machine has no NUMA,
+/// hwloc cannot read the topology, the operator asked for more cores
+/// than exist, or the OS refused to pin a thread or its memory.
#[derive(Debug, thiserror::Error)]
pub enum ShardingError {
#[error("Failed to detect topology: {msg}")]
@@ -56,6 +68,10 @@ pub enum ShardingError {
Other { msg: String },
}
+/// A snapshot of the machine's NUMA layout, read once from `hwloc`.
+///
+/// Holds how many NUMA nodes there are and, for each node, how many
+/// real (physical) cores and how many threads (logical cores) it has.
#[derive(Debug)]
pub struct NumaTopology {
topology: Topology,
@@ -65,6 +81,8 @@ pub struct NumaTopology {
}
impl NumaTopology {
+ /// Ask `hwloc` to read this machine's NUMA layout right now.
+ /// Errors if hwloc fails or the machine reports no NUMA nodes.
pub fn detect() -> Result<NumaTopology, ShardingError> {
let topology =
Topology::new().map_err(|e| ShardingError::TopologyDetection {
msg: e.to_string() })?;
@@ -110,10 +128,13 @@ impl NumaTopology {
})
}
+ /// How many real cores this node has. Returns `0` if no such node.
pub fn physical_cores_for_node(&self, node: usize) -> usize {
self.physical_cores_per_node.get(node).copied().unwrap_or(0)
}
+ /// How many threads (logical cores) this node has, hyperthreads
+ /// included. Returns `0` if no such node.
pub fn logical_cores_for_node(&self, node: usize) -> usize {
self.logical_cores_per_node.get(node).copied().unwrap_or(0)
}
@@ -161,6 +182,8 @@ impl NumaTopology {
}
}
+/// One shard's home: which CPU cores it may run on, and which NUMA
+/// node its memory should sit near (`None` means do not pin memory).
#[derive(Debug, Clone)]
pub struct ShardInfo {
pub cpu_set: HashSet<usize>,
@@ -168,6 +191,8 @@ pub struct ShardInfo {
}
impl ShardInfo {
+ /// Pin the calling thread to this shard's cores. On non-Linux this
+ /// does nothing (no-op). Empty core set also does nothing.
pub fn bind_cpu(&self) -> Result<(), ShardingError> {
#[cfg(target_os = "linux")]
{
@@ -196,6 +221,8 @@ impl ShardInfo {
Ok(())
}
+ /// Pin the calling thread's memory to this shard's NUMA node so
+ /// allocations stay local and fast. Does nothing if no node is set.
pub fn bind_memory(&self) -> Result<(), ShardingError> {
if let Some(node_id) = self.numa_node {
let topology = Topology::new().map_err(|err|
ShardingError::TopologyDetection {
@@ -230,12 +257,16 @@ impl ShardInfo {
}
}
+/// 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,
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> {
let topology = if matches!(allocation, CpuAllocation::NumaAware(_)) {
let numa_topology = NumaTopology::detect()?;
@@ -251,6 +282,8 @@ impl ShardAllocator {
})
}
+ /// Produce the final list of shards, one [`ShardInfo`] each, based
+ /// on the chosen [`CpuAllocation`]. This is the main entry point.
pub fn to_shard_assignments(&self) -> Result<Vec<ShardInfo>,
ShardingError> {
match &self.allocation {
CpuAllocation::All => {
