left_right/
read.rs

1use crate::sync::{fence, Arc, AtomicPtr, AtomicUsize, Ordering};
2use std::cell::Cell;
3use std::fmt;
4use std::marker::PhantomData;
5use std::ptr::NonNull;
6
7// To make [`WriteHandle`] and friends work.
8#[cfg(doc)]
9use crate::WriteHandle;
10
11mod guard;
12pub use guard::ReadGuard;
13
14mod factory;
15pub use factory::ReadHandleFactory;
16
17/// A read handle to a left-right guarded data structure.
18///
19/// To use a handle, first call [`enter`](Self::enter) to acquire a [`ReadGuard`]. This is similar
20/// to acquiring a `Mutex`, except that no exclusive lock is taken. All reads of the underlying
21/// data structure can then happen through the [`ReadGuard`] (which implements `Deref<Target =
22/// T>`).
23///
24/// Reads through a `ReadHandle` only see the changes up until the last time
25/// [`WriteHandle::publish`] was called. That is, even if a writer performs a number of
26/// modifications to the underlying data, those changes are not visible to reads until the writer
27/// calls [`publish`](crate::WriteHandle::publish).
28///
29/// `ReadHandle` is not `Sync`, which means that you cannot share a `ReadHandle` across many
30/// threads. This is because the coordination necessary to do so would significantly hamper the
31/// scalability of reads. If you had many reads go through one `ReadHandle`, they would need to
32/// coordinate among themselves for every read, which would lead to core contention and poor
33/// multi-core performance. By having `ReadHandle` not be `Sync`, you are forced to keep a
34/// `ReadHandle` per reader, which guarantees that you do not accidentally ruin your performance.
35///
36/// You can create a new, independent `ReadHandle` either by cloning an existing handle or by using
37/// a [`ReadHandleFactory`]. Note, however, that creating a new handle through either of these
38/// mechanisms _does_ take a lock, and may therefore become a bottleneck if you do it frequently.
39pub struct ReadHandle<T> {
40    pub(crate) inner: Arc<AtomicPtr<T>>,
41    pub(crate) epochs: crate::Epochs,
42    epoch: Arc<AtomicUsize>,
43    epoch_i: usize,
44    enters: Cell<usize>,
45
46    // `ReadHandle` is _only_ Send if T is Sync. If T is !Sync, then it's not okay for us to expose
47    // references to it to other threads! Since negative impls are not available on stable, we pull
48    // this little hack to make the type not auto-impl Send, and then explicitly add the impl when
49    // appropriate.
50    _unimpl_send: PhantomData<*const T>,
51}
52unsafe impl<T> Send for ReadHandle<T> where T: Sync {}
53
54impl<T> Drop for ReadHandle<T> {
55    fn drop(&mut self) {
56        // epoch must already be even for us to have &mut self,
57        // so okay to lock since we're not holding up the epoch anyway.
58        let e = self.epochs.lock().unwrap().remove(self.epoch_i);
59        assert!(Arc::ptr_eq(&e, &self.epoch));
60        assert_eq!(self.enters.get(), 0);
61    }
62}
63
64impl<T> fmt::Debug for ReadHandle<T> {
65    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
66        f.debug_struct("ReadHandle")
67            .field("epochs", &self.epochs)
68            .field("epoch", &self.epoch)
69            .finish()
70    }
71}
72
73impl<T> Clone for ReadHandle<T> {
74    fn clone(&self) -> Self {
75        ReadHandle::new_with_arc(Arc::clone(&self.inner), Arc::clone(&self.epochs))
76    }
77}
78
79impl<T> ReadHandle<T> {
80    pub(crate) fn new(inner: T, epochs: crate::Epochs) -> Self {
81        let store = Box::into_raw(Box::new(inner));
82        let inner = Arc::new(AtomicPtr::new(store));
83        Self::new_with_arc(inner, epochs)
84    }
85
86    fn new_with_arc(inner: Arc<AtomicPtr<T>>, epochs: crate::Epochs) -> Self {
87        // tell writer about our epoch tracker
88        let epoch = Arc::new(AtomicUsize::new(0));
89        // okay to lock, since we're not holding up the epoch
90        let epoch_i = epochs.lock().unwrap().insert(Arc::clone(&epoch));
91
92        Self {
93            epochs,
94            epoch,
95            epoch_i,
96            enters: Cell::new(0),
97            inner,
98            _unimpl_send: PhantomData,
99        }
100    }
101
102    /// Create a [`ReadHandleFactory`] which is `Send` & `Sync` and can be shared across threads to create
103    /// additional [`ReadHandle`] instances.
104    pub fn factory(&self) -> ReadHandleFactory<T> {
105        ReadHandleFactory {
106            inner: Arc::clone(&self.inner),
107            epochs: Arc::clone(&self.epochs),
108        }
109    }
110}
111
112impl<T> ReadHandle<T> {
113    /// Take out a guarded live reference to the read copy of the `T`.
114    ///
115    /// While the guard lives, the [`WriteHandle`] cannot proceed with a call to
116    /// [`WriteHandle::publish`], so no queued operations will become visible to _any_ reader.
117    ///
118    /// If the `WriteHandle` has been dropped, this function returns `None`.
119    pub fn enter(&self) -> Option<ReadGuard<'_, T>> {
120        let enters = self.enters.get();
121        if enters != 0 {
122            // We have already locked the epoch.
123            // Just give out another guard.
124            let r_handle = self.inner.load(Ordering::Acquire);
125            // since we previously bumped our epoch, this pointer will remain valid until we bump
126            // it again, which only happens when the last ReadGuard is dropped.
127            let r_handle = unsafe { r_handle.as_ref() };
128
129            return if let Some(r_handle) = r_handle {
130                self.enters.set(enters + 1);
131                Some(ReadGuard {
132                    handle: guard::ReadHandleState::from(self),
133                    t: r_handle,
134                })
135            } else {
136                unreachable!("if pointer is null, no ReadGuard should have been issued");
137            };
138        }
139
140        // once we update our epoch, the writer can no longer do a swap until we set the MSB to
141        // indicate that we've finished our read. however, we still need to deal with the case of a
142        // race between when the writer reads our epoch and when they decide to make the swap.
143        //
144        // assume that there is a concurrent writer. it just swapped the atomic pointer from A to
145        // B. the writer wants to modify A, and needs to know if that is safe. we can be in any of
146        // the following cases when we atomically swap out our epoch:
147        //
148        //  1. the writer has read our previous epoch twice
149        //  2. the writer has already read our previous epoch once
150        //  3. the writer has not yet read our previous epoch
151        //
152        // let's discuss each of these in turn.
153        //
154        //  1. since writers assume they are free to proceed if they read an epoch with MSB set
155        //     twice in a row, this is equivalent to case (2) below.
156        //  2. the writer will see our epoch change, and so will assume that we have read B. it
157        //     will therefore feel free to modify A. note that *another* pointer swap can happen,
158        //     back to A, but then the writer would be block on our epoch, and so cannot modify
159        //     A *or* B. consequently, using a pointer we read *after* the epoch swap is definitely
160        //     safe here.
161        //  3. the writer will read our epoch, notice that MSB is not set, and will keep reading,
162        //     continuing to observe that it is still not set until we finish our read. thus,
163        //     neither A nor B are being modified, and we can safely use either.
164        //
165        // in all cases, using a pointer we read *after* updating our epoch is safe.
166
167        // so, update our epoch tracker.
168        self.epoch.fetch_add(1, Ordering::AcqRel);
169
170        // ensure that the pointer read happens strictly after updating the epoch
171        fence(Ordering::SeqCst);
172
173        // then, atomically read pointer, and use the copy being pointed to
174        let r_handle = self.inner.load(Ordering::Acquire);
175
176        // since we bumped our epoch, this pointer will remain valid until we bump it again
177        let r_handle = unsafe { r_handle.as_ref() };
178
179        if let Some(r_handle) = r_handle {
180            // add a guard to ensure we restore read parity even if we panic
181            let enters = self.enters.get() + 1;
182            self.enters.set(enters);
183            Some(ReadGuard {
184                handle: guard::ReadHandleState::from(self),
185                t: r_handle,
186            })
187        } else {
188            // the writehandle has been dropped, and so has both copies,
189            // so restore parity and return None
190            self.epoch.fetch_add(1, Ordering::AcqRel);
191            None
192        }
193    }
194
195    /// Returns true if the [`WriteHandle`] has been dropped.
196    pub fn was_dropped(&self) -> bool {
197        self.inner.load(Ordering::Acquire).is_null()
198    }
199
200    /// Returns a raw pointer to the read copy of the data.
201    ///
202    /// Note that it is only safe to read through this pointer if you _know_ that the writer will
203    /// not start writing into it. This is most likely only the case if you are calling this method
204    /// from inside a method that holds `&mut WriteHandle`.
205    ///
206    /// Casting this pointer to `&mut` is never safe.
207    pub fn raw_handle(&self) -> Option<NonNull<T>> {
208        NonNull::new(self.inner.load(Ordering::Acquire))
209    }
210}
211
212/// `ReadHandle` cannot be shared across threads:
213///
214/// ```compile_fail
215/// use left_right::ReadHandle;
216///
217/// fn is_sync<T: Sync>() {
218///   // dummy function just used for its parameterized type bound
219/// }
220///
221/// // the line below will not compile as ReadHandle does not implement Sync
222///
223/// is_sync::<ReadHandle<u64>>()
224/// ```
225///
226/// But, it can be sent across threads:
227///
228/// ```
229/// use left_right::ReadHandle;
230///
231/// fn is_send<T: Send>() {
232///   // dummy function just used for its parameterized type bound
233/// }
234///
235/// is_send::<ReadHandle<u64>>()
236/// ```
237///
238/// As long as the wrapped type is `Sync` that is.
239///
240/// ```compile_fail
241/// use left_right::ReadHandle;
242///
243/// fn is_send<T: Send>() {}
244///
245/// is_send::<ReadHandle<std::cell::Cell<u64>>>()
246/// ```
247#[allow(dead_code)]
248struct CheckReadHandleSendNotSync;