arctic/concurrent/map.rs
1//! Auxiliary types for use with [`ConcurrentMap`][crate::concurrent::Map].
2
3use core::ops::ControlFlow;
4use core::ops::RangeFull;
5use core::sync::atomic::Ordering;
6
7#[cfg_attr(not(doc), expect(unused))]
8use crate::ConcurrentMap;
9use crate::Key;
10#[cfg_attr(not(doc), expect(unused))]
11use crate::SequentialMap;
12use crate::concurrent::Shard;
13use crate::concurrent::Smr;
14use crate::concurrent::Value;
15use crate::concurrent::iter;
16use crate::concurrent::smr;
17use crate::concurrent::smr::Guard as _;
18use crate::concurrent::value;
19use crate::raw::Cursor;
20use crate::raw::Edge;
21use crate::raw::cursor;
22use crate::raw::cursor::Path;
23use crate::raw::cursor::path;
24use crate::raw::edge::Meta as _;
25use crate::raw::key::Len as _;
26use crate::sequential;
27use crate::stat;
28
29/// See [`smr::Guard`].
30pub type Guard<'g, K, V, S> = <S as Smr<K, V>>::Guard<'g>;
31
32/// See [`value::Owned`].
33pub type Owned<'g, K, V, S> = value::Owned<Guard<'g, K, V, S>, V>;
34
35/// See [`value::Shared`].
36pub type Shared<'g, K, V, S> = value::Shared<Guard<'g, K, V, S>, V>;
37
38/// See [`value::Updated`].
39pub type Updated<'g, K, V, S> = value::Updated<Guard<'g, K, V, S>, V>;
40
41/// See [`value::Upserted`].
42pub type Upserted<'g, K, V, S> = value::Upserted<Guard<'g, K, V, S>, V>;
43
44/// Lock-free concurrent map that supports lexicographically ordered, non-linearizable range and prefix scans.
45///
46/// # Usage
47///
48/// Refer to [`SequentialMap`] for an introduction.
49/// The [`ConcurrentMap`] API differs in three ways: concurrent operations,
50/// safe memory reclamation, and advanced point operations.
51///
52/// ## Concurrent operations
53///
54/// Unlike [`SequentialMap`], an instance of [`ConcurrentMap`] can be shared
55/// and modified concurrently across threads. Methods that usually require a mutable reference
56/// (e.g., [`SequentialMap::upsert`]) instead use atomics to synchronize internally,
57/// allowing them to take an immutable reference (e.g., [`ConcurrentMap::upsert`]).
58///
59/// Note that scan operations are not linearizable. They do, however,
60/// satisfy weaker guarantees: (a) scans observe keys at most once, in order;
61/// and (b) scans observe all keys within bounds that were inserted before
62/// the scan starts, and were not removed before the scan ends.
63///
64/// ## Safe memory reclamation
65///
66/// In order to provide wait-free reads, [`ConcurrentMap`] requires
67/// a safe memory reclamation (SMR) mechanism to detect when
68/// allocations are safe to free. This results in the following API changes:
69///
70/// 1. Values are always returned behind guards. For example,
71/// while a successful [`sequential::Map::update`] returns ownership of
72/// the old value, a successful [`ConcurrentMap::update`] instead returns an [`Updated`]
73/// guard that allows references to the old and new value.
74///
75/// The guard may have other restrictions depending on the SMR implementation:
76/// for example, epoch-based SMR cannot free any memory while a guard is alive,
77/// and hazard keys currently only support holding a single guard at a time.
78///
79/// 2. Values behind guards are always read-only. This can be worked around by
80/// either using a value type with internal synchronization (e.g., `Box<Mutex<T>>`),
81/// or by obtaining a mutable reference to [`ConcurrentMap`] and then using the
82/// sequential API via [`ConcurrentMap::as_sequential`].
83///
84/// 3. Values distinguish between inline (e.g., integers) and indirect (e.g., `Box<T>`).
85/// In short, we return [`Value::Borrowed`] instead of `&V`, because the memory location
86/// where `V` itself is stored may be concurrently updated.
87/// (See [`Value`] for more information.)
88///
89/// ## Advanced point operations
90///
91/// Point operations can internally fail and retry under contention.
92/// We give the caller control over retries by providing variants of point
93/// operations (ending in suffix `_with`, e.g.,
94/// [`ConcurrentMap::update_with`]) that
95/// take a closure.
96///
97/// This can be used to efficiently implement lazy value initialization,
98/// or synchronization logic where the next value is computed from the
99/// current value, and then atomically inserted or updated.
100pub struct Map<K: Key, V: Value, S = smr::Default> {
101 smr: S,
102 seq: sequential::Map<K, V>,
103}
104
105impl<K: Key, V: Value, S: Default> Default for Map<K, V, S> {
106 fn default() -> Self {
107 Self::new()
108 }
109}
110
111impl<K: Key, V: Value, S: Default> Map<K, V, S> {
112 /// Construct an empty map with the default safe memory reclamation state.
113 pub fn new() -> Self {
114 Self::with_smr(S::default())
115 }
116}
117
118impl<K: Key, V: Value, S> Map<K, V, S> {
119 /// Construct an empty map with the given safe memory reclamation state.
120 pub const fn with_smr(smr: S) -> Self {
121 Self {
122 smr,
123 seq: sequential::Map::<K, V>::new(),
124 }
125 }
126}
127
128/// # Basic operations
129impl<K: Key, V: Value, S: Smr<K, V>> Map<K, V, S> {
130 /// Get a mutable view as a [`SequentialMap`] for temporary access to a more
131 /// efficient and flexible single-threaded API. For permanent access, use
132 /// [`From`].
133 ///
134 /// This method is safe because `&mut` guarantees this thread holds the
135 /// only reference to the underlying map.
136 ///
137 /// # Examples
138 ///
139 /// ```rust
140 /// use core::ops::ControlFlow;
141 /// use core::convert::Infallible;
142 /// use std::thread;
143 ///
144 /// use arctic::concurrent::smr;
145 /// use arctic::ConcurrentMap;
146 /// use arctic::Order;
147 /// use arctic::sequential;
148 ///
149 /// let mut map = ConcurrentMap::<u32, u64>::default();
150 ///
151 /// // Concurrently insert into map
152 /// thread::scope(|scope| {
153 /// let map = ↦
154 /// for id in 0..8 {
155 /// scope.spawn(move || {
156 /// map.insert(id, id as u64).expect("Key is not present");
157 /// });
158 /// }
159 /// });
160 ///
161 /// // Access sequential entry API
162 /// map.as_sequential()
163 /// .entry(8)
164 /// .or_insert(8);
165 ///
166 /// // Access sequential mutable iteration API
167 /// map.as_sequential()
168 /// .range_mut(5..=12)
169 /// .entries_mut(Order::Ascend)
170 /// .try_fold((), |(), (key, value)| {
171 /// assert!(key >= 5);
172 /// assert!(key <= 8, "Inserted up to 8");
173 /// assert_eq!(key, *value as u32);
174 /// *value += 1;
175 /// ControlFlow::<Infallible>::Continue(())
176 /// });
177 ///
178 /// // Sanity check that mutations are visible from concurrent map
179 /// let mut len = 0;
180 /// map.all()
181 /// .entries(Order::Descend)
182 /// .try_fold((), |(), (key, value)|{
183 /// let expected = if key >= 5 { key + 1 } else { key };
184 /// assert_eq!(*value as u32, expected);
185 /// len += 1;
186 /// ControlFlow::<Infallible>::Continue(())
187 /// });
188 /// assert_eq!(len, 9);
189 /// ```
190 #[inline]
191 pub fn as_sequential(&mut self) -> &mut sequential::Map<K, V> {
192 &mut self.seq
193 }
194
195 /// Get an immutable reference to the underlying safe memory reclamation state.
196 #[inline]
197 pub fn smr(&self) -> &S {
198 &self.smr
199 }
200
201 /// Get a mutable reference to the underlying safe memory reclamation state.
202 #[inline]
203 pub fn smr_mut(&mut self) -> &mut S {
204 &mut self.smr
205 }
206}
207
208/// # Point operations
209///
210/// This set of operations operates on a single key-value pair.
211///
212/// These operations are linearizable.
213impl<K: Key, V: Value, S: Smr<K, V>> Map<K, V, S> {
214 /// Returns whether `key` has an associated value.
215 ///
216 /// # Examples
217 ///
218 /// ```rust
219 /// use arctic::ConcurrentMap;
220 ///
221 /// let mut map = ConcurrentMap::<u64, u64>::new();
222 /// map.insert(1, 2).expect("Key is not present");
223 /// assert!(map.contains_key(&1));
224 /// assert!(!map.contains_key(&2));
225 /// ```
226 pub fn contains_key(&self, key: &K::Borrowed) -> bool {
227 let reader = K::Read::from(key);
228 let mut guard = self.smr.guard(reader);
229 unsafe { self.get_raw(&mut guard, reader) }.is_some()
230 }
231
232 /// Returns an immutable reference to the value associated with `key`.
233 ///
234 /// For a mutable reference, see [`ConcurrentMap::as_sequential`] and
235 /// [`SequentialMap::get_mut`].
236 /// There is no way to safely get a mutable reference to a value from an immutable [`Map`].
237 ///
238 /// # Examples
239 ///
240 /// ```rust
241 /// use arctic::ConcurrentMap;
242 ///
243 /// let map = ConcurrentMap::<u64, u64>::default();
244 /// let key = 64;
245 ///
246 /// assert!(map.get(&key).is_none());
247 ///
248 /// match map.insert(key, 3) {
249 /// Err(_) => unreachable!(),
250 /// Ok(new) => assert_eq!(*new, 3),
251 /// }
252 ///
253 /// match map.get(&key) {
254 /// None => unreachable!(),
255 /// Some(value) => assert_eq!(*value, 3),
256 /// }
257 /// ```
258 pub fn get<'g>(&'g self, key: &K::Borrowed) -> Option<Shared<'g, K, V, S>> {
259 let reader = K::Read::from(key);
260 let mut guard = self.smr.guard(reader);
261 let value = unsafe { self.get_raw(&mut guard, reader)? };
262 Some(unsafe { Shared::<'_, K, V, S>::wrap(guard, value) })
263 }
264
265 /// If there is no value associated with `key`, associate it with `value`.
266 ///
267 /// <div class="warning">
268 ///
269 /// This is **not** the same behavior as the standard library
270 /// (e.g., [`std::collections::BTreeMap::insert`]); see [`Map::upsert`] if
271 /// an existing value should be updated instead.)
272 ///
273 /// </div>
274 ///
275 /// Returns `Ok(&new_value)` if the insert succeeded,
276 /// or else `Err((&old_value, new_value))` if there is an existing
277 /// `old_value` associated with the key.
278 ///
279 /// See [`ConcurrentMap::insert_with`] for dynamic control flow and value construction.
280 ///
281 /// # Examples
282 ///
283 /// ```rust
284 /// use arctic::key::Str;
285 /// use arctic::key::NonNull;
286 /// use arctic::ConcurrentMap;
287 ///
288 /// let map = ConcurrentMap::<&'static Str<NonNull>, u64>::default();
289 /// let key = Str::new("korlex").expect("No null byte");
290 ///
291 /// // Key is not present, insert succeeds
292 /// match map.insert(key, 3) {
293 /// Err(_) => unreachable!(),
294 /// Ok(new) => assert_eq!(*new, 3),
295 /// }
296 ///
297 /// // Key is present, insert fails
298 /// match map.insert(key, 5) {
299 /// Err((old, new)) => {
300 /// assert_eq!(*old, 3);
301 /// assert_eq!(new, 5);
302 /// }
303 /// Ok(_) => unreachable!(),
304 /// }
305 /// ```
306 #[expect(clippy::type_complexity)]
307 pub fn insert<'g, 'k>(
308 &'g self,
309 key: K::Insert<'k>,
310 value: V,
311 ) -> Result<Shared<'g, K, V, S>, (Shared<'g, K, V, S>, V)> {
312 let mut value = Some(value);
313 self.insert_with(key, || value.take().expect("Call thunk once"))
314 .map_err(|(shared, initial)| {
315 (
316 shared,
317 value
318 .xor(initial)
319 .expect("Value must be in thunk or initial"),
320 )
321 })
322 }
323
324 /// Unconditionally associate `key` with `value`.
325 ///
326 /// Returns an [`Upserted`] guard that provides immutable references
327 /// to the (optional) old value and the newly updated (or inserted) value.
328 ///
329 /// See [`ConcurrentMap::upsert_with`] for dynamic control flow and value construction.
330 ///
331 /// # Examples
332 ///
333 /// ```rust
334 /// use arctic::key::BoxedStr;
335 /// use arctic::key::Terminated;
336 /// use arctic::key::Str;
337 /// use arctic::ConcurrentMap;
338 ///
339 /// let map = ConcurrentMap::<BoxedStr<Terminated<b'\n'>>, u64>::default();
340 /// let key = Str::new("arqad\n").expect("Newline terminated");
341 ///
342 /// // Key is not present, upsert performs insert
343 /// let upserted = map.upsert(key, 3);
344 /// assert_eq!(upserted.old(), None);
345 /// assert_eq!(*upserted.new(), 3);
346 ///
347 /// // Key is present, upsert performs update
348 /// let upserted = map.upsert(key, 5);
349 /// assert_eq!(upserted.old().copied(), Some(3));
350 /// assert_eq!(*upserted.new(), 5);
351 /// ```
352 pub fn upsert<'k>(&self, key: K::Insert<'k>, value: V) -> Upserted<'_, K, V, S> {
353 match self.upsert_with(key, Some(value), |_, new| {
354 ControlFlow::<(), _>::Continue(new.take().expect("Value is always initialized"))
355 }) {
356 Upsert::Success(upserted) => upserted,
357 Upsert::Break { .. } => unreachable!(),
358 }
359 }
360
361 /// If there is a value associated with `key`, update it to `value`.
362 ///
363 /// Returns `Ok((&old_value, &new_value))` if the update succeeded,
364 /// or else `Err(new_value)` if there was no old value associated with `key`.
365 ///
366 /// See [`ConcurrentMap::update_with`] for dynamic control flow and value construction.
367 ///
368 /// # Examples
369 ///
370 /// ```rust
371 /// use arctic::ConcurrentMap;
372 ///
373 /// let map = ConcurrentMap::<u32, Box<u64>>::default();
374 ///
375 /// match map.update(&37, Box::new(5)) {
376 /// Err(new) => assert_eq!(*new, 5),
377 /// Ok(_) => unreachable!(),
378 /// }
379 ///
380 /// match map.insert(37, Box::new(3)) {
381 /// Err(_) => unreachable!(),
382 /// Ok(new) => assert_eq!(*new, 3),
383 /// }
384 ///
385 /// match map.update(&37, Box::new(5)) {
386 /// Err(_) => unreachable!(),
387 /// Ok(updated) => {
388 /// assert_eq!(*updated.old(), 3);
389 /// assert_eq!(*updated.new(), 5);
390 /// },
391 /// }
392 /// ```
393 pub fn update<'g>(&'g self, key: &K::Borrowed, value: V) -> Result<Updated<'g, K, V, S>, V> {
394 match self.update_with(key, Some(value), |_, initial| {
395 ControlFlow::<(), _>::Continue(initial.take().expect("Value is always initialized"))
396 }) {
397 Update::Absent { new: Some(initial) } => Err(initial),
398 Update::Success(updated) => Ok(updated),
399 Update::Absent { new: None } | Update::Break { .. } => unreachable!(),
400 }
401 }
402
403 /// If there is a value associated with `key`, remove it from the map,
404 /// recursively removing empty tree nodes.
405 ///
406 /// This method is slow because it must keep a traversal stack, and scan and
407 /// delete empty nodes. See [`ConcurrentMap::remove_non_recursive`]
408 /// for a faster, but potentially memory-intensive alternative.
409 ///
410 /// Returns `Some(&old_value)` if the remove succeeded, or else `None` if
411 /// there was no old value associated with `key`.
412 ///
413 /// See [`ConcurrentMap::remove_with`] for dynamic control flow.
414 ///
415 /// # Examples
416 ///
417 /// ```rust
418 /// use arctic::ConcurrentMap;
419 ///
420 /// let map = ConcurrentMap::<u128, u64>::default();
421 /// let key = 0xabc;
422 ///
423 /// assert!(map.remove(&key).is_none());
424 /// map.insert(key, 5).expect("Key is not present");
425 /// match map.remove(&key) {
426 /// None => unreachable!(),
427 /// Some(removed) => assert_eq!(*removed, 5),
428 /// }
429 /// ```
430 pub fn remove<'g>(&'g self, key: &K::Borrowed) -> Option<Owned<'g, K, V, S>> {
431 match self.remove_with(key, |_| ControlFlow::Continue(())) {
432 Remove::Absent => None,
433 Remove::Success { old } => Some(old),
434 Remove::Break { old: _ } => unreachable!(),
435 }
436 }
437
438 /// If there is a value associated with `key`, remove it from the map,
439 /// **without** recursively removing empty tree nodes.
440 ///
441 /// <div class="warning">
442 ///
443 /// This method is much faster than [`ConcurrentMap::remove`],
444 /// because no traversal
445 /// stack or node scanning and replacement is necessary; however, it means
446 /// the memory usage of the tree is no longer correlated with the number of
447 /// keys and values it contains.
448 ///
449 /// This method should only be used if removals are rare or removed keys
450 /// are expected to be reinserted.
451 //
452 /// </div>
453 ///
454 /// Returns `Some(&old_value)` if the remove succeeded, or else `None` if
455 /// there was no old value associated with `key`.
456 ///
457 /// See [`ConcurrentMap::remove_non_recursive_with`] for dynamic control flow.
458 pub fn remove_non_recursive(&self, key: &K::Borrowed) -> Option<Owned<'_, K, V, S>> {
459 match self.remove_non_recursive_with(key, |_| ControlFlow::Continue(())) {
460 Remove::Absent => None,
461 Remove::Success { old } => Some(old),
462 Remove::Break { old: _ } => unreachable!(),
463 }
464 }
465}
466
467/// # Scan operations
468///
469/// This set of operations allows the caller to select a subtree
470/// (by prefix or range) for non-linearizable iteration.
471impl<K, V, S> Map<K, V, S>
472where
473 K: Key,
474 V: Value,
475 S: Smr<K, V>,
476{
477 /// Get an immutable reference to the entire tree.
478 ///
479 /// # Examples
480 ///
481 /// ```rust
482 /// use arctic::ConcurrentMap;
483 /// use arctic::Order;
484 ///
485 /// let map = ConcurrentMap::<u64, u64>::default();
486 /// map.insert(1, 2).expect("Key not present");
487 /// map.insert(3, 4).expect("Key not present");
488 ///
489 /// assert_eq!(map.all().entries(Order::Ascend).count(), 2);
490 /// ```
491 pub fn all(&self) -> iter::Shard<'_, 'static, K, V, RangeFull, Guard<'_, K, V, S>> {
492 let guard = self.smr.guard(K::Read::default());
493 unsafe { Shard::new(guard, self.seq.raw.all()) }
494 }
495
496 /// Get an immutable reference to the subtree of keys beginning with `prefix`.
497 ///
498 /// # Examples
499 ///
500 /// ```rust
501 /// use arctic::concurrent;
502 /// use arctic::ConcurrentMap;
503 /// use arctic::key::BoxedStr;
504 /// use arctic::key::NonNull;
505 /// use arctic::key::Str;
506 /// use arctic::Order;
507 ///
508 /// let map = ConcurrentMap::<BoxedStr<NonNull>, Box<u64>>::default();
509 ///
510 /// for (key, value) in [("prefix-one", 3), ("prefix-two", 2), ("three", 1)] {
511 /// map.insert(
512 /// Str::new(key).expect("No null byte"),
513 /// Box::new(value),
514 /// ).expect("Key not present");
515 /// }
516 ///
517 /// // Get all key value pairs where key starts with prefix
518 /// //
519 /// // Need a temporary binding here since lifetimes of references
520 /// // returned from iterators is tied to this shard
521 /// //
522 /// // Note: prefix does not need to satisfy any particular invariants;
523 /// // can be invalid UTF-8 or contain null or terminator bytes
524 /// let prefix = map.prefix("prefix");
525 ///
526 /// let entries: concurrent::EntryIter<_, _, _> = prefix.entries(Order::Ascend);
527 ///
528 /// // WARNING: using `entries` as `Iterator` requires cloning keys,
529 /// // which is expensive here due to BoxedStr keys
530 /// assert_eq!(entries.count(), 2);
531 ///
532 /// // Can use lending iterator API to avoid cloning
533 /// let mut entries: concurrent::EntryIter<_, _, _> = prefix.entries(Order::Ascend);
534 /// while let Some((key, _)) = entries.lend() {
535 /// assert!(key.as_str().starts_with("prefix"));
536 /// }
537 /// ```
538 pub fn prefix<'g, 'k>(
539 &'g self,
540 prefix: impl Into<K::Read<'k>>,
541 ) -> iter::Shard<'g, 'k, K, V, RangeFull, Guard<'g, K, V, S>> {
542 let prefix = prefix.into();
543 let guard = self.smr.guard(prefix);
544 unsafe { Shard::new(guard, self.seq.raw.prefix(prefix)) }
545 }
546
547 /// Get an immutable reference to the subtree of keys within `range`.
548 ///
549 /// # Examples
550 ///
551 /// ```rust
552 /// use arctic::ConcurrentMap;
553 /// use arctic::Order;
554 ///
555 /// let map = ConcurrentMap::<u64, u64>::default();
556 /// map.insert(1, 2).expect("Key not present");
557 /// map.insert(3, 4).expect("Key not present");
558 /// map.insert(5, 6).expect("Key not present");
559 ///
560 /// let range = map.range(3..=7);
561 ///
562 /// for (key, value) in range.entries(Order::Descend) {
563 /// assert!((3..=7).contains(&key));
564 /// }
565 /// ```
566 pub fn range<'g, 'k, R>(&'g self, range: R) -> iter::Shard<'g, 'k, K, V, R, Guard<'g, K, V, S>>
567 where
568 R: crate::raw::iter::Range<K::Read<'k>>,
569 {
570 let prefix = range.common_prefix();
571 let guard = self.smr.guard(prefix);
572 unsafe { Shard::new(guard, self.seq.raw.range(range, prefix)) }
573 }
574}
575
576/// # Advanced point operations
577///
578/// This set of operations extends the point operations to take a closure,
579/// allowing the caller to dynamically break out of an operation or lazily
580/// allocate a value. Importantly, this closure can observe the value
581/// currently associated with a key before deciding what to do, which enables
582/// more complex coordination in a concurrent setting.
583///
584/// For example, a concurrent counter could use
585/// [`ConcurrentMap::upsert_with`] to either
586/// insert one or update the current count by one, or an index could use
587/// [`ConcurrentMap::remove_with`] to
588/// remove a value only if it hasn't been concurrently updated.
589///
590/// These operations are linearizable.
591impl<K, V, S> Map<K, V, S>
592where
593 K: Key,
594 V: Value,
595 S: Smr<K, V>,
596{
597 /// If there is no value associated with `key`, call the provided `insert` closure
598 /// to compute a new value.
599 ///
600 /// The closure is called at most once, even under contention; the value will be
601 /// reused once allocated.
602 ///
603 /// Returns `Ok(&new_value)` if the insert succeeded,
604 /// or else `Err((&old_value, new_value))` if there is an existing
605 /// `old_value` associated with the key. `new_value` is `None`
606 /// if the closure was never called, or `Some` if this insert
607 /// was pre-empted by a concurrent insert to the same key.
608 ///
609 /// # Examples
610 ///
611 /// ```rust
612 /// use core::ops::ControlFlow;
613 ///
614 /// use arctic::ConcurrentMap;
615 /// use arctic::key::BoxedStr;
616 /// use arctic::key::NonNull;
617 /// use arctic::key::Str;
618 ///
619 /// let map = ConcurrentMap::<BoxedStr<NonNull>, Box<u64>>::default();
620 /// let key = Str::new("zipir").expect("No null byte");
621 ///
622 /// // Key not present, new value lazily allocated
623 /// match map.insert_with(key, || Box::new(10)) {
624 /// Ok(new) => {
625 /// assert_eq!(*new, 10);
626 /// }
627 /// Err(_) => unreachable!(),
628 /// }
629 ///
630 /// // Key present, new value not allocated
631 /// match map.insert_with(key, || Box::new(15)) {
632 /// Ok(_) => unreachable!(),
633 /// Err((old, new)) => {
634 /// assert_eq!(*old, 10);
635 /// assert!(new.is_none());
636 /// },
637 /// }
638 /// ```
639 #[expect(clippy::type_complexity)]
640 pub fn insert_with<'g, 'k, F>(
641 &'g self,
642 key: K::Insert<'k>,
643 insert: F,
644 ) -> Result<Shared<'g, K, V, S>, (Shared<'g, K, V, S>, Option<V>)>
645 where
646 F: FnOnce() -> V,
647 {
648 let mut thunk = Some(insert);
649
650 match self.upsert_with(key, None, |old, new| match old {
651 None => ControlFlow::Continue(match new.take() {
652 None => (thunk.take().expect("Call thunk once"))(),
653 Some(new) => new,
654 }),
655 Some(_) => ControlFlow::Break(()),
656 }) {
657 Upsert::Success(upserted) => Ok(upserted
658 .try_into_inserted()
659 .unwrap_or_else(|_| unreachable!("Continue on `None`"))),
660 Upsert::Break { old, new } => Err((old.expect("Break on `Some`"), new)),
661 }
662 }
663
664 /// Associate `key` with `value`, calling the provided `upsert` closure to
665 /// break or compute a new value.
666 ///
667 /// The closure may be called multiple times under contention,
668 /// and takes an immutable reference to the current value (if there is one), as well as `initial`
669 /// (on the first call) or `Some(prev_value)` (on subsequent calls); use [`Option::take`]
670 /// to move out of the option.
671 ///
672 /// Returns an [`Upsert`] enum.
673 ///
674 /// # Examples
675 ///
676 /// ```rust
677 /// use core::ops::ControlFlow;
678 ///
679 /// use arctic::ConcurrentMap;
680 /// use arctic::concurrent::map::Upsert;
681 ///
682 /// let map = ConcurrentMap::<u16, Box<u64>>::default();
683 /// let key = 20;
684 ///
685 /// // Key not present, closure continues, new value lazily allocated
686 /// match map.upsert_with(key, None, |old, new| {
687 /// assert!(old.is_none());
688 /// assert!(new.is_none());
689 /// ControlFlow::Continue(Box::new(9))
690 /// }) {
691 /// Upsert::Success(upserted) => {
692 /// assert!(upserted.old().is_none());
693 /// assert_eq!(*upserted.new(), 9);
694 /// },
695 /// Upsert::Break { .. } => unreachable!(),
696 /// }
697 ///
698 /// // Key present, closure breaks, new value not allocated
699 /// match map.upsert_with(key, None, |old, new| {
700 /// assert!(old.copied() == Some(9));
701 /// assert!(new.is_none());
702 /// ControlFlow::Break(())
703 /// }) {
704 /// Upsert::Success(_) => unreachable!(),
705 /// Upsert::Break { old, new } => {
706 /// assert_eq!(old.as_deref().copied(), Some(9));
707 /// assert!(new.is_none());
708 /// },
709 /// }
710 ///
711 /// // Key present, closure continues, new value lazily allocated (and reused under contention)
712 /// match map.upsert_with(key, None, |old, new| {
713 /// let next = old.copied().unwrap_or(0) + 1;
714 ///
715 /// ControlFlow::Continue(
716 /// new.take()
717 /// // Reuse allocation under contention
718 /// .map(|mut new: Box<u64>| {
719 /// *new = next;
720 /// new
721 /// })
722 /// // Allocate new value
723 /// .unwrap_or_else(|| Box::new(next)))
724 /// }) {
725 /// Upsert::Success(updated) => {
726 /// assert_eq!(updated.old().copied(), Some(9));
727 /// assert_eq!(*updated.new(), 10);
728 /// }
729 /// _ => unreachable!(),
730 /// }
731 /// ```
732 pub fn upsert_with<'g, 'k, F>(
733 &'g self,
734 key: K::Insert<'k>,
735 mut initial: Option<V>,
736 mut upsert: F,
737 ) -> Upsert<'g, K, V, S>
738 where
739 F: FnMut(Option<&V::Borrowed>, &mut Option<V>) -> ControlFlow<(), V>,
740 {
741 let reader = K::insert_as_read(key);
742 let mut guard = self.smr.guard(reader);
743
744 // NOTE: this is a macro so we get disjoint mutable borrows of `initial`
745 macro_rules! upsert {
746 () => {
747 |old: Option<u64>, new: Option<u64>| {
748 initial = new.map(|new| V::from_raw_unchecked(new));
749
750 match upsert(
751 old.as_ref().map(|old| V::borrow_from_raw_unchecked(old)),
752 &mut initial,
753 ) {
754 ControlFlow::Continue(new) => ControlFlow::Continue(new.into_raw()),
755 ControlFlow::Break(()) => ControlFlow::Break(()),
756 }
757 }
758 };
759 }
760
761 let upsert = match if cfg!(feature = "opt-no-path") {
762 Err(initial.take().map(V::into_raw))
763 } else {
764 unsafe {
765 self.upsert_with_optimistic(
766 &mut guard,
767 reader,
768 initial.take().map(V::into_raw),
769 upsert!(),
770 )
771 }
772 } {
773 Ok(upsert) => upsert,
774 Err(initial) => unsafe {
775 self.upsert_with_pessimistic(&mut guard, reader, initial, upsert!())
776 },
777 };
778
779 match upsert {
780 UpsertRaw::Success { old, new } => {
781 Upsert::Success(unsafe { Upserted::<K, V, S>::wrap(guard, old, new) })
782 }
783 UpsertRaw::Break { old } => Upsert::Break {
784 old: old.map(|old| unsafe { Shared::<K, V, S>::wrap(guard, old) }),
785 new: initial,
786 },
787 }
788 }
789
790 /// If there is a value associated with `key`, call the provided `update` closure
791 /// to break or compute a new value.
792 ///
793 /// The closure may be called multiple times under contention,
794 /// and takes an immutable reference to the current value, as well as `initial`
795 /// (on the first call) or `Some(prev_value)` (on subsequent calls); use [`Option::take`]
796 /// to move out of the option.
797 ///
798 /// Returns an [`Update`] enum.
799 ///
800 /// # Examples
801 ///
802 /// ```rust
803 /// use core::ops::ControlFlow;
804 ///
805 /// use arctic::ConcurrentMap;
806 /// use arctic::concurrent::map::Update;
807 ///
808 /// let map = ConcurrentMap::<u64, Box<u64>>::default();
809 /// let key = 5;
810 ///
811 /// // Key not present, closure never called, new value not allocated
812 /// match map.update_with(&key, None, |_, _| unreachable!()) {
813 /// Update::Absent { new } => assert!(new.is_none()),
814 /// Update::Success { .. } | Update::Break { .. } => unreachable!(),
815 /// }
816 ///
817 /// map.insert(key, Box::new(29)).expect("Key not present");
818 ///
819 /// // Key present, closure breaks, new value not allocated
820 /// match map.update_with(&key, None, |_, _| ControlFlow::Break(())) {
821 /// Update::Break { old, new } => {
822 /// assert_eq!(*old, 29);
823 /// assert!(new.is_none());
824 /// }
825 /// Update::Absent { .. } | Update::Success { .. } => unreachable!(),
826 /// }
827 ///
828 /// // Key present, closure continues, new value lazily allocated (and reused under contention)
829 /// match map.update_with(&key, None, |old, new| {
830 /// ControlFlow::Continue(
831 /// new.take()
832 /// // Reuse allocation under contention
833 /// .map(|mut new: Box<u64>| {
834 /// *new = *old + 1;
835 /// new
836 /// })
837 /// // Allocate new value
838 /// .unwrap_or_else(|| Box::new(*old + 1)))
839 /// }) {
840 /// Update::Success(updated) => {
841 /// assert_eq!(*updated.old(), 29);
842 /// assert_eq!(*updated.new(), 30);
843 /// }
844 /// Update::Absent { .. } | Update::Break { .. } => unreachable!(),
845 /// }
846 /// ```
847 pub fn update_with<'g, F>(
848 &'g self,
849 key: &K::Borrowed,
850 mut initial: Option<V>,
851 mut update: F,
852 ) -> Update<'g, K, V, S>
853 where
854 F: FnMut(&V::Borrowed, &mut Option<V>) -> ControlFlow<(), V>,
855 {
856 let reader = K::Read::from(key);
857 let mut guard = self.smr.guard(reader);
858
859 // NOTE: this is a macro so we get disjoint mutable borrows of `initial`
860 macro_rules! update {
861 () => {
862 |old: u64, new: Option<u64>| {
863 initial = new.map(|new| V::from_raw_unchecked(new));
864
865 match update(V::borrow_from_raw_unchecked(&old), &mut initial) {
866 ControlFlow::Continue(new) => ControlFlow::Continue(new.into_raw()),
867 ControlFlow::Break(()) => ControlFlow::Break(()),
868 }
869 }
870 };
871 }
872
873 let update = match if cfg!(feature = "opt-no-path") {
874 Err(initial.take().map(V::into_raw))
875 } else {
876 unsafe {
877 self.update_with_optimistic(
878 &mut guard,
879 reader,
880 initial.take().map(V::into_raw),
881 update!(),
882 )
883 }
884 } {
885 Ok(update) => update,
886 Err(initial) => unsafe {
887 self.update_with_pessimistic(&mut guard, reader, initial, update!())
888 },
889 };
890
891 match update {
892 UpdateRaw::Absent { new } => Update::Absent {
893 new: new.map(|new| unsafe { V::from_raw_unchecked(new) }),
894 },
895 UpdateRaw::Success { old, new } => {
896 Update::Success(unsafe { Updated::<K, V, S>::wrap(guard, old, new) })
897 }
898 UpdateRaw::Break { old } => Update::Break {
899 old: unsafe { Shared::<K, V, S>::wrap(guard, old) },
900 new: initial,
901 },
902 }
903 }
904
905 /// If there is a value associated with `key`, call `remove` to determine whether
906 /// to remove the value, recursively removing empty tree nodes.
907 ///
908 /// Returns a [`Remove`] enum.
909 ///
910 /// See also: [`ConcurrentMap::remove`],
911 /// [`ConcurrentMap::remove_non_recursive`],
912 /// [`ConcurrentMap::remove_non_recursive_with`].
913 ///
914 /// # Examples
915 ///
916 /// ```rust
917 /// use core::ops::ControlFlow;
918 ///
919 /// use arctic::ConcurrentMap;
920 /// use arctic::concurrent::map::Remove;
921 ///
922 /// let map = ConcurrentMap::<u128, u64>::default();
923 /// let key = 0xfeed;
924 ///
925 /// // Key not present, closure never called
926 /// match map.remove_with(&key, |_| unreachable!()) {
927 /// Remove::Absent => (),
928 /// Remove::Success { .. } | Remove::Break { .. } => unreachable!(),
929 /// }
930 ///
931 /// map.insert(key, 1).expect("Key not present");
932 ///
933 /// // Key present, closure breaks, value not removed
934 /// match map.remove_with(&key, |old| {
935 /// assert_eq!(*old, 1);
936 /// ControlFlow::Break(())
937 /// }) {
938 /// Remove::Break { old } => assert_eq!(*old, 1),
939 /// Remove::Absent | Remove::Success { .. } => unreachable!(),
940 /// }
941 ///
942 /// assert_eq!(map.get(&key).as_deref().copied(), Some(1));
943 ///
944 /// // Key present, closure continues, value removed
945 /// match map.remove_with(&key, |old| {
946 /// if *old > 0 {
947 /// ControlFlow::Continue(())
948 /// } else {
949 /// ControlFlow::Break(())
950 /// }
951 /// }) {
952 /// Remove::Success { old } => assert_eq!(*old, 1),
953 /// Remove::Absent | Remove::Break { .. } => unreachable!(),
954 /// }
955 ///
956 /// assert!(map.get(&key).is_none());
957 /// ```
958 pub fn remove_with<'g, F>(&'g self, key: &K::Borrowed, mut remove: F) -> Remove<'g, K, V, S>
959 where
960 F: FnMut(&V::Borrowed) -> ControlFlow<(), ()>,
961 {
962 let reader = K::Read::from(key);
963 let mut guard = self.smr.guard(reader);
964 let Ok(remove) = unsafe {
965 self.remove_with_raw::<true, path::Full<_>, _>(&mut guard, reader, |value| {
966 remove(V::borrow_from_raw_unchecked(&value))
967 })
968 };
969
970 match remove {
971 RemoveRaw::Absent => Remove::Absent,
972 RemoveRaw::Success { old } => Remove::Success {
973 old: unsafe { Owned::<K, V, S>::wrap(guard, old) },
974 },
975 RemoveRaw::Break { old } => Remove::Break {
976 old: unsafe { Shared::<K, V, S>::wrap(guard, old) },
977 },
978 }
979 }
980
981 /// If there is a value associated with `key`, call `remove` to determine whether
982 /// to remove the value, **without** recursively removing empty tree nodes.
983 ///
984 /// <div class="warning">
985 ///
986 /// See warning on [`Map::remove_non_recursive`].
987 ///
988 /// </div>
989 ///
990 /// Returns a [`Remove`] enum.
991 ///
992 /// See also: [`ConcurrentMap::remove`],
993 /// [`ConcurrentMap::remove_with`],
994 /// [`ConcurrentMap::remove_non_recursive`].
995 pub fn remove_non_recursive_with<F>(
996 &self,
997 key: &K::Borrowed,
998 mut remove: F,
999 ) -> Remove<'_, K, V, S>
1000 where
1001 F: FnMut(&V::Borrowed) -> ControlFlow<(), ()>,
1002 {
1003 let reader = K::Read::from(key);
1004 let mut guard = self.smr.guard(reader);
1005 let mut remove = |value: u64| remove(unsafe { V::borrow_from_raw_unchecked(&value) });
1006
1007 let remove = match if cfg!(feature = "opt-no-path") {
1008 Err(())
1009 } else {
1010 unsafe { self.remove_non_recursive_with_optimistic(&mut guard, reader, &mut remove) }
1011 } {
1012 Ok(remove) => remove,
1013 Err(()) => unsafe {
1014 self.remove_non_recursive_with_pessimistic(&mut guard, reader, &mut remove)
1015 },
1016 };
1017
1018 match remove {
1019 RemoveRaw::Absent => Remove::Absent,
1020 RemoveRaw::Success { old } => Remove::Success {
1021 old: unsafe { Owned::<K, V, S>::wrap(guard, old) },
1022 },
1023 RemoveRaw::Break { old } => Remove::Break {
1024 old: unsafe { Shared::<K, V, S>::wrap(guard, old) },
1025 },
1026 }
1027 }
1028}
1029
1030/// Outcome of a call to [`ConcurrentMap::upsert_with`].
1031pub enum Upsert<'g, K, V, S>
1032where
1033 K: Key,
1034 V: Value + 'g,
1035 S: Smr<K, V> + 'g,
1036{
1037 /// Value was successfully upserted.
1038 Success(Upserted<'g, K, V, S>),
1039 /// Closure returned [`core::ops::ControlFlow::Break`].
1040 Break {
1041 /// Latest value observed by closure.
1042 old: Option<Shared<'g, K, V, S>>,
1043 /// Latest value passed as argument or returned from closure.
1044 new: Option<V>,
1045 },
1046}
1047
1048/// Type-erased version of [`Upsert`].
1049enum UpsertRaw {
1050 Success { old: Option<u64>, new: u64 },
1051 Break { old: Option<u64> },
1052}
1053
1054/// Outcome of a call to [`ConcurrentMap::update_with`].
1055pub enum Update<'g, K, V, S>
1056where
1057 K: Key,
1058 V: Value + 'g,
1059 S: Smr<K, V> + 'g,
1060{
1061 /// Key was not present.
1062 Absent {
1063 /// Latest value passed as argument or returned from closure.
1064 new: Option<V>,
1065 },
1066 /// Value was successfully updated.
1067 Success(Updated<'g, K, V, S>),
1068 /// Closure returned [`core::ops::ControlFlow::Break`].
1069 Break {
1070 /// Latest value observed by closure.
1071 old: Shared<'g, K, V, S>,
1072 /// Latest value passed as argument or returned from closure.
1073 new: Option<V>,
1074 },
1075}
1076
1077/// Type-erased version of [`Update`].
1078enum UpdateRaw {
1079 Absent { new: Option<u64> },
1080 Success { old: u64, new: u64 },
1081 Break { old: u64 },
1082}
1083
1084/// Outcome of a call to [`ConcurrentMap::remove_with`].
1085pub enum Remove<'g, K, V, S>
1086where
1087 K: Key,
1088 V: Value + 'g,
1089 S: Smr<K, V> + 'g,
1090{
1091 /// Key was not present.
1092 Absent,
1093 /// Value was successfully removed.
1094 Success {
1095 /// Value that was removed.
1096 old: Owned<'g, K, V, S>,
1097 },
1098 /// Closure returned [`core::ops::ControlFlow::Break`].
1099 Break {
1100 /// Latest value observed by closure.
1101 old: Shared<'g, K, V, S>,
1102 },
1103}
1104
1105/// Type-erased version of [`Remove`].
1106enum RemoveRaw {
1107 Absent,
1108 Success { old: u64 },
1109 Break { old: u64 },
1110}
1111
1112/// # Private implementations
1113///
1114/// These methods erase value types and accept arbitrary key readers and SMR guards.
1115/// This reduces monomorphization and allows a future `concurrent::Set` implementation
1116/// to reuse this logic, at the cost of reducing type safety.
1117///
1118/// # Safety
1119///
1120/// Caller must guarantee:
1121/// - `_guard` protects nodes and values under `reader` for its lifetime.
1122/// - When inserting or upserting, `reader` preserves the prefix property.
1123/// - `initial` and every value returned from a closure was created via `V::into_raw`.
1124impl<K, V, S> Map<K, V, S>
1125where
1126 K: Key,
1127 V: Value,
1128 S: Smr<K, V>,
1129{
1130 #[inline]
1131 unsafe fn get_raw<'g>(&'g self, _guard: &mut S::Guard<'g>, reader: K::Read<'_>) -> Option<u64> {
1132 unsafe {
1133 let mut cursor = self.seq.raw.cursor::<path::Discard<_>>(reader);
1134 let walk = cursor.edge().load_packed(Ordering::Relaxed);
1135 cursor
1136 .traverse_value(walk)
1137 .map(|cursor::Value { value, edge: _ }| {
1138 if V::INDIRECT {
1139 // Synchronizes with release compare_exchanges in
1140 // `upsert_with_raw` and `update_with_raw`.
1141 crate::sync::atomic::fence(Ordering::Acquire);
1142 }
1143
1144 value
1145 })
1146 }
1147 }
1148
1149 #[inline]
1150 unsafe fn upsert_with_optimistic<'g, 'k, F>(
1151 &'g self,
1152 guard: &mut S::Guard<'g>,
1153 reader: K::Read<'k>,
1154 initial: Option<u64>,
1155 upsert: F,
1156 ) -> Result<UpsertRaw, Option<u64>>
1157 where
1158 F: FnMut(Option<u64>, Option<u64>) -> ControlFlow<(), u64>,
1159 {
1160 unsafe { self.upsert_with_raw::<path::Point<_>, _>(guard, reader, initial, upsert) }
1161 }
1162
1163 #[cold]
1164 unsafe fn upsert_with_pessimistic<'g, 'k, F>(
1165 &'g self,
1166 guard: &mut S::Guard<'g>,
1167 reader: K::Read<'k>,
1168 initial: Option<u64>,
1169 upsert: F,
1170 ) -> UpsertRaw
1171 where
1172 F: FnMut(Option<u64>, Option<u64>) -> ControlFlow<(), u64>,
1173 {
1174 stat::increment(stat::Counter::InsertPessimistic);
1175 unsafe { self.upsert_with_raw::<path::Full<_>, _>(guard, reader, initial, upsert) }
1176 .expect("path::Retain::PopError is Infallible")
1177 }
1178
1179 #[inline]
1180 unsafe fn upsert_with_raw<'g, 'k, P, F>(
1181 &'g self,
1182 guard: &mut S::Guard<'g>,
1183 reader: K::Read<'k>,
1184 mut initial: Option<u64>,
1185 mut upsert: F,
1186 ) -> Result<UpsertRaw, Option<u64>>
1187 where
1188 P: Path<K::Read<'k>>,
1189 F: FnMut(Option<u64>, Option<u64>) -> ControlFlow<(), u64>,
1190 {
1191 let mut cursor = unsafe { self.seq.raw.cursor::<P>(reader) };
1192 let mut walk = cursor.edge().load_packed(Ordering::Relaxed);
1193
1194 loop {
1195 match unsafe { cursor.traverse_insert(walk) } {
1196 cursor::Insert::Value {
1197 value: old_value,
1198 edge: old_edge,
1199 } => {
1200 if V::INDIRECT {
1201 // Synchronizes with release compare_exchanges in
1202 // `upsert_with_raw` and `update_with_raw`.
1203 crate::sync::atomic::fence(Ordering::Acquire);
1204 }
1205
1206 let new_value = match upsert(old_value, initial) {
1207 ControlFlow::Continue(new_value) => new_value,
1208 ControlFlow::Break(()) => {
1209 return Ok(UpsertRaw::Break { old: old_value });
1210 }
1211 };
1212
1213 if old_edge.meta().is_frozen() {
1214 // Restore value and fall through to freeze
1215 initial = Some(new_value);
1216 } else {
1217 let (new_edge, _) = cursor.create_path(old_edge, new_value);
1218 match cursor.edge().compare_exchange_packed(
1219 old_edge,
1220 new_edge,
1221 // Technically, if `new_edge` is an inline value, this could be relaxed.
1222 // Since it's likely to be a node, conservatively default to release.
1223 Ordering::Release,
1224 Ordering::Relaxed,
1225 ) {
1226 Ok(_) => {
1227 return Ok(UpsertRaw::Success {
1228 old: old_value,
1229 new: new_value,
1230 });
1231 }
1232 Err(conflict) => {
1233 if let Some(node) = new_edge.as_node() {
1234 unsafe {
1235 stat::increment(stat::Counter::FreeConflict);
1236 node.deallocate_recursive::<K::Edge>();
1237 }
1238 }
1239
1240 initial = Some(new_value);
1241 walk = conflict;
1242 continue;
1243 }
1244 }
1245 }
1246 }
1247 cursor::Insert::Replace {
1248 node: old_node,
1249 edge: old_edge,
1250 } if !old_edge.meta().is_frozen() => {
1251 let (smo, new_edge) = unsafe {
1252 old_node.freeze::<K::Edge>();
1253 old_node.replace(old_edge.meta())
1254 };
1255 match cursor.edge().compare_exchange_packed(
1256 old_edge,
1257 new_edge,
1258 Ordering::Release,
1259 Ordering::Relaxed,
1260 ) {
1261 Ok(_) => {
1262 unsafe { guard.retire_node(cursor.len().bits(), old_node.into_raw()) };
1263 walk = new_edge;
1264 }
1265 Err(conflict) => {
1266 // Does not go through SMR because `new` is still thread-local
1267 if smo.is_allocate() {
1268 let node = new_edge.as_node().expect("Allocating SMO creates node");
1269 unsafe {
1270 stat::increment(stat::Counter::FreeConflict);
1271 node.deallocate();
1272 }
1273 }
1274 walk = conflict;
1275 }
1276 }
1277
1278 continue;
1279 }
1280
1281 // Fall through to freeze
1282 cursor::Insert::Replace { .. } => (),
1283 }
1284
1285 walk = self.freeze(guard, &mut cursor).map_err(|_| initial)?;
1286 }
1287 }
1288
1289 #[inline]
1290 unsafe fn update_with_optimistic<'g, F>(
1291 &'g self,
1292 guard: &mut S::Guard<'g>,
1293 reader: K::Read<'_>,
1294 initial: Option<u64>,
1295 update: F,
1296 ) -> Result<UpdateRaw, Option<u64>>
1297 where
1298 F: FnMut(u64, Option<u64>) -> ControlFlow<(), u64>,
1299 {
1300 unsafe { self.update_with_raw::<path::Point<_>, _>(guard, reader, initial, update) }
1301 }
1302
1303 #[cold]
1304 unsafe fn update_with_pessimistic<'g, F>(
1305 &'g self,
1306 guard: &mut S::Guard<'g>,
1307 reader: K::Read<'_>,
1308 initial: Option<u64>,
1309 update: F,
1310 ) -> UpdateRaw
1311 where
1312 F: FnMut(u64, Option<u64>) -> ControlFlow<(), u64>,
1313 {
1314 stat::increment(stat::Counter::UpdatePessimistic);
1315 unsafe { self.update_with_raw::<path::Full<_>, _>(guard, reader, initial, update) }
1316 .expect("path::Retain::PopError is Infallible")
1317 }
1318
1319 #[inline]
1320 unsafe fn update_with_raw<'g, 'k, P, F>(
1321 &'g self,
1322 guard: &mut S::Guard<'g>,
1323 reader: K::Read<'k>,
1324 mut initial: Option<u64>,
1325 mut update: F,
1326 ) -> Result<UpdateRaw, Option<u64>>
1327 where
1328 P: Path<K::Read<'k>>,
1329 F: FnMut(u64, Option<u64>) -> ControlFlow<(), u64>,
1330 {
1331 let mut cursor = unsafe { self.seq.raw.cursor::<P>(reader) };
1332 let mut walk = cursor.edge().load_packed(Ordering::Relaxed);
1333
1334 loop {
1335 let cursor::Value {
1336 value: old_value,
1337 edge: old_edge,
1338 } = match unsafe { cursor.traverse_value(walk) } {
1339 None => return Ok(UpdateRaw::Absent { new: initial }),
1340 Some(update) if !update.edge.meta().is_frozen() => update,
1341 Some(_) => {
1342 walk = self.freeze(guard, &mut cursor).map_err(|_| initial)?;
1343 continue;
1344 }
1345 };
1346
1347 if V::INDIRECT {
1348 // Synchronizes with release compare_exchanges in
1349 // `upsert_with_raw` and `update_with_raw`.
1350 crate::sync::atomic::fence(Ordering::Acquire);
1351 }
1352
1353 let new_value = match update(old_value, initial) {
1354 ControlFlow::Continue(new_value) => new_value,
1355 ControlFlow::Break(()) => {
1356 return Ok(UpdateRaw::Break { old: old_value });
1357 }
1358 };
1359
1360 match cursor.edge().compare_exchange_packed(
1361 old_edge,
1362 Edge::new_value(old_edge.meta(), new_value),
1363 if V::INDIRECT {
1364 Ordering::Release
1365 } else {
1366 Ordering::Relaxed
1367 },
1368 Ordering::Relaxed,
1369 ) {
1370 Ok(_) => {
1371 return Ok(UpdateRaw::Success {
1372 old: old_value,
1373 new: new_value,
1374 });
1375 }
1376 Err(conflict) => {
1377 initial = Some(new_value);
1378 walk = conflict;
1379 }
1380 }
1381 }
1382 }
1383
1384 #[inline]
1385 unsafe fn remove_non_recursive_with_optimistic<'g, F>(
1386 &'g self,
1387 guard: &mut S::Guard<'g>,
1388 reader: K::Read<'_>,
1389 remove: F,
1390 ) -> Result<RemoveRaw, ()>
1391 where
1392 F: FnMut(u64) -> ControlFlow<(), ()>,
1393 {
1394 unsafe { self.remove_with_raw::<false, path::Point<_>, _>(guard, reader, remove) }
1395 }
1396
1397 #[cold]
1398 unsafe fn remove_non_recursive_with_pessimistic<'g, F>(
1399 &'g self,
1400 guard: &mut S::Guard<'g>,
1401 reader: K::Read<'_>,
1402 remove: F,
1403 ) -> RemoveRaw
1404 where
1405 F: FnMut(u64) -> ControlFlow<(), ()>,
1406 {
1407 let Ok(remove) =
1408 unsafe { self.remove_with_raw::<false, path::Full<_>, _>(guard, reader, remove) };
1409 remove
1410 }
1411
1412 #[inline]
1413 unsafe fn remove_with_raw<'g, 'k, const RECURSIVE: bool, P, F>(
1414 &'g self,
1415 guard: &mut S::Guard<'g>,
1416 reader: K::Read<'k>,
1417 mut remove: F,
1418 ) -> Result<RemoveRaw, P::PopError>
1419 where
1420 P: Path<K::Read<'k>>,
1421 F: FnMut(u64) -> ControlFlow<(), ()>,
1422 {
1423 let mut cursor = unsafe { self.seq.raw.cursor::<P>(reader) };
1424 let mut walk = cursor.edge().load_packed(Ordering::Relaxed);
1425
1426 let (value, edge) = loop {
1427 let cursor::Value { value, edge } = match unsafe { cursor.traverse_value(walk) } {
1428 None => return Ok(RemoveRaw::Absent),
1429 Some(update) if !update.edge.meta().is_frozen() => update,
1430 Some(_) => {
1431 walk = self.freeze(guard, &mut cursor)?;
1432 continue;
1433 }
1434 };
1435
1436 if V::INDIRECT {
1437 // Synchronizes with release compare_exchanges in
1438 // `upsert_with_raw` and `update_with_raw`.
1439 crate::sync::atomic::fence(Ordering::Acquire);
1440 }
1441
1442 match remove(value) {
1443 ControlFlow::Continue(()) => (),
1444 ControlFlow::Break(()) => {
1445 return Ok(RemoveRaw::Break { old: value });
1446 }
1447 }
1448
1449 match cursor.edge().compare_exchange_packed(
1450 edge,
1451 Edge::NULL,
1452 // Relaxed because publishing `Edge::NULL`
1453 Ordering::Relaxed,
1454 Ordering::Relaxed,
1455 ) {
1456 Ok(_) => break (value, edge),
1457 Err(conflict) => walk = conflict,
1458 }
1459 };
1460
1461 if RECURSIVE {
1462 let mut trim = edge.meta().len().into();
1463 let mut pop = 0;
1464
1465 'pop: while let Some((mut old_len, old_node)) =
1466 cursor.pop().expect("Recursive remove requires path")
1467 {
1468 if unsafe { old_node.len::<K::Edge>() } > 1 {
1469 break 'pop;
1470 }
1471
1472 cursor.trim(K::Len::BYTE + trim);
1473 pop += 1;
1474
1475 let mut old_edge = cursor.edge().load_packed(Ordering::Relaxed);
1476
1477 'freeze: loop {
1478 let addr = cursor.edge();
1479
1480 match unsafe { cursor.freeze(old_len, old_node, old_edge) }
1481 .expect("Recursive remove requires path")
1482 {
1483 // Fall through to `traverse_node`
1484 cursor::Freeze::Traverse { edge } => {
1485 old_edge = edge;
1486 }
1487 cursor::Freeze::Success {
1488 old_node: node,
1489 new_edge,
1490 } => {
1491 if let Some(node) = node {
1492 unsafe { guard.retire_node(cursor.len().bits(), node.into_raw()) };
1493 }
1494
1495 // `freeze` did not pop, so we (or someone else) replaced `old_node`
1496 if core::ptr::eq(cursor.edge(), addr) {
1497 trim = old_len.into();
1498 continue 'pop;
1499 }
1500
1501 old_edge = new_edge;
1502 }
1503 }
1504
1505 // Traverse down to `old_node`
1506 match cursor.traverse_node(old_edge) {
1507 Ok(edge) => {
1508 old_len = edge.meta().len();
1509 old_edge = edge;
1510 continue 'freeze;
1511 }
1512 Err(len) => {
1513 // If not found, pop to the closest parent node
1514 trim = len;
1515 continue 'pop;
1516 }
1517 }
1518 }
1519 }
1520
1521 stat::record(stat::Record::RemovePop, pop);
1522 }
1523
1524 Ok(RemoveRaw::Success { old: value })
1525 }
1526
1527 fn freeze<'g, 'k, P>(
1528 &'g self,
1529 guard: &mut S::Guard<'g>,
1530 cursor: &mut Cursor<K::Read<'k>, P>,
1531 ) -> Result<ribbit::Packed<Edge<K::Edge>>, P::PopError>
1532 where
1533 P: Path<K::Read<'k>>,
1534 {
1535 let (old_len, old_node) = cursor.pop()?.expect("Root edge cannot be frozen");
1536
1537 match unsafe {
1538 cursor.freeze(
1539 old_len,
1540 old_node,
1541 // Need to load here since we just popped
1542 cursor.edge().load_packed(Ordering::Relaxed),
1543 )
1544 }? {
1545 cursor::Freeze::Traverse { edge }
1546 | cursor::Freeze::Success {
1547 old_node: None,
1548 new_edge: edge,
1549 } => Ok(edge),
1550 cursor::Freeze::Success {
1551 old_node: Some(node),
1552 new_edge,
1553 } => {
1554 unsafe { guard.retire_node(cursor.len().bits(), node.into_raw()) };
1555 Ok(new_edge)
1556 }
1557 }
1558 }
1559}
1560
1561impl<K, V, S> From<sequential::Map<K, V>> for Map<K, V, S>
1562where
1563 K: Key,
1564 V: Value,
1565 S: Default,
1566{
1567 #[inline]
1568 fn from(seq: sequential::Map<K, V>) -> Self {
1569 Self {
1570 smr: S::default(),
1571 seq,
1572 }
1573 }
1574}
1575
1576impl<K, V, S> From<Map<K, V, S>> for sequential::Map<K, V>
1577where
1578 K: Key,
1579 V: Value,
1580{
1581 #[inline]
1582 fn from(map: Map<K, V, S>) -> sequential::Map<K, V> {
1583 map.seq
1584 }
1585}
1586
1587#[cfg(test)]
1588mod tests {
1589 use core::convert::Infallible;
1590 use core::ops::ControlFlow;
1591
1592 use crate::Order;
1593 use crate::concurrent::Map;
1594 use crate::key::BoxedSlice;
1595 use crate::key::BoxedStr;
1596 use crate::key::NonNull;
1597 use crate::key::Slice;
1598 use crate::key::Str;
1599 use crate::key::Terminated;
1600 use crate::raw::key::Read as _;
1601
1602 #[test]
1603 fn smoke() {
1604 let map = Map::<BoxedStr<NonNull>, _>::default();
1605 map.upsert(unsafe { Slice::new_unchecked("abcd") }, 1u64);
1606 assert_eq!(
1607 map.get(unsafe { Slice::new_unchecked("abcd") })
1608 .as_deref()
1609 .copied(),
1610 Some(1)
1611 );
1612 }
1613
1614 #[test]
1615 fn smoke_u64_key() {
1616 let map = Map::<[u8; 8], _>::default();
1617 let key = 0xdeadbeefu64.to_be_bytes();
1618 map.upsert(&key, 1u64);
1619 assert_eq!(map.get(&key).as_deref().copied(), Some(1));
1620 }
1621
1622 #[test]
1623 fn smoke_value_ref() {
1624 let values = [0, 1, 2, 3, 4, 5];
1625 let map = Map::<u64, &u64>::default();
1626
1627 for (key, value) in values.iter().enumerate() {
1628 map.upsert(key as u64, value);
1629 }
1630
1631 #[expect(clippy::needless_range_loop)]
1632 for key in 0..values.len() {
1633 let value = map.get(&(key as u64)).as_deref().copied().unwrap();
1634 assert!(core::ptr::eq(value, &values[key]));
1635 }
1636 }
1637
1638 #[test]
1639 fn smoke_value_box() {
1640 let values = [0, 1, 2, 3, 4, 5];
1641 let map = Map::<u64, Box<u64>>::default();
1642
1643 for (key, value) in values.iter().enumerate() {
1644 map.upsert(key as u64, Box::new(*value));
1645 }
1646
1647 std::thread::scope(|scope| {
1648 for _ in 0..8 {
1649 scope.spawn(|| {
1650 for key in (0..values.len()).cycle().take(100_000) {
1651 let value = map.get(&(key as u64)).as_deref().copied().unwrap();
1652 assert_eq!(key, value as usize);
1653 }
1654 });
1655 }
1656 });
1657
1658 // TODO: multiple hazards?
1659 // let a = map.get(3);
1660 // let b = map.get(5);
1661 // assert_ne!(a.as_deref(), b.as_deref());
1662
1663 for key in 0..values.len() {
1664 let value = map.get(&(key as u64)).as_deref().copied().unwrap();
1665 assert_eq!(key, value as usize);
1666 }
1667 }
1668
1669 #[test]
1670 fn scan_value() {
1671 let map = Map::<u64, _>::default();
1672 let key = 1u64;
1673 map.upsert(key, 2u64);
1674 assert_eq!(
1675 map.range(1u64..=1u64)
1676 .entries(Order::Ascend)
1677 .collect::<Vec<_>>(),
1678 vec![(1, 2)]
1679 );
1680 }
1681
1682 #[test]
1683 fn scan_node3() {
1684 insert_all(0u64..3);
1685 }
1686
1687 #[test]
1688 fn scan_node256() {
1689 insert_all(0u64..256);
1690 }
1691
1692 #[test]
1693 fn scan_gap() {
1694 let map = insert_all((0u64..512).step_by(2));
1695 assert_eq!(
1696 map.range(256u64..=511u64)
1697 .entries(Order::Ascend)
1698 .collect::<Vec<_>>(),
1699 (256..512)
1700 .step_by(2)
1701 .map(|key| (key, key / 2))
1702 .collect::<Vec<_>>()
1703 );
1704 }
1705
1706 #[test]
1707 fn node3_overwrite() {
1708 let mut map = Map::<u64, _>::default();
1709
1710 for value in [1u64, 2, 3] {
1711 map.upsert(1, value);
1712 assert_eq!(map.get(&1).as_deref().copied(), Some(value));
1713 }
1714
1715 assert_eq!(map.as_sequential().all().entries(Order::Ascend).count(), 1);
1716
1717 map.as_sequential()
1718 .all()
1719 .entries(Order::Ascend)
1720 .try_fold((), |(), (key, value)| {
1721 assert_eq!(key, 1);
1722 assert_eq!(*value, 3);
1723 ControlFlow::<Infallible>::Continue(())
1724 });
1725 }
1726
1727 #[test]
1728 fn node3_reverse() {
1729 insert_all((0u16..3).rev());
1730 }
1731
1732 #[test]
1733 fn node3_full() {
1734 insert_all(0u16..3);
1735 }
1736
1737 #[test]
1738 fn node3_expand() {
1739 insert_all(0u16..4);
1740 }
1741
1742 #[test]
1743 fn node15_full() {
1744 insert_all(0u16..15);
1745 }
1746
1747 #[test]
1748 fn node15_expand() {
1749 insert_all(0u16..16);
1750 }
1751
1752 #[test]
1753 fn node47_full() {
1754 insert_all(0u16..47);
1755 }
1756
1757 #[test]
1758 fn node47_expand() {
1759 insert_all(0u16..61);
1760 }
1761
1762 #[test]
1763 fn node256_full() {
1764 insert_all(0u16..=255);
1765 }
1766
1767 #[test]
1768 fn range_reverse() {
1769 let map = Map::<u64, _>::default();
1770
1771 for key in [5, 1, 4, 3, 2] {
1772 map.upsert(key, key);
1773 assert_eq!(map.get(&key).as_deref().copied(), Some(key));
1774 }
1775
1776 assert_eq!(
1777 map.range(2..=4).entries(Order::Descend).collect::<Vec<_>>(),
1778 vec![(4, 4), (3, 3), (2, 2)]
1779 );
1780 }
1781
1782 #[test]
1783 fn split_edges() {
1784 let mut key = (1..100).collect::<Vec<_>>();
1785 insert_all(core::iter::from_fn(|| {
1786 if key.is_empty() {
1787 None
1788 } else {
1789 let mut next = key.clone();
1790 next.push(0);
1791 key.pop();
1792 let next = next.into_boxed_slice();
1793 Some(BoxedSlice::<Terminated<0>>::new(next).unwrap())
1794 }
1795 }));
1796 }
1797
1798 #[test]
1799 fn one_long_key() {
1800 insert_all([BoxedStr::<NonNull>::new("a".repeat(1000)).unwrap()]);
1801 }
1802
1803 #[test]
1804 fn short_key() {
1805 insert_all([BoxedStr::<NonNull>::new("\n".to_string()).unwrap()]);
1806 }
1807
1808 #[test]
1809 fn two_long_keys() {
1810 insert_all([
1811 BoxedStr::<NonNull>::new("a".repeat(1000)).unwrap(),
1812 BoxedStr::<NonNull>::new("b".repeat(1000)).unwrap(),
1813 ]);
1814 }
1815
1816 #[test]
1817 fn smoke_key_slice() {
1818 let keys = ["ad", "abc"];
1819 let map = crate::concurrent::Map::<&Str<NonNull>, u64>::new();
1820 map.insert(Str::new(keys[0]).unwrap(), 0)
1821 .unwrap_or_else(|(_, _)| panic!());
1822 map.insert(Str::new(keys[1]).unwrap(), 1)
1823 .unwrap_or_else(|(_, _)| panic!());
1824
1825 let temp = "adabc";
1826 assert_eq!(
1827 map.get(Str::new(&temp[..2]).unwrap()).as_deref().copied(),
1828 Some(0)
1829 );
1830 assert_eq!(
1831 map.get(Str::new(&temp[2..]).unwrap()).as_deref().copied(),
1832 Some(1)
1833 );
1834 }
1835
1836 #[test]
1837 fn key_slice_long_prefix() {
1838 let keys = (0..10)
1839 .map(|i| "a".repeat(100) + &i.to_string())
1840 .collect::<Vec<_>>();
1841 let map = crate::concurrent::Map::<&Slice<NonNull>, u64>::new();
1842 for (i, key) in keys.iter().enumerate() {
1843 map.insert(Slice::new(key.as_bytes()).unwrap(), i as u64)
1844 .unwrap();
1845 }
1846 for (i, key) in keys.iter().enumerate() {
1847 assert_eq!(
1848 map.get(Slice::new(key.as_bytes()).unwrap())
1849 .as_deref()
1850 .copied(),
1851 Some(i as u64)
1852 );
1853 }
1854 }
1855
1856 fn insert_all<I, K>(iter: I) -> Map<K, u64>
1857 where
1858 I: IntoIterator<Item = K>,
1859 K: crate::Key + Clone + Ord + core::fmt::Debug,
1860 {
1861 let mut keys = iter
1862 .into_iter()
1863 .enumerate()
1864 .map(|(index, key)| (key, index as u64))
1865 .collect::<Vec<_>>();
1866
1867 let mut map = Map::default();
1868
1869 for (key, value) in &keys {
1870 map.upsert(key.as_insert(), *value);
1871 assert_eq!(map.get(key.borrow()).as_deref().copied(), Some(*value));
1872 }
1873
1874 for (key, value) in &keys {
1875 assert_eq!(map.get(key.borrow()).as_deref().copied(), Some(*value));
1876 }
1877
1878 let mut iter = map.as_sequential().all().entries(Order::Ascend);
1879 let mut count = 0;
1880 while iter.lend().is_some() {
1881 count += 1;
1882 }
1883 drop(iter);
1884
1885 assert_eq!(count, keys.len());
1886
1887 keys.sort_by(|(l, _), (r, _)| l.cmp(r));
1888
1889 // Sequential iteration
1890 map.as_sequential()
1891 .all()
1892 .entries(Order::Ascend)
1893 .zip(&keys)
1894 .for_each(|((lk, lv), (rk, rv))| {
1895 assert_eq!(lk, *rk);
1896 assert_eq!(*lv, *rv);
1897 });
1898
1899 let Some(((first, _), (last, _))) = keys.first().zip(keys.last()) else {
1900 return map;
1901 };
1902
1903 // Concurrent prefix scan, non-linearizable
1904 map.prefix(K::Read::from(first.borrow()).common_prefix(K::Read::from(last.borrow())))
1905 .entries(Order::Descend)
1906 .zip(keys.iter().rev())
1907 .for_each(|((lk, lv), (rk, rv))| {
1908 assert_eq!(lk, *rk);
1909 assert_eq!(lv, *rv);
1910 });
1911
1912 // Concurrent range scan, non-linearizable
1913 let mut i = 0;
1914 map.range(first.borrow()..=last.borrow())
1915 .entries(Order::Descend)
1916 .zip(keys.iter().rev())
1917 .for_each(|((lk, lv), (rk, rv))| {
1918 i += 1;
1919 assert_eq!(lk, *rk);
1920 assert_eq!(lv, *rv);
1921 });
1922 assert_eq!(i, keys.len());
1923
1924 map
1925 }
1926}