Merge pull request #63 from greatest-ape/cpu-pinning-2022-03-30

Improve CPU pinning

.github/workflows/cargo-build-and-test.yml

@@ -20,8 +20,8 @@ jobs:
     - name: Build
       run: |
         cargo build --verbose -p aquatic_udp --features "cpu-pinning"
-        cargo build --verbose -p aquatic_http --features "cpu-pinning"
-        cargo build --verbose -p aquatic_ws --features "cpu-pinning"
+        cargo build --verbose -p aquatic_http
+        cargo build --verbose -p aquatic_ws
     - name: Run tests
       run: cargo test --verbose --workspace --all-targets
 

Cargo.lock

@@ -75,6 +75,8 @@ dependencies = [
  "anyhow",
  "aquatic_toml_config",
  "arc-swap",
+ "duplicate",
+ "glommio",
  "hashbrown 0.12.0",
  "hex",
  "hwloc",
@@ -103,6 +105,7 @@ dependencies = [
  "futures-rustls",
  "glommio",
  "itoa 1.0.1",
+ "libc",
  "log",
  "memchr",
  "mimalloc",
@@ -874,6 +877,16 @@ version = "0.15.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "77c90badedccf4105eca100756a0b1289e191f6fcbdadd3cee1d2f614f97da8f"
 
+[[package]]
+name = "duplicate"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4a2535fcf1437c9de0d91a3921351c474258abdb6440cd03bb5a7cf0547e7214"
+dependencies = [
+ "heck 0.4.0",
+ "proc-macro-error",
+]
+
 [[package]]
 name = "either"
 version = "1.6.1"
@@ -1299,6 +1312,12 @@ dependencies = [
  "unicode-segmentation",
 ]
 
+[[package]]
+name = "heck"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2540771e65fc8cb83cd6e8a237f70c319bd5c29f78ed1084ba5d50eeac86f7f9"
+
 [[package]]
 name = "hermit-abi"
 version = "0.1.19"
@@ -2026,6 +2045,30 @@ dependencies = [
  "nix",
 ]
 
+[[package]]
+name = "proc-macro-error"
+version = "1.0.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "da25490ff9892aab3fcf7c36f08cfb902dd3e71ca0f9f9517bea02a73a5ce38c"
+dependencies = [
+ "proc-macro-error-attr",
+ "proc-macro2",
+ "quote",
+ "syn",
+ "version_check",
+]
+
+[[package]]
+name = "proc-macro-error-attr"
+version = "1.0.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a1be40180e52ecc98ad80b184934baf3d0d29f979574e439af5a55274b35f869"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "version_check",
+]
+
 [[package]]
 name = "proc-macro2"
 version = "1.0.36"
@@ -2607,7 +2650,7 @@ checksum = "eee35713129561f5e55c554bba1c378e2a7e67f81257b7311183de98c50e6f94"
 dependencies = [
  "dotenv",
  "either",
- "heck",
+ "heck 0.3.3",
  "once_cell",
  "proc-macro2",
  "quote",

aquatic_common/Cargo.toml

@@ -12,7 +12,8 @@ readme = "../README.md"
 name = "aquatic_common"
 
 [features]
-cpu-pinning = ["hwloc", "libc"]
+with-glommio = ["glommio"]
+with-hwloc = ["hwloc"]
 rustls-config = ["rustls", "rustls-pemfile"]
 
 [dependencies]
@@ -21,18 +22,20 @@ aquatic_toml_config = { version = "0.2.0", path = "../aquatic_toml_config" }
 ahash = "0.7"
 anyhow = "1"
 arc-swap = "1"
+duplicate = "0.4"
 hashbrown = "0.12"
 hex = "0.4"
 indexmap-amortized = "1"
+libc = "0.2"
 log = "0.4"
 privdrop = "0.5"
 rand = { version = "0.8", features = ["small_rng"] }
 serde = { version = "1", features = ["derive"] }
 
-# cpu-pinning
+# Optional
+glommio = { version = "0.7", optional = true }
 hwloc = { version = "0.5", optional = true }
-libc = { version = "0.2", optional = true }
 
 # rustls-config
 rustls = { version = "0.20", optional = true }
-rustls-pemfile = { version = "0.3", optional = true }
+rustls-pemfile = { version = "0.3", optional = true }

aquatic_common/src/cpu_pinning.rs

@@ -1,42 +1,88 @@
+//! Experimental CPU pinning
+
 use aquatic_toml_config::TomlConfig;
-use hwloc::{CpuSet, ObjectType, Topology, CPUBIND_THREAD};
 use serde::{Deserialize, Serialize};
 
-#[derive(Clone, Debug, PartialEq, TomlConfig, Serialize, Deserialize)]
+#[derive(Clone, Copy, Debug, PartialEq, TomlConfig, Serialize, Deserialize)]
 #[serde(rename_all = "lowercase")]
-pub enum CpuPinningMode {
+pub enum CpuPinningDirection {
     Ascending,
     Descending,
 }
 
-impl Default for CpuPinningMode {
+impl Default for CpuPinningDirection {
     fn default() -> Self {
         Self::Ascending
     }
 }
 
-#[derive(Clone, Debug, PartialEq, TomlConfig, Deserialize)]
-pub struct CpuPinningConfig {
-    pub active: bool,
-    pub mode: CpuPinningMode,
-    pub core_offset: usize,
+#[cfg(feature = "with-glommio")]
+#[derive(Clone, Copy, Debug, PartialEq, TomlConfig, Serialize, Deserialize)]
+#[serde(rename_all = "kebab-case")]
+pub enum HyperThreadMapping {
+    System,
+    Subsequent,
+    Split,
 }
 
-impl Default for CpuPinningConfig {
+#[cfg(feature = "with-glommio")]
+impl Default for HyperThreadMapping {
     fn default() -> Self {
-        Self {
-            active: false,
-            mode: Default::default(),
-            core_offset: 0,
-        }
+        Self::System
     }
 }
 
-impl CpuPinningConfig {
-    pub fn default_for_load_test() -> Self {
-        Self {
-            mode: CpuPinningMode::Descending,
-            ..Default::default()
+pub trait CpuPinningConfig {
+    fn active(&self) -> bool;
+    fn direction(&self) -> CpuPinningDirection;
+    #[cfg(feature = "with-glommio")]
+    fn hyperthread(&self) -> HyperThreadMapping;
+    fn core_offset(&self) -> usize;
+}
+
+// Do these shenanigans for compatibility with aquatic_toml_config
+#[duplicate::duplicate_item(
+    mod_name struct_name cpu_pinning_direction;
+    [asc] [CpuPinningConfigAsc] [CpuPinningDirection::Ascending];
+    [desc] [CpuPinningConfigDesc] [CpuPinningDirection::Descending];
+)]
+pub mod mod_name {
+    use super::*;
+
+    /// Experimental cpu pinning
+    #[derive(Clone, Debug, PartialEq, TomlConfig, Deserialize)]
+    pub struct struct_name {
+        pub active: bool,
+        pub direction: CpuPinningDirection,
+        #[cfg(feature = "with-glommio")]
+        pub hyperthread: HyperThreadMapping,
+        pub core_offset: usize,
+    }
+
+    impl Default for struct_name {
+        fn default() -> Self {
+            Self {
+                active: false,
+                direction: cpu_pinning_direction,
+                #[cfg(feature = "with-glommio")]
+                hyperthread: Default::default(),
+                core_offset: 0,
+            }
+        }
+    }
+    impl CpuPinningConfig for struct_name {
+        fn active(&self) -> bool {
+            self.active
+        }
+        fn direction(&self) -> CpuPinningDirection {
+            self.direction
+        }
+        #[cfg(feature = "with-glommio")]
+        fn hyperthread(&self) -> HyperThreadMapping {
+            self.hyperthread
+        }
+        fn core_offset(&self) -> usize {
+            self.core_offset
         }
     }
 }
@@ -45,38 +91,194 @@ impl CpuPinningConfig {
 pub enum WorkerIndex {
     SocketWorker(usize),
     RequestWorker(usize),
-    Other,
+    Util,
 }
 
 impl WorkerIndex {
-    fn get_core_index(
-        self,
-        config: &CpuPinningConfig,
+    pub fn get_core_index<C: CpuPinningConfig>(
+        &self,
+        config: &C,
         socket_workers: usize,
-        core_count: usize,
+        request_workers: usize,
+        num_cores: usize,
     ) -> usize {
         let ascending_index = match self {
-            Self::Other => config.core_offset,
-            Self::SocketWorker(index) => config.core_offset + 1 + index,
-            Self::RequestWorker(index) => config.core_offset + 1 + socket_workers + index,
+            Self::SocketWorker(index) => config.core_offset() + index,
+            Self::RequestWorker(index) => config.core_offset() + socket_workers + index,
+            Self::Util => config.core_offset() + socket_workers + request_workers,
         };
 
-        match config.mode {
-            CpuPinningMode::Ascending => ascending_index,
-            CpuPinningMode::Descending => core_count - 1 - ascending_index,
+        let max_core_index = num_cores - 1;
+
+        let ascending_index = ascending_index.min(max_core_index);
+
+        match config.direction() {
+            CpuPinningDirection::Ascending => ascending_index,
+            CpuPinningDirection::Descending => max_core_index - ascending_index,
         }
     }
 }
 
+#[cfg(feature = "with-glommio")]
+pub mod glommio {
+    use ::glommio::{CpuSet, Placement};
+
+    use super::*;
+
+    fn get_cpu_set() -> anyhow::Result<CpuSet> {
+        CpuSet::online().map_err(|err| anyhow::anyhow!("Couldn't get CPU set: {:#}", err))
+    }
+
+    fn get_num_cpu_cores() -> anyhow::Result<usize> {
+        get_cpu_set()?
+            .iter()
+            .map(|l| l.core)
+            .max()
+            .map(|index| index + 1)
+            .ok_or(anyhow::anyhow!("CpuSet is empty"))
+    }
+
+    fn logical_cpus_string(cpu_set: &CpuSet) -> String {
+        let mut logical_cpus = cpu_set.iter().map(|l| l.cpu).collect::<Vec<usize>>();
+
+        logical_cpus.sort_unstable();
+
+        logical_cpus
+            .into_iter()
+            .map(|cpu| cpu.to_string())
+            .collect::<Vec<String>>()
+            .join(", ")
+    }
+
+    fn get_worker_cpu_set<C: CpuPinningConfig>(
+        config: &C,
+        socket_workers: usize,
+        request_workers: usize,
+        worker_index: WorkerIndex,
+    ) -> anyhow::Result<CpuSet> {
+        let num_cpu_cores = get_num_cpu_cores()?;
+
+        let core_index =
+            worker_index.get_core_index(config, socket_workers, request_workers, num_cpu_cores);
+
+        let too_many_workers = match (&config.hyperthread(), &config.direction()) {
+            (
+                HyperThreadMapping::Split | HyperThreadMapping::Subsequent,
+                CpuPinningDirection::Ascending,
+            ) => core_index >= num_cpu_cores / 2,
+            (
+                HyperThreadMapping::Split | HyperThreadMapping::Subsequent,
+                CpuPinningDirection::Descending,
+            ) => core_index < num_cpu_cores / 2,
+            (_, _) => false,
+        };
+
+        if too_many_workers {
+            return Err(anyhow::anyhow!("CPU pinning: total number of workers (including the single utility worker) can not exceed number of virtual CPUs / 2 - core_offset in this hyperthread mapping mode"));
+        }
+
+        let cpu_set = match config.hyperthread() {
+            HyperThreadMapping::System => get_cpu_set()?.filter(|l| l.core == core_index),
+            HyperThreadMapping::Split => match config.direction() {
+                CpuPinningDirection::Ascending => get_cpu_set()?
+                    .filter(|l| l.cpu == core_index || l.cpu == core_index + num_cpu_cores / 2),
+                CpuPinningDirection::Descending => get_cpu_set()?
+                    .filter(|l| l.cpu == core_index || l.cpu == core_index - num_cpu_cores / 2),
+            },
+            HyperThreadMapping::Subsequent => {
+                let cpu_index_offset = match config.direction() {
+                    // 0 -> 0 and 1
+                    // 1 -> 2 and 3
+                    // 2 -> 4 and 5
+                    CpuPinningDirection::Ascending => core_index * 2,
+                    // 15 -> 14 and 15
+                    // 14 -> 12 and 13
+                    // 13 -> 10 and 11
+                    CpuPinningDirection::Descending => num_cpu_cores - 2 * (num_cpu_cores - core_index),
+                };
+
+                get_cpu_set()?
+                    .filter(|l| l.cpu == cpu_index_offset || l.cpu == cpu_index_offset + 1)
+            }
+        };
+
+        if cpu_set.is_empty() {
+            Err(anyhow::anyhow!(
+                "CPU pinning: produced empty CPU set for {:?}. Try decreasing number of workers",
+                worker_index
+            ))
+        } else {
+            ::log::info!(
+                "Logical CPUs for {:?}: {}",
+                worker_index,
+                logical_cpus_string(&cpu_set)
+            );
+
+            Ok(cpu_set)
+        }
+    }
+
+    pub fn get_worker_placement<C: CpuPinningConfig>(
+        config: &C,
+        socket_workers: usize,
+        request_workers: usize,
+        worker_index: WorkerIndex,
+    ) -> anyhow::Result<Placement> {
+        if config.active() {
+            let cpu_set =
+                get_worker_cpu_set(config, socket_workers, request_workers, worker_index)?;
+
+            Ok(Placement::Fenced(cpu_set))
+        } else {
+            Ok(Placement::Unbound)
+        }
+    }
+
+    pub fn set_affinity_for_util_worker<C: CpuPinningConfig>(
+        config: &C,
+        socket_workers: usize,
+        request_workers: usize,
+    ) -> anyhow::Result<()> {
+        let worker_cpu_set =
+            get_worker_cpu_set(config, socket_workers, request_workers, WorkerIndex::Util)?;
+
+        unsafe {
+            let mut set: libc::cpu_set_t = ::std::mem::zeroed();
+
+            for cpu_location in worker_cpu_set {
+                libc::CPU_SET(cpu_location.cpu, &mut set);
+            }
+
+            let status = libc::pthread_setaffinity_np(
+                libc::pthread_self(),
+                ::std::mem::size_of::<libc::cpu_set_t>(),
+                &set,
+            );
+
+            if status != 0 {
+                return Err(anyhow::Error::new(::std::io::Error::from_raw_os_error(
+                    status,
+                )));
+            }
+        }
+
+        Ok(())
+    }
+}
+
 /// Pin current thread to a suitable core
 ///
 /// Requires hwloc (`apt-get install libhwloc-dev`)
-pub fn pin_current_if_configured_to(
-    config: &CpuPinningConfig,
+#[cfg(feature = "with-hwloc")]
+pub fn pin_current_if_configured_to<C: CpuPinningConfig>(
+    config: &C,
     socket_workers: usize,
+    request_workers: usize,
     worker_index: WorkerIndex,
 ) {
-    if config.active {
+    use hwloc::{CpuSet, ObjectType, Topology, CPUBIND_THREAD};
+
+    if config.active() {
         let mut topology = Topology::new();
 
         let core_cpu_sets: Vec<CpuSet> = topology
@@ -86,7 +288,10 @@ pub fn pin_current_if_configured_to(
             .map(|core| core.allowed_cpuset().expect("hwloc: get core cpu set"))
             .collect();
 
-        let core_index = worker_index.get_core_index(config, socket_workers, core_cpu_sets.len());
+        let num_cores = core_cpu_sets.len();
+
+        let core_index =
+            worker_index.get_core_index(config, socket_workers, request_workers, num_cores);
 
         let cpu_set = core_cpu_sets
             .get(core_index)

aquatic_common/src/lib.rs

@@ -5,7 +5,6 @@ use ahash::RandomState;
 use rand::Rng;
 
 pub mod access_list;
-#[cfg(feature = "cpu-pinning")]
 pub mod cpu_pinning;
 pub mod privileges;
 #[cfg(feature = "rustls-config")]

aquatic_http/Cargo.toml

@@ -15,12 +15,9 @@ name = "aquatic_http"
 [[bin]]
 name = "aquatic_http"
 
-[features]
-cpu-pinning = ["aquatic_common/cpu-pinning"]
-
 [dependencies]
 aquatic_cli_helpers = { version = "0.2.0", path = "../aquatic_cli_helpers" }
-aquatic_common = { version = "0.2.0", path = "../aquatic_common", features = ["rustls-config"] }
+aquatic_common = { version = "0.2.0", path = "../aquatic_common", features = ["rustls-config", "with-glommio"] }
 aquatic_http_protocol = { version = "0.2.0", path = "../aquatic_http_protocol" }
 aquatic_toml_config = { version = "0.2.0", path = "../aquatic_toml_config" }
 
@@ -31,6 +28,7 @@ futures-lite = "1"
 futures-rustls = "0.22"
 glommio = "0.7"
 itoa = "1"
+libc = "0.2"
 log = "0.4"
 mimalloc = { version = "0.1", default-features = false }
 memchr = "2"

aquatic_http/src/config.rs

@@ -1,6 +1,6 @@
 use std::{net::SocketAddr, path::PathBuf};
 
-use aquatic_common::{access_list::AccessListConfig, privileges::PrivilegeConfig};
+use aquatic_common::{access_list::AccessListConfig, privileges::PrivilegeConfig, cpu_pinning::asc::CpuPinningConfigAsc};
 use aquatic_toml_config::TomlConfig;
 use serde::Deserialize;
 
@@ -23,8 +23,7 @@ pub struct Config {
     pub cleaning: CleaningConfig,
     pub privileges: PrivilegeConfig,
     pub access_list: AccessListConfig,
-    #[cfg(feature = "cpu-pinning")]
-    pub cpu_pinning: aquatic_common::cpu_pinning::CpuPinningConfig,
+    pub cpu_pinning: CpuPinningConfigAsc,
 }
 
 impl Default for Config {
@@ -38,7 +37,6 @@ impl Default for Config {
             cleaning: CleaningConfig::default(),
             privileges: PrivilegeConfig::default(),
             access_list: AccessListConfig::default(),
-            #[cfg(feature = "cpu-pinning")]
             cpu_pinning: Default::default(),
         }
     }

aquatic_http/src/lib.rs

@@ -1,7 +1,10 @@
-#[cfg(feature = "cpu-pinning")]
-use aquatic_common::cpu_pinning::{pin_current_if_configured_to, WorkerIndex};
 use aquatic_common::{
-    access_list::update_access_list, privileges::drop_privileges_after_socket_binding,
+    access_list::update_access_list,
+    cpu_pinning::{
+        glommio::{get_worker_placement, set_affinity_for_util_worker},
+        WorkerIndex,
+    },
+    privileges::drop_privileges_after_socket_binding,
     rustls_config::create_rustls_config,
 };
 use common::State;
@@ -34,12 +37,13 @@ pub fn run(config: Config) -> ::anyhow::Result<()> {
         ::std::thread::spawn(move || run_inner(config, state));
     }
 
-    #[cfg(feature = "cpu-pinning")]
-    pin_current_if_configured_to(
-        &config.cpu_pinning,
-        config.socket_workers,
-        WorkerIndex::Other,
-    );
+    if config.cpu_pinning.active {
+        set_affinity_for_util_worker(
+            &config.cpu_pinning,
+            config.socket_workers,
+            config.request_workers,
+        )?;
+    }
 
     for signal in &mut signals {
         match signal {
@@ -76,16 +80,15 @@ pub fn run_inner(config: Config, state: State) -> anyhow::Result<()> {
         let response_mesh_builder = response_mesh_builder.clone();
         let num_bound_sockets = num_bound_sockets.clone();
 
-        let builder = LocalExecutorBuilder::default().name("socket");
+        let placement = get_worker_placement(
+            &config.cpu_pinning,
+            config.socket_workers,
+            config.request_workers,
+            WorkerIndex::SocketWorker(i),
+        )?;
+        let builder = LocalExecutorBuilder::new(placement).name("socket");
 
         let executor = builder.spawn(move || async move {
-            #[cfg(feature = "cpu-pinning")]
-            pin_current_if_configured_to(
-                &config.cpu_pinning,
-                config.socket_workers,
-                WorkerIndex::SocketWorker(i),
-            );
-
             workers::socket::run_socket_worker(
                 config,
                 state,
@@ -106,16 +109,15 @@ pub fn run_inner(config: Config, state: State) -> anyhow::Result<()> {
         let request_mesh_builder = request_mesh_builder.clone();
         let response_mesh_builder = response_mesh_builder.clone();
 
-        let builder = LocalExecutorBuilder::default().name("request");
+        let placement = get_worker_placement(
+            &config.cpu_pinning,
+            config.socket_workers,
+            config.request_workers,
+            WorkerIndex::RequestWorker(i),
+        )?;
+        let builder = LocalExecutorBuilder::new(placement).name("request");
 
         let executor = builder.spawn(move || async move {
-            #[cfg(feature = "cpu-pinning")]
-            pin_current_if_configured_to(
-                &config.cpu_pinning,
-                config.socket_workers,
-                WorkerIndex::RequestWorker(i),
-            );
-
             workers::request::run_request_worker(
                 config,
                 state,
@@ -135,12 +137,13 @@ pub fn run_inner(config: Config, state: State) -> anyhow::Result<()> {
     )
     .unwrap();
 
-    #[cfg(feature = "cpu-pinning")]
-    pin_current_if_configured_to(
-        &config.cpu_pinning,
-        config.socket_workers,
-        WorkerIndex::Other,
-    );
+    if config.cpu_pinning.active {
+        set_affinity_for_util_worker(
+            &config.cpu_pinning,
+            config.socket_workers,
+            config.request_workers,
+        )?;
+    }
 
     for executor in executors {
         executor

aquatic_http_load_test/Cargo.toml

@@ -12,12 +12,9 @@ readme = "../README.md"
 [[bin]]
 name = "aquatic_http_load_test"
 
-[features]
-cpu-pinning = ["aquatic_common/cpu-pinning"]
-
 [dependencies]
 aquatic_cli_helpers = { version = "0.2.0", path = "../aquatic_cli_helpers" }
-aquatic_common = { version = "0.2.0", path = "../aquatic_common" }
+aquatic_common = { version = "0.2.0", path = "../aquatic_common", features = ["with-glommio"] }
 aquatic_http_protocol = { version = "0.2.0", path = "../aquatic_http_protocol" }
 aquatic_toml_config = { version = "0.2.0", path = "../aquatic_toml_config" }
 

aquatic_http_load_test/src/config.rs

@@ -1,6 +1,7 @@
 use std::net::SocketAddr;
 
 use aquatic_cli_helpers::LogLevel;
+use aquatic_common::cpu_pinning::desc::CpuPinningConfigDesc;
 use aquatic_toml_config::TomlConfig;
 use serde::Deserialize;
 
@@ -22,8 +23,7 @@ pub struct Config {
     pub url_suffix: String,
     pub duration: usize,
     pub torrents: TorrentConfig,
-    #[cfg(feature = "cpu-pinning")]
-    pub cpu_pinning: aquatic_common::cpu_pinning::CpuPinningConfig,
+    pub cpu_pinning: CpuPinningConfigDesc,
 }
 
 impl aquatic_cli_helpers::Config for Config {
@@ -61,8 +61,7 @@ impl Default for Config {
             url_suffix: "".into(),
             duration: 0,
             torrents: TorrentConfig::default(),
-            #[cfg(feature = "cpu-pinning")]
-            cpu_pinning: aquatic_common::cpu_pinning::CpuPinningConfig::default_for_load_test(),
+            cpu_pinning: Default::default(),
         }
     }
 }

aquatic_http_load_test/src/main.rs

@@ -3,8 +3,8 @@ use std::thread;
 use std::time::{Duration, Instant};
 
 use ::glommio::LocalExecutorBuilder;
-#[cfg(feature = "cpu-pinning")]
-use aquatic_common::cpu_pinning::{pin_current_if_configured_to, WorkerIndex};
+use aquatic_common::cpu_pinning::glommio::{get_worker_placement, set_affinity_for_util_worker};
+use aquatic_common::cpu_pinning::WorkerIndex;
 use rand::prelude::*;
 use rand_distr::Pareto;
 
@@ -55,8 +55,6 @@ fn run(config: Config) -> ::anyhow::Result<()> {
         pareto: Arc::new(pareto),
     };
 
-    // Start socket workers
-
     let tls_config = create_tls_config().unwrap();
 
     for i in 0..config.num_workers {
@@ -64,26 +62,23 @@ fn run(config: Config) -> ::anyhow::Result<()> {
         let tls_config = tls_config.clone();
         let state = state.clone();
 
-        LocalExecutorBuilder::default()
-            .spawn(move || async move {
-                #[cfg(feature = "cpu-pinning")]
-                pin_current_if_configured_to(
-                    &config.cpu_pinning,
-                    config.num_workers,
-                    WorkerIndex::SocketWorker(i),
-                );
+        let placement = get_worker_placement(
+            &config.cpu_pinning,
+            config.num_workers,
+            0,
+            WorkerIndex::SocketWorker(i),
+        )?;
 
+        LocalExecutorBuilder::new(placement)
+            .spawn(move || async move {
                 run_socket_thread(config, tls_config, state).await.unwrap();
             })
             .unwrap();
     }
 
-    #[cfg(feature = "cpu-pinning")]
-    pin_current_if_configured_to(
-        &config.cpu_pinning,
-        config.num_workers as usize,
-        WorkerIndex::Other,
-    );
+    if config.cpu_pinning.active {
+        set_affinity_for_util_worker(&config.cpu_pinning, config.num_workers, 0)?;
+    }
 
     monitor_statistics(state, &config);
 

aquatic_udp/Cargo.toml

@@ -16,7 +16,7 @@ name = "aquatic_udp"
 name = "aquatic_udp"
 
 [features]
-cpu-pinning = ["aquatic_common/cpu-pinning"]
+cpu-pinning = ["aquatic_common/with-hwloc"]
 
 [dependencies]
 aquatic_cli_helpers = { version = "0.2.0", path = "../aquatic_cli_helpers" }

aquatic_udp/src/config.rs

@@ -35,7 +35,7 @@ pub struct Config {
     pub privileges: PrivilegeConfig,
     pub access_list: AccessListConfig,
     #[cfg(feature = "cpu-pinning")]
-    pub cpu_pinning: aquatic_common::cpu_pinning::CpuPinningConfig,
+    pub cpu_pinning: aquatic_common::cpu_pinning::asc::CpuPinningConfigAsc,
 }
 
 impl Default for Config {

aquatic_udp/src/lib.rs

@@ -75,6 +75,7 @@ pub fn run(config: Config) -> ::anyhow::Result<()> {
                 pin_current_if_configured_to(
                     &config.cpu_pinning,
                     config.socket_workers,
+                    config.request_workers,
                     WorkerIndex::RequestWorker(i),
                 );
 
@@ -104,6 +105,7 @@ pub fn run(config: Config) -> ::anyhow::Result<()> {
                 pin_current_if_configured_to(
                     &config.cpu_pinning,
                     config.socket_workers,
+                    config.request_workers,
                     WorkerIndex::SocketWorker(i),
                 );
 
@@ -130,7 +132,8 @@ pub fn run(config: Config) -> ::anyhow::Result<()> {
                 pin_current_if_configured_to(
                     &config.cpu_pinning,
                     config.socket_workers,
-                    WorkerIndex::Other,
+                    config.request_workers,
+                    WorkerIndex::Util,
                 );
 
                 workers::statistics::run_statistics_worker(config, state);
@@ -149,7 +152,8 @@ pub fn run(config: Config) -> ::anyhow::Result<()> {
     pin_current_if_configured_to(
         &config.cpu_pinning,
         config.socket_workers,
-        WorkerIndex::Other,
+        config.request_workers,
+        WorkerIndex::Util,
     );
 
     for signal in &mut signals {

aquatic_udp_load_test/Cargo.toml

@@ -9,12 +9,12 @@ repository = "https://github.com/greatest-ape/aquatic"
 keywords = ["udp", "benchmark", "peer-to-peer", "torrent", "bittorrent"]
 readme = "../README.md"
 
+[features]
+cpu-pinning = ["aquatic_common/with-hwloc"]
+
 [[bin]]
 name = "aquatic_udp_load_test"
 
-[features]
-cpu-pinning = ["aquatic_common/cpu-pinning"]
-
 [dependencies]
 aquatic_cli_helpers = { version = "0.2.0", path = "../aquatic_cli_helpers" }
 aquatic_common = { version = "0.2.0", path = "../aquatic_common" }

aquatic_udp_load_test/src/config.rs

@@ -4,7 +4,7 @@ use serde::Deserialize;
 
 use aquatic_cli_helpers::LogLevel;
 #[cfg(feature = "cpu-pinning")]
-use aquatic_common::cpu_pinning::CpuPinningConfig;
+use aquatic_common::cpu_pinning::desc::CpuPinningConfigDesc;
 use aquatic_toml_config::TomlConfig;
 
 /// aquatic_udp_load_test configuration
@@ -24,7 +24,7 @@ pub struct Config {
     pub network: NetworkConfig,
     pub requests: RequestConfig,
     #[cfg(feature = "cpu-pinning")]
-    pub cpu_pinning: CpuPinningConfig,
+    pub cpu_pinning: CpuPinningConfigDesc,
 }
 
 impl Default for Config {
@@ -37,7 +37,7 @@ impl Default for Config {
             network: NetworkConfig::default(),
             requests: RequestConfig::default(),
             #[cfg(feature = "cpu-pinning")]
-            cpu_pinning: CpuPinningConfig::default_for_load_test(),
+            cpu_pinning: Default::default(),
         }
     }
 }

aquatic_udp_load_test/src/main.rs

@@ -84,6 +84,7 @@ fn run(config: Config) -> ::anyhow::Result<()> {
             pin_current_if_configured_to(
                 &config.cpu_pinning,
                 config.workers as usize,
+                0,
                 WorkerIndex::SocketWorker(i as usize),
             );
 
@@ -95,7 +96,8 @@ fn run(config: Config) -> ::anyhow::Result<()> {
     pin_current_if_configured_to(
         &config.cpu_pinning,
         config.workers as usize,
-        WorkerIndex::Other,
+        0,
+        WorkerIndex::Util,
     );
 
     monitor_statistics(state, &config);

aquatic_ws/Cargo.toml

@@ -9,19 +9,15 @@ repository = "https://github.com/greatest-ape/aquatic"
 keywords = ["webtorrent", "websocket", "peer-to-peer", "torrent", "bittorrent"]
 readme = "../README.md"
 
-
 [lib]
 name = "aquatic_ws"
 
 [[bin]]
 name = "aquatic_ws"
 
-[features]
-cpu-pinning = ["aquatic_common/cpu-pinning"]
-
 [dependencies]
 aquatic_cli_helpers = { version = "0.2.0", path = "../aquatic_cli_helpers" }
-aquatic_common = { version = "0.2.0", path = "../aquatic_common", features = ["rustls-config"] }
+aquatic_common = { version = "0.2.0", path = "../aquatic_common", features = ["rustls-config", "with-glommio"] }
 aquatic_toml_config = { version = "0.2.0", path = "../aquatic_toml_config" }
 aquatic_ws_protocol = { version = "0.2.0", path = "../aquatic_ws_protocol" }
 

aquatic_ws/src/config.rs

@@ -1,8 +1,7 @@
 use std::net::SocketAddr;
 use std::path::PathBuf;
 
-#[cfg(feature = "cpu-pinning")]
-use aquatic_common::cpu_pinning::CpuPinningConfig;
+use aquatic_common::cpu_pinning::asc::CpuPinningConfigAsc;
 use aquatic_common::{access_list::AccessListConfig, privileges::PrivilegeConfig};
 use serde::Deserialize;
 
@@ -26,8 +25,7 @@ pub struct Config {
     pub cleaning: CleaningConfig,
     pub privileges: PrivilegeConfig,
     pub access_list: AccessListConfig,
-    #[cfg(feature = "cpu-pinning")]
-    pub cpu_pinning: CpuPinningConfig,
+    pub cpu_pinning: CpuPinningConfigAsc,
 }
 
 impl Default for Config {
@@ -41,7 +39,6 @@ impl Default for Config {
             cleaning: CleaningConfig::default(),
             privileges: PrivilegeConfig::default(),
             access_list: AccessListConfig::default(),
-            #[cfg(feature = "cpu-pinning")]
             cpu_pinning: Default::default(),
         }
     }

aquatic_ws/src/lib.rs

@@ -4,13 +4,13 @@ pub mod workers;
 
 use std::sync::{atomic::AtomicUsize, Arc};
 
+use aquatic_common::cpu_pinning::glommio::{get_worker_placement, set_affinity_for_util_worker};
+use aquatic_common::cpu_pinning::WorkerIndex;
 use aquatic_common::rustls_config::create_rustls_config;
 use glommio::{channels::channel_mesh::MeshBuilder, prelude::*};
 use signal_hook::{consts::SIGUSR1, iterator::Signals};
 
 use aquatic_common::access_list::update_access_list;
-#[cfg(feature = "cpu-pinning")]
-use aquatic_common::cpu_pinning::{pin_current_if_configured_to, WorkerIndex};
 use aquatic_common::privileges::drop_privileges_after_socket_binding;
 
 use common::*;
@@ -35,12 +35,13 @@ pub fn run(config: Config) -> ::anyhow::Result<()> {
         ::std::thread::spawn(move || run_workers(config, state));
     }
 
-    #[cfg(feature = "cpu-pinning")]
-    pin_current_if_configured_to(
-        &config.cpu_pinning,
-        config.socket_workers,
-        WorkerIndex::Other,
-    );
+    if config.cpu_pinning.active {
+        set_affinity_for_util_worker(
+            &config.cpu_pinning,
+            config.socket_workers,
+            config.request_workers,
+        )?;
+    }
 
     for signal in &mut signals {
         match signal {
@@ -77,16 +78,15 @@ fn run_workers(config: Config, state: State) -> anyhow::Result<()> {
         let response_mesh_builder = response_mesh_builder.clone();
         let num_bound_sockets = num_bound_sockets.clone();
 
-        let builder = LocalExecutorBuilder::default().name("socket");
+        let placement = get_worker_placement(
+            &config.cpu_pinning,
+            config.socket_workers,
+            config.request_workers,
+            WorkerIndex::SocketWorker(i),
+        )?;
+        let builder = LocalExecutorBuilder::new(placement).name("socket");
 
         let executor = builder.spawn(move || async move {
-            #[cfg(feature = "cpu-pinning")]
-            pin_current_if_configured_to(
-                &config.cpu_pinning,
-                config.socket_workers,
-                WorkerIndex::SocketWorker(i),
-            );
-
             workers::socket::run_socket_worker(
                 config,
                 state,
@@ -107,16 +107,15 @@ fn run_workers(config: Config, state: State) -> anyhow::Result<()> {
         let request_mesh_builder = request_mesh_builder.clone();
         let response_mesh_builder = response_mesh_builder.clone();
 
-        let builder = LocalExecutorBuilder::default().name("request");
+        let placement = get_worker_placement(
+            &config.cpu_pinning,
+            config.socket_workers,
+            config.request_workers,
+            WorkerIndex::RequestWorker(i),
+        )?;
+        let builder = LocalExecutorBuilder::new(placement).name("request");
 
         let executor = builder.spawn(move || async move {
-            #[cfg(feature = "cpu-pinning")]
-            pin_current_if_configured_to(
-                &config.cpu_pinning,
-                config.socket_workers,
-                WorkerIndex::RequestWorker(i),
-            );
-
             workers::request::run_request_worker(
                 config,
                 state,
@@ -136,12 +135,13 @@ fn run_workers(config: Config, state: State) -> anyhow::Result<()> {
     )
     .unwrap();
 
-    #[cfg(feature = "cpu-pinning")]
-    pin_current_if_configured_to(
-        &config.cpu_pinning,
-        config.socket_workers,
-        WorkerIndex::Other,
-    );
+    if config.cpu_pinning.active {
+        set_affinity_for_util_worker(
+            &config.cpu_pinning,
+            config.socket_workers,
+            config.request_workers,
+        )?;
+    }
 
     for executor in executors {
         executor

aquatic_ws_load_test/Cargo.toml

@@ -12,12 +12,9 @@ readme = "../README.md"
 [[bin]]
 name = "aquatic_ws_load_test"
 
-[features]
-cpu-pinning = ["aquatic_common/cpu-pinning"]
-
 [dependencies]
 aquatic_cli_helpers = { version = "0.2.0", path = "../aquatic_cli_helpers" }
-aquatic_common = { version = "0.2.0", path = "../aquatic_common", features = ["rustls-config"] }
+aquatic_common = { version = "0.2.0", path = "../aquatic_common", features = ["with-glommio"] }
 aquatic_toml_config = { version = "0.2.0", path = "../aquatic_toml_config" }
 aquatic_ws_protocol = { version = "0.2.0", path = "../aquatic_ws_protocol" }
 

aquatic_ws_load_test/src/config.rs

@@ -1,8 +1,7 @@
 use std::net::SocketAddr;
 
 use aquatic_cli_helpers::LogLevel;
-#[cfg(feature = "cpu-pinning")]
-use aquatic_common::cpu_pinning::CpuPinningConfig;
+use aquatic_common::cpu_pinning::desc::CpuPinningConfigDesc;
 use aquatic_toml_config::TomlConfig;
 use serde::Deserialize;
 
@@ -17,8 +16,7 @@ pub struct Config {
     pub connection_creation_interval_ms: u64,
     pub duration: usize,
     pub torrents: TorrentConfig,
-    #[cfg(feature = "cpu-pinning")]
-    pub cpu_pinning: CpuPinningConfig,
+    pub cpu_pinning: CpuPinningConfigDesc,
 }
 
 impl aquatic_cli_helpers::Config for Config {
@@ -37,8 +35,7 @@ impl Default for Config {
             connection_creation_interval_ms: 10,
             duration: 0,
             torrents: TorrentConfig::default(),
-            #[cfg(feature = "cpu-pinning")]
-            cpu_pinning: CpuPinningConfig::default_for_load_test(),
+            cpu_pinning: Default::default(),
         }
     }
 }

aquatic_ws_load_test/src/main.rs

@@ -2,8 +2,8 @@ use std::sync::{atomic::Ordering, Arc};
 use std::thread;
 use std::time::{Duration, Instant};
 
-#[cfg(feature = "cpu-pinning")]
-use aquatic_common::cpu_pinning::{pin_current_if_configured_to, WorkerIndex};
+use aquatic_common::cpu_pinning::glommio::{get_worker_placement, set_affinity_for_util_worker};
+use aquatic_common::cpu_pinning::WorkerIndex;
 use glommio::LocalExecutorBuilder;
 use rand::prelude::*;
 use rand_distr::Pareto;
@@ -59,26 +59,23 @@ fn run(config: Config) -> ::anyhow::Result<()> {
         let tls_config = tls_config.clone();
         let state = state.clone();
 
-        LocalExecutorBuilder::default()
-            .spawn(move || async move {
-                #[cfg(feature = "cpu-pinning")]
-                pin_current_if_configured_to(
-                    &config.cpu_pinning,
-                    config.num_workers,
-                    WorkerIndex::SocketWorker(i),
-                );
+        let placement = get_worker_placement(
+            &config.cpu_pinning,
+            config.num_workers,
+            0,
+            WorkerIndex::SocketWorker(i),
+        )?;
 
+        LocalExecutorBuilder::new(placement)
+            .spawn(move || async move {
                 run_socket_thread(config, tls_config, state).await.unwrap();
             })
             .unwrap();
     }
 
-    #[cfg(feature = "cpu-pinning")]
-    pin_current_if_configured_to(
-        &config.cpu_pinning,
-        config.num_workers as usize,
-        WorkerIndex::Other,
-    );
+    if config.cpu_pinning.active {
+        set_affinity_for_util_worker(&config.cpu_pinning, config.num_workers, 0)?;
+    }
 
     monitor_statistics(state, &config);