From fb607ac0c24a98d548b66f3b95f871cde87fc321 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Joakim=20Frosteg=C3=A5rd?= Date: Wed, 30 Mar 2022 22:33:23 +0200 Subject: [PATCH 1/7] Improve CPU pinning --- Cargo.lock | 2 + aquatic_common/Cargo.toml | 8 +- aquatic_common/src/cpu_pinning.rs | 114 ++++++++++++++++++++++++--- aquatic_common/src/lib.rs | 1 - aquatic_http/Cargo.toml | 6 +- aquatic_http/src/config.rs | 2 - aquatic_http/src/lib.rs | 65 +++++++-------- aquatic_http_load_test/Cargo.toml | 5 +- aquatic_http_load_test/src/config.rs | 2 - aquatic_http_load_test/src/main.rs | 31 +++----- aquatic_udp/Cargo.toml | 2 +- aquatic_udp/src/lib.rs | 8 +- aquatic_udp_load_test/Cargo.toml | 6 +- aquatic_udp_load_test/src/main.rs | 4 +- aquatic_ws/Cargo.toml | 6 +- aquatic_ws/src/config.rs | 3 - aquatic_ws/src/lib.rs | 60 +++++++------- aquatic_ws_load_test/Cargo.toml | 5 +- aquatic_ws_load_test/src/config.rs | 3 - aquatic_ws_load_test/src/main.rs | 29 +++---- 20 files changed, 219 insertions(+), 143 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 89f5e0e..8b801f9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -75,6 +75,7 @@ dependencies = [ "anyhow", "aquatic_toml_config", "arc-swap", + "glommio", "hashbrown 0.12.0", "hex", "hwloc", @@ -101,6 +102,7 @@ dependencies = [ "futures-rustls", "glommio", "itoa 1.0.1", + "libc", "log", "memchr", "mimalloc", diff --git a/aquatic_common/Cargo.toml b/aquatic_common/Cargo.toml index 1a1790f..4c12b3a 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"] [dependencies] aquatic_toml_config = "0.2.0" @@ -23,11 +24,12 @@ arc-swap = "1" 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 } \ No newline at end of file diff --git a/aquatic_common/src/cpu_pinning.rs b/aquatic_common/src/cpu_pinning.rs index 6f0a9d9..48eac60 100644 --- a/aquatic_common/src/cpu_pinning.rs +++ b/aquatic_common/src/cpu_pinning.rs @@ -1,5 +1,4 @@ use aquatic_toml_config::TomlConfig; -use hwloc::{CpuSet, ObjectType, Topology, CPUBIND_THREAD}; use serde::{Deserialize, Serialize}; #[derive(Clone, Debug, PartialEq, TomlConfig, Serialize, Deserialize)] @@ -15,6 +14,7 @@ impl Default for CpuPinningMode { } } +/// Experimental CPU pinning #[derive(Clone, Debug, PartialEq, TomlConfig, Deserialize)] pub struct CpuPinningConfig { pub active: bool, @@ -45,37 +45,128 @@ impl CpuPinningConfig { pub enum WorkerIndex { SocketWorker(usize), RequestWorker(usize), - Other, + Util, } impl WorkerIndex { - fn get_core_index( - self, + pub fn get_core_index( + &self, config: &CpuPinningConfig, 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, }; + let max_core_index = num_cores - 1; + + let ascending_index = ascending_index.min(max_core_index); + match config.mode { CpuPinningMode::Ascending => ascending_index, - CpuPinningMode::Descending => core_count - 1 - ascending_index, + CpuPinningMode::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() + .ok_or(anyhow::anyhow!("CpuSet is empty")) + } + + fn get_worker_cpu_set( + config: &CpuPinningConfig, + 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); + + Ok(get_cpu_set()?.filter(|l| l.core == core_index)) + } + + pub fn get_worker_placement( + config: &CpuPinningConfig, + 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: &CpuPinningConfig, + socket_workers: usize, + request_workers: usize, + ) -> anyhow::Result<()> { + let worker_cpu_set = + get_worker_cpu_set(&config, socket_workers, request_workers, WorkerIndex::Util)?; + + let logical_cpus: Vec = worker_cpu_set.iter().map(|l| l.cpu).collect(); + + unsafe { + let mut set: libc::cpu_set_t = ::std::mem::zeroed(); + + for cpu in logical_cpus { + libc::CPU_SET(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`) +#[cfg(feature = "with-hwloc")] pub fn pin_current_if_configured_to( config: &CpuPinningConfig, socket_workers: usize, + request_workers: usize, worker_index: WorkerIndex, ) { + use hwloc::{CpuSet, ObjectType, Topology, CPUBIND_THREAD}; + if config.active { let mut topology = Topology::new(); @@ -86,7 +177,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 6995382..cbeb1c2 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; diff --git a/aquatic_http/Cargo.toml b/aquatic_http/Cargo.toml index 86b4b5a..62ca64e 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 = "0.2.0" -aquatic_common = "0.2.0" +aquatic_common = { version = "0.2.0", features = ["with-glommio"] } aquatic_http_protocol = "0.2.0" aquatic_toml_config = "0.2.0" @@ -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..f122c12 100644 --- a/aquatic_http/src/config.rs +++ b/aquatic_http/src/config.rs @@ -23,7 +23,6 @@ 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, } @@ -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 b484a53..55f6673 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, }; use common::{State, TlsConfig}; use glommio::{channels::channel_mesh::MeshBuilder, prelude::*}; @@ -37,12 +40,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 18200ce..d818bf4 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 = "0.2.0" -aquatic_common = "0.2.0" +aquatic_common = { version = "0.2.0", features = ["with-glommio"] } aquatic_http_protocol = "0.2.0" aquatic_toml_config = "0.2.0" diff --git a/aquatic_http_load_test/src/config.rs b/aquatic_http_load_test/src/config.rs index 6f88e79..be0e24d 100644 --- a/aquatic_http_load_test/src/config.rs +++ b/aquatic_http_load_test/src/config.rs @@ -20,7 +20,6 @@ pub struct Config { pub connection_creation_interval_ms: u64, pub duration: usize, pub torrents: TorrentConfig, - #[cfg(feature = "cpu-pinning")] pub cpu_pinning: aquatic_common::cpu_pinning::CpuPinningConfig, } @@ -58,7 +57,6 @@ impl Default for Config { connection_creation_interval_ms: 10, duration: 0, torrents: TorrentConfig::default(), - #[cfg(feature = "cpu-pinning")] cpu_pinning: aquatic_common::cpu_pinning::CpuPinningConfig::default_for_load_test(), } } 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 e8b2e8c..9c9dc74 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 = "0.2.0" 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 93f5c40..82ce25d 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 = "0.2.0" aquatic_common = "0.2.0" 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 af02221..e4c51bb 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 = "0.2.0" -aquatic_common = "0.2.0" +aquatic_common = { version = "0.2.0", features = ["with-glommio"] } aquatic_toml_config = "0.2.0" aquatic_ws_protocol = "0.2.0" diff --git a/aquatic_ws/src/config.rs b/aquatic_ws/src/config.rs index 95ab65a..fdff2b9 100644 --- a/aquatic_ws/src/config.rs +++ b/aquatic_ws/src/config.rs @@ -1,7 +1,6 @@ use std::net::SocketAddr; use std::path::PathBuf; -#[cfg(feature = "cpu-pinning")] use aquatic_common::cpu_pinning::CpuPinningConfig; use aquatic_common::{access_list::AccessListConfig, privileges::PrivilegeConfig}; use serde::Deserialize; @@ -26,7 +25,6 @@ pub struct Config { pub cleaning: CleaningConfig, pub privileges: PrivilegeConfig, pub access_list: AccessListConfig, - #[cfg(feature = "cpu-pinning")] pub cpu_pinning: CpuPinningConfig, } @@ -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 9e3deab..82b6471 100644 --- a/aquatic_ws/src/lib.rs +++ b/aquatic_ws/src/lib.rs @@ -6,12 +6,12 @@ use std::fs::File; use std::io::BufReader; 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 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::*; @@ -36,12 +36,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 { @@ -75,16 +76,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, @@ -105,16 +105,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, @@ -134,12 +133,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 e6b58d4..0f7ee30 100644 --- a/aquatic_ws_load_test/Cargo.toml +++ b/aquatic_ws_load_test/Cargo.toml @@ -12,13 +12,10 @@ readme = "../README.md" [[bin]] name = "aquatic_ws_load_test" -[features] -cpu-pinning = ["aquatic_common/cpu-pinning"] - [dependencies] async-tungstenite = "0.17" aquatic_cli_helpers = "0.2.0" -aquatic_common = "0.2.0" +aquatic_common = { version = "0.2.0", features = ["with-glommio"] } aquatic_toml_config = "0.2.0" aquatic_ws_protocol = "0.2.0" diff --git a/aquatic_ws_load_test/src/config.rs b/aquatic_ws_load_test/src/config.rs index 258423b..0fa311b 100644 --- a/aquatic_ws_load_test/src/config.rs +++ b/aquatic_ws_load_test/src/config.rs @@ -1,7 +1,6 @@ use std::net::SocketAddr; use aquatic_cli_helpers::LogLevel; -#[cfg(feature = "cpu-pinning")] use aquatic_common::cpu_pinning::CpuPinningConfig; use aquatic_toml_config::TomlConfig; use serde::Deserialize; @@ -17,7 +16,6 @@ pub struct Config { pub connection_creation_interval_ms: u64, pub duration: usize, pub torrents: TorrentConfig, - #[cfg(feature = "cpu-pinning")] pub cpu_pinning: CpuPinningConfig, } @@ -37,7 +35,6 @@ 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(), } } 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); From 3dc9068dd25c09f09a74b551b340e498987e3cff Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Joakim=20Frosteg=C3=A5rd?= Date: Thu, 31 Mar 2022 15:06:44 +0200 Subject: [PATCH 2/7] cpu pinning: add hyperthread mapping modes (glommio only) --- aquatic_common/src/cpu_pinning.rs | 95 +++++++++++++++++++++++++++++-- 1 file changed, 90 insertions(+), 5 deletions(-) diff --git a/aquatic_common/src/cpu_pinning.rs b/aquatic_common/src/cpu_pinning.rs index 48eac60..baae0cb 100644 --- a/aquatic_common/src/cpu_pinning.rs +++ b/aquatic_common/src/cpu_pinning.rs @@ -14,11 +14,29 @@ impl Default for CpuPinningMode { } } +#[cfg(feature = "with-glommio")] +#[derive(Clone, Debug, PartialEq, TomlConfig, Serialize, Deserialize)] +#[serde(rename_all = "kebab-case")] +pub enum HyperThreadMapping { + System, + Subsequent, + Split, +} + +#[cfg(feature = "with-glommio")] +impl Default for HyperThreadMapping { + fn default() -> Self { + Self::System + } +} + /// Experimental CPU pinning #[derive(Clone, Debug, PartialEq, TomlConfig, Deserialize)] pub struct CpuPinningConfig { pub active: bool, pub mode: CpuPinningMode, + #[cfg(feature = "with-glommio")] + pub hyperthread: HyperThreadMapping, pub core_offset: usize, } @@ -27,6 +45,8 @@ impl Default for CpuPinningConfig { Self { active: false, mode: Default::default(), + #[cfg(feature = "with-glommio")] + hyperthread: Default::default(), core_offset: 0, } } @@ -88,9 +108,22 @@ pub mod glommio { .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: &CpuPinningConfig, socket_workers: usize, @@ -102,7 +135,61 @@ pub mod glommio { let core_index = worker_index.get_core_index(&config, socket_workers, request_workers, num_cpu_cores); - Ok(get_cpu_set()?.filter(|l| l.core == core_index)) + let too_many_workers = match (&config.hyperthread, &config.mode) { + ( + HyperThreadMapping::Split | HyperThreadMapping::Subsequent, + CpuPinningMode::Ascending, + ) => core_index >= num_cpu_cores / 2, + ( + HyperThreadMapping::Split | HyperThreadMapping::Subsequent, + CpuPinningMode::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.mode { + CpuPinningMode::Ascending => get_cpu_set()? + .filter(|l| l.cpu == core_index || l.cpu == core_index + num_cpu_cores / 2), + CpuPinningMode::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.mode { + // 0 -> 0 and 1 + // 1 -> 2 and 3 + // 2 -> 4 and 5 + CpuPinningMode::Ascending => core_index * 2, + // 15 -> 14 and 15 + // 14 -> 12 and 13 + // 13 -> 10 and 11 + CpuPinningMode::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( @@ -129,13 +216,11 @@ pub mod glommio { let worker_cpu_set = get_worker_cpu_set(&config, socket_workers, request_workers, WorkerIndex::Util)?; - let logical_cpus: Vec = worker_cpu_set.iter().map(|l| l.cpu).collect(); - unsafe { let mut set: libc::cpu_set_t = ::std::mem::zeroed(); - for cpu in logical_cpus { - libc::CPU_SET(cpu, &mut set); + for cpu_location in worker_cpu_set { + libc::CPU_SET(cpu_location.cpu, &mut set); } let status = libc::pthread_setaffinity_np( From e2ee050233cdfe835b15686f91edff6b43006e9d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Joakim=20Frosteg=C3=A5rd?= Date: Thu, 31 Mar 2022 15:20:41 +0200 Subject: [PATCH 3/7] Fix GitHub CI --- .github/workflows/cargo-build-and-test.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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 From 6c149331dc0ef6a94c5b21d88b6b85892cfa5ec5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Joakim=20Frosteg=C3=A5rd?= Date: Mon, 4 Apr 2022 22:38:47 +0200 Subject: [PATCH 4/7] Rewrite CpuPinningConfig implementation to support aquatic_toml_config --- Cargo.lock | 41 +++++++++ aquatic_common/Cargo.toml | 1 + aquatic_common/src/cpu_pinning.rs | 119 ++++++++++++++++----------- aquatic_http/src/config.rs | 4 +- aquatic_http_load_test/src/config.rs | 5 +- aquatic_udp/src/config.rs | 2 +- aquatic_udp_load_test/src/config.rs | 6 +- aquatic_ws/src/config.rs | 4 +- aquatic_ws_load_test/src/config.rs | 6 +- 9 files changed, 127 insertions(+), 61 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 8b801f9..138db60 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -75,6 +75,7 @@ dependencies = [ "anyhow", "aquatic_toml_config", "arc-swap", + "duplicate", "glommio", "hashbrown 0.12.0", "hex", @@ -703,6 +704,16 @@ dependencies = [ "crypto-common", ] +[[package]] +name = "duplicate" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4a2535fcf1437c9de0d91a3921351c474258abdb6440cd03bb5a7cf0547e7214" +dependencies = [ + "heck", + "proc-macro-error", +] + [[package]] name = "either" version = "1.6.1" @@ -1065,6 +1076,12 @@ dependencies = [ "serde", ] +[[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" @@ -1583,6 +1600,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" diff --git a/aquatic_common/Cargo.toml b/aquatic_common/Cargo.toml index 4c12b3a..40b9427 100644 --- a/aquatic_common/Cargo.toml +++ b/aquatic_common/Cargo.toml @@ -21,6 +21,7 @@ aquatic_toml_config = "0.2.0" ahash = "0.7" anyhow = "1" arc-swap = "1" +duplicate = "0.4" hashbrown = "0.12" hex = "0.4" indexmap-amortized = "1" diff --git a/aquatic_common/src/cpu_pinning.rs b/aquatic_common/src/cpu_pinning.rs index baae0cb..1e7bccc 100644 --- a/aquatic_common/src/cpu_pinning.rs +++ b/aquatic_common/src/cpu_pinning.rs @@ -1,7 +1,9 @@ +//! Experimental CPU pinning + use aquatic_toml_config::TomlConfig; 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 { Ascending, @@ -15,7 +17,7 @@ impl Default for CpuPinningMode { } #[cfg(feature = "with-glommio")] -#[derive(Clone, Debug, PartialEq, TomlConfig, Serialize, Deserialize)] +#[derive(Clone, Copy, Debug, PartialEq, TomlConfig, Serialize, Deserialize)] #[serde(rename_all = "kebab-case")] pub enum HyperThreadMapping { System, @@ -30,33 +32,54 @@ impl Default for HyperThreadMapping { } } -/// Experimental CPU pinning -#[derive(Clone, Debug, PartialEq, TomlConfig, Deserialize)] -pub struct CpuPinningConfig { - pub active: bool, - pub mode: CpuPinningMode, - #[cfg(feature = "with-glommio")] - pub hyperthread: HyperThreadMapping, - pub core_offset: usize, +pub trait CpuPinningConfig { + fn active(&self) -> bool; + fn mode(&self) -> CpuPinningMode; + fn hyperthread(&self) -> HyperThreadMapping; + fn core_offset(&self) -> usize; } -impl Default for CpuPinningConfig { - fn default() -> Self { - Self { - active: false, - mode: Default::default(), - #[cfg(feature = "with-glommio")] - hyperthread: Default::default(), - core_offset: 0, +// Do these shenanigans for compatibility with aquatic_toml_config +#[duplicate::duplicate_item( + mod_name struct_name direction; + [asc] [CpuPinningConfigAsc] [CpuPinningMode::Ascending]; + [desc] [CpuPinningConfigDesc] [CpuPinningMode::Descending]; +)] +pub mod mod_name { + use super::*; + + #[derive(Clone, Debug, PartialEq, TomlConfig, Deserialize)] + pub struct struct_name { + pub active: bool, + pub mode: CpuPinningMode, + #[cfg(feature = "with-glommio")] + pub hyperthread: HyperThreadMapping, + pub core_offset: usize, + } + + impl Default for struct_name { + fn default() -> Self { + Self { + active: false, + mode: direction, + #[cfg(feature = "with-glommio")] + hyperthread: Default::default(), + core_offset: 0, + } } } -} - -impl CpuPinningConfig { - pub fn default_for_load_test() -> Self { - Self { - mode: CpuPinningMode::Descending, - ..Default::default() + impl CpuPinningConfig for struct_name { + fn active(&self) -> bool { + self.active + } + fn mode(&self) -> CpuPinningMode { + self.mode + } + fn hyperthread(&self) -> HyperThreadMapping { + self.hyperthread + } + fn core_offset(&self) -> usize { + self.core_offset } } } @@ -69,24 +92,24 @@ pub enum WorkerIndex { } impl WorkerIndex { - pub fn get_core_index( + pub fn get_core_index( &self, - config: &CpuPinningConfig, + config: &C, socket_workers: usize, request_workers: usize, num_cores: usize, ) -> usize { let ascending_index = match self { - 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, + 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, }; let max_core_index = num_cores - 1; let ascending_index = ascending_index.min(max_core_index); - match config.mode { + match config.mode() { CpuPinningMode::Ascending => ascending_index, CpuPinningMode::Descending => max_core_index - ascending_index, } @@ -124,8 +147,8 @@ pub mod glommio { .join(", ") } - fn get_worker_cpu_set( - config: &CpuPinningConfig, + fn get_worker_cpu_set( + config: &C, socket_workers: usize, request_workers: usize, worker_index: WorkerIndex, @@ -133,9 +156,9 @@ pub mod glommio { let num_cpu_cores = get_num_cpu_cores()?; let core_index = - worker_index.get_core_index(&config, socket_workers, request_workers, num_cpu_cores); + worker_index.get_core_index(config, socket_workers, request_workers, num_cpu_cores); - let too_many_workers = match (&config.hyperthread, &config.mode) { + let too_many_workers = match (&config.hyperthread(), &config.mode()) { ( HyperThreadMapping::Split | HyperThreadMapping::Subsequent, CpuPinningMode::Ascending, @@ -151,16 +174,16 @@ pub mod glommio { 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 { + let cpu_set = match config.hyperthread() { HyperThreadMapping::System => get_cpu_set()?.filter(|l| l.core == core_index), - HyperThreadMapping::Split => match config.mode { + HyperThreadMapping::Split => match config.mode() { CpuPinningMode::Ascending => get_cpu_set()? .filter(|l| l.cpu == core_index || l.cpu == core_index + num_cpu_cores / 2), CpuPinningMode::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.mode { + let cpu_index_offset = match config.mode() { // 0 -> 0 and 1 // 1 -> 2 and 3 // 2 -> 4 and 5 @@ -192,15 +215,15 @@ pub mod glommio { } } - pub fn get_worker_placement( - config: &CpuPinningConfig, + pub fn get_worker_placement( + config: &C, socket_workers: usize, request_workers: usize, worker_index: WorkerIndex, ) -> anyhow::Result { - if config.active { + if config.active() { let cpu_set = - get_worker_cpu_set(&config, socket_workers, request_workers, worker_index)?; + get_worker_cpu_set(config, socket_workers, request_workers, worker_index)?; Ok(Placement::Fenced(cpu_set)) } else { @@ -208,13 +231,13 @@ pub mod glommio { } } - pub fn set_affinity_for_util_worker( - config: &CpuPinningConfig, + 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)?; + get_worker_cpu_set(config, socket_workers, request_workers, WorkerIndex::Util)?; unsafe { let mut set: libc::cpu_set_t = ::std::mem::zeroed(); @@ -244,15 +267,15 @@ pub mod glommio { /// /// Requires hwloc (`apt-get install libhwloc-dev`) #[cfg(feature = "with-hwloc")] -pub fn pin_current_if_configured_to( - config: &CpuPinningConfig, +pub fn pin_current_if_configured_to( + config: &C, socket_workers: usize, request_workers: usize, worker_index: WorkerIndex, ) { use hwloc::{CpuSet, ObjectType, Topology, CPUBIND_THREAD}; - if config.active { + if config.active() { let mut topology = Topology::new(); let core_cpu_sets: Vec = topology diff --git a/aquatic_http/src/config.rs b/aquatic_http/src/config.rs index f122c12..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,7 +23,7 @@ pub struct Config { pub cleaning: CleaningConfig, pub privileges: PrivilegeConfig, pub access_list: AccessListConfig, - pub cpu_pinning: aquatic_common::cpu_pinning::CpuPinningConfig, + pub cpu_pinning: CpuPinningConfigAsc, } impl Default for Config { diff --git a/aquatic_http_load_test/src/config.rs b/aquatic_http_load_test/src/config.rs index be0e24d..994913d 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; @@ -20,7 +21,7 @@ pub struct Config { pub connection_creation_interval_ms: u64, pub duration: usize, pub torrents: TorrentConfig, - pub cpu_pinning: aquatic_common::cpu_pinning::CpuPinningConfig, + pub cpu_pinning: CpuPinningConfigDesc, } impl aquatic_cli_helpers::Config for Config { @@ -57,7 +58,7 @@ impl Default for Config { connection_creation_interval_ms: 10, duration: 0, torrents: TorrentConfig::default(), - cpu_pinning: aquatic_common::cpu_pinning::CpuPinningConfig::default_for_load_test(), + cpu_pinning: Default::default(), } } } 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_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_ws/src/config.rs b/aquatic_ws/src/config.rs index fdff2b9..b1ea961 100644 --- a/aquatic_ws/src/config.rs +++ b/aquatic_ws/src/config.rs @@ -1,7 +1,7 @@ use std::net::SocketAddr; use std::path::PathBuf; -use aquatic_common::cpu_pinning::CpuPinningConfig; +use aquatic_common::cpu_pinning::asc::CpuPinningConfigAsc; use aquatic_common::{access_list::AccessListConfig, privileges::PrivilegeConfig}; use serde::Deserialize; @@ -25,7 +25,7 @@ pub struct Config { pub cleaning: CleaningConfig, pub privileges: PrivilegeConfig, pub access_list: AccessListConfig, - pub cpu_pinning: CpuPinningConfig, + pub cpu_pinning: CpuPinningConfigAsc, } impl Default for Config { diff --git a/aquatic_ws_load_test/src/config.rs b/aquatic_ws_load_test/src/config.rs index 0fa311b..9949c65 100644 --- a/aquatic_ws_load_test/src/config.rs +++ b/aquatic_ws_load_test/src/config.rs @@ -1,7 +1,7 @@ use std::net::SocketAddr; use aquatic_cli_helpers::LogLevel; -use aquatic_common::cpu_pinning::CpuPinningConfig; +use aquatic_common::cpu_pinning::desc::CpuPinningConfigDesc; use aquatic_toml_config::TomlConfig; use serde::Deserialize; @@ -16,7 +16,7 @@ pub struct Config { pub connection_creation_interval_ms: u64, pub duration: usize, pub torrents: TorrentConfig, - pub cpu_pinning: CpuPinningConfig, + pub cpu_pinning: CpuPinningConfigDesc, } impl aquatic_cli_helpers::Config for Config { @@ -35,7 +35,7 @@ impl Default for Config { connection_creation_interval_ms: 10, duration: 0, torrents: TorrentConfig::default(), - cpu_pinning: CpuPinningConfig::default_for_load_test(), + cpu_pinning: Default::default(), } } } From ffce413217d510058431cd75a9f65e4e0e7179de Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Joakim=20Frosteg=C3=A5rd?= Date: Mon, 4 Apr 2022 23:00:07 +0200 Subject: [PATCH 5/7] aquatic_common: fix build error with aquatic_udp --- aquatic_common/src/cpu_pinning.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/aquatic_common/src/cpu_pinning.rs b/aquatic_common/src/cpu_pinning.rs index 1e7bccc..ffb1019 100644 --- a/aquatic_common/src/cpu_pinning.rs +++ b/aquatic_common/src/cpu_pinning.rs @@ -35,6 +35,7 @@ impl Default for HyperThreadMapping { pub trait CpuPinningConfig { fn active(&self) -> bool; fn mode(&self) -> CpuPinningMode; + #[cfg(feature = "with-glommio")] fn hyperthread(&self) -> HyperThreadMapping; fn core_offset(&self) -> usize; } @@ -75,6 +76,7 @@ pub mod mod_name { fn mode(&self) -> CpuPinningMode { self.mode } + #[cfg(feature = "with-glommio")] fn hyperthread(&self) -> HyperThreadMapping { self.hyperthread } From 76ccd8ba55f0076fab3eaba8da6d03653f0d8eee Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Joakim=20Frosteg=C3=A5rd?= Date: Mon, 4 Apr 2022 23:04:09 +0200 Subject: [PATCH 6/7] aquatic_common: rename CpuPinningMode to CpuPinningDirection --- aquatic_common/src/cpu_pinning.rs | 44 +++++++++++++++---------------- 1 file changed, 22 insertions(+), 22 deletions(-) diff --git a/aquatic_common/src/cpu_pinning.rs b/aquatic_common/src/cpu_pinning.rs index ffb1019..00abbb6 100644 --- a/aquatic_common/src/cpu_pinning.rs +++ b/aquatic_common/src/cpu_pinning.rs @@ -5,12 +5,12 @@ use serde::{Deserialize, Serialize}; #[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 } @@ -34,7 +34,7 @@ impl Default for HyperThreadMapping { pub trait CpuPinningConfig { fn active(&self) -> bool; - fn mode(&self) -> CpuPinningMode; + fn direction(&self) -> CpuPinningDirection; #[cfg(feature = "with-glommio")] fn hyperthread(&self) -> HyperThreadMapping; fn core_offset(&self) -> usize; @@ -42,9 +42,9 @@ pub trait CpuPinningConfig { // Do these shenanigans for compatibility with aquatic_toml_config #[duplicate::duplicate_item( - mod_name struct_name direction; - [asc] [CpuPinningConfigAsc] [CpuPinningMode::Ascending]; - [desc] [CpuPinningConfigDesc] [CpuPinningMode::Descending]; + mod_name struct_name cpu_pinning_direction; + [asc] [CpuPinningConfigAsc] [CpuPinningDirection::Ascending]; + [desc] [CpuPinningConfigDesc] [CpuPinningDirection::Descending]; )] pub mod mod_name { use super::*; @@ -52,7 +52,7 @@ pub mod mod_name { #[derive(Clone, Debug, PartialEq, TomlConfig, Deserialize)] pub struct struct_name { pub active: bool, - pub mode: CpuPinningMode, + pub direction: CpuPinningDirection, #[cfg(feature = "with-glommio")] pub hyperthread: HyperThreadMapping, pub core_offset: usize, @@ -62,7 +62,7 @@ pub mod mod_name { fn default() -> Self { Self { active: false, - mode: direction, + direction: cpu_pinning_direction, #[cfg(feature = "with-glommio")] hyperthread: Default::default(), core_offset: 0, @@ -73,8 +73,8 @@ pub mod mod_name { fn active(&self) -> bool { self.active } - fn mode(&self) -> CpuPinningMode { - self.mode + fn direction(&self) -> CpuPinningDirection { + self.direction } #[cfg(feature = "with-glommio")] fn hyperthread(&self) -> HyperThreadMapping { @@ -111,9 +111,9 @@ impl WorkerIndex { let ascending_index = ascending_index.min(max_core_index); - match config.mode() { - CpuPinningMode::Ascending => ascending_index, - CpuPinningMode::Descending => max_core_index - ascending_index, + match config.direction() { + CpuPinningDirection::Ascending => ascending_index, + CpuPinningDirection::Descending => max_core_index - ascending_index, } } } @@ -160,14 +160,14 @@ pub mod glommio { let core_index = worker_index.get_core_index(config, socket_workers, request_workers, num_cpu_cores); - let too_many_workers = match (&config.hyperthread(), &config.mode()) { + let too_many_workers = match (&config.hyperthread(), &config.direction()) { ( HyperThreadMapping::Split | HyperThreadMapping::Subsequent, - CpuPinningMode::Ascending, + CpuPinningDirection::Ascending, ) => core_index >= num_cpu_cores / 2, ( HyperThreadMapping::Split | HyperThreadMapping::Subsequent, - CpuPinningMode::Descending, + CpuPinningDirection::Descending, ) => core_index < num_cpu_cores / 2, (_, _) => false, }; @@ -178,22 +178,22 @@ pub mod glommio { let cpu_set = match config.hyperthread() { HyperThreadMapping::System => get_cpu_set()?.filter(|l| l.core == core_index), - HyperThreadMapping::Split => match config.mode() { - CpuPinningMode::Ascending => get_cpu_set()? + HyperThreadMapping::Split => match config.direction() { + CpuPinningDirection::Ascending => get_cpu_set()? .filter(|l| l.cpu == core_index || l.cpu == core_index + num_cpu_cores / 2), - CpuPinningMode::Descending => get_cpu_set()? + 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.mode() { + let cpu_index_offset = match config.direction() { // 0 -> 0 and 1 // 1 -> 2 and 3 // 2 -> 4 and 5 - CpuPinningMode::Ascending => core_index * 2, + CpuPinningDirection::Ascending => core_index * 2, // 15 -> 14 and 15 // 14 -> 12 and 13 // 13 -> 10 and 11 - CpuPinningMode::Descending => num_cpu_cores - 2 * (num_cpu_cores - core_index), + CpuPinningDirection::Descending => num_cpu_cores - 2 * (num_cpu_cores - core_index), }; get_cpu_set()? From 621f45e84c9615ee8b1d869db9b3d0eb53c22756 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Joakim=20Frosteg=C3=A5rd?= Date: Mon, 4 Apr 2022 23:04:56 +0200 Subject: [PATCH 7/7] aquatic_common: add doc comments for CpuPinningConfig structs --- aquatic_common/src/cpu_pinning.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/aquatic_common/src/cpu_pinning.rs b/aquatic_common/src/cpu_pinning.rs index 00abbb6..b82d5bb 100644 --- a/aquatic_common/src/cpu_pinning.rs +++ b/aquatic_common/src/cpu_pinning.rs @@ -49,6 +49,7 @@ pub trait CpuPinningConfig { pub mod mod_name { use super::*; + /// Experimental cpu pinning #[derive(Clone, Debug, PartialEq, TomlConfig, Deserialize)] pub struct struct_name { pub active: bool,