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