From 16bf2bd9cd477faf1f7b9f415ff9154a34e26fe1 Mon Sep 17 00:00:00 2001 From: mohakgoel1 Date: Sun, 7 Jun 2026 18:38:08 +0530 Subject: [PATCH 1/3] minimum correct ordering changes Signed-off-by: mohakgoel1 --- src/pool/builder.rs | 10 ++-- src/pool/spawn.rs | 103 +++++++++++++++++++++++++++++++++------- src/queue/multilevel.rs | 18 +++---- src/task/future.rs | 14 +++--- 4 files changed, 107 insertions(+), 38 deletions(-) diff --git a/src/pool/builder.rs b/src/pool/builder.rs index 09752a0..869d6ab 100644 --- a/src/pool/builder.rs +++ b/src/pool/builder.rs @@ -58,7 +58,7 @@ impl Clone for SchedConfig { SchedConfig { max_thread_count: self.max_thread_count, min_thread_count: self.min_thread_count, - core_thread_count: AtomicUsize::new(self.core_thread_count.load(Ordering::SeqCst)), + core_thread_count: AtomicUsize::new(self.core_thread_count.load(Ordering::Acquire)), max_inplace_spin: self.max_inplace_spin, max_idle_time: self.max_idle_time, max_wait_time: self.max_wait_time, @@ -166,7 +166,7 @@ impl Builder { if count > 0 { self.sched_config .core_thread_count - .store(count, Ordering::SeqCst); + .store(count, Ordering::Release); } self } @@ -241,15 +241,15 @@ impl Builder { T: TaskCell + Send, { assert!(self.sched_config.min_thread_count <= self.sched_config.max_thread_count); - let core_thread_count = self.sched_config.core_thread_count.load(Ordering::SeqCst); + let core_thread_count = self.sched_config.core_thread_count.load(Ordering::Acquire); if core_thread_count == 0 || core_thread_count > self.sched_config.max_thread_count { self.sched_config .core_thread_count - .store(self.sched_config.max_thread_count, Ordering::SeqCst); + .store(self.sched_config.max_thread_count, Ordering::Release); } else if core_thread_count < self.sched_config.min_thread_count { self.sched_config .core_thread_count - .store(self.sched_config.min_thread_count, Ordering::SeqCst); + .store(self.sched_config.min_thread_count, Ordering::Release); } let (injector, local_queues) = queue::build(queue_type, self.sched_config.max_thread_count); let core = Arc::new(QueueCore::new(injector, self.sched_config.clone())); diff --git a/src/pool/spawn.rs b/src/pool/spawn.rs index 2512013..8e74414 100644 --- a/src/pool/spawn.rs +++ b/src/pool/spawn.rs @@ -14,6 +14,74 @@ use std::sync::{ }; use std::time::Instant; +#[cfg(any( + target_arch = "x86_64", + target_arch = "aarch64", + target_arch = "powerpc64", +))] +const CACHE_LINE_SIZE: usize = 128; + +#[cfg(target_arch = "arm")] +const CACHE_LINE_SIZE: usize = 64; + +#[cfg(not(any( + target_arch = "x86_64", + target_arch = "aarch64", + target_arch = "powerpc64", + target_arch = "arm", +)))] +const CACHE_LINE_SIZE: usize = 64; + +#[cfg(any( + target_arch = "x86_64", + target_arch = "aarch64", + target_arch = "powerpc64", +))] +#[repr(C, align(128))] +struct CacheAligned { + value: AtomicUsize, + _pad: [u8; CACHE_LINE_SIZE - std::mem::size_of::()], +} + +#[cfg(target_arch = "arm")] +#[repr(C, align(64))] +struct CacheAligned { + value: AtomicUsize, + _pad: [u8; CACHE_LINE_SIZE - std::mem::size_of::()], +} + +#[cfg(not(any( + target_arch = "x86_64", + target_arch = "aarch64", + target_arch = "powerpc64", + target_arch = "arm", +)))] + +// We intentionally use a custom CacheAligned struct instead of +// crossbeam_utils::CachePadded because CachePadded aligns to only +// 32 bytes on 32-bit arm targets, which may not fully prevent false +// sharing on all arm32 cache line sizes. This explicit struct uses +// platform-specific #[repr(C, align(N))] to guarantee correct +// alignment on every target: +// aarch64, x86-64, powerpc64 -> 128 bytes +// arm (32-bit) -> 64 bytes +// all others -> 64 bytes + +#[repr(C, align(64))] +struct CacheAligned { + value: AtomicUsize, + _pad: [u8; CACHE_LINE_SIZE - std::mem::size_of::()], +} + +impl CacheAligned { + fn new(v: usize) -> Self { + CacheAligned { + value: AtomicUsize::new(v), + _pad: [0u8; CACHE_LINE_SIZE - std::mem::size_of::()], + } + } +} + /// An usize is used to trace the threads that are working actively. /// To save additional memory and atomic operation, the number and /// shutdown hint are merged into one number in the following format @@ -36,9 +104,10 @@ pub fn is_shutdown(cnt: usize) -> bool { /// /// Every thread pool instance should have one and only `QueueCore`. It's /// saved in an `Arc` and shared between all worker threads and remote handles. + pub(crate) struct QueueCore { global_queue: TaskInjector, - active_workers: AtomicUsize, + active_workers: CacheAligned, config: SchedConfig, } @@ -46,7 +115,7 @@ impl QueueCore { pub fn new(global_queue: TaskInjector, config: SchedConfig) -> QueueCore { QueueCore { global_queue, - active_workers: AtomicUsize::new(config.max_thread_count << WORKER_COUNT_SHIFT), + active_workers: CacheAligned::new(config.max_thread_count << WORKER_COUNT_SHIFT), config, } } @@ -56,8 +125,8 @@ impl QueueCore { /// If the method is going to wake up any threads, source is used to trace who triggers /// the action. pub fn ensure_workers(&self, source: usize) { - let cnt = self.active_workers.load(Ordering::SeqCst); - if (cnt >> WORKER_COUNT_SHIFT) >= self.config.core_thread_count.load(Ordering::SeqCst) + let cnt = self.active_workers.value.load(Ordering::Acquire); + if (cnt >> WORKER_COUNT_SHIFT) >= self.config.core_thread_count.load(Ordering::Acquire) || is_shutdown(cnt) { return; @@ -74,7 +143,7 @@ impl QueueCore { parking_lot_core::unpark_filter( addr, |p: ParkToken| { - if !unparked_once && p.0 <= self.config.core_thread_count.load(Ordering::SeqCst) + if !unparked_once && p.0 <= self.config.core_thread_count.load(Ordering::Acquire) { unparked_once = true; FilterOp::Unpark @@ -91,7 +160,7 @@ impl QueueCore { /// /// `source` is used to trace who triggers the action. pub fn mark_shutdown(&self, source: usize) { - self.active_workers.fetch_or(SHUTDOWN_BIT, Ordering::SeqCst); + self.active_workers.value.fetch_or(SHUTDOWN_BIT, Ordering::AcqRel); let addr = self as *const QueueCore as usize; unsafe { parking_lot_core::unpark_all(addr, UnparkToken(source)); @@ -100,7 +169,7 @@ impl QueueCore { /// Checks if the thread pool is shutting down. pub fn is_shutdown(&self) -> bool { - let cnt = self.active_workers.load(Ordering::SeqCst); + let cnt = self.active_workers.value.load(Ordering::Acquire); is_shutdown(cnt) } @@ -108,17 +177,17 @@ impl QueueCore { /// /// It can be marked as sleep only when the pool is not shutting down. pub fn mark_sleep(&self) -> bool { - let mut cnt = self.active_workers.load(Ordering::SeqCst); + let mut cnt = self.active_workers.value.load(Ordering::Acquire); loop { if is_shutdown(cnt) { return false; } - match self.active_workers.compare_exchange_weak( + match self.active_workers.value.compare_exchange_weak( cnt, cnt - WORKER_COUNT_BASE, - Ordering::SeqCst, - Ordering::SeqCst, + Ordering::AcqRel, + Ordering::Acquire, ) { Ok(_) => return true, Err(n) => cnt = n, @@ -128,13 +197,13 @@ impl QueueCore { /// Marks current thread as woken up states. pub fn mark_woken(&self) { - let mut cnt = self.active_workers.load(Ordering::SeqCst); + let mut cnt = self.active_workers.value.load(Ordering::Acquire); loop { - match self.active_workers.compare_exchange_weak( + match self.active_workers.value.compare_exchange_weak( cnt, cnt + WORKER_COUNT_BASE, - Ordering::SeqCst, - Ordering::SeqCst, + Ordering::AcqRel, + Ordering::Acquire, ) { Ok(_) => return, Err(n) => cnt = n, @@ -151,7 +220,7 @@ impl QueueCore { } self.config .core_thread_count - .store(new_thread_count, Ordering::SeqCst); + .store(new_thread_count, Ordering::Release); } pub fn config(&self) -> &SchedConfig { @@ -355,7 +424,7 @@ impl Local { marked_sleep = true; // If this thread is above core_thread_count, go to sleep // without popping so scaled-down threads don't keep working. - if id > self.core.config.core_thread_count.load(Ordering::SeqCst) { + if id > self.core.config.core_thread_count.load(Ordering::Acquire) { return true; } fail_point!("worker-pop-or-sleep-before-validate-pop"); diff --git a/src/queue/multilevel.rs b/src/queue/multilevel.rs index de4fb70..7c7d76b 100644 --- a/src/queue/multilevel.rs +++ b/src/queue/multilevel.rs @@ -496,7 +496,7 @@ impl LevelManager { fn maybe_adjust_chance(&self) { if self .adjusting - .compare_exchange(false, true, SeqCst, SeqCst) + .compare_exchange(false, true, AcqRel, Acquire) .is_err() { return; @@ -507,7 +507,7 @@ impl LevelManager { let total_diff = total - self.last_total_elapsed_us.get(); if total_diff < ADJUST_CHANCE_INTERVAL_US { // Needn't change it now. - self.adjusting.store(false, SeqCst); + self.adjusting.store(false, Release); return; } let level0 = self.level0_elapsed_us.get(); @@ -549,13 +549,13 @@ impl LevelManager { }, ); self.max_level_queue_steal_size - .store(new_steal_count, SeqCst); + .store(new_steal_count, Release); for (i, c) in self.last_exec_tasks_per_level.iter().enumerate() { c.set(cur_total_tasks_per_level[i]); } } - self.adjusting.store(false, SeqCst); + self.adjusting.store(false, Release); } } @@ -686,7 +686,7 @@ impl TaskElapsedMap { } fn get_elapsed(&self, key: u64) -> Arc { - let new_index = self.new_index.load(SeqCst); + let new_index = self.new_index.load(Acquire); let new_map = &self.maps[new_index]; let old_map = &self.maps[new_index ^ 1]; if let Some(v) = new_map.get(&key) { @@ -714,14 +714,14 @@ impl TaskElapsedMap { fn try_cleanup(&self) -> Option { self.cleaning_up - .compare_exchange(false, true, SeqCst, SeqCst) + .compare_exchange(false, true, AcqRel, Acquire) .map(|_| { - let old_index = self.new_index.load(SeqCst) ^ 1; + let old_index = self.new_index.load(Acquire) ^ 1; self.maps[old_index].clear(); - self.new_index.store(old_index, SeqCst); + self.new_index.store(old_index, Release); let now = now(); *self.last_cleanup_time.lock().unwrap() = now; - self.cleaning_up.store(false, SeqCst); + self.cleaning_up.store(false, Release); now }) .ok() diff --git a/src/task/future.rs b/src/task/future.rs index 1b5d884..92c508c 100644 --- a/src/task/future.rs +++ b/src/task/future.rs @@ -14,7 +14,7 @@ use std::pin::Pin; use std::ptr::NonNull; use std::sync::atomic::{ AtomicU8, AtomicUsize, - Ordering::{Acquire, Relaxed, Release, SeqCst}, + Ordering::{AcqRel, Acquire, Relaxed, Release}, }; use std::sync::{atomic, Arc}; use std::task::{Context, Poll, RawWaker, RawWakerVTable, Waker}; @@ -243,13 +243,13 @@ unsafe fn drop_raw(this: *const ()) { } unsafe fn wake_impl(task: Cow<'_, TaskCell>) { - let mut status = task.status().load(SeqCst); + let mut status = task.status().load(Acquire); loop { match status { IDLE => { match task .status() - .compare_exchange_weak(IDLE, NOTIFIED, SeqCst, SeqCst) + .compare_exchange_weak(IDLE, NOTIFIED, AcqRel, Acquire) { Ok(_) => { wake_task(task, false); @@ -261,7 +261,7 @@ unsafe fn wake_impl(task: Cow<'_, TaskCell>) { POLLING => { match task .status() - .compare_exchange_weak(POLLING, NOTIFIED, SeqCst, SeqCst) + .compare_exchange_weak(POLLING, NOTIFIED, AcqRel, Acquire) { Ok(_) => break, Err(cur) => status = cur, @@ -390,9 +390,9 @@ impl crate::pool::Runner for Runner { let mut cx = waker_ref.to_context(); let mut repoll_times = 0; loop { - task_cell.status().store(POLLING, SeqCst); + task_cell.status().store(POLLING, Release); if task_cell.poll(&mut cx).is_ready() { - task_cell.status().store(COMPLETED, SeqCst); + task_cell.status().store(COMPLETED, Release); return true; } let extras = { &mut *task_cell.extras().get() }; @@ -406,7 +406,7 @@ impl crate::pool::Runner for Runner { } match task_cell .status() - .compare_exchange(POLLING, IDLE, SeqCst, SeqCst) + .compare_exchange(POLLING, IDLE, AcqRel, Acquire) { Ok(_) => return false, Err(NOTIFIED) => { From cffc85208b62ef91df06ebdbe59710909f77196b Mon Sep 17 00:00:00 2001 From: mohakgoel1 Date: Tue, 28 Jul 2026 21:03:41 +0530 Subject: [PATCH 2/3] Cover 256-byte AArch64 cache lines Signed-off-by: mohakgoel1 --- src/pool/spawn.rs | 39 ++++++++++++++++++++++----------------- 1 file changed, 22 insertions(+), 17 deletions(-) diff --git a/src/pool/spawn.rs b/src/pool/spawn.rs index 8e74414..7af8f5d 100644 --- a/src/pool/spawn.rs +++ b/src/pool/spawn.rs @@ -14,11 +14,21 @@ use std::sync::{ }; use std::time::Instant; -#[cfg(any( - target_arch = "x86_64", - target_arch = "aarch64", - target_arch = "powerpc64", -))] +// We intentionally use a custom CacheAligned struct instead of +// crossbeam_utils::CachePadded because CachePadded aligns to only +// 32 bytes on 32-bit arm targets, which may not fully prevent false +// sharing on all arm32 cache line sizes. This explicit struct uses +// platform-specific #[repr(C, align(N))] to guarantee correct +// alignment on every target: +// aarch64 -> 256 bytes (conservative: covers Fujitsu A64FX 256-byte +// cache lines; safe for Neoverse/Graviton 128-byte lines) +// x86-64, powerpc64 -> 128 bytes +// arm (32-bit) -> 64 bytes +// all others -> 64 bytes +#[cfg(target_arch = "aarch64")] +const CACHE_LINE_SIZE: usize = 256; + +#[cfg(any(target_arch = "x86_64", target_arch = "powerpc64"))] const CACHE_LINE_SIZE: usize = 128; #[cfg(target_arch = "arm")] @@ -32,9 +42,15 @@ const CACHE_LINE_SIZE: usize = 64; )))] const CACHE_LINE_SIZE: usize = 64; +#[cfg(target_arch = "aarch64")] +#[repr(C, align(256))] +struct CacheAligned { + value: AtomicUsize, + _pad: [u8; CACHE_LINE_SIZE - std::mem::size_of::()], +} + #[cfg(any( target_arch = "x86_64", - target_arch = "aarch64", target_arch = "powerpc64", ))] #[repr(C, align(128))] @@ -56,17 +72,6 @@ struct CacheAligned { target_arch = "powerpc64", target_arch = "arm", )))] - -// We intentionally use a custom CacheAligned struct instead of -// crossbeam_utils::CachePadded because CachePadded aligns to only -// 32 bytes on 32-bit arm targets, which may not fully prevent false -// sharing on all arm32 cache line sizes. This explicit struct uses -// platform-specific #[repr(C, align(N))] to guarantee correct -// alignment on every target: -// aarch64, x86-64, powerpc64 -> 128 bytes -// arm (32-bit) -> 64 bytes -// all others -> 64 bytes - #[repr(C, align(64))] struct CacheAligned { value: AtomicUsize, From 9d513e53f8347517a1d3b28e98e6efb9392f810f Mon Sep 17 00:00:00 2001 From: mohakgoel1 Date: Sat, 15 Aug 2026 15:06:57 +0530 Subject: [PATCH 3/3] Replace custom CacheAligned with crossbeam_utils::CachePadded for active_workers Signed-off-by: mohakgoel1 --- src/pool/spawn.rs | 95 +++++++---------------------------------------- 1 file changed, 13 insertions(+), 82 deletions(-) diff --git a/src/pool/spawn.rs b/src/pool/spawn.rs index 7af8f5d..84807e9 100644 --- a/src/pool/spawn.rs +++ b/src/pool/spawn.rs @@ -6,6 +6,7 @@ use crate::pool::SchedConfig; use crate::queue::{Extras, LocalQueue, Pop, PopResult, TaskCell, TaskInjector, WithExtras}; +use crossbeam_utils::CachePadded; use fail::fail_point; use parking_lot_core::{FilterOp, ParkResult, ParkToken, UnparkToken}; use std::sync::{ @@ -14,78 +15,8 @@ use std::sync::{ }; use std::time::Instant; -// We intentionally use a custom CacheAligned struct instead of -// crossbeam_utils::CachePadded because CachePadded aligns to only -// 32 bytes on 32-bit arm targets, which may not fully prevent false -// sharing on all arm32 cache line sizes. This explicit struct uses -// platform-specific #[repr(C, align(N))] to guarantee correct -// alignment on every target: -// aarch64 -> 256 bytes (conservative: covers Fujitsu A64FX 256-byte -// cache lines; safe for Neoverse/Graviton 128-byte lines) -// x86-64, powerpc64 -> 128 bytes -// arm (32-bit) -> 64 bytes -// all others -> 64 bytes -#[cfg(target_arch = "aarch64")] -const CACHE_LINE_SIZE: usize = 256; - -#[cfg(any(target_arch = "x86_64", target_arch = "powerpc64"))] -const CACHE_LINE_SIZE: usize = 128; - -#[cfg(target_arch = "arm")] -const CACHE_LINE_SIZE: usize = 64; - -#[cfg(not(any( - target_arch = "x86_64", - target_arch = "aarch64", - target_arch = "powerpc64", - target_arch = "arm", -)))] -const CACHE_LINE_SIZE: usize = 64; - -#[cfg(target_arch = "aarch64")] -#[repr(C, align(256))] -struct CacheAligned { - value: AtomicUsize, - _pad: [u8; CACHE_LINE_SIZE - std::mem::size_of::()], -} - -#[cfg(any( - target_arch = "x86_64", - target_arch = "powerpc64", -))] -#[repr(C, align(128))] -struct CacheAligned { - value: AtomicUsize, - _pad: [u8; CACHE_LINE_SIZE - std::mem::size_of::()], -} - -#[cfg(target_arch = "arm")] -#[repr(C, align(64))] -struct CacheAligned { - value: AtomicUsize, - _pad: [u8; CACHE_LINE_SIZE - std::mem::size_of::()], -} - -#[cfg(not(any( - target_arch = "x86_64", - target_arch = "aarch64", - target_arch = "powerpc64", - target_arch = "arm", -)))] -#[repr(C, align(64))] -struct CacheAligned { - value: AtomicUsize, - _pad: [u8; CACHE_LINE_SIZE - std::mem::size_of::()], -} - -impl CacheAligned { - fn new(v: usize) -> Self { - CacheAligned { - value: AtomicUsize::new(v), - _pad: [0u8; CACHE_LINE_SIZE - std::mem::size_of::()], - } - } -} +// crossbeam_utils::CachePadded is used to isolate active_workers to its +// own cache line, preventing false sharing with global_queue in QueueCore. /// An usize is used to trace the threads that are working actively. /// To save additional memory and atomic operation, the number and @@ -112,7 +43,7 @@ pub fn is_shutdown(cnt: usize) -> bool { pub(crate) struct QueueCore { global_queue: TaskInjector, - active_workers: CacheAligned, + active_workers: CachePadded, config: SchedConfig, } @@ -120,7 +51,7 @@ impl QueueCore { pub fn new(global_queue: TaskInjector, config: SchedConfig) -> QueueCore { QueueCore { global_queue, - active_workers: CacheAligned::new(config.max_thread_count << WORKER_COUNT_SHIFT), + active_workers: CachePadded::new(AtomicUsize::new(config.max_thread_count << WORKER_COUNT_SHIFT)), config, } } @@ -130,7 +61,7 @@ impl QueueCore { /// If the method is going to wake up any threads, source is used to trace who triggers /// the action. pub fn ensure_workers(&self, source: usize) { - let cnt = self.active_workers.value.load(Ordering::Acquire); + let cnt = self.active_workers.load(Ordering::Acquire); if (cnt >> WORKER_COUNT_SHIFT) >= self.config.core_thread_count.load(Ordering::Acquire) || is_shutdown(cnt) { @@ -165,7 +96,7 @@ impl QueueCore { /// /// `source` is used to trace who triggers the action. pub fn mark_shutdown(&self, source: usize) { - self.active_workers.value.fetch_or(SHUTDOWN_BIT, Ordering::AcqRel); + self.active_workers.fetch_or(SHUTDOWN_BIT, Ordering::AcqRel); let addr = self as *const QueueCore as usize; unsafe { parking_lot_core::unpark_all(addr, UnparkToken(source)); @@ -174,7 +105,7 @@ impl QueueCore { /// Checks if the thread pool is shutting down. pub fn is_shutdown(&self) -> bool { - let cnt = self.active_workers.value.load(Ordering::Acquire); + let cnt = self.active_workers.load(Ordering::Acquire); is_shutdown(cnt) } @@ -182,13 +113,13 @@ impl QueueCore { /// /// It can be marked as sleep only when the pool is not shutting down. pub fn mark_sleep(&self) -> bool { - let mut cnt = self.active_workers.value.load(Ordering::Acquire); + let mut cnt = self.active_workers.load(Ordering::Acquire); loop { if is_shutdown(cnt) { return false; } - match self.active_workers.value.compare_exchange_weak( + match self.active_workers.compare_exchange_weak( cnt, cnt - WORKER_COUNT_BASE, Ordering::AcqRel, @@ -202,9 +133,9 @@ impl QueueCore { /// Marks current thread as woken up states. pub fn mark_woken(&self) { - let mut cnt = self.active_workers.value.load(Ordering::Acquire); + let mut cnt = self.active_workers.load(Ordering::Acquire); loop { - match self.active_workers.value.compare_exchange_weak( + match self.active_workers.compare_exchange_weak( cnt, cnt + WORKER_COUNT_BASE, Ordering::AcqRel, @@ -393,7 +324,7 @@ impl Local { } pub(crate) fn is_scaled_down_worker(&self) -> bool { - self.id > self.core.config.core_thread_count.load(Ordering::SeqCst) + self.id > self.core.config.core_thread_count.load(Ordering::Acquire) } pub(crate) fn drain(&mut self) {