From 20d282a788ca75215f9371129da56fb2ad1e6c36 Mon Sep 17 00:00:00 2001 From: Eliza Weisman Date: Sun, 2 Oct 2022 11:44:12 -0700 Subject: [PATCH 1/7] test(util): add a `test_dbg!` macro --- util/Cargo.toml | 3 +++ util/src/macros.rs | 31 +++++++++++++++++++++++++++++++ 2 files changed, 34 insertions(+) diff --git a/util/Cargo.toml b/util/Cargo.toml index a9651d06..0f9b840f 100644 --- a/util/Cargo.toml +++ b/util/Cargo.toml @@ -27,6 +27,9 @@ tracing = { git = "https://github.com/tokio-rs/tracing", default_features = fals [target.'cfg(loom)'.dependencies] loom = "0.5.5" +[target.'cfg(loom)'.dev-dependencies] +tracing_01 = { package = "tracing", version = "0.1.36" } + [package.metadata.docs.rs] all-features = true rustdoc-args = ["--cfg", "docsrs"] diff --git a/util/src/macros.rs b/util/src/macros.rs index 928563d6..f41fca63 100644 --- a/util/src/macros.rs +++ b/util/src/macros.rs @@ -66,3 +66,34 @@ macro_rules! unreachable_unchecked { } }); } + +#[cfg(all(test, not(loom)))] +macro_rules! test_dbg { + ($x:expr) => { + match $x { + x => { + tracing::debug!("{} = {x:?}", stringify!($x)); + x + } + } + }; +} + +#[cfg(all(test, loom))] +macro_rules! test_dbg { + ($x:expr) => { + match $x { + x => { + tracing_01::debug!("{} = {x:?}", stringify!($x)); + x + } + } + }; +} + +#[cfg(not(test))] +macro_rules! test_dbg { + ($x:expr) => { + $x + }; +} From a02dcae576868fc3414b06ed79323ee20bb54177 Mon Sep 17 00:00:00 2001 From: Eliza Weisman Date: Sun, 2 Oct 2022 13:01:55 -0700 Subject: [PATCH 2/7] wip --- Cargo.lock | 1 + util/src/loom.rs | 8 ++ util/src/sync.rs | 2 + util/src/sync/spin/backoff.rs | 10 ++ util/src/sync/spin/mutex.rs | 4 +- util/src/sync/tearable.rs | 171 ++++++++++++++++++++++++++++++++++ 6 files changed, 194 insertions(+), 2 deletions(-) create mode 100644 util/src/sync/tearable.rs diff --git a/Cargo.lock b/Cargo.lock index 3cc93a0d..010d62a5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -987,6 +987,7 @@ dependencies = [ "loom", "mycelium-bitfield", "proptest", + "tracing 0.1.36", "tracing 0.2.0", "tracing-subscriber 0.3.0", ] diff --git a/util/src/loom.rs b/util/src/loom.rs index 693c0737..f07870bb 100644 --- a/util/src/loom.rs +++ b/util/src/loom.rs @@ -10,6 +10,14 @@ mod inner { mod inner { #![allow(dead_code)] + #[cfg(test)] + pub(crate) use std::thread; + + #[cfg(test)] + pub(crate) fn model(f: impl Fn()) { + f() + } + pub(crate) mod alloc { /// Track allocations, detecting leaks #[derive(Debug, Default)] diff --git a/util/src/sync.rs b/util/src/sync.rs index c3e4b3cb..14da7dd3 100644 --- a/util/src/sync.rs +++ b/util/src/sync.rs @@ -9,6 +9,8 @@ pub use core::sync::atomic; pub mod cell; pub mod once; pub mod spin; +pub mod tearable; + #[doc(inline)] pub use self::once::{InitOnce, Lazy}; diff --git a/util/src/sync/spin/backoff.rs b/util/src/sync/spin/backoff.rs index 2ec76b3b..12be5b03 100644 --- a/util/src/sync/spin/backoff.rs +++ b/util/src/sync/spin/backoff.rs @@ -50,10 +50,20 @@ impl Backoff { #[inline(always)] pub fn spin(&mut self) { // Issue 2^exp pause instructions. + #[cfg(not(loom))] for _ in 0..(1 << self.exp) { hint::spin_loop(); } + #[cfg(loom)] + { + // when `loom` is in use, we only issue the hint once, because + // otherwise, loom will do a bunch of meaningless thread switches. + // Issue 2^exp pause instructions. + test_dbg!(1 << self.exp); + hint::spin_loop(); + } + if self.exp < self.max { self.exp += 1 } diff --git a/util/src/sync/spin/mutex.rs b/util/src/sync/spin/mutex.rs index a62e7309..9b812477 100644 --- a/util/src/sync/spin/mutex.rs +++ b/util/src/sync/spin/mutex.rs @@ -203,9 +203,9 @@ impl<'a, T: fmt::Display> fmt::Display for MutexGuard<'a, T> { } } -#[cfg(all(test, loom))] +#[cfg(test)] mod tests { - use loom::thread; + use crate::loom::{self, thread}; use std::prelude::v1::*; use std::sync::Arc; diff --git a/util/src/sync/tearable.rs b/util/src/sync/tearable.rs new file mode 100644 index 00000000..90928426 --- /dev/null +++ b/util/src/sync/tearable.rs @@ -0,0 +1,171 @@ +use crate::sync::{ + atomic::{AtomicU32, Ordering::*}, + hint, + spin::Backoff, +}; + +/// An efficient +/// [sequence lock]: https://en.wikipedia.org/wiki/Seqlock +pub struct TearableU64 { + seq: AtomicU32, + high: AtomicU32, + low: AtomicU32, +} + +pub struct TryWriteError(T); + +impl TearableU64 { + loom_const_fn! { + #[must_use] + pub(crate) fn new(value: u64) -> Self { + let (high, low) = split(value); + Self { + seq: AtomicU32::new(0), + high: AtomicU32::new(high), + low: AtomicU32::new(low), + } + } + } + + /// Reads the data stored inside the sequence lock. + /// + /// This method may spin if the lock is currently being written to. + pub fn load(&self) -> u64 { + loop { + let mut preread = self.seq.load(Acquire); + + let mut backoff = Backoff::new(); + while is_writing(test_dbg!(preread)) { + backoff.spin(); + preread = self.seq.load(Acquire); + } + + let low = self.low.load(Acquire); + let high = self.high.load(Acquire); + + let postread = self.seq.load(Acquire); + + // if the sequence numbers match, we didn't observe a torn write. + // return it! + if test_dbg!(preread) == test_dbg!(postread) { + return unsplit(high, low); + } + + // in the case we *did* observe a torn write, we only issue one spin + // loop hint, rather than backing off, because if we try another + // read, we should get something reasonable (unless it's being + // written to again). + hint::spin_loop() + } + } + + /// Writes a new value to the data stored inside the lock. + /// + /// This method may spin if the lock is currently being written to. + pub fn store(&self, value: u64) { + let mut seq = self.seq.load(Relaxed); + let mut backoff = Backoff::new(); + + // wait for a current write to complete + while test_dbg!(is_writing(seq)) { + backoff.spin(); + seq = self.seq.load(Relaxed); + } + + // increment the sequence number by one to indicate that we're starting + // a write. + while let Err(actual) = + test_dbg!(self + .seq + .compare_exchange_weak(seq, seq.wrapping_add(1), Acquire, Relaxed)) + { + seq = actual; + hint::spin_loop(); + } + + let (high, low) = split(value); + self.low.store(low, Release); + self.high.store(high, Release); + + // increment the sequence number again to indicate that we have finished + // a write. + self.seq.store(seq.wrapping_add(2), Release); + } + + // pub fn try_store(&self, value: u64) -> Result<(), TryWriteError> { + // // increment the sequence number by one to indicate that we're starting + // // a write. + // let seq = self.seq.fetch_add(1, Relaxed); + // if is_writing(seq) { + // return Err(TryWriteError(value)); + // } + + // let next = seq.wrapping_add(1); + // self.seq + // .compare_exchange(seq, next, Acquire, Relaxed) + // .map_err(|_| TryWriteError(value))?; + + // let (high, low) = split(value); + // self.low.store(low, Release); + // self.high.store(high, Release); + + // // increment the sequence number again to indicate that we have finished + // // a write. + // self.seq.store(seq.wrapping_add(2), Release); + // Ok(()) + // } +} + +/// Returns `true` if a sequence number indicates that a write is in progress. + +#[inline(always)] +const fn is_writing(seq: u32) -> bool { + seq & 1 == 1 +} + +const fn split(val: u64) -> (u32, u32) { + ((val >> 32) as u32, val as u32) +} + +const fn unsplit(high: u32, low: u32) -> u64 { + ((high as u64) << 32) | (low as u64) +} + +#[cfg(test)] +mod tests { + use crate::loom::{self, thread}; + use std::sync::Arc; + + use super::*; + + #[test] + fn spmc() { + const VALS: &[u64] = &[0, u64::MAX, u32::MAX as u64 + 1]; + const READERS: usize = 2; + + loom::model(|| { + let t = Arc::new(TearableU64::new(0)); + + let threads = (0..READERS) + .map(|_| { + let t = t.clone(); + thread::spawn(move || { + for _ in 0..READERS { + let value = test_dbg!(t.load()); + assert!(VALS.contains(&value)); + } + }) + }) + .collect::>(); + + for &value in &VALS[1..] { + t.store(test_dbg!(value)); + thread::yield_now(); + } + + for thread in threads { + thread.join().unwrap() + } + }); + } +} From 8878f16dca9ac0dd09bff4ed7395c616b87c0605 Mon Sep 17 00:00:00 2001 From: Eliza Weisman Date: Tue, 4 Oct 2022 09:31:19 -0700 Subject: [PATCH 3/7] wip --- Cargo.lock | 1 + util/Cargo.toml | 4 +++- util/src/lib.rs | 17 +++++++++++++++++ util/src/loom.rs | 28 ++++++++++++++++++++++++++++ util/src/sync/tearable.rs | 7 +++++-- 5 files changed, 54 insertions(+), 3 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 010d62a5..f4f7d4fb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -990,6 +990,7 @@ dependencies = [ "tracing 0.1.36", "tracing 0.2.0", "tracing-subscriber 0.3.0", + "tracing-subscriber 0.3.15", ] [[package]] diff --git a/util/Cargo.toml b/util/Cargo.toml index 0f9b840f..9e66f596 100644 --- a/util/Cargo.toml +++ b/util/Cargo.toml @@ -21,7 +21,7 @@ mycelium-bitfield = { path = "../bitfield", default-features = false } [dev-dependencies] proptest = "1" -tracing-subscriber = { git = "https://github.com/tokio-rs/tracing" } +tracing-subscriber = { git = "https://github.com/tokio-rs/tracing", features = ["fmt", "env-filter"] } tracing = { git = "https://github.com/tokio-rs/tracing", default_features = false, features = ["attributes", "std"] } [target.'cfg(loom)'.dependencies] @@ -29,6 +29,8 @@ loom = "0.5.5" [target.'cfg(loom)'.dev-dependencies] tracing_01 = { package = "tracing", version = "0.1.36" } +tracing-subscriber_03 = { package = "tracing-subscriber", version = "0.3.15", features = ["fmt", "env-filter"] } + [package.metadata.docs.rs] all-features = true diff --git a/util/src/lib.rs b/util/src/lib.rs index 6398ff39..56e6b92a 100644 --- a/util/src/lib.rs +++ b/util/src/lib.rs @@ -21,3 +21,20 @@ pub(crate) mod loom; pub use self::macros::*; pub use cordyceps as intrusive; pub use mycelium_bitfield as bits; + +#[cfg(test)] +pub(crate) mod test_util { + #[cfg(not(loom))] + pub(crate) fn trace_init() -> impl Drop { + use tracing_subscriber::{EnvFilter, prelude::*}; + let filter = EnvFilter::from_env("LOOM_LOG"); + tracing_subscriber::fmt().with_test_writer().without_time().with_env_filter(filter).set_default() + } + + #[cfg(loom)] + pub(crate) fn trace_init() -> impl Drop { + use tracing_subscriber_03::{EnvFilter, prelude::*}; + let filter = EnvFilter::from_env("LOOM_LOG"); + tracing_subscriber_03::fmt().with_test_writer().without_time().with_env_filter(filter).set_default() + } +} \ No newline at end of file diff --git a/util/src/loom.rs b/util/src/loom.rs index f07870bb..87c819b0 100644 --- a/util/src/loom.rs +++ b/util/src/loom.rs @@ -18,6 +18,34 @@ mod inner { f() } + + #[cfg(test)] + pub(crate) mod model { + #[non_exhaustive] + #[derive(Default)] + pub(crate) struct Builder { + pub(crate) max_threads: usize, + pub(crate) max_branches: usize, + pub(crate) max_permutations: Option, + // pub(crate) max_duration: Option, + pub(crate) preemption_bound: Option, + // pub(crate) checkpoint_file: Option, + pub(crate) checkpoint_interval: usize, + pub(crate) location: bool, + pub(crate) log: bool, + } + + impl Builder { + pub(crate) fn new() -> Self { + Self::default() + } + + pub(crate) fn check(&self, f: impl Fn()) { + super::model(f) + } + } + } + pub(crate) mod alloc { /// Track allocations, detecting leaks #[derive(Debug, Default)] diff --git a/util/src/sync/tearable.rs b/util/src/sync/tearable.rs index 90928426..20c4ef42 100644 --- a/util/src/sync/tearable.rs +++ b/util/src/sync/tearable.rs @@ -142,8 +142,11 @@ mod tests { fn spmc() { const VALS: &[u64] = &[0, u64::MAX, u32::MAX as u64 + 1]; const READERS: usize = 2; - - loom::model(|| { + let _trace = crate::test_util::trace_init(); + + let mut builder = loom::model::Builder::new(); + builder.max_branches = 10_000; + builder.check(|| { let t = Arc::new(TearableU64::new(0)); let threads = (0..READERS) From bf45d85e3498411b62232119ef7ba24f2af8ef47 Mon Sep 17 00:00:00 2001 From: Eliza Weisman Date: Wed, 5 Oct 2022 09:24:55 -0700 Subject: [PATCH 4/7] wip --- justfile | 6 +- util/src/sync/tearable.rs | 154 ++++++++++++++++++++------------------ 2 files changed, 88 insertions(+), 72 deletions(-) diff --git a/justfile b/justfile index b5686561..2809fb8c 100755 --- a/justfile +++ b/justfile @@ -26,6 +26,9 @@ _fmt := if env_var_or_default("GITHUB_ACTIONS", "") != "true" { "" } else { ``` } +_loom_max_preemptions := env_var_or_default("LOOM_MAX_PREEMPTIONS", "2") +_loom_max_branches := env_var_or_default("LOOM_MAX_BRANCHES", "10000") + # default recipe to display help information default: @echo "justfile for Mycelium" @@ -110,6 +113,7 @@ loom crate='' *args='': _get-nextest export RUSTFLAGS="--cfg loom ${RUSTFLAGS:-}" export LOOM_MAX_PREEMPTIONS="${LOOM_MAX_PREEMPTIONS:-2}" + export LOOM_MAX_BRANCHES="${LOOM_MAX_BRANCHES:-10000}" export LOOM_LOG="${LOOM_LOG:-mycelium=trace,maitake=trace,cordyceps=trace,debug}" # if logging is enabled, also enable location tracking. @@ -120,7 +124,7 @@ loom crate='' *args='': _get-nextest status "Disabled" "logging and location tracking" fi - status "Configured" "loom, LOOM_MAX_PREEMPTIONS=${LOOM_MAX_PREEMPTIONS}" + status "Configured" "loom, LOOM_MAX_PREEMPTIONS=${LOOM_MAX_PREEMPTIONS}, LOOM_MAX_BRANCHES=${LOOM_MAX_BRANCHES}" if [[ "${LOOM_CHECKPOINT_FILE:-}" ]]; then export LOOM_CHECKPOINT_FILE="${LOOM_CHECKPOINT_FILE:-}" diff --git a/util/src/sync/tearable.rs b/util/src/sync/tearable.rs index 20c4ef42..b3015e7c 100644 --- a/util/src/sync/tearable.rs +++ b/util/src/sync/tearable.rs @@ -1,10 +1,13 @@ use crate::sync::{ atomic::{AtomicU32, Ordering::*}, hint, - spin::Backoff, }; -/// An efficient +/// An efficient shareable 64-bit integer value, based on a [sequence lock]. +/// +/// This type is intended for cases where an atomic 64-bit value is needed on +/// platforms that do not support atomic operations on 64-bit memory locations. +/// /// [sequence lock]: https://en.wikipedia.org/wiki/Seqlock pub struct TearableU64 { seq: AtomicU32, @@ -12,8 +15,6 @@ pub struct TearableU64 { low: AtomicU32, } -pub struct TryWriteError(T); - impl TearableU64 { loom_const_fn! { #[must_use] @@ -27,28 +28,31 @@ impl TearableU64 { } } - /// Reads the data stored inside the sequence lock. + /// Loads the current value. /// - /// This method may spin if the lock is currently being written to. + /// This method may spin if a write is in progress or a torn write is + /// observed. + #[must_use] pub fn load(&self) -> u64 { loop { - let mut preread = self.seq.load(Acquire); - - let mut backoff = Backoff::new(); - while is_writing(test_dbg!(preread)) { - backoff.spin(); - preread = self.seq.load(Acquire); - } - - let low = self.low.load(Acquire); - let high = self.high.load(Acquire); - - let postread = self.seq.load(Acquire); - - // if the sequence numbers match, we didn't observe a torn write. - // return it! - if test_dbg!(preread) == test_dbg!(postread) { - return unsplit(high, low); + // snapshot the sequence number before reading the value, waiting + // for a write to complete if one is in progress. + let preread = self.seq.load(Acquire); + + // wait for a current write to complete + if !is_writing(test_dbg!(preread)) { + // read the number + let low = self.low.load(Acquire); + let high = self.high.load(Acquire); + + // snapshot the sequence number again after the value has been read. + let postread = self.seq.load(Acquire); + + // if the sequence numbers match, we didn't observe a torn write. + // return it! + if test_dbg!(preread) == test_dbg!(postread) { + return unsplit(high, low); + } } // in the case we *did* observe a torn write, we only issue one spin @@ -59,61 +63,65 @@ impl TearableU64 { } } - /// Writes a new value to the data stored inside the lock. - /// - /// This method may spin if the lock is currently being written to. pub fn store(&self, value: u64) { - let mut seq = self.seq.load(Relaxed); - let mut backoff = Backoff::new(); + let mut curr = self.seq.load(Relaxed); + loop { + // is a write in progress? + if is_writing(test_dbg!(curr)) { + hint::spin_loop(); + curr = self.seq.load(Relaxed); + continue; + } - // wait for a current write to complete - while test_dbg!(is_writing(seq)) { - backoff.spin(); - seq = self.seq.load(Relaxed); + match self.write(curr, value) { + // write succeeded! + Ok(_) => return, + // no joy, try again. + Err(actual) => curr = actual, + } + + hint::spin_loop(); } + } + + pub fn try_store(&self, value: u64) -> Result<(), u64> { + let mut curr = self.seq.load(Relaxed); + loop { + // is a write in progress? + if is_writing(test_dbg!(curr)) { + return Err(value); + } + + match self.write(curr, value) { + // write succeeded! + Ok(_) => return Ok(()), + // no joy, try again. + Err(actual) => curr = actual, + } - // increment the sequence number by one to indicate that we're starting - // a write. - while let Err(actual) = - test_dbg!(self - .seq - .compare_exchange_weak(seq, seq.wrapping_add(1), Acquire, Relaxed)) - { - seq = actual; hint::spin_loop(); } + } + + fn write(&self, curr: u32, value: u64) -> Result<(), u32> { + // try to increment the sequence number by one to indicate that + // we're starting a write. + test_dbg!(self + .seq + .compare_exchange_weak(curr, curr.wrapping_add(1), Acquire, Relaxed))?; + // incremented the sequence number, go ahead and do a write let (high, low) = split(value); self.low.store(low, Release); self.high.store(high, Release); // increment the sequence number again to indicate that we have finished // a write. - self.seq.store(seq.wrapping_add(2), Release); + self.seq.store(curr.wrapping_add(2), Release); + Ok(()) } - // pub fn try_store(&self, value: u64) -> Result<(), TryWriteError> { - // // increment the sequence number by one to indicate that we're starting - // // a write. - // let seq = self.seq.fetch_add(1, Relaxed); - // if is_writing(seq) { - // return Err(TryWriteError(value)); - // } - - // let next = seq.wrapping_add(1); - // self.seq - // .compare_exchange(seq, next, Acquire, Relaxed) - // .map_err(|_| TryWriteError(value))?; - - // let (high, low) = split(value); - // self.low.store(low, Release); - // self.high.store(high, Release); - - // // increment the sequence number again to indicate that we have finished - // // a write. - // self.seq.store(seq.wrapping_add(2), Release); - // Ok(()) - // } + // fn start_write(&self, curr: u32) -> Result<(), u32> {} } /// Returns `true` if a sequence number indicates that a write is in progress. @@ -139,21 +147,25 @@ mod tests { use super::*; #[test] - fn spmc() { - const VALS: &[u64] = &[0, u64::MAX, u32::MAX as u64 + 1]; - const READERS: usize = 2; + fn doesnt_tear() { + // multiple reader tests hit the loom branch limit really easily, since + // loom's scheduler may find a path through the execution where yielding + // in the spin loop makes it ping-pong back and forth between the two + // reader threads forever, without executing the writer again. this is + // annoying. so, just test with one reader in `cfg!(loom)`. + const READERS: usize = if cfg!(loom) { 1 } else { 4 }; + const VALS: &[u64] = &[0, u64::MAX, u32::MAX as u64 - 1]; + let _trace = crate::test_util::trace_init(); - - let mut builder = loom::model::Builder::new(); - builder.max_branches = 10_000; - builder.check(|| { + + loom::model(|| { let t = Arc::new(TearableU64::new(0)); let threads = (0..READERS) .map(|_| { let t = t.clone(); thread::spawn(move || { - for _ in 0..READERS { + for _ in 0..VALS.len() { let value = test_dbg!(t.load()); assert!(VALS.contains(&value)); } From 0efc572d195b09a1e2e27395e3f6b3c58cf952dc Mon Sep 17 00:00:00 2001 From: Eliza Weisman Date: Wed, 5 Oct 2022 09:50:56 -0700 Subject: [PATCH 5/7] moar terrible stuff --- util/src/sync/tearable.rs | 150 ++++++++++++++++++++++++++++++-------- 1 file changed, 121 insertions(+), 29 deletions(-) diff --git a/util/src/sync/tearable.rs b/util/src/sync/tearable.rs index b3015e7c..3d531fe5 100644 --- a/util/src/sync/tearable.rs +++ b/util/src/sync/tearable.rs @@ -3,6 +3,8 @@ use crate::sync::{ hint, }; +use core::convert::TryFrom; + /// An efficient shareable 64-bit integer value, based on a [sequence lock]. /// /// This type is intended for cases where an atomic 64-bit value is needed on @@ -103,25 +105,62 @@ impl TearableU64 { } } + pub fn fetch_add(&self, value: u64) -> u64 { + let mut curr = self.seq.load(Relaxed); + loop { + // is a write in progress? + if is_writing(test_dbg!(curr)) { + hint::spin_loop(); + curr = self.seq.load(Relaxed); + continue; + } + + if let Err(actual) = self.try_start_write(curr) { + curr = actual; + hint::spin_loop(); + continue; + } + + let (high, low) = split(value); + let prev_low = self.low.fetch_add(low, AcqRel); + let overflow = (prev_low as u64 + low as u64).saturating_sub(u32::MAX as u64); + let prev_high = match u32::try_from(high as u64 + overflow) { + Ok(high) => self.high.fetch_add(high, Release), + Err(_) => todo!("wrap around"), + }; + + self.finish_write(curr); + return unsplit(prev_high, prev_low); + } + } + fn write(&self, curr: u32, value: u64) -> Result<(), u32> { - // try to increment the sequence number by one to indicate that - // we're starting a write. - test_dbg!(self - .seq - .compare_exchange_weak(curr, curr.wrapping_add(1), Acquire, Relaxed))?; + self.try_start_write(curr)?; // incremented the sequence number, go ahead and do a write let (high, low) = split(value); self.low.store(low, Release); self.high.store(high, Release); - // increment the sequence number again to indicate that we have finished - // a write. - self.seq.store(curr.wrapping_add(2), Release); + self.finish_write(curr); Ok(()) } - // fn start_write(&self, curr: u32) -> Result<(), u32> {} + /// Try to increment the sequence number by one to indicate that + /// we're starting a write. + #[inline(always)] + fn try_start_write(&self, curr: u32) -> Result { + test_dbg!(self + .seq + .compare_exchange_weak(curr, curr.wrapping_add(1), Acquire, Relaxed)) + } + + // Increment the sequence number again to indicate that we have finished + // a write. + #[inline(always)] + fn finish_write(&self, curr: u32) { + test_dbg!(self.seq.store(curr.wrapping_add(2), Release)); + } } /// Returns `true` if a sequence number indicates that a write is in progress. @@ -145,40 +184,93 @@ mod tests { use std::sync::Arc; use super::*; + const U32_MAX: u64 = u32::MAX as u64; + + // multiple reader tests hit the loom branch limit really easily, since + // loom's scheduler may find a path through the execution where yielding + // in the spin loop makes it ping-pong back and forth between the two + // reader threads forever, without executing the writer again. this is + // annoying. so, just test with one reader in `cfg!(loom)`. + const READERS: usize = if cfg!(loom) { 1 } else { 4 }; + + fn spawn_readers( + tearable: &Arc, + vals: &'static [u64], + ) -> Vec> { + (0..READERS) + .map(|_| { + let t = tearable.clone(); + thread::spawn(move || { + for _ in 0..vals.len() { + let value = test_dbg!(t.load()); + assert!(vals.contains(&value)); + } + }) + }) + .collect::>() + } #[test] fn doesnt_tear() { - // multiple reader tests hit the loom branch limit really easily, since - // loom's scheduler may find a path through the execution where yielding - // in the spin loop makes it ping-pong back and forth between the two - // reader threads forever, without executing the writer again. this is - // annoying. so, just test with one reader in `cfg!(loom)`. - const READERS: usize = if cfg!(loom) { 1 } else { 4 }; - const VALS: &[u64] = &[0, u64::MAX, u32::MAX as u64 - 1]; + const VALS: &[u64] = &[0, u64::MAX, U32_MAX - 1]; let _trace = crate::test_util::trace_init(); loom::model(|| { let t = Arc::new(TearableU64::new(0)); - - let threads = (0..READERS) - .map(|_| { - let t = t.clone(); - thread::spawn(move || { - for _ in 0..VALS.len() { - let value = test_dbg!(t.load()); - assert!(VALS.contains(&value)); - } - }) - }) - .collect::>(); + let readers = spawn_readers(&t, VALS); for &value in &VALS[1..] { t.store(test_dbg!(value)); thread::yield_now(); } - for thread in threads { + for thread in readers { + thread.join().unwrap() + } + }); + } + + #[test] + fn fetch_add_would_tear() { + const VALS: &[u64] = &[U32_MAX, U32_MAX + 1]; + + let _trace = crate::test_util::trace_init(); + + loom::model(|| { + let t = Arc::new(TearableU64::new(U32_MAX)); + let readers = spawn_readers(&t, VALS); + + let prev = t.fetch_add(1); + thread::yield_now(); + assert_eq!(prev, U32_MAX); + + for thread in readers { + thread.join().unwrap() + } + }); + } + + #[test] + fn fetch_add_no_tear() { + const VALS: &[u64] = &[0, U32_MAX - 1, U32_MAX]; + + let _trace = crate::test_util::trace_init(); + + loom::model(|| { + let t = Arc::new(TearableU64::new(0)); + + let readers = spawn_readers(&t, VALS); + + let prev = t.fetch_add(U32_MAX - 1); + thread::yield_now(); + assert_eq!(prev, 0); + + let prev = t.fetch_add(1); + thread::yield_now(); + assert_eq!(prev, U32_MAX - 1); + + for thread in readers { thread.join().unwrap() } }); From 37e04bb798e94298de4a92bbda4b1c2e7e9804d9 Mon Sep 17 00:00:00 2001 From: Eliza Weisman Date: Fri, 7 Oct 2022 09:52:31 -0700 Subject: [PATCH 6/7] add polyfill api, docs --- util/src/sync/tearable.rs | 302 ++++++++++++++++++++++++++++++++++++-- 1 file changed, 291 insertions(+), 11 deletions(-) diff --git a/util/src/sync/tearable.rs b/util/src/sync/tearable.rs index 3d531fe5..3480b736 100644 --- a/util/src/sync/tearable.rs +++ b/util/src/sync/tearable.rs @@ -1,14 +1,46 @@ -use crate::sync::{ - atomic::{AtomicU32, Ordering::*}, - hint, +use crate::{ + fmt, + sync::{ + atomic::{AtomicU32, Ordering::*}, + hint, + }, }; -use core::convert::TryFrom; - -/// An efficient shareable 64-bit integer value, based on a [sequence lock]. +/// An efficient shareable 64-bit integer value, based on a [sequence +/// lock]. +/// +/// This type is intended for cases where an atomic 64-bit value is +/// needed on platforms that do not support atomic operations on 64-bit +/// memory locations. +/// +/// In addition, a [polyfill type](AtomicU64Polyfill), which uses 64-bit atomic +/// operations on platforms that support them, and tearable 32-bit atomic +/// operations otherwise, is also provided. This type can be used to provide a +/// more efficient implementation when 64-bit atomics are available, while still +/// exposing the same API when they are not. /// -/// This type is intended for cases where an atomic 64-bit value is needed on -/// platforms that do not support atomic operations on 64-bit memory locations. +/// # Implementation Details +/// +/// A [sequence lock] is a form of reader-writer lock which works by +/// allowing writers to load the locked data at any time, using a +/// sequence number which is incremented both before and after a write +/// operation to determine if a given read of the locked data has +/// observed a torn write. In essence, the sequence lock deliberately +/// performs a potentially racy read, but only *uses* the result of the +/// read if it did not observe a torn write. +/// +/// In Rust, sequence locks for arbitrary data cannot currently be +/// implemented soundly, as an unsynchronized (non-atomic) read of a +/// memory location during a write to the same location isu ndefined +/// behavior, even if the *result* of such a read is not actually +/// observed. However, this type is *not* unsound (and in fact does not +/// involve any unsafe code), as it stores a pair of [`AtomicU32`] +/// values, rather than arbitrary data. Therefore, reading the data is +/// always a pair of atomic operations, and potential tearing occurs +/// only between loading the two atomic values. This is potentially +/// racy, but it is not a *data race*, so the seqeunce lock +/// implementation is sound *when specialized specifically for atomic +/// integers*. /// /// [sequence lock]: https://en.wikipedia.org/wiki/Seqlock pub struct TearableU64 { @@ -17,8 +49,19 @@ pub struct TearableU64 { low: AtomicU32, } +pub use polyfill::AtomicU64Polyfill; + +/// Error returned by [`TearableU64::try_store`] that indicates another write +/// operation is in progress. +/// +/// This error contains the value that we were attempting to write, which can be +/// retrieved using [`WriteInProgress::into_inner`]. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct WriteInProgress(u64); + impl TearableU64 { loom_const_fn! { + /// Returns a new `TearableU64` with the provided current `value`. #[must_use] pub(crate) fn new(value: u64) -> Self { let (high, low) = split(value); @@ -65,6 +108,18 @@ impl TearableU64 { } } + /// Store `value` in this `TearableU64`. + /// + /// This method will spin if another write operation ([`store`], [`swap`], + /// or [`try_store`]) is concurrently in progress, and it will spin until + /// that write operation completes. Therefore, this method is **not** safe + /// for use in an interrupt handler if writes to a `TearableU64` may occur + /// outside of that interrupt handler. For use in an interrupt handler, + /// consider the fallible [`try_store`] method instead. + /// + /// [`store`]: Self::store + /// [`swap`]: Self::swap + /// [`try_store`]: Self::try_store pub fn store(&self, value: u64) { let mut curr = self.seq.load(Relaxed); loop { @@ -81,11 +136,26 @@ impl TearableU64 { // no joy, try again. Err(actual) => curr = actual, } - - hint::spin_loop(); } } + /// Attempt to store `value` in this `TearableU64`, failing if another write + /// operation ([`store`], [`swap`], or [`try_store`]) is concurrently in + /// progress. + /// + /// This method will not spin if a write is in progress, but may retry a + /// failed `compare_exchange_weak`. It is safe to call this method in an + /// interrupt handler, however, as it will not spin indefinitely if a write + /// operation is interrupted. + /// + /// # Returns + /// + /// - `Ok(())` if the value was successfully stored. + /// - `Err(`[`WriteInProgress`]`)` if another write is in progress + /// + /// [`store`]: Self::store + /// [`swap`]: Self::swap + /// [`try_store`]: Self::try_store pub fn try_store(&self, value: u64) -> Result<(), u64> { let mut curr = self.seq.load(Relaxed); loop { @@ -100,12 +170,44 @@ impl TearableU64 { // no joy, try again. Err(actual) => curr = actual, } + } + } + + /// Store a new value, returning the previous one. + /// + /// This method may spin if another write is in progress. + #[must_use = "if the return value of `TearableU64::swap` is not used, consider using `TearableU64::store` instead"] + pub fn swap(&self, value: u64) -> u64 { + let mut curr = self.seq.load(Relaxed); + loop { + // is a write in progress? + if is_writing(test_dbg!(curr)) { + hint::spin_loop(); + curr = self.seq.load(Relaxed); + continue; + } - hint::spin_loop(); + // try to start a write + if let Err(actual) = self.try_start_write(curr) { + curr = actual; + hint::spin_loop(); + continue; + } + + // write started! + let (high, low) = split(value); + let prev_low = self.low.swap(low, AcqRel); + let prev_high = self.high.swap(high, AcqRel); + + self.finish_write(curr); + return unsplit(prev_high, prev_low); } } + // TODO(eliza): finish this + /* pub fn fetch_add(&self, value: u64) -> u64 { + use core::convert::TryFrom; let mut curr = self.seq.load(Relaxed); loop { // is a write in progress? @@ -133,6 +235,7 @@ impl TearableU64 { return unsplit(prev_high, prev_low); } } + */ fn write(&self, curr: u32, value: u64) -> Result<(), u32> { self.try_start_write(curr)?; @@ -149,6 +252,7 @@ impl TearableU64 { /// Try to increment the sequence number by one to indicate that /// we're starting a write. #[inline(always)] + #[cfg_attr(loom, track_caller)] fn try_start_write(&self, curr: u32) -> Result { test_dbg!(self .seq @@ -158,11 +262,45 @@ impl TearableU64 { // Increment the sequence number again to indicate that we have finished // a write. #[inline(always)] + #[cfg_attr(loom, track_caller)] fn finish_write(&self, curr: u32) { test_dbg!(self.seq.store(curr.wrapping_add(2), Release)); } } +impl fmt::Debug for TearableU64 { + #[inline] + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("TearableU64") + .field("value", &self.load()) + .field("seq", &self.seq.load(Relaxed)) + .finish() + } +} + +// === impl WriteInProgress === + +impl fmt::Display for WriteInProgress { + #[inline] + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + f, + "failed to store {}; another write is in progress", + self.0 + ) + } +} + +impl WriteInProgress { + /// Returns the value that the caller of [`TearableU64::try_store`] + /// attempted to store when this error was returned. + pub fn into_inner(self) -> u64 { + self.0 + } +} + +// === helpers === + /// Returns `true` if a sequence number indicates that a write is in progress. #[inline(always)] @@ -170,14 +308,153 @@ const fn is_writing(seq: u32) -> bool { seq & 1 == 1 } +/// Split a 64-bit integer into a `(high, low)` pair of 32-bit integers. +#[inline(always)] const fn split(val: u64) -> (u32, u32) { ((val >> 32) as u32, val as u32) } +/// Combine a `(high, low)` pair of 32-bit integers into a 64-bit integer. const fn unsplit(high: u32, low: u32) -> u64 { ((high as u64) << 32) | (low as u64) } +/// `AtomicU64` version of the polyfill. +// NOTE: `target_arch` values of "arm", "mips", and +// "powerpc" refer specifically to the 32-bit versions +// of those architectures; the 64-bit architectures get +// the `target_arch` strings "aarch64", "mips64", and +// "powerpc64", respectively. +#[cfg(not(any( + target_arch = "arm", + target_arch = "mips", + target_arch = "powerpc", + target_arch = "riscv32", +)))] +mod polyfill { + use super::*; + use crate::sync::atomic::AtomicU64; + + /// A polyfill implementation of an [`AtomicU64`] that is available + /// regardless of whether or not the target architecture supports 64-bit + /// atomic operations. + /// + /// This type provides a limited subset of the [`AtomicU64`] API. When + /// 64-bit atomic operations are available, this type is implemented using + /// an [`AtomicU64`]. Otherwise, it is implemented using a [`TearableU64`]. + /// + /// This version of `AtomicU64Polyfill` was compiled for a target + /// architecture that supports 64-bit atomic operations. + #[derive(Debug)] + pub struct AtomicU64Polyfill(AtomicU64); + + impl AtomicU64Polyfill { + loom_const_fn! { + #[must_use] + pub fn new(value: u64) -> Self { + Self(AtomicU64::new(value)) + } + } + + /// Load the current value in the cell. + /// + /// This method always performs an [`Acquire`] load. + #[inline] + #[must_use] + #[cfg_attr(loom, track_caller)] + pub fn load(&self) -> u64 { + self.0.load(Acquire) + } + + /// Store `value` in the cell. + /// + /// This method always performs a [`Release`] store. + #[inline] + #[cfg_attr(loom, track_caller)] + pub fn store(&self, value: u64) { + self.0.store(value, Acquire) + } + + /// Store `value` in the cell, returning the previous value. + /// + /// This method always performs an [`AcqRel`] swap. + #[inline] + #[must_use] + #[cfg_attr(loom, track_caller)] + pub fn swap(&self, value: u64) -> u64 { + self.0.swap(value, AcqRel) + } + } +} + +/// Version of the polyfill without `AtomicU64`. +// NOTE: `target_arch` values of "arm", "mips", and +// "powerpc" refer specifically to the 32-bit versions +// of those architectures; the 64-bit architectures get +// the `target_arch` strings "aarch64", "mips64", and +// "powerpc64", respectively. +#[cfg(any( + target_arch = "arm", + target_arch = "mips", + target_arch = "powerpc", + target_arch = "riscv32", +))] +mod polyfill { + use super::*; + + /// A polyfill implementation of an [`AtomicU64`] that is available + /// regardless of whether or not the target architecture supports 64-bit + /// atomic operations. + /// + /// This type provides a limited subset of the [`AtomicU64`] API. When + /// 64-bit atomic operations are available, this type is implemented using + /// an [`AtomicU64`]. Otherwise, it is implemented using a [`TearableU64`]. + /// + /// This version of `AtomicU64Polyfill` was compiled for a target + /// architecture that does not support 64-bit atomic operations, and + /// therefore uses a [`TearableU64`]. + #[derive(Debug)] + pub struct AtomicU64Polyfill(TearableU64); + + impl AtomicU64Polyfill { + loom_const_fn! { + #[must_use] + pub fn new(value: u64) -> Self { + Self(TearableU64::new(value)) + } + } + + /// Load the current value in the cell. + /// + /// This method always performs an [`Acquire`] load. + #[inline] + #[must_use] + #[cfg_attr(loom, track_caller)] + pub fn load(&self) -> u64 { + self.0.load() + } + + /// Store `value` in the cell. + /// + /// This method always performs a [`Release`] store. + #[inline] + #[cfg_attr(loom, track_caller)] + pub fn store(&self, value: u64) { + self.0.store(value) + } + + /// Store `value` in the cell, returning the previous value. + /// + /// This method always performs an [`AcqRel`] swap. + #[inline] + #[must_use] + #[cfg_attr(loom, track_caller)] + pub fn swap(&self, value: u64) -> u64 { + self.0.swap(value) + } + } +} + #[cfg(test)] mod tests { use crate::loom::{self, thread}; @@ -231,6 +508,8 @@ mod tests { }); } + // TODO(eliza): put this back when you finish `fetch_add`... + /* #[test] fn fetch_add_would_tear() { const VALS: &[u64] = &[U32_MAX, U32_MAX + 1]; @@ -275,4 +554,5 @@ mod tests { } }); } + */ } From b520dbed743162b882fd7dce7d7991c72c944d2f Mon Sep 17 00:00:00 2001 From: Eliza Weisman Date: Sat, 8 Oct 2022 09:27:27 -0700 Subject: [PATCH 7/7] wip --- util/src/sync/tearable.rs | 119 +++++++++++++++++++++++++++++++------- 1 file changed, 98 insertions(+), 21 deletions(-) diff --git a/util/src/sync/tearable.rs b/util/src/sync/tearable.rs index 3480b736..942186c2 100644 --- a/util/src/sync/tearable.rs +++ b/util/src/sync/tearable.rs @@ -204,10 +204,8 @@ impl TearableU64 { } } - // TODO(eliza): finish this - /* - pub fn fetch_add(&self, value: u64) -> u64 { - use core::convert::TryFrom; + // TODO(eliza): add a fetch_add for 64-bit vals... + pub fn fetch_add_u32(&self, value: u32) -> u64 { let mut curr = self.seq.load(Relaxed); loop { // is a write in progress? @@ -223,19 +221,32 @@ impl TearableU64 { continue; } - let (high, low) = split(value); - let prev_low = self.low.fetch_add(low, AcqRel); - let overflow = (prev_low as u64 + low as u64).saturating_sub(u32::MAX as u64); - let prev_high = match u32::try_from(high as u64 + overflow) { - Ok(high) => self.high.fetch_add(high, Release), - Err(_) => todo!("wrap around"), + let prev_low = self.low.fetch_add(value, AcqRel); + let overflow = ((prev_low as u64 + value as u64) >> 32) as u32; + let prev_high = if overflow > 0 { + let prev_high = self.high.fetch_add(overflow, AcqRel); + // did adding to the high half wrap? + if prev_high as u64 + overflow as u64 > u32::MAX as u64 { + // NOTE(eliza): this doesn't *need* to be a swap, but it's + // nice to be able to check that nobody else is writing... + let _low = self.low.swap(prev_high.wrapping_add(overflow), AcqRel); + debug_assert_eq!( + _low, + prev_low.wrapping_add(value), + "fetch_add_u32: low value should not be modified \ + concurrently, since we haven't incremented the seq \ + number to finish writing yet. this is a bug!" + ); + } + prev_high + } else { + self.high.load(Acquire) }; self.finish_write(curr); return unsplit(prev_high, prev_low); } } - */ fn write(&self, curr: u32, value: u64) -> Result<(), u32> { self.try_start_write(curr)?; @@ -372,7 +383,7 @@ mod polyfill { #[inline] #[cfg_attr(loom, track_caller)] pub fn store(&self, value: u64) { - self.0.store(value, Acquire) + self.0.store(value, Release) } /// Store `value` in the cell, returning the previous value. @@ -480,7 +491,12 @@ mod tests { thread::spawn(move || { for _ in 0..vals.len() { let value = test_dbg!(t.load()); - assert!(vals.contains(&value)); + assert!( + vals.contains(&value), + "\n value: {:?}\n vals: {:?}", + value, + vals + ); } }) }) @@ -508,10 +524,8 @@ mod tests { }); } - // TODO(eliza): put this back when you finish `fetch_add`... - /* #[test] - fn fetch_add_would_tear() { + fn fetch_add_u32_would_tear() { const VALS: &[u64] = &[U32_MAX, U32_MAX + 1]; let _trace = crate::test_util::trace_init(); @@ -520,7 +534,7 @@ mod tests { let t = Arc::new(TearableU64::new(U32_MAX)); let readers = spawn_readers(&t, VALS); - let prev = t.fetch_add(1); + let prev = t.fetch_add_u32(1); thread::yield_now(); assert_eq!(prev, U32_MAX); @@ -531,7 +545,7 @@ mod tests { } #[test] - fn fetch_add_no_tear() { + fn fetch_add_u32_no_tear() { const VALS: &[u64] = &[0, U32_MAX - 1, U32_MAX]; let _trace = crate::test_util::trace_init(); @@ -541,11 +555,32 @@ mod tests { let readers = spawn_readers(&t, VALS); - let prev = t.fetch_add(U32_MAX - 1); + let prev = t.fetch_add_u32(u32::MAX - 1); thread::yield_now(); assert_eq!(prev, 0); - let prev = t.fetch_add(1); + let prev = t.fetch_add_u32(1); + thread::yield_now(); + assert_eq!(prev, U32_MAX - 1); + + for thread in readers { + thread.join().unwrap() + } + }); + } + + #[test] + fn fetch_add_u32_wrap_low() { + const VALS: &[u64] = &[U32_MAX - 1, U32_MAX + 99]; + + let _trace = crate::test_util::trace_init(); + + loom::model(|| { + let t = Arc::new(TearableU64::new(U32_MAX - 1)); + + let readers = spawn_readers(&t, VALS); + + let prev = t.fetch_add_u32(100); thread::yield_now(); assert_eq!(prev, U32_MAX - 1); @@ -554,5 +589,47 @@ mod tests { } }); } - */ + + #[test] + fn fetch_add_u32_wrap_high() { + const VALS: &[u64] = &[u64::MAX, 100]; + + let _trace = crate::test_util::trace_init(); + + loom::model(|| { + let t = Arc::new(TearableU64::new(u64::MAX)); + + let readers = spawn_readers(&t, VALS); + + let prev = t.fetch_add_u32(100); + thread::yield_now(); + assert_eq!(prev, u64::MAX); + + for thread in readers { + thread.join().unwrap() + } + }); + } + + #[test] + fn fetch_add_u32_high_maxed() { + const HIGH_MAX: u64 = U32_MAX << 32; + const VALS: &[u64] = &[HIGH_MAX, HIGH_MAX + 100]; + + let _trace = crate::test_util::trace_init(); + + loom::model(|| { + let t = Arc::new(TearableU64::new(HIGH_MAX)); + + let readers = spawn_readers(&t, VALS); + + let prev = t.fetch_add_u32(100); + thread::yield_now(); + assert_eq!(prev, HIGH_MAX); + + for thread in readers { + thread.join().unwrap() + } + }); + } }