diff --git a/.github/workflows/cargo-build-and-test.yml b/.github/workflows/cargo-build-and-test.yml index ca49e4e..a3c98ea 100644 --- a/.github/workflows/cargo-build-and-test.yml +++ b/.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 diff --git a/Cargo.lock b/Cargo.lock index 0a8e40a..4c33496 100644 --- a/Cargo.lock +++ b/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", diff --git a/aquatic_common/Cargo.toml b/aquatic_common/Cargo.toml index bb12b7f..1f6eb26 100644 --- a/aquatic_common/Cargo.toml +++ b/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 } \ No newline at end of file +rustls-pemfile = { version = "0.3", optional = true } diff --git a/aquatic_common/src/cpu_pinning.rs b/aquatic_common/src/cpu_pinning.rs index 6f0a9d9..b82d5bb 100644 --- a/aquatic_common/src/cpu_pinning.rs +++ b/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( + &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::online().map_err(|err| anyhow::anyhow!("Couldn't get CPU set: {:#}", err)) + } + + fn get_num_cpu_cores() -> anyhow::Result { + 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::>(); + + logical_cpus.sort_unstable(); + + logical_cpus + .into_iter() + .map(|cpu| cpu.to_string()) + .collect::>() + .join(", ") + } + + fn get_worker_cpu_set( + config: &C, + socket_workers: usize, + request_workers: usize, + worker_index: WorkerIndex, + ) -> anyhow::Result { + 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( + config: &C, + socket_workers: usize, + request_workers: usize, + worker_index: WorkerIndex, + ) -> anyhow::Result { + 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( + 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::(), + &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( + 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 = 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) diff --git a/aquatic_common/src/lib.rs b/aquatic_common/src/lib.rs index 687a547..b243cfd 100644 --- a/aquatic_common/src/lib.rs +++ b/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")] diff --git a/aquatic_http/Cargo.toml b/aquatic_http/Cargo.toml index c654787..b9b1152 100644 --- a/aquatic_http/Cargo.toml +++ b/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" diff --git a/aquatic_http/src/config.rs b/aquatic_http/src/config.rs index 1da0542..c94cf1e 100644 --- a/aquatic_http/src/config.rs +++ b/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(), } } diff --git a/aquatic_http/src/lib.rs b/aquatic_http/src/lib.rs index 0f92da5..28ca996 100644 --- a/aquatic_http/src/lib.rs +++ b/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 diff --git a/aquatic_http_load_test/Cargo.toml b/aquatic_http_load_test/Cargo.toml index ca4f073..625a157 100644 --- a/aquatic_http_load_test/Cargo.toml +++ b/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" } diff --git a/aquatic_http_load_test/src/config.rs b/aquatic_http_load_test/src/config.rs index a581e89..d8906df 100644 --- a/aquatic_http_load_test/src/config.rs +++ b/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(), } } } diff --git a/aquatic_http_load_test/src/main.rs b/aquatic_http_load_test/src/main.rs index 3cdef32..b7e7978 100644 --- a/aquatic_http_load_test/src/main.rs +++ b/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); diff --git a/aquatic_udp/Cargo.toml b/aquatic_udp/Cargo.toml index b98e16e..970b717 100644 --- a/aquatic_udp/Cargo.toml +++ b/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" } diff --git a/aquatic_udp/src/config.rs b/aquatic_udp/src/config.rs index 6e23599..01a0917 100644 --- a/aquatic_udp/src/config.rs +++ b/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 { diff --git a/aquatic_udp/src/lib.rs b/aquatic_udp/src/lib.rs index 7814f6f..d56d743 100644 --- a/aquatic_udp/src/lib.rs +++ b/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 { diff --git a/aquatic_udp_load_test/Cargo.toml b/aquatic_udp_load_test/Cargo.toml index 50c215b..31029b7 100644 --- a/aquatic_udp_load_test/Cargo.toml +++ b/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" } diff --git a/aquatic_udp_load_test/src/config.rs b/aquatic_udp_load_test/src/config.rs index bb99dbe..e803eb8 100644 --- a/aquatic_udp_load_test/src/config.rs +++ b/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(), } } } diff --git a/aquatic_udp_load_test/src/main.rs b/aquatic_udp_load_test/src/main.rs index 9cdff4e..dd6ad57 100644 --- a/aquatic_udp_load_test/src/main.rs +++ b/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); diff --git a/aquatic_ws/Cargo.toml b/aquatic_ws/Cargo.toml index ee23f08..0add3fa 100644 --- a/aquatic_ws/Cargo.toml +++ b/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" } diff --git a/aquatic_ws/src/config.rs b/aquatic_ws/src/config.rs index 95ab65a..b1ea961 100644 --- a/aquatic_ws/src/config.rs +++ b/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(), } } diff --git a/aquatic_ws/src/lib.rs b/aquatic_ws/src/lib.rs index ad8028d..0ba6bbc 100644 --- a/aquatic_ws/src/lib.rs +++ b/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 diff --git a/aquatic_ws_load_test/Cargo.toml b/aquatic_ws_load_test/Cargo.toml index f2d521e..82496b7 100644 --- a/aquatic_ws_load_test/Cargo.toml +++ b/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" } diff --git a/aquatic_ws_load_test/src/config.rs b/aquatic_ws_load_test/src/config.rs index 258423b..9949c65 100644 --- a/aquatic_ws_load_test/src/config.rs +++ b/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(), } } } diff --git a/aquatic_ws_load_test/src/main.rs b/aquatic_ws_load_test/src/main.rs index 78bc24d..f4f16eb 100644 --- a/aquatic_ws_load_test/src/main.rs +++ b/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);