uqa_core/
atomic.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Conditional atomic updates using the compare/exchange interface supported by the MSRV.
8
9use std::sync::atomic::{AtomicI32, AtomicU64, Ordering};
10
11macro_rules! updater {
12    ($name:ident, $atomic:ty, $value:ty) => {
13        /// Return the previous value after a successful update, or the observed value when the callback returns `None`. The callback can run repeatedly after contention; the orderings apply to the successful exchange and failed reads respectively.
14        pub fn $name(
15            atomic: &$atomic,
16            set_order: Ordering,
17            fetch_order: Ordering,
18            mut update: impl FnMut($value) -> Option<$value>,
19        ) -> Result<$value, $value> {
20            let mut previous = atomic.load(fetch_order);
21            loop {
22                let next = update(previous).ok_or(previous)?;
23                match atomic.compare_exchange_weak(previous, next, set_order, fetch_order) {
24                    Ok(value) => return Ok(value),
25                    Err(value) => previous = value,
26                }
27            }
28        }
29    };
30}
31
32updater!(try_update_u64, AtomicU64, u64);
33updater!(try_update_i32, AtomicI32, i32);
34
35#[cfg(test)]
36mod tests {
37    use super::*;
38
39    #[test]
40    fn exhausted_unsigned_and_signed_identities_do_not_wrap() {
41        let unsigned = AtomicU64::new(u64::MAX);
42        assert_eq!(
43            try_update_u64(&unsigned, Ordering::Relaxed, Ordering::Relaxed, |id| id
44                .checked_add(1)),
45            Err(u64::MAX)
46        );
47        assert_eq!(unsigned.load(Ordering::Relaxed), u64::MAX);
48        let signed = AtomicI32::new(i32::MAX);
49        assert_eq!(
50            try_update_i32(&signed, Ordering::Relaxed, Ordering::Relaxed, |id| id
51                .checked_add(1)),
52            Err(i32::MAX)
53        );
54        assert_eq!(signed.load(Ordering::Relaxed), i32::MAX);
55        let zero = AtomicU64::new(0);
56        assert_eq!(
57            try_update_u64(&zero, Ordering::AcqRel, Ordering::Acquire, |n| n
58                .checked_sub(1)),
59            Err(0)
60        );
61        assert_eq!(zero.load(Ordering::Acquire), 0);
62    }
63
64    #[test]
65    fn changed_value_is_reloaded_before_retrying_the_callback() {
66        let atomic = AtomicU64::new(1);
67        let mut first = true;
68        let previous = try_update_u64(&atomic, Ordering::AcqRel, Ordering::Acquire, |value| {
69            if first {
70                first = false;
71                atomic.store(9, Ordering::Release);
72            }
73            value.checked_add(1)
74        });
75        assert_eq!(previous, Ok(9));
76        assert_eq!(atomic.load(Ordering::Acquire), 10);
77    }
78
79    #[test]
80    fn concurrent_updates_preserve_every_increment() {
81        let atomic = AtomicI32::new(0);
82        std::thread::scope(|scope| {
83            for _ in 0..4 {
84                scope.spawn(|| {
85                    for _ in 0..256 {
86                        try_update_i32(&atomic, Ordering::Relaxed, Ordering::Relaxed, |n| {
87                            n.checked_add(1)
88                        })
89                        .unwrap();
90                    }
91                });
92            }
93        });
94        assert_eq!(atomic.load(Ordering::Relaxed), 1024);
95    }
96}
uqa_core/atomic.rs

uqa_core/
atomic.rs
