low_latency_data_structures/seqlock/
lock.rs

1use std::cell::UnsafeCell;
2use std::sync::Arc;
3use std::sync::atomic::{AtomicU64, Ordering, fence};
4
5use crossbeam_utils::CachePadded;
6
7use crate::seqlock::reader::Reader;
8use crate::seqlock::writer::Writer;
9
10/// Creates a new seqlock initialised with `initial_value`.
11///
12/// Returns a paired `(Writer, Reader)`. Additional readers can be obtained
13/// by cloning the [`Reader`].
14///
15/// # Examples
16///
17/// ```
18/// use low_latency_data_structures::seqlock::new;
19///
20/// let (writer, reader) = new(0u64);
21/// assert_eq!(reader.read(), 0);
22/// writer.write(42);
23/// assert_eq!(reader.read(), 42);
24/// ```
25pub fn new<T>(initial_value: T) -> (Writer<T>, Reader<T>)
26where
27    T: bytemuck::AnyBitPattern,
28{
29    let sl: Arc<SeqLock<T>> = Arc::new(
30        SeqLockInner {
31            data: UnsafeCell::new(initial_value),
32            seq: AtomicU64::new(0),
33        }
34        .into(),
35    );
36    let writer = Writer::new(sl.clone());
37    let reader = Reader::new(sl);
38    (writer, reader)
39}
40
41pub(super) type SeqLock<T> = CachePadded<SeqLockInner<T>>;
42
43pub(super) struct SeqLockInner<T> {
44    // `data` first, `seq` after: mirrors Rigtorp's layout so a small T and
45    // the seq counter land on the same cache line under the CachePadded
46    // wrapper. Reader touches one line per load; writer touches one line
47    // per store.
48    data: UnsafeCell<T>,
49    seq: AtomicU64,
50}
51
52// SAFETY: UnsafeCell is not Sync but access is synchronised through the
53// write()/read() methods using the seq-number protocol. Torn reads of `data`
54// would be undefined under the abstract C11/C++11 model, but `T:
55// AnyBitPattern` guarantees every bit pattern materialises a valid T, so the
56// worst-case observed value is stale, never UB. The seq check around the
57// read rejects stale or torn values before they leave `Reader::read`.
58unsafe impl<T> Sync for SeqLockInner<T> where T: bytemuck::AnyBitPattern {}
59
60impl<T> SeqLockInner<T>
61where
62    T: bytemuck::AnyBitPattern,
63{
64    #[inline]
65    pub(super) fn write(&self, value: T) {
66        let s = self.seq.load(Ordering::Relaxed);
67        self.seq.store(s.wrapping_add(1), Ordering::Relaxed);
68        // ARM: prevents the write from reordering above the s load
69        fence(Ordering::Release);
70        // SAFETY: see the unsafe impl Sync for SeqLockInner above.
71        // write_volatile also pins the store: it cannot be reordered with
72        // the surrounding seq stores by the compiler.
73        unsafe { self.data.get().write_volatile(value) };
74        self.seq.store(s.wrapping_add(2), Ordering::Release);
75    }
76
77    #[inline]
78    pub(super) fn read(&self) -> T {
79        loop {
80            let s1 = self.seq.load(Ordering::Acquire);
81            if s1 & 1 == 1 {
82                continue;
83            }
84            // SAFETY: see the unsafe impl Sync for SeqLockInner above. The
85            // torn read is filtered by the s1==s2 check below.
86            let read_attempt = unsafe { self.data.get().read_volatile() };
87            // ARM: prevents the read from reordering below the s2 load
88            fence(Ordering::Acquire);
89            let s2 = self.seq.load(Ordering::Relaxed);
90            if s1 == s2 {
91                return read_attempt;
92            }
93        }
94    }
95}
96
97#[cfg(feature = "tests_basic")]
98#[cfg(test)]
99mod tests {
100    use std::sync::Barrier;
101    use std::thread;
102
103    use super::*;
104
105    #[test]
106    fn first_read_returns_initial_value() {
107        for init_value in [42, 1, 23] {
108            let (_, reader) = new(init_value);
109            assert_eq!(reader.read(), init_value);
110        }
111    }
112
113    #[test]
114    fn single_thread_reads_what_it_has_written() {
115        let (writer, reader) = new(0);
116        for value in [42, 1, 23] {
117            writer.write(value);
118            assert_eq!(reader.read(), value);
119        }
120    }
121
122    #[test]
123    fn single_writer_multi_reader() {
124        const READERS: usize = 3;
125        const VALUES: [u64; 4] = [1, 4, 7, 9];
126        let (writer, reader) = new(0);
127        let reader = Arc::new(reader);
128        let barrier = Arc::new(Barrier::new(READERS + 1));
129        let writer_h = {
130            let barrier = barrier.clone();
131            thread::spawn(move || {
132                // wait for readers to read the initial value
133                barrier.wait();
134                for value in VALUES {
135                    writer.write(value);
136                    // waits for all readers to get the value
137                    barrier.wait();
138                }
139            })
140        };
141        let readers_hs: Vec<_> = (0..READERS)
142            .map(|_| {
143                let barrier = barrier.clone();
144                let reader = reader.clone();
145                thread::spawn(move || {
146                    let mut collected: Vec<u64> = Vec::with_capacity(VALUES.len() + 1);
147                    loop {
148                        if collected.len() == VALUES.len() + 1 {
149                            break;
150                        }
151                        let item = reader.read();
152                        if let Some(&last) = collected.last()
153                            && last == item
154                        {
155                            continue;
156                        } else {
157                            collected.push(item);
158                            barrier.wait();
159                        }
160                    }
161                    collected
162                })
163            })
164            .collect();
165        writer_h.join().unwrap();
166        let readers_results: Vec<Vec<u64>> =
167            readers_hs.into_iter().map(|h| h.join().unwrap()).collect();
168        assert!(
169            readers_results
170                .iter()
171                .all(|result| result.len() == VALUES.len() + 1 && &result[1..] == &VALUES[..])
172        );
173    }
174}
175
176#[cfg(test)]
177#[cfg(feature = "tests_dhat")]
178mod tests_dhat {
179    use std::hint::black_box;
180
181    use super::*;
182
183    #[test]
184    fn hot_path_without_allocations() {
185        let _profiler = dhat::Profiler::builder().testing().build();
186        let (writer, reader) = new(0u64);
187        const ITERS: u64 = 10_000_000;
188
189        // Warm up to absorb any one-time platform allocations: lazy symbol
190        // resolution in the dynamic linker, libstd TLS init, debug-build
191        // slow-path stubs that LLVM emits for overflow/panic messages. None
192        // of these scale with iteration count; the per-iteration regression
193        // we actually care about would dwarf the slack budget below.
194        for i in 0..10_000u64 {
195            writer.write(black_box(i));
196            black_box(reader.read());
197        }
198
199        let before = dhat::HeapStats::get();
200        for i in 0..ITERS {
201            writer.write(black_box(i));
202            black_box(reader.read());
203        }
204        let after = dhat::HeapStats::get();
205
206        let allocs = after.total_blocks - before.total_blocks;
207        assert!(
208            allocs < 64,
209            "hot path allocated {allocs} blocks over {ITERS} iterations; \
210             a real regression would be O(ITERS), this slack absorbs O(1) \
211             platform noise (lazy linker, debug stubs)"
212        );
213    }
214}
low_latency_data_structures/seqlock/lock.rs

low_latency_data_structures/seqlock/
lock.rs
