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;
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 unsafe {
763 self.upsert_with_optimistic(
764 &mut guard,
765 reader,
766 initial.take().map(V::into_raw),
767 upsert!(),
768 )
769 }
770 } else {
771 Err(())
772 } {
773 Ok(upsert) => upsert,
774 Err(()) => unsafe {
775 self.upsert_with_pessimistic(
776 &mut guard,
777 reader,
778 initial.take().map(V::into_raw),
779 upsert!(),
780 )
781 },
782 };
783
784 match upsert {
785 UpsertRaw::Success { old, new } => {
786 Upsert::Success(unsafe { Upserted::<K, V, S>::wrap(guard, old, new) })
787 }
788 UpsertRaw::Break { old } => Upsert::Break {
789 old: old.map(|old| unsafe { Shared::<K, V, S>::wrap(guard, old) }),
790 new: initial,
791 },
792 }
793 }
794
795 /// If there is a value associated with `key`, call the provided `update` closure
796 /// to break or compute a new value.
797 ///
798 /// The closure may be called multiple times under contention,
799 /// and takes an immutable reference to the current value, as well as `initial`
800 /// (on the first call) or `Some(prev_value)` (on subsequent calls); use [`Option::take`]
801 /// to move out of the option.
802 ///
803 /// Returns an [`Update`] enum.
804 ///
805 /// # Examples
806 ///
807 /// ```rust
808 /// use core::ops::ControlFlow;
809 ///
810 /// use arctic::ConcurrentMap;
811 /// use arctic::concurrent::map::Update;
812 ///
813 /// let map = ConcurrentMap::<u64, Box<u64>>::default();
814 /// let key = 5;
815 ///
816 /// // Key not present, closure never called, new value not allocated
817 /// match map.update_with(&key, None, |_, _| unreachable!()) {
818 /// Update::Absent { new } => assert!(new.is_none()),
819 /// Update::Success { .. } | Update::Break { .. } => unreachable!(),
820 /// }
821 ///
822 /// map.insert(key, Box::new(29)).expect("Key not present");
823 ///
824 /// // Key present, closure breaks, new value not allocated
825 /// match map.update_with(&key, None, |_, _| ControlFlow::Break(())) {
826 /// Update::Break { old, new } => {
827 /// assert_eq!(*old, 29);
828 /// assert!(new.is_none());
829 /// }
830 /// Update::Absent { .. } | Update::Success { .. } => unreachable!(),
831 /// }
832 ///
833 /// // Key present, closure continues, new value lazily allocated (and reused under contention)
834 /// match map.update_with(&key, None, |old, new| {
835 /// ControlFlow::Continue(
836 /// new.take()
837 /// // Reuse allocation under contention
838 /// .map(|mut new: Box<u64>| {
839 /// *new = *old + 1;
840 /// new
841 /// })
842 /// // Allocate new value
843 /// .unwrap_or_else(|| Box::new(*old + 1)))
844 /// }) {
845 /// Update::Success(updated) => {
846 /// assert_eq!(*updated.old(), 29);
847 /// assert_eq!(*updated.new(), 30);
848 /// }
849 /// Update::Absent { .. } | Update::Break { .. } => unreachable!(),
850 /// }
851 /// ```
852 pub fn update_with<'g, F>(
853 &'g self,
854 key: &K::Borrowed,
855 mut initial: Option<V>,
856 mut update: F,
857 ) -> Update<'g, K, V, S>
858 where
859 F: FnMut(&V::Borrowed, &mut Option<V>) -> ControlFlow<(), V>,
860 {
861 let reader = K::Read::from(key);
862 let mut guard = self.smr.guard(reader);
863
864 // NOTE: this is a macro so we get disjoint mutable borrows of `initial`
865 macro_rules! update {
866 () => {
867 |old: u64, new: Option<u64>| {
868 initial = new.map(|new| V::from_raw_unchecked(new));
869
870 match update(V::borrow_from_raw_unchecked(&old), &mut initial) {
871 ControlFlow::Continue(new) => ControlFlow::Continue(new.into_raw()),
872 ControlFlow::Break(()) => ControlFlow::Break(()),
873 }
874 }
875 };
876 }
877
878 let update = match if cfg!(feature = "opt-no-path") {
879 unsafe {
880 self.update_with_optimistic(
881 &mut guard,
882 reader,
883 initial.take().map(V::into_raw),
884 update!(),
885 )
886 }
887 } else {
888 Err(())
889 } {
890 Ok(update) => update,
891 Err(()) => unsafe {
892 self.update_with_pessimistic(
893 &mut guard,
894 reader,
895 initial.take().map(V::into_raw),
896 update!(),
897 )
898 },
899 };
900
901 match update {
902 UpdateRaw::Absent { new } => Update::Absent {
903 new: new.map(|new| unsafe { V::from_raw_unchecked(new) }),
904 },
905 UpdateRaw::Success { old, new } => {
906 Update::Success(unsafe { Updated::<K, V, S>::wrap(guard, old, new) })
907 }
908 UpdateRaw::Break { old } => Update::Break {
909 old: unsafe { Shared::<K, V, S>::wrap(guard, old) },
910 new: initial,
911 },
912 }
913 }
914
915 /// If there is a value associated with `key`, call `remove` to determine whether
916 /// to remove the value, recursively removing empty tree nodes.
917 ///
918 /// Returns a [`Remove`] enum.
919 ///
920 /// See also: [`ConcurrentMap::remove`],
921 /// [`ConcurrentMap::remove_non_recursive`],
922 /// [`ConcurrentMap::remove_non_recursive_with`].
923 ///
924 /// # Examples
925 ///
926 /// ```rust
927 /// use core::ops::ControlFlow;
928 ///
929 /// use arctic::ConcurrentMap;
930 /// use arctic::concurrent::map::Remove;
931 ///
932 /// let map = ConcurrentMap::<u128, u64>::default();
933 /// let key = 0xfeed;
934 ///
935 /// // Key not present, closure never called
936 /// match map.remove_with(&key, |_| unreachable!()) {
937 /// Remove::Absent => (),
938 /// Remove::Success { .. } | Remove::Break { .. } => unreachable!(),
939 /// }
940 ///
941 /// map.insert(key, 1).expect("Key not present");
942 ///
943 /// // Key present, closure breaks, value not removed
944 /// match map.remove_with(&key, |old| {
945 /// assert_eq!(*old, 1);
946 /// ControlFlow::Break(())
947 /// }) {
948 /// Remove::Break { old } => assert_eq!(*old, 1),
949 /// Remove::Absent | Remove::Success { .. } => unreachable!(),
950 /// }
951 ///
952 /// assert_eq!(map.get(&key).as_deref().copied(), Some(1));
953 ///
954 /// // Key present, closure continues, value removed
955 /// match map.remove_with(&key, |old| {
956 /// if *old > 0 {
957 /// ControlFlow::Continue(())
958 /// } else {
959 /// ControlFlow::Break(())
960 /// }
961 /// }) {
962 /// Remove::Success { old } => assert_eq!(*old, 1),
963 /// Remove::Absent | Remove::Break { .. } => unreachable!(),
964 /// }
965 ///
966 /// assert!(map.get(&key).is_none());
967 /// ```
968 pub fn remove_with<'g, F>(&'g self, key: &K::Borrowed, mut remove: F) -> Remove<'g, K, V, S>
969 where
970 F: FnMut(&V::Borrowed) -> ControlFlow<(), ()>,
971 {
972 let reader = K::Read::from(key);
973 let mut guard = self.smr.guard(reader);
974 let Ok(remove) = unsafe {
975 self.remove_with_raw::<true, path::Retain<_>, _>(&mut guard, reader, |value| {
976 remove(V::borrow_from_raw_unchecked(&value))
977 })
978 };
979
980 match remove {
981 RemoveRaw::Absent => Remove::Absent,
982 RemoveRaw::Success { old } => Remove::Success {
983 old: unsafe { Owned::<K, V, S>::wrap(guard, old) },
984 },
985 RemoveRaw::Break { old } => Remove::Break {
986 old: unsafe { Shared::<K, V, S>::wrap(guard, old) },
987 },
988 }
989 }
990
991 /// If there is a value associated with `key`, call `remove` to determine whether
992 /// to remove the value, **without** recursively removing empty tree nodes.
993 ///
994 /// <div class="warning">
995 ///
996 /// See warning on [`Map::remove_non_recursive`].
997 ///
998 /// </div>
999 ///
1000 /// Returns a [`Remove`] enum.
1001 ///
1002 /// See also: [`ConcurrentMap::remove`],
1003 /// [`ConcurrentMap::remove_with`],
1004 /// [`ConcurrentMap::remove_non_recursive`].
1005 pub fn remove_non_recursive_with<F>(
1006 &self,
1007 key: &K::Borrowed,
1008 mut remove: F,
1009 ) -> Remove<'_, K, V, S>
1010 where
1011 F: FnMut(&V::Borrowed) -> ControlFlow<(), ()>,
1012 {
1013 let reader = K::Read::from(key);
1014 let mut guard = self.smr.guard(reader);
1015 let mut remove = |value: u64| remove(unsafe { V::borrow_from_raw_unchecked(&value) });
1016
1017 let remove = match if cfg!(feature = "opt-no-path") {
1018 unsafe { self.remove_non_recursive_with_optimistic(&mut guard, reader, &mut remove) }
1019 } else {
1020 Err(())
1021 } {
1022 Ok(remove) => remove,
1023 Err(()) => unsafe {
1024 self.remove_non_recursive_with_pessimistic(&mut guard, reader, &mut remove)
1025 },
1026 };
1027
1028 match remove {
1029 RemoveRaw::Absent => Remove::Absent,
1030 RemoveRaw::Success { old } => Remove::Success {
1031 old: unsafe { Owned::<K, V, S>::wrap(guard, old) },
1032 },
1033 RemoveRaw::Break { old } => Remove::Break {
1034 old: unsafe { Shared::<K, V, S>::wrap(guard, old) },
1035 },
1036 }
1037 }
1038}
1039
1040/// Outcome of a call to [`ConcurrentMap::upsert_with`].
1041pub enum Upsert<'g, K, V, S>
1042where
1043 K: Key,
1044 V: Value + 'g,
1045 S: Smr<K, V> + 'g,
1046{
1047 /// Value was successfully upserted.
1048 Success(Upserted<'g, K, V, S>),
1049 /// Closure returned [`core::ops::ControlFlow::Break`].
1050 Break {
1051 /// Latest value observed by closure.
1052 old: Option<Shared<'g, K, V, S>>,
1053 /// Latest value passed as argument or returned from closure.
1054 new: Option<V>,
1055 },
1056}
1057
1058/// Type-erased version of [`Upsert`].
1059enum UpsertRaw {
1060 Success { old: Option<u64>, new: u64 },
1061 Break { old: Option<u64> },
1062}
1063
1064/// Outcome of a call to [`ConcurrentMap::update_with`].
1065pub enum Update<'g, K, V, S>
1066where
1067 K: Key,
1068 V: Value + 'g,
1069 S: Smr<K, V> + 'g,
1070{
1071 /// Key was not present.
1072 Absent {
1073 /// Latest value passed as argument or returned from closure.
1074 new: Option<V>,
1075 },
1076 /// Value was successfully updated.
1077 Success(Updated<'g, K, V, S>),
1078 /// Closure returned [`core::ops::ControlFlow::Break`].
1079 Break {
1080 /// Latest value observed by closure.
1081 old: Shared<'g, K, V, S>,
1082 /// Latest value passed as argument or returned from closure.
1083 new: Option<V>,
1084 },
1085}
1086
1087/// Type-erased version of [`Update`].
1088enum UpdateRaw {
1089 Absent { new: Option<u64> },
1090 Success { old: u64, new: u64 },
1091 Break { old: u64 },
1092}
1093
1094/// Outcome of a call to [`ConcurrentMap::remove_with`].
1095pub enum Remove<'g, K, V, S>
1096where
1097 K: Key,
1098 V: Value + 'g,
1099 S: Smr<K, V> + 'g,
1100{
1101 /// Key was not present.
1102 Absent,
1103 /// Value was successfully removed.
1104 Success {
1105 /// Value that was removed.
1106 old: Owned<'g, K, V, S>,
1107 },
1108 /// Closure returned [`core::ops::ControlFlow::Break`].
1109 Break {
1110 /// Latest value observed by closure.
1111 old: Shared<'g, K, V, S>,
1112 },
1113}
1114
1115/// Type-erased version of [`Remove`].
1116enum RemoveRaw {
1117 Absent,
1118 Success { old: u64 },
1119 Break { old: u64 },
1120}
1121
1122/// # Private implementations
1123///
1124/// These methods erase value types and accept arbitrary key readers and SMR guards.
1125/// This reduces monomorphization and allows a future `concurrent::Set` implementation
1126/// to reuse this logic, at the cost of reducing type safety.
1127///
1128/// # Safety
1129///
1130/// Caller must guarantee:
1131/// - `_guard` protects nodes and values under `reader` for its lifetime.
1132/// - When inserting or upserting, `reader` preserves the prefix property.
1133/// - `initial` and every value returned from a closure was created via `V::into_raw`.
1134impl<K, V, S> Map<K, V, S>
1135where
1136 K: Key,
1137 V: Value,
1138 S: Smr<K, V>,
1139{
1140 #[inline]
1141 unsafe fn get_raw<'g>(&'g self, _guard: &mut S::Guard<'g>, reader: K::Read<'_>) -> Option<u64> {
1142 unsafe { self.seq.raw.cursor::<path::Discard>(reader).traverse_get() }
1143 }
1144
1145 #[inline]
1146 unsafe fn upsert_with_optimistic<'g, 'k, F>(
1147 &'g self,
1148 guard: &mut S::Guard<'g>,
1149 reader: K::Read<'k>,
1150 initial: Option<u64>,
1151 upsert: F,
1152 ) -> Result<UpsertRaw, ()>
1153 where
1154 F: FnMut(Option<u64>, Option<u64>) -> ControlFlow<(), u64>,
1155 {
1156 unsafe { self.upsert_with_raw::<path::Discard, _>(guard, reader, initial, upsert) }
1157 }
1158
1159 #[cold]
1160 unsafe fn upsert_with_pessimistic<'g, 'k, F>(
1161 &'g self,
1162 guard: &mut S::Guard<'g>,
1163 reader: K::Read<'k>,
1164 initial: Option<u64>,
1165 upsert: F,
1166 ) -> UpsertRaw
1167 where
1168 F: FnMut(Option<u64>, Option<u64>) -> ControlFlow<(), u64>,
1169 {
1170 stat::increment(stat::Counter::InsertPessimistic);
1171 let Ok(upsert) =
1172 unsafe { self.upsert_with_raw::<path::Retain<_>, _>(guard, reader, initial, upsert) };
1173 upsert
1174 }
1175
1176 #[inline]
1177 unsafe fn upsert_with_raw<'g, 'k, P, F>(
1178 &'g self,
1179 guard: &mut S::Guard<'g>,
1180 reader: K::Read<'k>,
1181 mut initial: Option<u64>,
1182 mut upsert: F,
1183 ) -> Result<UpsertRaw, P::PopError>
1184 where
1185 P: Path<K::Read<'k>>,
1186 F: FnMut(Option<u64>, Option<u64>) -> ControlFlow<(), u64>,
1187 {
1188 let mut cursor = unsafe { self.seq.raw.cursor::<P>(reader) };
1189
1190 loop {
1191 match cursor.traverse_insert() {
1192 cursor::Insert::Value {
1193 value: old_value,
1194 edge: old_edge,
1195 } => {
1196 let new_value = match upsert(old_value, initial) {
1197 ControlFlow::Continue(new_value) => new_value,
1198 ControlFlow::Break(()) => {
1199 return Ok(UpsertRaw::Break { old: old_value });
1200 }
1201 };
1202
1203 if old_edge.meta().is_frozen() {
1204 // Restore value and fall through to freeze
1205 initial = Some(new_value);
1206 } else {
1207 let (new_edge, _) = cursor.create_path(old_edge, new_value);
1208 match cursor.edge().compare_exchange_packed(
1209 old_edge,
1210 new_edge,
1211 Ordering::AcqRel,
1212 Ordering::Acquire,
1213 ) {
1214 Ok(_) => {
1215 return Ok(UpsertRaw::Success {
1216 old: old_value,
1217 new: new_value,
1218 });
1219 }
1220 Err(_) => {
1221 if let Some(node) = new_edge.as_node() {
1222 unsafe {
1223 stat::increment(stat::Counter::FreeConflict);
1224 node.deallocate_recursive::<K::Edge>();
1225 }
1226 }
1227
1228 initial = Some(new_value);
1229 continue;
1230 }
1231 }
1232 }
1233 }
1234 cursor::Insert::Replace {
1235 node: old_node,
1236 edge: old_edge,
1237 } if !old_edge.meta().is_frozen() => {
1238 let (smo, new_edge) = unsafe {
1239 old_node.freeze::<K::Edge>();
1240 old_node.replace(old_edge.meta())
1241 };
1242 match cursor.edge().compare_exchange_packed(
1243 old_edge,
1244 new_edge,
1245 Ordering::AcqRel,
1246 Ordering::Acquire,
1247 ) {
1248 Ok(_) => {
1249 unsafe { guard.retire_node(cursor.len().bits(), old_node.into_raw()) };
1250 }
1251 Err(_) => {
1252 // Does not go through SMR because `new` is still thread-local
1253 if smo.is_allocate() {
1254 let node = new_edge.as_node().expect("Allocating SMO creates node");
1255 unsafe {
1256 stat::increment(stat::Counter::FreeConflict);
1257 node.deallocate();
1258 }
1259 }
1260 }
1261 }
1262
1263 continue;
1264 }
1265
1266 // Fall through to freeze
1267 cursor::Insert::Replace { .. } => (),
1268 }
1269
1270 match cursor.freeze()? {
1271 None => (),
1272 Some(node) => unsafe { guard.retire_node(cursor.len().bits(), node.into_raw()) },
1273 }
1274 }
1275 }
1276
1277 #[inline]
1278 unsafe fn update_with_optimistic<'g, F>(
1279 &'g self,
1280 guard: &mut S::Guard<'g>,
1281 reader: K::Read<'_>,
1282 initial: Option<u64>,
1283 update: F,
1284 ) -> Result<UpdateRaw, ()>
1285 where
1286 F: FnMut(u64, Option<u64>) -> ControlFlow<(), u64>,
1287 {
1288 unsafe { self.update_with_raw::<path::Discard, _>(guard, reader, initial, update) }
1289 }
1290
1291 #[cold]
1292 unsafe fn update_with_pessimistic<'g, F>(
1293 &'g self,
1294 guard: &mut S::Guard<'g>,
1295 reader: K::Read<'_>,
1296 initial: Option<u64>,
1297 update: F,
1298 ) -> UpdateRaw
1299 where
1300 F: FnMut(u64, Option<u64>) -> ControlFlow<(), u64>,
1301 {
1302 stat::increment(stat::Counter::UpdatePessimistic);
1303 let Ok(update) =
1304 unsafe { self.update_with_raw::<path::Retain<_>, _>(guard, reader, initial, update) };
1305 update
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, P::PopError>
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::Update {
1324 value: old_value,
1325 edge: old_edge,
1326 } = match cursor.traverse_update() {
1327 None => return Ok(UpdateRaw::Absent { new: initial }),
1328 Some(update) if !update.edge.meta().is_frozen() => update,
1329 Some(_) => match cursor.freeze()? {
1330 None => continue,
1331 Some(node) => unsafe {
1332 guard.retire_node(cursor.len().bits(), node.into_raw());
1333 continue;
1334 },
1335 },
1336 };
1337
1338 let new_value = match update(old_value, initial) {
1339 ControlFlow::Continue(new_value) => new_value,
1340 ControlFlow::Break(()) => {
1341 return Ok(UpdateRaw::Break { old: old_value });
1342 }
1343 };
1344
1345 match cursor.edge().compare_exchange_packed(
1346 old_edge,
1347 Edge::new_value(old_edge.meta(), new_value),
1348 Ordering::AcqRel,
1349 Ordering::Acquire,
1350 ) {
1351 Ok(_) => {
1352 return Ok(UpdateRaw::Success {
1353 old: old_value,
1354 new: new_value,
1355 });
1356 }
1357 Err(_) => {
1358 initial = Some(new_value);
1359 }
1360 }
1361 }
1362 }
1363
1364 #[inline]
1365 unsafe fn remove_non_recursive_with_optimistic<'g, F>(
1366 &'g self,
1367 guard: &mut S::Guard<'g>,
1368 reader: K::Read<'_>,
1369 remove: F,
1370 ) -> Result<RemoveRaw, ()>
1371 where
1372 F: FnMut(u64) -> ControlFlow<(), ()>,
1373 {
1374 unsafe { self.remove_with_raw::<false, path::Discard, _>(guard, reader, remove) }
1375 }
1376
1377 #[cold]
1378 unsafe fn remove_non_recursive_with_pessimistic<'g, F>(
1379 &'g self,
1380 guard: &mut S::Guard<'g>,
1381 reader: K::Read<'_>,
1382 remove: F,
1383 ) -> RemoveRaw
1384 where
1385 F: FnMut(u64) -> ControlFlow<(), ()>,
1386 {
1387 let Ok(remove) =
1388 unsafe { self.remove_with_raw::<false, path::Retain<_>, _>(guard, reader, remove) };
1389 remove
1390 }
1391
1392 #[inline]
1393 unsafe fn remove_with_raw<'g, 'k, const RECURSIVE: bool, P, F>(
1394 &'g self,
1395 guard: &mut S::Guard<'g>,
1396 reader: K::Read<'k>,
1397 mut remove: F,
1398 ) -> Result<RemoveRaw, P::PopError>
1399 where
1400 P: Path<K::Read<'k>>,
1401 F: FnMut(u64) -> ControlFlow<(), ()>,
1402 {
1403 let mut cursor = unsafe { self.seq.raw.cursor::<P>(reader) };
1404
1405 let (value, edge) = loop {
1406 let cursor::Update { value, edge } = match cursor.traverse_update() {
1407 None => return Ok(RemoveRaw::Absent),
1408 Some(update) if !update.edge.meta().is_frozen() => update,
1409 Some(_) => match cursor.freeze()? {
1410 None => continue,
1411 Some(node) => unsafe {
1412 guard.retire_node(cursor.len().bits(), node.into_raw());
1413 continue;
1414 },
1415 },
1416 };
1417
1418 match remove(value) {
1419 ControlFlow::Continue(()) => (),
1420 ControlFlow::Break(()) => {
1421 return Ok(RemoveRaw::Break { old: value });
1422 }
1423 }
1424
1425 if cursor
1426 .edge()
1427 .compare_exchange_packed(edge, Edge::NULL, Ordering::AcqRel, Ordering::Acquire)
1428 .is_ok()
1429 {
1430 break (value, edge);
1431 }
1432 };
1433
1434 if RECURSIVE {
1435 let mut trim = edge.meta().len();
1436
1437 'outer: while let Some(target) = cursor
1438 .pop()
1439 .unwrap_or_else(|_| panic!("Recursive remove requires path"))
1440 {
1441 if unsafe { target.len::<K::Edge>() } > 1 {
1442 break 'outer;
1443 }
1444
1445 cursor.trim(K::Len::BYTE + trim.into());
1446
1447 loop {
1448 let old = match cursor.traverse_prefix() {
1449 None => break 'outer,
1450 Some(old) if !old.meta().is_frozen() => old,
1451 Some(_) => match cursor.freeze() {
1452 Err(_) => unreachable!("Recursive remove requires path"),
1453 Ok(None) => continue,
1454 Ok(Some(node)) => unsafe {
1455 guard.retire_node(cursor.len().bits(), node.into_raw());
1456 continue;
1457 },
1458 },
1459 };
1460
1461 let (smo, new) = match old.child() {
1462 None => break 'outer,
1463 Some(edge::Child::Value(_)) => unreachable!("Prefix precondition"),
1464 Some(edge::Child::Node(node)) if node == target => unsafe {
1465 node.freeze::<K::Edge>();
1466 node.replace(old.meta())
1467 },
1468 // Must have been replaced by someone else
1469 Some(edge::Child::Node(_)) => break 'outer,
1470 };
1471
1472 match cursor.edge().compare_exchange_packed(
1473 old,
1474 new,
1475 Ordering::AcqRel,
1476 Ordering::Acquire,
1477 ) {
1478 Ok(old) => {
1479 unsafe { guard.retire_node(cursor.len().bits(), target.into_raw()) };
1480 trim = old.meta().len();
1481 continue 'outer;
1482 }
1483 Err(_) => {
1484 if smo.is_allocate()
1485 && let Some(node) = new.as_node()
1486 {
1487 stat::increment(stat::Counter::FreeConflict);
1488 unsafe { node.deallocate() };
1489 }
1490 }
1491 }
1492 }
1493 }
1494 }
1495
1496 Ok(RemoveRaw::Success { old: value })
1497 }
1498}
1499
1500impl<K, V, S> From<sequential::Map<K, V>> for Map<K, V, S>
1501where
1502 K: Key,
1503 V: Value,
1504 S: Default,
1505{
1506 #[inline]
1507 fn from(seq: sequential::Map<K, V>) -> Self {
1508 Self {
1509 smr: S::default(),
1510 seq,
1511 }
1512 }
1513}
1514
1515impl<K, V, S> From<Map<K, V, S>> for sequential::Map<K, V>
1516where
1517 K: Key,
1518 V: Value,
1519{
1520 #[inline]
1521 fn from(map: Map<K, V, S>) -> sequential::Map<K, V> {
1522 map.seq
1523 }
1524}
1525
1526#[cfg(test)]
1527mod tests {
1528 use core::convert::Infallible;
1529 use core::ops::ControlFlow;
1530
1531 use crate::Order;
1532 use crate::concurrent::Map;
1533 use crate::key::BoxedSlice;
1534 use crate::key::BoxedStr;
1535 use crate::key::NonNull;
1536 use crate::key::Slice;
1537 use crate::key::Str;
1538 use crate::key::Terminated;
1539 use crate::raw::key::Read as _;
1540
1541 #[test]
1542 fn smoke() {
1543 let map = Map::<BoxedStr<NonNull>, _>::default();
1544 map.upsert(unsafe { Slice::new_unchecked("abcd") }, 1u64);
1545 assert_eq!(
1546 map.get(unsafe { Slice::new_unchecked("abcd") })
1547 .as_deref()
1548 .copied(),
1549 Some(1)
1550 );
1551 }
1552
1553 #[test]
1554 fn smoke_u64_key() {
1555 let map = Map::<[u8; 8], _>::default();
1556 let key = 0xdeadbeefu64.to_be_bytes();
1557 map.upsert(&key, 1u64);
1558 assert_eq!(map.get(&key).as_deref().copied(), Some(1));
1559 }
1560
1561 #[test]
1562 fn smoke_value_ref() {
1563 let values = [0, 1, 2, 3, 4, 5];
1564 let map = Map::<u64, &u64>::default();
1565
1566 for (key, value) in values.iter().enumerate() {
1567 map.upsert(key as u64, value);
1568 }
1569
1570 #[expect(clippy::needless_range_loop)]
1571 for key in 0..values.len() {
1572 let value = map.get(&(key as u64)).as_deref().copied().unwrap();
1573 assert!(core::ptr::eq(value, &values[key]));
1574 }
1575 }
1576
1577 #[test]
1578 fn smoke_value_box() {
1579 let values = [0, 1, 2, 3, 4, 5];
1580 let map = Map::<u64, Box<u64>>::default();
1581
1582 for (key, value) in values.iter().enumerate() {
1583 map.upsert(key as u64, Box::new(*value));
1584 }
1585
1586 std::thread::scope(|scope| {
1587 for _ in 0..8 {
1588 scope.spawn(|| {
1589 for key in (0..values.len()).cycle().take(100_000) {
1590 let value = map.get(&(key as u64)).as_deref().copied().unwrap();
1591 assert_eq!(key, value as usize);
1592 }
1593 });
1594 }
1595 });
1596
1597 // TODO: multiple hazards?
1598 // let a = map.get(3);
1599 // let b = map.get(5);
1600 // assert_ne!(a.as_deref(), b.as_deref());
1601
1602 for key in 0..values.len() {
1603 let value = map.get(&(key as u64)).as_deref().copied().unwrap();
1604 assert_eq!(key, value as usize);
1605 }
1606 }
1607
1608 #[test]
1609 fn scan_value() {
1610 let map = Map::<u64, _>::default();
1611 let key = 1u64;
1612 map.upsert(key, 2u64);
1613 assert_eq!(
1614 map.range(1u64..=1u64)
1615 .entries(Order::Ascend)
1616 .collect::<Vec<_>>(),
1617 vec![(1, 2)]
1618 );
1619 }
1620
1621 #[test]
1622 fn scan_node3() {
1623 insert_all(0u64..3);
1624 }
1625
1626 #[test]
1627 fn scan_node256() {
1628 insert_all(0u64..256);
1629 }
1630
1631 #[test]
1632 fn scan_gap() {
1633 let map = insert_all((0u64..512).step_by(2));
1634 assert_eq!(
1635 map.range(256u64..=511u64)
1636 .entries(Order::Ascend)
1637 .collect::<Vec<_>>(),
1638 (256..512)
1639 .step_by(2)
1640 .map(|key| (key, key / 2))
1641 .collect::<Vec<_>>()
1642 );
1643 }
1644
1645 #[test]
1646 fn node3_overwrite() {
1647 let mut map = Map::<u64, _>::default();
1648
1649 for value in [1u64, 2, 3] {
1650 map.upsert(1, value);
1651 assert_eq!(map.get(&1).as_deref().copied(), Some(value));
1652 }
1653
1654 assert_eq!(map.as_sequential().all().entries(Order::Ascend).count(), 1);
1655
1656 map.as_sequential()
1657 .all()
1658 .entries(Order::Ascend)
1659 .try_fold((), |(), (key, value)| {
1660 assert_eq!(key, 1);
1661 assert_eq!(*value, 3);
1662 ControlFlow::<Infallible>::Continue(())
1663 });
1664 }
1665
1666 #[test]
1667 fn node3_reverse() {
1668 insert_all((0u16..3).rev());
1669 }
1670
1671 #[test]
1672 fn node3_full() {
1673 insert_all(0u16..3);
1674 }
1675
1676 #[test]
1677 fn node3_expand() {
1678 insert_all(0u16..4);
1679 }
1680
1681 #[test]
1682 fn node15_full() {
1683 insert_all(0u16..15);
1684 }
1685
1686 #[test]
1687 fn node15_expand() {
1688 insert_all(0u16..16);
1689 }
1690
1691 #[test]
1692 fn node47_full() {
1693 insert_all(0u16..47);
1694 }
1695
1696 #[test]
1697 fn node47_expand() {
1698 insert_all(0u16..61);
1699 }
1700
1701 #[test]
1702 fn node256_full() {
1703 insert_all(0u16..=255);
1704 }
1705
1706 #[test]
1707 fn range_reverse() {
1708 let map = Map::<u64, _>::default();
1709
1710 for key in [5, 1, 4, 3, 2] {
1711 map.upsert(key, key);
1712 assert_eq!(map.get(&key).as_deref().copied(), Some(key));
1713 }
1714
1715 assert_eq!(
1716 map.range(2..=4).entries(Order::Descend).collect::<Vec<_>>(),
1717 vec![(4, 4), (3, 3), (2, 2)]
1718 );
1719 }
1720
1721 #[test]
1722 fn split_edges() {
1723 let mut key = (1..100).collect::<Vec<_>>();
1724 insert_all(core::iter::from_fn(|| {
1725 if key.is_empty() {
1726 None
1727 } else {
1728 let mut next = key.clone();
1729 next.push(0);
1730 key.pop();
1731 let next = next.into_boxed_slice();
1732 Some(BoxedSlice::<Terminated<0>>::new(next).unwrap())
1733 }
1734 }));
1735 }
1736
1737 #[test]
1738 fn one_long_key() {
1739 insert_all([BoxedStr::<NonNull>::new("a".repeat(1000)).unwrap()]);
1740 }
1741
1742 #[test]
1743 fn short_key() {
1744 insert_all([BoxedStr::<NonNull>::new("\n".to_string()).unwrap()]);
1745 }
1746
1747 #[test]
1748 fn two_long_keys() {
1749 insert_all([
1750 BoxedStr::<NonNull>::new("a".repeat(1000)).unwrap(),
1751 BoxedStr::<NonNull>::new("b".repeat(1000)).unwrap(),
1752 ]);
1753 }
1754
1755 #[test]
1756 fn smoke_key_slice() {
1757 let keys = ["ad", "abc"];
1758 let map = crate::concurrent::Map::<&Str<NonNull>, u64>::new();
1759 map.insert(Str::new(keys[0]).unwrap(), 0)
1760 .unwrap_or_else(|(_, _)| panic!());
1761 map.insert(Str::new(keys[1]).unwrap(), 1)
1762 .unwrap_or_else(|(_, _)| panic!());
1763
1764 let temp = "adabc";
1765 assert_eq!(
1766 map.get(Str::new(&temp[..2]).unwrap()).as_deref().copied(),
1767 Some(0)
1768 );
1769 assert_eq!(
1770 map.get(Str::new(&temp[2..]).unwrap()).as_deref().copied(),
1771 Some(1)
1772 );
1773 }
1774
1775 #[test]
1776 fn key_slice_long_prefix() {
1777 let keys = (0..10)
1778 .map(|i| "a".repeat(100) + &i.to_string())
1779 .collect::<Vec<_>>();
1780 let map = crate::concurrent::Map::<&Slice<NonNull>, u64>::new();
1781 for (i, key) in keys.iter().enumerate() {
1782 map.insert(Slice::new(key.as_bytes()).unwrap(), i as u64)
1783 .unwrap();
1784 }
1785 for (i, key) in keys.iter().enumerate() {
1786 assert_eq!(
1787 map.get(Slice::new(key.as_bytes()).unwrap())
1788 .as_deref()
1789 .copied(),
1790 Some(i as u64)
1791 );
1792 }
1793 }
1794
1795 fn insert_all<I, K>(iter: I) -> Map<K, u64>
1796 where
1797 I: IntoIterator<Item = K>,
1798 K: crate::Key + Clone + Ord + core::fmt::Debug,
1799 {
1800 let mut keys = iter
1801 .into_iter()
1802 .enumerate()
1803 .map(|(index, key)| (key, index as u64))
1804 .collect::<Vec<_>>();
1805
1806 let mut map = Map::default();
1807
1808 for (key, value) in &keys {
1809 map.upsert(key.as_insert(), *value);
1810 assert_eq!(map.get(key.borrow()).as_deref().copied(), Some(*value));
1811 }
1812
1813 for (key, value) in &keys {
1814 assert_eq!(map.get(key.borrow()).as_deref().copied(), Some(*value));
1815 }
1816
1817 let mut iter = map.as_sequential().all().entries(Order::Ascend);
1818 let mut count = 0;
1819 while iter.lend().is_some() {
1820 count += 1;
1821 }
1822 drop(iter);
1823
1824 assert_eq!(count, keys.len());
1825
1826 keys.sort_by(|(l, _), (r, _)| l.cmp(r));
1827
1828 // Sequential iteration
1829 map.as_sequential()
1830 .all()
1831 .entries(Order::Ascend)
1832 .zip(&keys)
1833 .for_each(|((lk, lv), (rk, rv))| {
1834 assert_eq!(lk, *rk);
1835 assert_eq!(*lv, *rv);
1836 });
1837
1838 let Some(((first, _), (last, _))) = keys.first().zip(keys.last()) else {
1839 return map;
1840 };
1841
1842 // Concurrent prefix scan, non-linearizable
1843 map.prefix(K::Read::from(first.borrow()).common_prefix(K::Read::from(last.borrow())))
1844 .entries(Order::Descend)
1845 .zip(keys.iter().rev())
1846 .for_each(|((lk, lv), (rk, rv))| {
1847 assert_eq!(lk, *rk);
1848 assert_eq!(lv, *rv);
1849 });
1850
1851 // Concurrent range scan, non-linearizable
1852 let mut i = 0;
1853 map.range(first.borrow()..=last.borrow())
1854 .entries(Order::Descend)
1855 .zip(keys.iter().rev())
1856 .for_each(|((lk, lv), (rk, rv))| {
1857 i += 1;
1858 assert_eq!(lk, *rk);
1859 assert_eq!(lv, *rv);
1860 });
1861 assert_eq!(i, keys.len());
1862
1863 map
1864 }
1865}