orx_concurrent_option/concurrent_option.rs
1use crate::{handle::Handle, mut_handle::MutHandle, states::StateU8};
2use core::{cell::UnsafeCell, mem::MaybeUninit, sync::atomic::AtomicU8};
3
4/// ConcurrentOption is a thread-safe and lock-free read-write option type.
5///
6/// ## ConcurrentOption Methods In Groups
7///
8/// ConcurrentOption methods are based on the standard Option with minor differences in order to better fit concurrent programs.
9///
10/// For example, instead of `fn map<U, F>(self, f: F) -> Option<U>`
11/// * ConcurrentOption implements `fn map<U, F>(&self, f: F) -> Option<U>` which is specialized to map over the reference while guaranteeing the lack of data race.
12/// * Note that the prior result can trivially be obtained by `maybe.exclusive_take().map(f)` when we have the ownership.
13///
14/// ### ⬤ Methods requiring self or &mut self
15///
16/// These methods are safe by the borrow checker and they behave similar to the original variants.
17///
18/// In order to separate them from the thread-safe versions, methods requiring `&mut self` are prefixed with **exclusive_**.
19///
20/// Some such methods are `unwrap`, `expect`, `exclusive_mut` or `exclusive_take`.
21///
22/// ### ⬤ Thread safe versions of mutating methods
23///
24/// Thread safe variants of mutating methods are available and they can be safely be called with a shared `&self` reference.
25///
26/// Some examples are `take`, `take_if`, `replace`, etc.
27///
28/// These methods guarantee that there exist no other mutation or no reading during the mutation.
29///
30/// ### ⬤ Thread safe versions of read methods
31///
32/// Thread safe variants of methods which access the underlying value to calculate a result are available.
33///
34/// Some examples are `is_some`, `map`, `and_then`, etc.
35///
36/// These methods guarantee that there exist no mutation while reading the data.
37///
38/// ### ⬤ Partially thread safe methods
39///
40/// Methods which return a shared reference `&T` or mutable reference `&mut T` to the underlying value of the optional are marked as `unsafe`.
41///
42/// These methods internally guarantee the creation of a valid reference in the absence of a data race. In this sense, they are thread safe.
43///
44/// On the other hand, since they return the reference, the reference is leaked outside the type. A succeeding mutation might lead to a data race, and hence, to an undefined behavior.
45///
46/// Some example methods are `as_ref`, `as_deref`, `insert`, etc.
47///
48/// ### ⬤ Methods to allow manual control on concurrency
49///
50/// ConcurrentOption also exposes methods which accepts a `core::sync::atomic::Ordering` and gives the control to the caller. These methods are suffixed with **with_order**, except for the state.
51///
52/// Some such methods are `state`, `as_ref_with_order`, `get_raw_with_order`, `clone_with_order`, etc.
53///
54/// ## Examples
55///
56/// ### Concurrent Read & Write
57///
58/// The following example demonstrates the ease of concurrently mutating the state of the option while safely reading the underlying data with multiple reader and writer threads.
59///
60/// ```rust
61/// use orx_concurrent_option::*;
62/// use std::time::Duration;
63///
64/// enum MutOperation {
65/// InitializeIfNone,
66/// UpdateIfSome,
67/// Replace,
68/// Take,
69/// TakeIf,
70/// }
71///
72/// impl MutOperation {
73/// fn new(i: usize) -> Self {
74/// match i % 5 {
75/// 0 => Self::InitializeIfNone,
76/// 1 => Self::UpdateIfSome,
77/// 2 => Self::Replace,
78/// 3 => Self::Take,
79/// _ => Self::TakeIf,
80/// }
81/// }
82/// }
83///
84/// let num_readers = 8;
85/// let num_writers = 8;
86///
87/// let values = vec![ConcurrentOption::<String>::none(); 8];
88///
89/// std::thread::scope(|s| {
90/// for _ in 0..num_readers {
91/// s.spawn(|| {
92/// for _ in 0..100 {
93/// std::thread::sleep(Duration::from_millis(100));
94/// let mut num_chars = 0;
95/// for maybe in &values {
96/// // concurrently access the value
97/// num_chars += maybe.map(|x| x.len()).unwrap_or(0);
98/// }
99/// assert!(num_chars <= 100);
100/// }
101/// });
102/// }
103///
104/// for _ in 0..num_writers {
105/// s.spawn(|| {
106/// for i in 0..100 {
107/// std::thread::sleep(Duration::from_millis(100));
108/// let e = i % values.len();
109///
110/// // concurrently update the option
111/// match MutOperation::new(i) {
112/// MutOperation::InitializeIfNone => {
113/// values[e].initialize_if_none(e.to_string());
114/// }
115/// MutOperation::UpdateIfSome => {
116/// values[e].update_if_some(|x| *x = format!("{}!", x));
117/// }
118/// MutOperation::Replace => {
119/// values[e].replace(e.to_string());
120/// }
121/// MutOperation::Take => {
122/// _ = values[e].take();
123/// }
124/// MutOperation::TakeIf => _ = values[e].take_if(|x| x.len() < 2),
125/// }
126/// let e = i % values.len();
127/// _ = values[e].initialize_if_none(e.to_string());
128/// }
129/// });
130/// }
131/// })
132/// ```
133///
134/// ### Concurrent Initialize & Read
135///
136/// A common use case for option is to model a delayed initialization; rather than concurrent mutation. In other words, we start with a None variant and at some point we receive the value and convert our option to Some(value), which will then stay as Some(value) throughout its lifetime.
137///
138/// This scenario demonstrates a use case where we can safely leak a reference outside the optional:
139/// * All references provided by ConcurrentOption are valid and data race free at the point they are obtained. In other words, we can only obtain a reference after the value is initialized; i.e., the option becomes Some(value).
140/// * Since we will never mutate the option after initialization, we can safely keep a reference to it without a concern about a data race.
141/// * However, no further mutation is our promise and responsibility as the caller. ConcurrentOption has no control over the leaked references; and hence, obtaining the reference is through the unsafe `as_ref` method.
142///
143/// For this scenario, we can make use of two matching methods:
144/// * `initialize_if_none` is a thread safe method to initialize the value of the option to the given value. It is safe to call the method on a Some variant, it will have no impact. Further, it makes sure that no reader can access the value until it is completely initialized.
145/// * `as_ref` method returns a reference to the underlying value if the option is a Some variant. Otherwise, if the value has not been initialized, we will safely receive None. Note that we could also use `as_ref_with_order` paired up with `Acquire` or `SeqCst` ordering if we want to model the access ordering manually.
146///
147/// ```rust
148/// use orx_concurrent_option::*;
149///
150/// fn reader(maybe: &ConcurrentOption<String>) {
151/// let mut is_none_at_least_once = false;
152/// let mut is_seven_at_least_once = false;
153/// for _ in 0..100 {
154/// std::thread::sleep(std::time::Duration::from_millis(100));
155///
156/// let read = unsafe { maybe.as_ref() };
157/// let is_none = read.is_none();
158/// let is_seven = read == Some(&7.to_string());
159///
160/// assert!(is_none || is_seven);
161///
162/// is_none_at_least_once |= is_none;
163/// is_seven_at_least_once |= is_seven;
164/// }
165/// assert!(is_none_at_least_once && is_seven_at_least_once);
166/// }
167///
168/// fn initializer(maybe: &ConcurrentOption<String>) {
169/// for _ in 0..50 {
170/// // wait for a while to simulate a delay
171/// std::thread::sleep(std::time::Duration::from_millis(100));
172/// }
173///
174/// let _ = maybe.initialize_if_none(7.to_string());
175///
176/// for _ in 0..50 {
177/// // it is safe to call `initialize_if_none` on Some variant
178/// // it will do nothing
179/// let inserted = maybe.initialize_if_none(1_000_000.to_string());
180/// assert!(!inserted);
181/// }
182/// }
183///
184/// let num_readers = 8;
185/// let num_writers = 8;
186///
187/// let maybe = ConcurrentOption::<String>::none();
188/// let maybe_ref = &maybe;
189///
190/// std::thread::scope(|s| {
191/// for _ in 0..num_readers {
192/// s.spawn(|| reader(maybe_ref));
193/// }
194/// for _ in 0..num_writers {
195/// s.spawn(|| initializer(maybe_ref));
196/// }
197/// });
198///
199/// assert_eq!(maybe.unwrap(), 7.to_string());
200/// ```
201pub struct ConcurrentOption<T> {
202 pub(crate) value: UnsafeCell<MaybeUninit<T>>,
203 pub(crate) state: AtomicU8,
204}
205
206impl<T> ConcurrentOption<T> {
207 pub(crate) fn get_handle(
208 &self,
209 initial_state: StateU8,
210 success_state: StateU8,
211 ) -> Option<Handle<'_>> {
212 Handle::get(&self.state, initial_state, success_state)
213 }
214
215 #[inline(always)]
216 pub(crate) fn spin_get_handle(
217 &self,
218 initial_state: StateU8,
219 success_state: StateU8,
220 ) -> Option<Handle<'_>> {
221 Handle::spin_get(&self.state, initial_state, success_state)
222 }
223
224 /// Provides the mut handle on the value of the optional:
225 /// * the optional must be in the `initial_state` for this method to succeed,
226 /// * the optional will be brought to `success_state` once the handle is dropped.
227 ///
228 /// # Safety
229 ///
230 /// This method is unsafe since the handle provides direct access to the underlying
231 /// value, skipping thread-safety guarantees.
232 pub unsafe fn mut_handle(
233 &self,
234 initial_state: StateU8,
235 success_state: StateU8,
236 ) -> Option<MutHandle<'_, T>> {
237 MutHandle::spin_get(self, initial_state, success_state)
238 }
239
240 /// Compares `self` and `other` while holding both in the reserved state for the entire
241 /// duration of the comparison, so that a concurrent mutation of either option cannot race
242 /// with the read of its value.
243 ///
244 /// `some_some` is only called with valid references to the underlying values of both options,
245 /// and only while both options are locked; it must not be able to observe or cause any further
246 /// mutation of either option.
247 pub(crate) fn locked_compare<R>(
248 &self,
249 other: &Self,
250 none_none: R,
251 some_none: R,
252 none_some: R,
253 some_some: impl FnOnce(&T, &T) -> R,
254 ) -> R {
255 if core::ptr::eq(self, other) {
256 // avoid locking the same option twice, which would deadlock
257 return match self.spin_get_handle(crate::states::SOME, crate::states::SOME) {
258 Some(handle) => {
259 let l = unsafe { (*self.value.get()).assume_init_ref() };
260 let result = some_some(l, l);
261 drop(handle);
262 result
263 }
264 None => none_none,
265 };
266 }
267
268 match self.spin_get_handle(crate::states::SOME, crate::states::SOME) {
269 None => match other.is_some() {
270 true => none_some,
271 false => none_none,
272 },
273 Some(handle_self) => {
274 let result = match other.spin_get_handle(crate::states::SOME, crate::states::SOME) {
275 Some(handle_other) => {
276 let l = unsafe { (*self.value.get()).assume_init_ref() };
277 let r = unsafe { (*other.value.get()).assume_init_ref() };
278 let out = some_some(l, r);
279 drop(handle_other);
280 out
281 }
282 None => some_none,
283 };
284 drop(handle_self);
285 result
286 }
287 }
288 }
289}
290
291unsafe impl<T: Send> Send for ConcurrentOption<T> {}
292
293unsafe impl<T: Sync> Sync for ConcurrentOption<T> {}