1use ::core::borrow::Borrow;
2use ::core::fmt::Debug;
3use ::core::iter::FusedIterator;
4use ::core::marker::PhantomData;
5use ::core::ops::{Bound, RangeBounds};
6use ::core::sync::atomic::{AtomicPtr, AtomicU64, Ordering};
7use alloc::boxed::Box;
8use alloc::collections::BTreeMap;
9use alloc::sync::Arc;
10use alloc::vec;
11use alloc::vec::Vec;
12use parking_lot::{
13 ArcRwLockReadGuard, ArcRwLockWriteGuard, Mutex, MutexGuard, RawRwLock, RwLock, RwLockReadGuard, RwLockWriteGuard,
14};
15
16use crate::cdc::change::ChangeEvent;
17use crate::concurrent::operation::*;
18use crate::core::constants::DEFAULT_INNER_SIZE;
19use crate::core::node::*;
20
21use super::r#ref::Ref;
22
23#[inline]
29fn yield_now() {
30 #[cfg(feature = "std")]
31 std::thread::yield_now();
32 #[cfg(not(feature = "std"))]
33 ::core::hint::spin_loop();
34}
35
36const ROOT_PUBLICATION_SPIN_LIMIT: usize = 16;
37const STABLE_READ_BLOCKING_FALLBACK_AFTER: usize = 2;
38const PUBLICATION_BACKLOG_DRAIN_THRESHOLD: usize = 64;
39
40type NodeIndex<T, Node> = BTreeMap<T, Arc<RwLock<Node>>>;
41
42const PUBLISHED_ROUTES_PER_CHUNK: usize = 128;
47const PUBLISHED_REBUILD_ROUTES_PER_CHUNK: usize = PUBLISHED_ROUTES_PER_CHUNK * 2 / 3;
50const PUBLISHED_ROUTE_MERGE_THRESHOLD: usize = PUBLISHED_ROUTES_PER_CHUNK * 3 / 4;
51
52struct PublishedChunk<T, Node> {
53 entries: Vec<(T, Arc<RwLock<Node>>)>,
54}
55
56impl<T: Clone, Node> Clone for PublishedChunk<T, Node> {
57 fn clone(&self) -> Self {
58 Self {
59 entries: self.entries.clone(),
60 }
61 }
62}
63
64struct PublishedNodeIndex<T, Node> {
65 chunks: Vec<Arc<PublishedChunk<T, Node>>>,
66 len: usize,
67}
68
69impl<T, Node> Clone for PublishedNodeIndex<T, Node> {
70 fn clone(&self) -> Self {
71 Self {
72 chunks: self.chunks.clone(),
73 len: self.len,
74 }
75 }
76}
77
78impl<T, Node> PublishedNodeIndex<T, Node>
79where
80 T: Ord + Clone,
81{
82 fn from_canonical(index: &NodeIndex<T, Node>) -> Self {
83 let mut chunks = Vec::with_capacity(index.len().div_ceil(PUBLISHED_REBUILD_ROUTES_PER_CHUNK));
84 let mut entries = Vec::with_capacity(PUBLISHED_REBUILD_ROUTES_PER_CHUNK);
85
86 for (key, node) in index {
87 entries.push((key.clone(), node.clone()));
88 if entries.len() == PUBLISHED_REBUILD_ROUTES_PER_CHUNK {
89 chunks.push(Arc::new(PublishedChunk { entries }));
90 entries = Vec::with_capacity(PUBLISHED_REBUILD_ROUTES_PER_CHUNK);
91 }
92 }
93 if !entries.is_empty() {
94 chunks.push(Arc::new(PublishedChunk { entries }));
95 }
96
97 Self {
98 chunks,
99 len: index.len(),
100 }
101 }
102
103 fn iter(&self) -> impl Iterator<Item = (&T, &Arc<RwLock<Node>>)> {
104 self.chunks
105 .iter()
106 .flat_map(|chunk| chunk.entries.iter().map(|(key, node)| (key, node)))
107 }
108
109 fn first_key_value(&self) -> Option<(&T, &Arc<RwLock<Node>>)> {
110 self.chunks.first()?.entries.first().map(|(key, node)| (key, node))
111 }
112
113 fn last_key_value(&self) -> Option<(&T, &Arc<RwLock<Node>>)> {
114 self.chunks.last()?.entries.last().map(|(key, node)| (key, node))
115 }
116
117 fn chunk_for<Q>(&self, key: &Q) -> usize
118 where
119 T: Borrow<Q>,
120 Q: Ord + ?Sized,
121 {
122 self.chunks.partition_point(|chunk| {
123 let max = &chunk.entries.last().expect("published chunks are non-empty").0;
124 <T as Borrow<Q>>::borrow(max) < key
125 })
126 }
127
128 fn first_for_bound<Q>(&self, bound: Bound<&Q>) -> Option<(&T, &Arc<RwLock<Node>>)>
129 where
130 T: Borrow<Q>,
131 Q: Ord + ?Sized,
132 {
133 let key = match bound {
134 Bound::Included(key) | Bound::Excluded(key) => key,
135 Bound::Unbounded => return self.first_key_value(),
136 };
137 let mut chunk_index = self.chunk_for(key);
138 while let Some(chunk) = self.chunks.get(chunk_index) {
139 let entry_index = chunk.entries.partition_point(|(candidate, _)| match bound {
140 Bound::Included(_) => <T as Borrow<Q>>::borrow(candidate) < key,
141 Bound::Excluded(_) => <T as Borrow<Q>>::borrow(candidate) <= key,
142 Bound::Unbounded => false,
143 });
144 if let Some((found, node)) = chunk.entries.get(entry_index) {
145 return Some((found, node));
146 }
147 chunk_index += 1;
148 }
149 None
150 }
151
152 fn insert(&mut self, key: T, node: Arc<RwLock<Node>>) -> Option<Arc<RwLock<Node>>> {
153 if self.chunks.is_empty() {
154 self.chunks.push(Arc::new(PublishedChunk {
155 entries: vec![(key, node)],
156 }));
157 self.len = 1;
158 return None;
159 }
160
161 let mut chunk_index = self.chunk_for(&key);
162 if chunk_index == self.chunks.len() {
163 chunk_index -= 1;
164 }
165 let chunk = Arc::make_mut(&mut self.chunks[chunk_index]);
166 match chunk.entries.binary_search_by(|(candidate, _)| candidate.cmp(&key)) {
167 Ok(index) => Some(::core::mem::replace(&mut chunk.entries[index].1, node)),
168 Err(index) => {
169 chunk.entries.insert(index, (key, node));
170 self.len += 1;
171 if chunk.entries.len() > PUBLISHED_ROUTES_PER_CHUNK {
172 let right = chunk.entries.split_off(chunk.entries.len() / 2);
173 self.chunks
174 .insert(chunk_index + 1, Arc::new(PublishedChunk { entries: right }));
175 }
176 None
177 }
178 }
179 }
180
181 fn remove<Q>(&mut self, key: &Q) -> Option<Arc<RwLock<Node>>>
182 where
183 T: Borrow<Q>,
184 Q: Ord + ?Sized,
185 {
186 let chunk_index = self.chunk_for(key);
187 let entry_index = self
188 .chunks
189 .get(chunk_index)?
190 .entries
191 .binary_search_by(|(candidate, _)| <T as Borrow<Q>>::borrow(candidate).cmp(key))
192 .ok()?;
193 let chunk = Arc::make_mut(&mut self.chunks[chunk_index]);
194 let (_, removed) = chunk.entries.remove(entry_index);
195 self.len -= 1;
196
197 if chunk.entries.is_empty() {
198 self.chunks.remove(chunk_index);
199 } else if chunk_index > 0
200 && self.chunks[chunk_index - 1].entries.len() + self.chunks[chunk_index].entries.len()
201 <= PUBLISHED_ROUTE_MERGE_THRESHOLD
202 {
203 let right = self.chunks.remove(chunk_index);
204 Arc::make_mut(&mut self.chunks[chunk_index - 1])
205 .entries
206 .extend(right.entries.iter().cloned());
207 } else if chunk_index + 1 < self.chunks.len()
208 && self.chunks[chunk_index].entries.len() + self.chunks[chunk_index + 1].entries.len()
209 <= PUBLISHED_ROUTE_MERGE_THRESHOLD
210 {
211 let right = self.chunks.remove(chunk_index + 1);
212 Arc::make_mut(&mut self.chunks[chunk_index])
213 .entries
214 .extend(right.entries.iter().cloned());
215 }
216
217 Some(removed)
218 }
219}
220
221#[inline]
222fn node_identity<Node>(node: &Arc<RwLock<Node>>) -> usize {
223 Arc::as_ptr(node) as usize
227}
228
229struct RetiredIndex<T, Node>(*mut PublishedNodeIndex<T, Node>);
230
231unsafe impl<T: Send, Node: Send> Send for RetiredIndex<T, Node> {}
239
240impl<T, Node> Drop for RetiredIndex<T, Node> {
241 fn drop(&mut self) {
242 unsafe { drop(Box::from_raw(self.0)) }
245 }
246}
247
248struct PublishedIndex<T, Node> {
249 current: AtomicPtr<PublishedNodeIndex<T, Node>>,
250 domain: ps_reclaim::Domain,
251}
252
253impl<T, Node> PublishedIndex<T, Node> {
254 fn new() -> Self {
255 Self {
256 current: AtomicPtr::new(Box::into_raw(Box::new(PublishedNodeIndex {
257 chunks: Vec::new(),
258 len: 0,
259 }))),
260 domain: ps_reclaim::Domain::new(),
261 }
262 }
263}
264
265impl<T, Node> PublishedIndex<T, Node>
266where
267 T: Ord + Clone + Send + 'static,
268 Node: Send + 'static,
269{
270 fn last_is(&self, node: &Arc<RwLock<Node>>) -> bool {
276 let current = self.current.load(Ordering::Acquire);
277 unsafe {
280 (&*current)
281 .last_key_value()
282 .is_some_and(|(_, published)| Arc::ptr_eq(published, node))
283 }
284 }
285
286 fn snapshot(&self) -> PublishedNodeIndex<T, Node> {
287 let current = self.current.load(Ordering::Acquire);
288 unsafe { (&*current).clone() }
291 }
292
293 fn replace(&self, replacement: PublishedNodeIndex<T, Node>) -> RetiredIndex<T, Node> {
294 let replacement = Box::into_raw(Box::new(replacement));
297 let retired = self.current.swap(replacement, Ordering::AcqRel);
298 RetiredIndex(retired)
299 }
300
301 fn retire(&self, retired: RetiredIndex<T, Node>) {
302 self.domain.retire(move || drop(retired));
305 }
306
307 fn advance(&self) {
308 if self.domain.pending() >= PUBLICATION_BACKLOG_DRAIN_THRESHOLD {
315 self.domain.advance();
316 }
317 }
318}
319
320impl<T, Node> Drop for PublishedIndex<T, Node> {
321 fn drop(&mut self) {
322 let current = *self.current.get_mut();
323 unsafe { drop(Box::from_raw(current)) }
326 }
327}
328
329pub(crate) struct Topology<T, Node> {
338 index: RwLock<NodeIndex<T, Node>>,
339 published_keys: Mutex<BTreeMap<usize, T>>,
343 published: PublishedIndex<T, Node>,
344 generation: AtomicU64,
347}
348
349impl<T, Node> Debug for Topology<T, Node> {
350 fn fmt(&self, formatter: &mut ::core::fmt::Formatter<'_>) -> ::core::fmt::Result {
351 formatter
352 .debug_struct("Topology")
353 .field("nodes", &self.index.read().len())
354 .field("generation", &self.generation.load(Ordering::Relaxed))
355 .finish()
356 }
357}
358
359impl<T, Node> Topology<T, Node> {
360 fn new() -> Self {
361 Self {
362 index: RwLock::new(BTreeMap::new()),
363 published_keys: Mutex::new(BTreeMap::new()),
364 published: PublishedIndex::new(),
365 generation: AtomicU64::new(0),
366 }
367 }
368
369 #[inline]
370 pub(crate) fn read(&self) -> RwLockReadGuard<'_, NodeIndex<T, Node>> {
371 self.index.read()
372 }
373}
374
375impl<T, Node> Topology<T, Node>
376where
377 T: Ord + Clone + Send + 'static,
378 Node: Send + 'static,
379{
380 #[inline]
381 fn write(&self) -> TopologyWriteGuard<'_, T, Node> {
382 let index = self.index.write();
383 self.generation.fetch_add(1, Ordering::AcqRel);
384 TopologyWriteGuard {
385 topology: self,
386 index: Some(index),
387 published: None,
388 published_keys: None,
389 publish: true,
390 dirty: false,
391 }
392 }
393
394 #[inline]
402 fn write_rekey(&self) -> TopologyWriteGuard<'_, T, Node> {
403 let index = self.index.write();
404 TopologyWriteGuard {
405 topology: self,
406 index: Some(index),
407 published: None,
408 published_keys: None,
409 publish: false,
410 dirty: false,
411 }
412 }
413
414 #[inline]
415 fn try_write(&self) -> Option<TopologyWriteGuard<'_, T, Node>> {
416 let index = self.index.try_write()?;
417 self.generation.fetch_add(1, Ordering::AcqRel);
418 Some(TopologyWriteGuard {
419 topology: self,
420 index: Some(index),
421 published: None,
422 published_keys: None,
423 publish: true,
424 dirty: false,
425 })
426 }
427}
428
429pub(crate) struct TopologyWriteGuard<'a, T, Node>
430where
431 T: Ord + Clone + Send + 'static,
432 Node: Send + 'static,
433{
434 topology: &'a Topology<T, Node>,
435 index: Option<RwLockWriteGuard<'a, NodeIndex<T, Node>>>,
438 published: Option<PublishedNodeIndex<T, Node>>,
439 published_keys: Option<MutexGuard<'a, BTreeMap<usize, T>>>,
440 publish: bool,
441 dirty: bool,
442}
443
444impl<T, Node> ::core::ops::Deref for TopologyWriteGuard<'_, T, Node>
445where
446 T: Ord + Clone + Send + 'static,
447 Node: Send + 'static,
448{
449 type Target = NodeIndex<T, Node>;
450
451 fn deref(&self) -> &Self::Target {
452 self.index.as_deref().expect("topology guard already released")
453 }
454}
455
456impl<'a, T, Node> TopologyWriteGuard<'a, T, Node>
457where
458 T: Ord + Clone + Send + 'static,
459 Node: Send + 'static,
460{
461 pub(crate) fn enable_publication(&mut self) {
462 if self.publish {
463 return;
464 }
465 debug_assert!(
466 !self.dirty,
467 "publication must be enabled before mutating an opt-out topology guard"
468 );
469 self.topology.generation.fetch_add(1, Ordering::AcqRel);
474 self.publish = true;
475 }
476
477 fn ensure_publication_snapshot(&mut self) {
478 if self.published.is_none() {
479 self.published_keys = Some(self.topology.published_keys.lock());
480 self.published = Some(self.topology.published.snapshot());
481 }
482 }
483
484 fn rebuild_publication(&mut self) {
491 if self.published_keys.is_none() {
492 self.published_keys = Some(self.topology.published_keys.lock());
493 }
494 let canonical = self.index.as_deref().expect("topology guard already released");
495 let rebuilt = PublishedNodeIndex::from_canonical(canonical);
496 let rebuilt_keys = canonical
497 .iter()
498 .map(|(key, node)| (node_identity(node), key.clone()))
499 .collect();
500 **self
501 .published_keys
502 .as_mut()
503 .expect("publication identity lock was initialized") = rebuilt_keys;
504 self.published = Some(rebuilt);
505 self.dirty = true;
506 }
507
508 pub(crate) fn is_last_node(&self, node: &Arc<RwLock<Node>>) -> bool {
509 self.index
510 .as_deref()
511 .and_then(BTreeMap::last_key_value)
512 .is_some_and(|(_, candidate)| Arc::ptr_eq(candidate, node))
513 }
514
515 pub(crate) fn can_rekey_last_node_without_publication(&self, node: &Arc<RwLock<Node>>, new_key: &T) -> bool {
520 let mut canonical = self
521 .index
522 .as_deref()
523 .expect("topology guard already released")
524 .iter()
525 .rev();
526 let Some((_, last_node)) = canonical.next() else {
527 return false;
528 };
529 if !Arc::ptr_eq(last_node, node) || !self.topology.published.last_is(node) {
530 return false;
531 }
532 match canonical.next() {
533 Some((predecessor, _)) => predecessor < new_key,
534 None => true,
535 }
536 }
537
538 pub(crate) fn rekey_last_node(&mut self, old_key: &T, new_key: T, node: Arc<RwLock<Node>>) {
546 debug_assert!(!self.publish, "last-node rekey must use the opt-out guard");
547 debug_assert!(
548 !self.dirty,
549 "an opt-out guard may perform only one explicit last-node rekey"
550 );
551 debug_assert!(
552 self.is_last_node(&node),
553 "only the canonical last node may skip publication"
554 );
555
556 let index = self.index.as_deref_mut().expect("topology guard already released");
557 let removed = index.remove(old_key);
558 debug_assert!(
559 removed.as_ref().is_some_and(|removed| Arc::ptr_eq(removed, &node)),
560 "last-node rekey must remove its expected canonical route"
561 );
562 let replaced = index.insert(new_key, node);
563 debug_assert!(
564 replaced.is_none(),
565 "last-node rekey must not collide with another canonical route"
566 );
567 self.dirty = true;
568 }
569
570 fn repair_last_route_before_insert(&mut self) {
573 debug_assert!(self.publish, "route insertion repair requires publication");
574 let Some((canonical_key, last_node)) = self
575 .index
576 .as_deref()
577 .expect("topology guard already released")
578 .last_key_value()
579 .map(|(key, node)| (key.clone(), node.clone()))
580 else {
581 return;
582 };
583
584 self.ensure_publication_snapshot();
585 let published_key = self
586 .published_keys
587 .as_ref()
588 .expect("publication identity map initialized")
589 .get(&node_identity(&last_node))
590 .cloned();
591 let Some(published_key) = published_key else {
592 self.rebuild_publication();
593 return;
594 };
595 if published_key == canonical_key {
596 return;
597 }
598
599 let repaired_consistently = {
600 let published = self.published.as_mut().expect("publication snapshot initialized");
601 let published_keys = self
602 .published_keys
603 .as_mut()
604 .expect("publication identity map initialized");
605 let removed = published.remove(&published_key);
606 let displaced = published.insert(canonical_key.clone(), last_node.clone());
607 published_keys.insert(node_identity(&last_node), canonical_key);
608 removed.is_some_and(|old_node| Arc::ptr_eq(&old_node, &last_node)) && displaced.is_none()
609 };
610 if !repaired_consistently {
611 self.rebuild_publication();
612 } else {
613 self.dirty = true;
614 }
615 }
616
617 pub(crate) fn insert(&mut self, key: T, node: Arc<RwLock<Node>>) -> Option<Arc<RwLock<Node>>> {
618 self.repair_last_route_before_insert();
619 self.insert_after_last_route_repair(key, node)
620 }
621
622 fn insert_after_last_route_repair(&mut self, key: T, node: Arc<RwLock<Node>>) -> Option<Arc<RwLock<Node>>> {
623 debug_assert!(self.publish, "generic topology insertion requires publication");
624 let replaced = self
625 .index
626 .as_deref_mut()
627 .expect("topology guard already released")
628 .insert(key.clone(), node.clone());
629 self.dirty = true;
630
631 if self.publish {
632 self.ensure_publication_snapshot();
633 if let Some(replaced) = &replaced {
634 let removed_consistently = {
635 let published = self.published.as_mut().expect("publication snapshot initialized");
636 let published_keys = self
637 .published_keys
638 .as_mut()
639 .expect("publication identity map initialized");
640 published_keys
641 .remove(&node_identity(replaced))
642 .and_then(|old_key| published.remove(&old_key))
643 .is_some_and(|old_node| Arc::ptr_eq(&old_node, replaced))
644 };
645 if !removed_consistently {
646 self.rebuild_publication();
647 return Some(replaced.clone());
648 }
649 }
650
651 let displaced = self
652 .published
653 .as_mut()
654 .expect("publication snapshot initialized")
655 .insert(key.clone(), node.clone());
656 self.published_keys
657 .as_mut()
658 .expect("publication identity map initialized")
659 .insert(node_identity(&node), key);
660 if displaced.is_some_and(|old_node| !Arc::ptr_eq(&old_node, &node)) {
661 self.rebuild_publication();
666 }
667 }
668
669 replaced
670 }
671
672 pub(crate) fn remove<Q>(&mut self, key: &Q) -> Option<Arc<RwLock<Node>>>
673 where
674 T: Borrow<Q>,
675 Q: Ord + ?Sized,
676 {
677 debug_assert!(self.publish, "generic topology removal requires publication");
678 let removed = self
679 .index
680 .as_deref_mut()
681 .expect("topology guard already released")
682 .remove(key)?;
683 self.dirty = true;
684
685 if self.publish {
686 self.ensure_publication_snapshot();
687 let removed_consistently = {
688 let published = self.published.as_mut().expect("publication snapshot initialized");
689 let published_keys = self
690 .published_keys
691 .as_mut()
692 .expect("publication identity map initialized");
693 published_keys
694 .remove(&node_identity(&removed))
695 .and_then(|route_key| published.remove::<T>(&route_key))
696 .is_some_and(|route_node| Arc::ptr_eq(&route_node, &removed))
697 };
698 if !removed_consistently {
699 self.rebuild_publication();
700 }
701 }
702
703 Some(removed)
704 }
705}
706
707impl<T, Node> Drop for TopologyWriteGuard<'_, T, Node>
708where
709 T: Ord + Clone + Send + 'static,
710 Node: Send + 'static,
711{
712 fn drop(&mut self) {
713 #[cfg(debug_assertions)]
714 if self.publish && self.dirty {
715 let canonical = self.index.as_deref().expect("topology guard exists while validating");
716 let published = self.published.as_ref().expect("publishing guard carries snapshot");
717 let published_keys = self
718 .published_keys
719 .as_deref()
720 .expect("publishing guard carries identity map");
721 debug_assert_eq!(
722 canonical.len(),
723 published.len,
724 "canonical and published node counts diverged"
725 );
726 debug_assert_eq!(
727 canonical.len(),
728 published_keys.len(),
729 "canonical and published identity counts diverged"
730 );
731 for ((_, canonical_node), (route_key, route_node)) in canonical.iter().zip(published.iter()) {
732 debug_assert!(
733 Arc::ptr_eq(route_node, canonical_node),
734 "canonical and published node order diverged"
735 );
736 debug_assert!(
737 published_keys.get(&node_identity(canonical_node)) == Some(route_key),
738 "published identity key does not match the route index"
739 );
740 }
741 }
742
743 if !self.publish {
744 drop(self.published_keys.take());
745 drop(self.index.take());
746 return;
747 }
748 if !self.dirty {
749 self.topology.generation.fetch_add(1, Ordering::Release);
750 drop(self.published_keys.take());
751 drop(self.index.take());
752 return;
753 }
754 let retired = self.topology.published.replace(
755 self.published
756 .take()
757 .expect("publishing guard must carry a read snapshot"),
758 );
759 self.topology.generation.fetch_add(1, Ordering::Release);
763 self.topology.published.retire(retired);
764 drop(self.published_keys.take());
765 drop(self.index.take());
766 self.topology.published.advance();
767 }
768}
769
770pub(crate) fn no_identity_adoption<T>(_stored: &T, _incoming: &mut T) {}
774
775fn first_for_borrowed_bound<'a, T, Q, V>(
782 index: &'a BTreeMap<T, V>,
783 bound: Bound<&Q>,
784 borrow_order_matches: bool,
785) -> Option<(&'a T, &'a V)>
786where
787 T: Ord + Borrow<Q>,
788 Q: Ord + ?Sized,
789{
790 if borrow_order_matches {
791 return index.range::<Q, _>((bound, Bound::Unbounded)).next();
792 }
793
794 index.iter().find(|(key, _)| match bound {
795 Bound::Included(value) => <T as Borrow<Q>>::borrow(key) >= value,
796 Bound::Excluded(value) => <T as Borrow<Q>>::borrow(key) > value,
797 Bound::Unbounded => true,
798 })
799}
800
801fn first_published_for_borrowed_bound<'a, T, Q, V>(
802 index: &'a PublishedNodeIndex<T, V>,
803 bound: Bound<&Q>,
804 borrow_order_matches: bool,
805) -> Option<(&'a T, &'a Arc<RwLock<V>>)>
806where
807 T: Ord + Clone + Borrow<Q>,
808 Q: Ord + ?Sized,
809{
810 if borrow_order_matches {
811 return index.first_for_bound(bound);
812 }
813
814 index.iter().find(|(key, _)| match bound {
815 Bound::Included(value) => <T as Borrow<Q>>::borrow(key) >= value,
816 Bound::Excluded(value) => <T as Borrow<Q>>::borrow(key) > value,
817 Bound::Unbounded => true,
818 })
819}
820
821fn node_for_borrowed_end<'a, T, Q, V>(
822 index: &'a BTreeMap<T, V>,
823 end: &Q,
824 borrow_order_matches: bool,
825) -> Option<(&'a T, &'a V)>
826where
827 T: Ord + Borrow<Q>,
828 Q: Ord + ?Sized,
829{
830 if borrow_order_matches {
831 return index
832 .range::<Q, _>((Bound::Included(end), Bound::Unbounded))
833 .next()
834 .or_else(|| index.last_key_value());
835 }
836
837 let mut last_equal = None;
838 for entry @ (key, _) in index {
839 match <T as Borrow<Q>>::borrow(key).cmp(end) {
840 ::core::cmp::Ordering::Less => {}
841 ::core::cmp::Ordering::Equal => last_equal = Some(entry),
842 ::core::cmp::Ordering::Greater => return Some(entry),
846 }
847 }
848 last_equal.or_else(|| index.last_key_value())
849}
850
851#[derive(Debug)]
907pub struct BTreeSet<T, Node = Vec<T>>
908where
909 T: Ord + Clone + 'static,
910 Node: NodeLike<T>,
911{
912 pub(crate) index: Topology<T, Node>,
918 node_capacity: usize,
919 borrow_order_matches: bool,
923 #[cfg(feature = "cdc")]
924 event_id: AtomicU64,
928}
929impl<T: Ord + Clone + 'static, Node: NodeLike<T>> Default for BTreeSet<T, Node> {
930 fn default() -> Self {
931 Self {
932 index: Topology::new(),
933 node_capacity: DEFAULT_INNER_SIZE,
934 borrow_order_matches: true,
935 #[cfg(feature = "cdc")]
936 event_id: AtomicU64::new(0),
937 }
938 }
939}
940
941impl<T, Node> BTreeSet<T, Node>
942where
943 T: Debug + Ord + Clone + Send,
944 Node: NodeLike<T> + Send + 'static,
945{
946 pub fn new() -> Self {
947 Self::default()
948 }
949 pub fn with_maximum_node_size(node_capacity: usize) -> Self {
959 Self {
960 index: Topology::new(),
961 node_capacity,
962 borrow_order_matches: true,
963 #[cfg(feature = "cdc")]
964 event_id: AtomicU64::new(0),
965 }
966 }
967 pub(crate) fn with_grouped_borrow_routing(mut self) -> Self {
968 self.borrow_order_matches = false;
969 self
970 }
971 pub fn attach_node(&self, node: Node) {
972 self.attach_nodes(::core::iter::once(node));
973 }
974
975 pub fn attach_nodes(&self, nodes: impl IntoIterator<Item = Node>) {
982 let mut nodes = nodes.into_iter().peekable();
983 if nodes.peek().is_none() {
984 return;
985 }
986
987 let mut index = self.index.write();
988 index.repair_last_route_before_insert();
989 for node in nodes {
990 let node_id = node
991 .max()
992 .cloned()
993 .expect("node should contain at least one value to be correct node");
994 index.insert_after_last_route_repair(node_id, Arc::new(RwLock::new(node)));
995 }
996 }
997
998 #[cfg(feature = "cdc")]
999 pub(crate) fn export_topology(&self) -> (usize, Vec<Vec<T>>) {
1000 let index = self.index.read();
1001 let nodes = index
1002 .values()
1003 .map(|node| node.read().iter().cloned().collect())
1004 .collect();
1005 (self.node_capacity, nodes)
1006 }
1007
1008 #[allow(clippy::type_complexity)]
1009 fn put_checked_inner<const EMIT_CDC: bool>(
1012 &self,
1013 value: T,
1014 adopt: fn(&T, &mut T),
1015 ) -> Result<(Option<T>, Vec<ChangeEvent<T>>), (ArcRwLockWriteGuard<RawRwLock, Node>, usize, T)> {
1016 loop {
1017 let mut cdc = vec![];
1018 let index = self.index.read();
1019 let target_node_entry = match index.range(value.clone()..).next() {
1020 Some(entry) => entry,
1021 None => {
1022 if let Some(last) = index.last_key_value() {
1023 last
1024 } else {
1025 drop(index);
1026 let mut spins = 0;
1027 let mut index = loop {
1028 if let Some(guard) = self.index.try_write() {
1029 break guard;
1030 }
1031 if spins >= ROOT_PUBLICATION_SPIN_LIMIT {
1032 break self.index.write();
1036 }
1037 spins += 1;
1038 ::core::hint::spin_loop();
1039 };
1040 if !index.is_empty() {
1043 continue;
1044 }
1045
1046 let mut first_node = Node::with_capacity(self.node_capacity);
1047 first_node.insert(value.clone());
1048
1049 #[cfg(feature = "cdc")]
1050 if EMIT_CDC {
1051 let node_insertion = ChangeEvent::CreateNode {
1052 event_id: self.event_id.fetch_add(1, Ordering::Relaxed).into(),
1055 max_value: value.clone(),
1056 };
1057 cdc.push(node_insertion);
1058 }
1059
1060 index.insert(value, Arc::new(RwLock::new(first_node)));
1061
1062 return Ok((None, cdc));
1063 }
1064 }
1065 };
1066
1067 let mut node_guard = target_node_entry.1.clone().write_arc();
1068
1069 #[allow(unused_assignments)]
1070 let mut operation = None;
1071 if !node_guard.need_to_split(self.node_capacity, &value) {
1072 let old_max = node_guard.max().cloned();
1073 let (inserted, idx) = NodeLike::insert(&mut *node_guard, value.clone());
1074 if inserted {
1075 #[cfg(feature = "cdc")]
1076 if EMIT_CDC {
1077 let node_element_insertion = ChangeEvent::InsertAt {
1078 event_id: self.event_id.fetch_add(1, Ordering::Relaxed).into(),
1081 max_value: old_max.clone().unwrap_or(value.clone()),
1082 value: value.clone(),
1083 index: idx,
1084 };
1085 cdc.push(node_element_insertion);
1086 }
1087
1088 if node_guard.max().cloned() == old_max {
1089 return Ok((None, cdc));
1090 }
1091
1092 operation = Some(Operation::UpdateMax(
1104 target_node_entry.1.clone(),
1105 target_node_entry.0.clone(),
1106 ));
1107 } else {
1108 return Err((node_guard, idx, old_max.unwrap()));
1109 }
1110 } else {
1111 operation = Some(Operation::Split(
1112 target_node_entry.1.clone(),
1113 target_node_entry.0.clone(),
1114 value.clone(),
1115 ));
1116 }
1117
1118 drop(node_guard);
1119 drop(index);
1120
1121 let op = operation.unwrap();
1122 let mut index = match &op {
1123 Operation::UpdateMax(_, _) => self.index.write_rekey(),
1124 Operation::Split(_, _, _) | Operation::MakeUnreachable(_, _) => self.index.write(),
1125 };
1126 match &op {
1127 Operation::Split(_, _, _) => {
1128 if let Ok((value, value_cdc)) = op.commit::<EMIT_CDC>(&mut index, adopt) {
1129 #[cfg(feature = "cdc")]
1130 if EMIT_CDC {
1131 for unassigned_event in value_cdc {
1132 let event_id = self.event_id.fetch_add(1, Ordering::Relaxed).into();
1133 cdc.push(unassigned_event.assign_id(event_id));
1134 }
1135 }
1136 return Ok((value, cdc));
1137 } else {
1138 continue;
1139 }
1140 }
1141 Operation::UpdateMax(_, _) => {
1142 return if let Ok((value, value_cdc)) = op.commit::<EMIT_CDC>(&mut index, adopt) {
1143 #[cfg(feature = "cdc")]
1144 if EMIT_CDC {
1145 for unassigned_event in value_cdc {
1146 let event_id = self.event_id.fetch_add(1, Ordering::Relaxed).into();
1147 cdc.push(unassigned_event.assign_id(event_id));
1148 }
1149 }
1150 Ok((value, cdc))
1151 } else {
1152 Ok((None, cdc))
1153 }
1154 }
1155 Operation::MakeUnreachable(_, _) => unreachable!(),
1156 }
1157 }
1158 }
1159 fn put_inner<const EMIT_CDC: bool>(&self, value: T, adopt: fn(&T, &mut T)) -> (Option<T>, Vec<ChangeEvent<T>>) {
1160 match self.put_checked_inner::<EMIT_CDC>(value.clone(), adopt) {
1161 Ok(res) => res,
1162 Err((mut node_guard, idx, max)) => {
1163 let mut value = value;
1167 if let Some(stored) = node_guard.get_ith(idx) {
1168 adopt(stored, &mut value);
1169 }
1170 let mut cdc = vec![];
1171 #[cfg(feature = "cdc")]
1172 if EMIT_CDC {
1173 if node_guard.len() == 1 {
1174 let node_removal = ChangeEvent::RemoveNode {
1175 event_id: self.event_id.fetch_add(1, Ordering::Relaxed).into(),
1178 max_value: max.clone(),
1179 };
1180 let node_insertion = ChangeEvent::CreateNode {
1181 event_id: self.event_id.fetch_add(1, Ordering::Relaxed).into(),
1183 max_value: value.clone(),
1184 };
1185 cdc.push(node_removal);
1186 cdc.push(node_insertion);
1187 } else if idx == node_guard.len() - 1 {
1188 let new_max = if node_guard.len() <= 1 {
1189 None
1190 } else {
1191 node_guard.get_ith(node_guard.len() - 2)
1192 };
1193 let node_element_removal = ChangeEvent::RemoveAt {
1194 event_id: self.event_id.fetch_add(1, Ordering::Relaxed).into(),
1197 max_value: max.clone(),
1198 value: value.clone(),
1199 index: idx,
1200 };
1201 let node_element_insertion = ChangeEvent::InsertAt {
1202 event_id: self.event_id.fetch_add(1, Ordering::Relaxed).into(),
1204 max_value: new_max.expect("length was checked so should be ok").clone(),
1205 value: value.clone(),
1206 index: idx,
1207 };
1208 cdc.push(node_element_removal);
1209 cdc.push(node_element_insertion);
1210 } else {
1211 let node_element_removal = ChangeEvent::RemoveAt {
1212 event_id: self.event_id.fetch_add(1, Ordering::Relaxed).into(),
1215 max_value: max.clone(),
1216 value: value.clone(),
1217 index: idx,
1218 };
1219 let node_element_insertion = ChangeEvent::InsertAt {
1220 event_id: self.event_id.fetch_add(1, Ordering::Relaxed).into(),
1222 max_value: max.clone(),
1223 value: value.clone(),
1224 index: idx,
1225 };
1226 cdc.push(node_element_removal);
1227 cdc.push(node_element_insertion);
1228 }
1229 }
1230
1231 (NodeLike::replace(&mut *node_guard, idx, value.clone()), cdc)
1232 }
1233 }
1234 }
1235
1236 pub(crate) fn put(&self, value: T) -> Option<T> {
1237 self.put_inner::<false>(value, no_identity_adoption).0
1238 }
1239
1240 #[cfg(feature = "multimap")]
1241 pub(crate) fn put_with(&self, value: T, adopt: fn(&T, &mut T)) -> Option<T> {
1242 self.put_inner::<false>(value, adopt).0
1243 }
1244
1245 #[allow(clippy::type_complexity)]
1246 pub(crate) fn put_checked(
1247 &self,
1248 value: T,
1249 ) -> Result<(Option<T>, Vec<ChangeEvent<T>>), (ArcRwLockWriteGuard<RawRwLock, Node>, usize, T)> {
1250 self.put_checked_inner::<false>(value, no_identity_adoption)
1251 }
1252
1253 pub(crate) fn put_cdc(&self, value: T) -> (Option<T>, Vec<ChangeEvent<T>>) {
1254 self.put_inner::<true>(value, no_identity_adoption)
1255 }
1256
1257 #[cfg(all(feature = "multimap", feature = "cdc"))]
1258 pub(crate) fn put_cdc_with(&self, value: T, adopt: fn(&T, &mut T)) -> (Option<T>, Vec<ChangeEvent<T>>) {
1259 self.put_inner::<true>(value, adopt)
1260 }
1261
1262 #[allow(clippy::type_complexity)]
1263 pub(crate) fn put_cdc_checked(
1264 &self,
1265 value: T,
1266 ) -> Result<(Option<T>, Vec<ChangeEvent<T>>), (ArcRwLockWriteGuard<RawRwLock, Node>, usize, T)> {
1267 self.put_checked_inner::<true>(value, no_identity_adoption)
1268 }
1269
1270 pub fn insert(&self, value: T) -> bool {
1291 self.put(value).is_none()
1292 }
1293 fn remove_inner<const EMIT_CDC: bool, Q>(&self, value: &Q) -> (Option<T>, Vec<ChangeEvent<T>>)
1296 where
1297 T: Borrow<Q>,
1298 Q: Ord + ?Sized,
1299 {
1300 let mut cdc = vec![];
1301 let index = self.index.read();
1302 if let Some(target_node_entry) =
1309 first_for_borrowed_bound(&index, Bound::Included(value), self.borrow_order_matches)
1310 .or_else(|| index.last_key_value())
1311 {
1312 let mut node_guard = target_node_entry.1.clone().write_arc();
1313 let old_max = node_guard.max().cloned();
1314 let deleted = NodeLike::delete(&mut *node_guard, value);
1315 if deleted.is_none() {
1316 return (None, cdc);
1317 }
1318 let (deleted, idx) = deleted.expect("should be ok as checked before");
1319
1320 let operation = if node_guard.len() > 0 {
1321 #[cfg(feature = "cdc")]
1322 if EMIT_CDC {
1323 let node_element_removal = ChangeEvent::RemoveAt {
1324 event_id: self.event_id.fetch_add(1, Ordering::Relaxed).into(),
1327 max_value: old_max.clone().expect("Max value should exist as Node is not empty"),
1328 value: deleted.clone(),
1329 index: idx,
1330 };
1331 cdc.push(node_element_removal);
1332 }
1333
1334 if old_max.as_ref() == node_guard.max() {
1335 return (Some(deleted), cdc);
1336 }
1337
1338 Some(Operation::UpdateMax(
1343 target_node_entry.1.clone(),
1344 target_node_entry.0.clone(),
1345 ))
1346 } else {
1347 Some(Operation::MakeUnreachable(
1348 target_node_entry.1.clone(),
1349 target_node_entry.0.clone(),
1350 ))
1351 };
1352
1353 drop(node_guard);
1354 drop(index);
1355
1356 let operation = operation.unwrap();
1357 let mut index = match &operation {
1358 Operation::UpdateMax(_, _) => self.index.write_rekey(),
1359 Operation::Split(_, _, _) | Operation::MakeUnreachable(_, _) => self.index.write(),
1360 };
1361
1362 return if let Ok((_, value_cdc)) = operation.commit::<EMIT_CDC>(&mut index, no_identity_adoption) {
1363 #[cfg(feature = "cdc")]
1364 if EMIT_CDC {
1365 for unassigned_event in value_cdc {
1366 let event_id = self.event_id.fetch_add(1, Ordering::Relaxed).into();
1367 cdc.push(unassigned_event.assign_id(event_id));
1368 }
1369 }
1370 (Some(deleted), cdc)
1371 } else {
1372 (Some(deleted), cdc)
1373 };
1374 }
1375
1376 (None, vec![])
1377 }
1378
1379 pub fn remove_cdc<Q>(&self, value: &Q) -> (Option<T>, Vec<ChangeEvent<T>>)
1380 where
1381 T: Borrow<Q>,
1382 Q: Ord + ?Sized,
1383 {
1384 self.remove_inner::<true, Q>(value)
1385 }
1386 pub fn remove<Q>(&self, value: &Q) -> Option<T>
1405 where
1406 T: Borrow<Q>,
1407 Q: Ord + ?Sized,
1408 {
1409 self.remove_inner::<false, Q>(value).0
1410 }
1411
1412 #[inline(always)]
1418 fn lock_node_for_value_optimistic<Q>(&self, value: &Q) -> Option<ArcRwLockReadGuard<RawRwLock, Node>>
1419 where
1420 T: Borrow<Q>,
1421 Q: Ord + ?Sized,
1422 {
1423 let node = {
1424 let index = self.index.read();
1425 match first_for_borrowed_bound(&index, Bound::Included(value), self.borrow_order_matches) {
1426 Some((_, node)) => Some(node.clone()),
1427 None => index
1428 .last_key_value()
1429 .map(|(_, node)| node.clone())
1430 .or_else(|| index.first_key_value().map(|(_, node)| node.clone())),
1431 }
1432 }?;
1433 Some(node.read_arc())
1434 }
1435
1436 #[inline(always)]
1443 fn lock_node_for_value<Q>(&self, value: &Q) -> Option<ArcRwLockReadGuard<RawRwLock, Node>>
1444 where
1445 T: Borrow<Q>,
1446 Q: Ord + ?Sized,
1447 {
1448 let mut retries = 0;
1449 let mut writer_spins = 0;
1450
1451 loop {
1452 let generation = self.index.generation.load(Ordering::Acquire);
1453 if !generation.is_multiple_of(2) {
1454 if writer_spins < ROOT_PUBLICATION_SPIN_LIMIT {
1455 writer_spins += 1;
1456 ::core::hint::spin_loop();
1457 } else {
1458 yield_now();
1459 }
1460 continue;
1461 }
1462 writer_spins = 0;
1463
1464 if retries >= STABLE_READ_BLOCKING_FALLBACK_AFTER {
1465 let index = self.index.read();
1469 let node = match first_for_borrowed_bound(&index, Bound::Included(value), self.borrow_order_matches) {
1470 Some((_, node)) => Some(node.clone()),
1471 None => index
1472 .last_key_value()
1473 .map(|(_, node)| node.clone())
1474 .or_else(|| index.first_key_value().map(|(_, node)| node.clone())),
1475 }?;
1476 let node_guard = node.read_arc();
1477 drop(index);
1478 return Some(node_guard);
1479 }
1480
1481 let pin = self.index.published.domain.pin();
1482 let snapshot = self.index.published.current.load(Ordering::Acquire);
1483 let index = unsafe { &*snapshot };
1486 let node =
1487 match first_published_for_borrowed_bound(index, Bound::Included(value), self.borrow_order_matches) {
1488 Some((_, node)) => Some(node.clone()),
1489 None => index
1490 .last_key_value()
1491 .map(|(_, node)| node.clone())
1492 .or_else(|| index.first_key_value().map(|(_, node)| node.clone())),
1493 };
1494 let Some(node) = node else {
1495 if self.index.generation.load(Ordering::Acquire) == generation {
1496 return None;
1497 }
1498 retries += 1;
1499 continue;
1500 };
1501
1502 drop(pin);
1506 let node_guard = node.read_arc();
1507 if self.index.generation.load(Ordering::Acquire) == generation {
1508 return Some(node_guard);
1509 }
1510 retries += 1;
1511 }
1512 }
1513
1514 #[inline(always)]
1515 fn get_with_guard<Q, R>(
1516 node_guard: ArcRwLockReadGuard<RawRwLock, Node>,
1517 value: &Q,
1518 read: impl FnOnce(&T) -> R,
1519 ) -> Option<R>
1520 where
1521 T: Borrow<Q>,
1522 Q: Ord + ?Sized,
1523 {
1524 let position = node_guard.try_select(value)?;
1525 node_guard.get_ith(position).map(read)
1526 }
1527
1528 #[inline(always)]
1529 pub(crate) fn get_with<Q, R>(&self, value: &Q, read: impl FnOnce(&T) -> R) -> Option<R>
1530 where
1531 T: Borrow<Q>,
1532 Q: Ord + ?Sized,
1533 {
1534 let mut retries = 0;
1535 let mut writer_spins = 0;
1536 let mut read = Some(read);
1537
1538 loop {
1539 let generation = self.index.generation.load(Ordering::Acquire);
1540 if !generation.is_multiple_of(2) {
1541 if writer_spins < ROOT_PUBLICATION_SPIN_LIMIT {
1542 writer_spins += 1;
1543 ::core::hint::spin_loop();
1544 } else {
1545 yield_now();
1546 }
1547 continue;
1548 }
1549 writer_spins = 0;
1550
1551 if retries >= STABLE_READ_BLOCKING_FALLBACK_AFTER {
1552 let index = self.index.read();
1553 let node = first_for_borrowed_bound(&index, Bound::Included(value), self.borrow_order_matches)
1554 .or_else(|| index.last_key_value())
1555 .or_else(|| index.first_key_value())
1556 .map(|(_, node)| node.clone())?;
1557 let node_guard = node.read_arc();
1558 drop(index);
1559 let position = node_guard.try_select(value)?;
1560 return node_guard
1561 .get_ith(position)
1562 .map(read.take().expect("read closure is consumed only on return"));
1563 }
1564
1565 let pin = self.index.published.domain.pin();
1566 let snapshot = self.index.published.current.load(Ordering::Acquire);
1567 let index = unsafe { &*snapshot };
1570 let node = first_published_for_borrowed_bound(index, Bound::Included(value), self.borrow_order_matches)
1571 .or_else(|| index.last_key_value())
1572 .or_else(|| index.first_key_value())
1573 .map(|(_, node)| node);
1574 let Some(node) = node else {
1575 if self.index.generation.load(Ordering::Acquire) == generation {
1576 return None;
1577 }
1578 retries += 1;
1579 continue;
1580 };
1581
1582 if let Some(node_guard) = node.try_read() {
1588 if self.index.generation.load(Ordering::Acquire) != generation {
1589 retries += 1;
1590 continue;
1591 }
1592 let position = node_guard.try_select(value)?;
1593 return node_guard
1594 .get_ith(position)
1595 .map(read.take().expect("read closure is consumed only on return"));
1596 }
1597
1598 let node = node.clone();
1599 drop(pin);
1600 let node_guard = node.read_arc();
1601 if self.index.generation.load(Ordering::Acquire) != generation {
1602 retries += 1;
1603 continue;
1604 }
1605 let position = node_guard.try_select(value)?;
1606 return node_guard
1607 .get_ith(position)
1608 .map(read.take().expect("read closure is consumed only on return"));
1609 }
1610 }
1611
1612 #[inline(always)]
1613 pub(crate) fn get_with_optimistic<Q, R>(&self, value: &Q, read: impl FnOnce(&T) -> R) -> Option<R>
1614 where
1615 T: Borrow<Q>,
1616 Q: Ord + ?Sized,
1617 {
1618 Self::get_with_guard(self.lock_node_for_value_optimistic(value)?, value, read)
1619 }
1620
1621 pub fn contains<Q>(&self, value: &Q) -> bool
1637 where
1638 T: Borrow<Q>,
1639 Q: Ord + ?Sized,
1640 {
1641 self.get_with(value, |_| ()).is_some()
1642 }
1643 pub fn get<'a, Q>(&'a self, value: &'a Q) -> Option<Ref<T, Node>>
1644 where
1645 T: Borrow<Q>,
1646 Q: Ord + ?Sized,
1647 {
1648 if let Some(node_guard) = self.lock_node_for_value(value) {
1649 let potential_position = node_guard.try_select(value);
1650
1651 if let Some(position) = potential_position {
1652 return Some(Ref {
1653 node_guard,
1654 position,
1655 phantom_data: PhantomData,
1656 });
1657 }
1658 }
1659
1660 None
1661 }
1662
1663 pub fn len(&self) -> usize {
1664 self.index.read().values().map(|node| node.read().len()).sum()
1665 }
1666 pub fn is_empty(&self) -> bool {
1667 self.index.read().values().all(|node| node.read().is_empty())
1668 }
1669 pub fn capacity(&self) -> usize {
1670 self.index
1671 .read()
1672 .values()
1673 .map(|node| {
1674 let guard = node.read();
1675 guard.capacity()
1676 })
1677 .sum()
1678 }
1679 pub fn node_count(&self) -> usize {
1680 self.index.read().len()
1681 }
1682}
1683
1684impl<T> FromIterator<T> for BTreeSet<T>
1685where
1686 T: Debug + Ord + Clone + Send,
1687{
1688 fn from_iter<K: IntoIterator<Item = T>>(iter: K) -> Self {
1689 let btree = BTreeSet::new();
1690 iter.into_iter().for_each(|item| {
1691 btree.insert(item);
1692 });
1693
1694 btree
1695 }
1696}
1697
1698impl<T, const N: usize> From<[T; N]> for BTreeSet<T>
1699where
1700 T: Debug + Ord + Clone + Send,
1701{
1702 fn from(value: [T; N]) -> Self {
1703 let btree: BTreeSet<T> = Default::default();
1704
1705 value.into_iter().for_each(|item| {
1706 btree.insert(item);
1707 });
1708
1709 btree
1710 }
1711}
1712
1713pub struct Iter<'a, T, Node>
1727where
1728 T: Debug + Ord + Clone + Send + 'static,
1729 Node: NodeLike<T> + Send + 'static,
1730{
1731 tree: &'a BTreeSet<T, Node>,
1732 current_front_batch: Option<alloc::vec::IntoIter<T>>,
1733 current_back_batch: Option<alloc::vec::IntoIter<T>>,
1734 exhausted_front_node: Option<Arc<RwLock<Node>>>,
1738 exhausted_back_node: Option<Arc<RwLock<Node>>>,
1739 front_partial: Option<(Arc<RwLock<Node>>, usize)>,
1746 back_partial: Option<(Arc<RwLock<Node>>, usize)>,
1747 front_batch_limit: usize,
1749 back_batch_limit: usize,
1750 current_front_value: Option<T>,
1751 current_back_value: Option<T>,
1752 met: bool,
1753}
1754
1755const INITIAL_BATCH: usize = 4;
1762
1763const MAX_BATCH: usize = 4096;
1766
1767impl<'a, T, Node> Iter<'a, T, Node>
1768where
1769 T: Debug + Ord + Clone + Send + 'static,
1770 Node: NodeLike<T> + Send + 'static,
1771{
1772 pub fn new(btree: &'a BTreeSet<T, Node>) -> Self {
1773 Self {
1778 tree: btree,
1779 current_front_batch: None,
1780 current_back_batch: None,
1781 exhausted_front_node: None,
1782 exhausted_back_node: None,
1783 front_partial: None,
1784 back_partial: None,
1785 front_batch_limit: INITIAL_BATCH,
1786 back_batch_limit: INITIAL_BATCH,
1787 current_front_value: None,
1788 current_back_value: None,
1789 met: false,
1790 }
1791 }
1792
1793 fn install_front_batch(&mut self) -> bool {
1818 let index = self.tree.index.read();
1819 let candidate = match self.current_front_value.as_ref() {
1820 Some(last_yielded) => index.range((Bound::Excluded(last_yielded), Bound::Unbounded)).next(),
1821 None => index.first_key_value(),
1822 };
1823 let entry = match (candidate, self.exhausted_front_node.as_ref()) {
1829 (Some((key, node)), Some(exhausted)) if Arc::ptr_eq(node, exhausted) => {
1830 index.range((Bound::Excluded(key), Bound::Unbounded)).next()
1831 }
1832 (candidate, _) => candidate,
1833 };
1834 let Some((_, entry)) = entry else {
1835 return false;
1836 };
1837 let node = entry.clone();
1838 let guard = node.read_arc();
1839 drop(index);
1840
1841 let rank_skip = self
1842 .current_front_value
1843 .as_ref()
1844 .and_then(|value| guard.rank(Bound::Excluded(value), true))
1845 .map_or(0, |rank| rank + 1);
1846 let partial_skip = match self.front_partial.as_ref() {
1866 Some((partial, taken)) if Arc::ptr_eq(partial, &node) => *taken,
1867 _ => 0,
1868 };
1869 let skip = if self.current_front_value.is_some() {
1870 rank_skip
1871 } else {
1872 partial_skip
1873 };
1874
1875 let available = guard.len().saturating_sub(skip);
1881 let take = available.min(self.front_batch_limit);
1882 let batch = guard.iter().skip(skip).take(take).cloned().collect::<Vec<_>>();
1883 drop(guard);
1884
1885 if take == available {
1886 self.exhausted_front_node = Some(node);
1888 self.front_partial = None;
1889 } else {
1890 self.exhausted_front_node = None;
1893 self.front_partial = Some((node, skip + take));
1894 }
1895 self.front_batch_limit = self.front_batch_limit.saturating_mul(2).min(MAX_BATCH);
1896 self.current_front_batch = Some(batch.into_iter());
1897 true
1898 }
1899
1900 fn install_back_batch(&mut self) -> bool {
1904 let index = self.tree.index.read();
1905 let candidate = match self.current_back_value.as_ref() {
1906 Some(last_yielded) => index
1907 .range((Bound::Included(last_yielded), Bound::Unbounded))
1908 .next()
1909 .or_else(|| index.last_key_value()),
1910 None => index.last_key_value(),
1911 };
1912 let entry = match (candidate, self.exhausted_back_node.as_ref()) {
1913 (Some((key, node)), Some(exhausted)) if Arc::ptr_eq(node, exhausted) => index.range(..key).next_back(),
1914 (candidate, _) => candidate,
1915 };
1916 let Some((_, entry)) = entry else {
1917 return false;
1918 };
1919 let node = entry.clone();
1920 let guard = node.read_arc();
1921 drop(index);
1922
1923 let truncate = self
1924 .current_back_value
1925 .as_ref()
1926 .and_then(|value| guard.rank(Bound::Excluded(value), false))
1927 .map_or(0, |rank| rank + 1);
1928 let partial_truncate = match self.back_partial.as_ref() {
1935 Some((partial, taken)) if Arc::ptr_eq(partial, &node) => *taken,
1936 _ => 0,
1937 };
1938 let truncate = if self.current_back_value.is_some() {
1939 truncate
1940 } else {
1941 partial_truncate
1942 };
1943 let available = guard.len().saturating_sub(truncate);
1944 let take = available.min(self.back_batch_limit);
1945 let skip = available - take;
1948 let batch = guard.iter().skip(skip).take(take).cloned().collect::<Vec<_>>();
1949 drop(guard);
1950
1951 if take == available {
1952 self.exhausted_back_node = Some(node);
1953 self.back_partial = None;
1954 } else {
1955 self.exhausted_back_node = None;
1956 self.back_partial = Some((node, truncate + take));
1957 }
1958 self.back_batch_limit = self.back_batch_limit.saturating_mul(2).min(MAX_BATCH);
1959 self.current_back_batch = Some(batch.into_iter());
1960 true
1961 }
1962}
1963
1964impl<'a, T, Node> Iterator for Iter<'a, T, Node>
1965where
1966 T: Debug + Ord + Clone + Send + 'static,
1967 Node: NodeLike<T> + Send + 'static,
1968{
1969 type Item = T;
1970
1971 fn next(&mut self) -> Option<Self::Item> {
1972 loop {
1973 if self.met {
1974 return None;
1975 }
1976
1977 if self.current_front_batch.is_none() && !self.install_front_batch() {
1978 return None;
1979 }
1980
1981 let batch = self.current_front_batch.as_mut().expect("installed above");
1982 if let Some(value) = batch.next() {
1983 if let Some(current_front_value) = self.current_front_value.as_ref() {
1989 if value.le(current_front_value) {
1990 continue;
1991 }
1992 }
1993 if let Some(current_back_value) = self.current_back_value.as_ref() {
1994 if value.ge(current_back_value) {
1995 self.met = true;
1996 return None;
1997 }
1998 }
1999 self.current_front_value = Some(value.clone());
2000 return Some(value);
2001 } else {
2002 self.current_front_batch = None;
2003 }
2004 }
2005 }
2006}
2007
2008impl<'a, T, Node> DoubleEndedIterator for Iter<'a, T, Node>
2009where
2010 T: Debug + Ord + Clone + Send + 'static,
2011 Node: NodeLike<T> + Send + 'static,
2012{
2013 fn next_back(&mut self) -> Option<Self::Item> {
2014 loop {
2015 if self.met {
2016 return None;
2017 }
2018
2019 if self.current_back_batch.is_none() && !self.install_back_batch() {
2020 return None;
2021 }
2022
2023 let batch = self.current_back_batch.as_mut().expect("installed above");
2024 if let Some(value) = batch.next_back() {
2025 if let Some(current_back_value) = self.current_back_value.as_ref() {
2029 if value.ge(current_back_value) {
2030 continue;
2031 }
2032 }
2033 if let Some(current_front_value) = self.current_front_value.as_ref() {
2034 if value.le(current_front_value) {
2035 self.met = true;
2036 return None;
2037 }
2038 }
2039 self.current_back_value = Some(value.clone());
2040 return Some(value);
2041 } else {
2042 self.current_back_batch = None;
2043 }
2044 }
2045 }
2046}
2047
2048impl<'a, T: Debug + Ord + Clone + Send, Node: NodeLike<T> + Send + 'static> FusedIterator for Iter<'a, T, Node> {}
2049
2050impl<'a, T, Node> IntoIterator for &'a BTreeSet<T, Node>
2051where
2052 T: Debug + Ord + Send + Clone,
2053 Node: NodeLike<T> + Send + 'static,
2054{
2055 type Item = T;
2056
2057 type IntoIter = Iter<'a, T, Node>;
2058
2059 fn into_iter(self) -> Self::IntoIter {
2060 Iter::new(self)
2061 }
2062}
2063
2064pub struct Range<'a, T, Node>
2067where
2068 T: Debug + Ord + Clone + Send + 'static,
2069 Node: NodeLike<T> + Send + 'static,
2070{
2071 iter: Iter<'a, T, Node>,
2072}
2073
2074impl<'a, T, Node> Range<'a, T, Node>
2075where
2076 T: Debug + Ord + Clone + Send + 'static,
2077 Node: NodeLike<T> + Send + 'static,
2078{
2079 pub fn new<Q, R>(btree: &'a BTreeSet<T, Node>, range: R) -> Self
2080 where
2081 T: Borrow<Q>,
2082 Q: Ord + ?Sized,
2083 R: RangeBounds<Q>,
2084 {
2085 let index = btree.index.read();
2086
2087 let start_bound = range.start_bound();
2088 let end_bound = range.end_bound();
2089 let mut met = match (start_bound, end_bound) {
2090 (Bound::Included(start), Bound::Included(end)) => start > end,
2091 (Bound::Included(start), Bound::Excluded(end))
2092 | (Bound::Excluded(start), Bound::Included(end))
2093 | (Bound::Excluded(start), Bound::Excluded(end)) => start >= end,
2094 _ => false,
2095 };
2096
2097 let current_front_entry = first_for_borrowed_bound(&index, start_bound, btree.borrow_order_matches);
2098
2099 let front_value = if let Some((front_key, front_node)) = current_front_entry {
2100 let front_guard = front_node.clone().read_arc();
2101 let rank = match start_bound {
2102 Bound::Included(v) => front_guard.rank(Bound::Included(v), true),
2103 Bound::Excluded(v) => front_guard.rank(Bound::Excluded(v), true),
2104 Bound::Unbounded => None,
2105 };
2106 if let Some(rank) = rank {
2107 let value = front_guard.iter().nth(rank).cloned();
2108 drop(front_guard);
2109
2110 value
2111 } else {
2112 drop(front_guard);
2118 if let Some((_, pre_front_node)) = index.range::<T, _>(..front_key).next_back() {
2119 let pre_front_guard = pre_front_node.clone().read_arc();
2120 pre_front_guard.iter().last().cloned()
2121 } else {
2122 None
2123 }
2124 }
2125 } else {
2126 None
2127 };
2128
2129 let current_back_entry = match end_bound {
2130 Bound::Included(end) | Bound::Excluded(end) => {
2131 node_for_borrowed_end(&index, end, btree.borrow_order_matches)
2132 }
2133 Bound::Unbounded => index.last_key_value(),
2134 };
2135
2136 let back_value = if let Some((back_key, back_node)) = current_back_entry {
2137 let back_guard = back_node.clone().read_arc();
2138 let rank = match end_bound {
2139 Bound::Included(v) => back_guard.rank(Bound::Included(v), false),
2140 Bound::Excluded(v) => back_guard.rank(Bound::Excluded(v), false),
2141 Bound::Unbounded => None,
2142 };
2143 if let Some(rank) = rank {
2144 let value = back_guard.iter().nth_back(rank).cloned();
2145 drop(back_guard);
2146
2147 value
2148 } else {
2149 drop(back_guard);
2151 if let Some((_, next_back_node)) = index
2152 .range::<T, _>((Bound::Excluded(back_key), Bound::Unbounded))
2153 .next()
2154 {
2155 let next_back_guard = next_back_node.clone().read_arc();
2156 next_back_guard.iter().next().cloned()
2157 } else {
2158 None
2159 }
2160 }
2161 } else {
2162 None
2163 };
2164
2165 if front_value.is_none() && back_value.is_none() {
2166 if start_bound != Bound::Unbounded || end_bound != Bound::Unbounded {
2168 if let Some(max) = index
2169 .last_key_value()
2170 .and_then(|(_, node)| node.clone().read_arc().max().cloned())
2171 {
2172 if let Bound::Included(v) = start_bound {
2173 if v > max.borrow() {
2174 met = true;
2175 }
2176 } else if let Bound::Excluded(v) = start_bound {
2177 if v >= max.borrow() {
2178 met = true;
2179 }
2180 }
2181 }
2182
2183 if let Some(min) = index
2184 .first_key_value()
2185 .and_then(|(_, node)| node.clone().read_arc().min().cloned())
2186 {
2187 if let Bound::Included(v) = end_bound {
2188 if v < min.borrow() {
2189 met = true;
2190 }
2191 } else if let Bound::Excluded(v) = end_bound {
2192 if v <= min.borrow() {
2193 met = true;
2194 }
2195 }
2196 }
2197 }
2198 }
2199
2200 Self {
2205 iter: Iter {
2206 tree: btree,
2207 current_front_batch: None,
2208 current_back_batch: None,
2209 exhausted_front_node: None,
2210 exhausted_back_node: None,
2211 front_partial: None,
2212 back_partial: None,
2213 front_batch_limit: INITIAL_BATCH,
2214 back_batch_limit: INITIAL_BATCH,
2215 current_front_value: front_value,
2216 current_back_value: back_value,
2217 met,
2218 },
2219 }
2220 }
2221}
2222
2223impl<'a, T, Node> Iterator for Range<'a, T, Node>
2224where
2225 T: Debug + Ord + Clone + Send + 'static,
2226 Node: NodeLike<T> + Send + 'static,
2227{
2228 type Item = T;
2229
2230 fn next(&mut self) -> Option<Self::Item> {
2231 self.iter.next()
2232 }
2233}
2234
2235impl<'a, T, Node> DoubleEndedIterator for Range<'a, T, Node>
2236where
2237 T: Debug + Ord + Clone + Send + 'static,
2238 Node: NodeLike<T> + Send + 'static,
2239{
2240 fn next_back(&mut self) -> Option<Self::Item> {
2241 self.iter.next_back()
2242 }
2243}
2244
2245impl<'a, T, Node> FusedIterator for Range<'a, T, Node>
2246where
2247 T: Debug + Ord + Clone + Send + 'static,
2248 Node: NodeLike<T> + Send + 'static,
2249{
2250}
2251
2252impl<'a, T, Node> BTreeSet<T, Node>
2253where
2254 T: Debug + Ord + Clone + Send + 'static,
2255 Node: NodeLike<T> + Send + 'static,
2256{
2257 pub fn iter(&'a self) -> Iter<'a, T, Node> {
2290 Iter::new(self)
2291 }
2292 pub fn range<Q, R>(&'a self, range: R) -> Range<'a, T, Node>
2320 where
2321 T: Borrow<Q>,
2322 Q: Ord + ?Sized,
2323 R: RangeBounds<Q>,
2324 {
2325 Range::new(self, range)
2326 }
2327}
2328
2329impl<T> BTreeSet<T>
2330where
2331 T: Debug + Ord + Clone + Send + 'static,
2332{
2333 pub fn remove_range<R, Q>(&self, range: R)
2334 where
2335 Q: Ord + ?Sized,
2336 T: Borrow<Q>,
2337 R: RangeBounds<Q>,
2338 {
2339 let mut detached_nodes = Vec::new();
2342 let mut index = self.index.write();
2343
2344 let start_bound = range.start_bound();
2345 let end_bound = range.end_bound();
2346
2347 let Some((front_key, front_node)) = first_for_borrowed_bound(&index, start_bound, self.borrow_order_matches)
2350 else {
2351 return;
2352 };
2353 let front_key = front_key.clone();
2354 let front_node = front_node.clone();
2355
2356 let back_entry = match end_bound {
2362 Bound::Included(end) | Bound::Excluded(end) => {
2363 node_for_borrowed_end(&index, end, self.borrow_order_matches)
2364 }
2365 Bound::Unbounded => index.last_key_value(),
2366 };
2367 let Some((back_key, back_node)) = back_entry else {
2368 return;
2369 };
2370 let back_key = back_key.clone();
2371 let back_node = back_node.clone();
2372 if back_key < front_key {
2373 return;
2375 }
2376
2377 let removed_prefix_len = |guard: &Vec<T>| -> usize {
2380 match end_bound {
2381 Bound::Included(end) => guard.rank(Bound::Excluded(end), true).map_or(0, |last| last + 1),
2382 Bound::Excluded(end) => guard.rank(Bound::Included(end), true).map_or(0, |last| last + 1),
2383 Bound::Unbounded => guard.len(),
2384 }
2385 };
2386
2387 if Arc::ptr_eq(&front_node, &back_node) {
2388 let mut guard = front_node.clone().write_arc();
2390 let front_position = guard.rank(start_bound, true).map_or(0, |last| last + 1);
2391 let back_position = removed_prefix_len(&guard);
2392 if back_position <= front_position {
2393 return;
2394 }
2395
2396 let original_len = guard.len();
2397 guard.drain(front_position..back_position);
2398 if back_position == original_len {
2399 index.remove::<T>(&front_key);
2402 if let Some(new_max) = guard.last().cloned() {
2403 index.insert(new_max, front_node);
2404 }
2405 }
2406 return;
2407 }
2408
2409 let mut front_guard = front_node.clone().write_arc();
2410 let mut back_guard = back_node.clone().write_arc();
2411 let front_position = front_guard.rank(start_bound, true).map_or(0, |last| last + 1);
2412 let back_position = removed_prefix_len(&back_guard);
2413
2414 let middle_keys = index
2416 .range::<T, _>((Bound::Excluded(&front_key), Bound::Excluded(&back_key)))
2417 .map(|(key, _)| key.clone())
2418 .collect::<Vec<_>>();
2419 for key in middle_keys {
2420 let node = index
2421 .remove::<T>(&key)
2422 .expect("middle key was collected under the write lock");
2423 let mut removed_node = node.write_arc();
2424 detached_nodes.push(::core::mem::take(&mut *removed_node));
2425 }
2426
2427 index.remove::<T>(&front_key);
2430 front_guard.drain(front_position..);
2431 if !front_guard.is_empty() {
2432 let new_front_max = front_guard.last().unwrap().clone();
2433 index.insert(new_front_max, front_node);
2434 }
2435
2436 if back_position >= back_guard.len() {
2439 index.remove::<T>(&back_key);
2440 back_guard.drain(..);
2441 } else if back_position > 0 {
2442 back_guard.drain(..back_position);
2443 }
2444 }
2445}
2446
2447#[cfg(test)]
2448mod tests {
2449 use crate::concurrent::operation::Operation;
2450 use crate::concurrent::set::{BTreeSet, Iter, DEFAULT_INNER_SIZE, INITIAL_BATCH};
2451 use crate::core::node::NodeLike;
2452 use parking_lot::RwLock;
2453 use rand::Rng;
2454 use std::collections::HashSet;
2455 use std::ops::Bound::{self, Included};
2456 use std::sync::mpsc;
2457 use std::sync::{Arc, Barrier, Mutex};
2458 use std::thread;
2459 use std::time::Duration;
2460
2461 #[test]
2463 fn test_node_size_two_preserves_all_u64_values() {
2464 let set = BTreeSet::<u64>::with_maximum_node_size(2);
2465
2466 for value in 0..10_u64 {
2467 set.insert(value);
2468 }
2469
2470 assert_eq!(set.iter().collect::<Vec<_>>(), (0..10).collect::<Vec<_>>());
2471 }
2472
2473 #[test]
2475 fn test_node_size_three_preserves_all_u8_values() {
2476 let set = BTreeSet::<u8>::with_maximum_node_size(3);
2477
2478 for value in 0..20_u8 {
2479 set.insert(value);
2480 }
2481
2482 assert_eq!(set.iter().collect::<Vec<_>>(), (0..20).collect::<Vec<_>>());
2483 }
2484
2485 #[test]
2486 fn concurrent_first_writers_preserve_disjoint_ranges() {
2487 const WRITERS: u64 = 8;
2488 const VALUES_PER_WRITER: u64 = 1_000;
2489
2490 let set = Arc::new(BTreeSet::<u64>::new());
2491 let start = Arc::new(Barrier::new(WRITERS as usize));
2492 let handles = (0..WRITERS)
2493 .map(|writer| {
2494 let set = Arc::clone(&set);
2495 let start = Arc::clone(&start);
2496 thread::spawn(move || {
2497 start.wait();
2498 let first = writer * VALUES_PER_WRITER;
2499 for value in first..first + VALUES_PER_WRITER {
2500 assert!(set.insert(value));
2501 }
2502 })
2503 })
2504 .collect::<Vec<_>>();
2505
2506 for handle in handles {
2507 handle.join().unwrap();
2508 }
2509
2510 let expected = (0..WRITERS * VALUES_PER_WRITER).collect::<Vec<_>>();
2511 assert_eq!(set.len(), expected.len());
2512 assert_eq!(set.iter().collect::<Vec<_>>(), expected);
2513 }
2514
2515 #[test]
2516 fn published_point_reads_remain_definitive_across_splits() {
2517 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
2518
2519 const STABLE_KEYS: usize = 256;
2520 const FINAL_KEYS: usize = 4_096;
2521 const READERS: usize = 4;
2522
2523 let set = Arc::new(BTreeSet::<usize>::with_maximum_node_size(8));
2524 for key in 0..STABLE_KEYS {
2525 set.insert(key);
2526 }
2527
2528 let start = Arc::new(Barrier::new(READERS + 1));
2529 let done = Arc::new(AtomicBool::new(false));
2530 let published_up_to = Arc::new(AtomicUsize::new(STABLE_KEYS - 1));
2531 let readers = (0..READERS)
2532 .map(|reader| {
2533 let set = Arc::clone(&set);
2534 let start = Arc::clone(&start);
2535 let done = Arc::clone(&done);
2536 let published_up_to = Arc::clone(&published_up_to);
2537 thread::spawn(move || {
2538 start.wait();
2539 let mut probe = reader;
2540 while !done.load(Ordering::Acquire) {
2541 let key = probe % STABLE_KEYS;
2542 assert_eq!(set.get_with(&key, |value| *value), Some(key));
2543 let newest_acknowledged = published_up_to.load(Ordering::Acquire);
2544 assert_eq!(
2545 set.get_with(&newest_acknowledged, |value| *value),
2546 Some(newest_acknowledged),
2547 "an acknowledged insert disappeared from the published route"
2548 );
2549 probe += READERS;
2550 }
2551 })
2552 })
2553 .collect::<Vec<_>>();
2554
2555 start.wait();
2556 for key in STABLE_KEYS..FINAL_KEYS {
2557 set.insert(key);
2558 published_up_to.store(key, Ordering::Release);
2559 }
2560 done.store(true, Ordering::Release);
2561
2562 for reader in readers {
2563 reader.join().unwrap();
2564 }
2565 assert_eq!(set.len(), FINAL_KEYS);
2566 for key in 0..FINAL_KEYS {
2567 assert_eq!(set.get_with(&key, |value| *value), Some(key));
2568 }
2569 }
2570
2571 #[test]
2572 fn published_pointer_survives_reclamation_interleavings() {
2573 use std::sync::atomic::{AtomicBool, Ordering};
2574
2575 let set = Arc::new(BTreeSet::<usize>::with_maximum_node_size(2));
2576 for key in 0..8 {
2577 set.insert(key);
2578 }
2579 let start = Arc::new(Barrier::new(2));
2580 let done = Arc::new(AtomicBool::new(false));
2581
2582 let reader_set = Arc::clone(&set);
2583 let reader_start = Arc::clone(&start);
2584 let reader_done = Arc::clone(&done);
2585 let reader = thread::spawn(move || {
2586 reader_start.wait();
2587 let mut probe = 0;
2588 while !reader_done.load(Ordering::Acquire) {
2589 let key = probe % 8;
2590 assert_eq!(reader_set.get_with(&key, |value| *value), Some(key));
2591 probe += 1;
2592 }
2593 });
2594
2595 start.wait();
2596 for key in 8..80 {
2597 set.insert(key);
2598 }
2599 for key in 8..80 {
2600 assert_eq!(set.remove(&key), Some(key));
2601 }
2602 done.store(true, Ordering::Release);
2603 reader.join().unwrap();
2604
2605 for key in 0..8 {
2606 assert_eq!(set.get_with(&key, |value| *value), Some(key));
2607 }
2608 }
2609
2610 #[test]
2611 fn published_route_chunks_split_and_merge_without_losing_keys() {
2612 let set = BTreeSet::<usize>::with_maximum_node_size(2);
2613 for key in 0..600 {
2614 assert!(set.insert(key));
2615 }
2616 for key in (0..600).step_by(2) {
2617 assert_eq!(set.remove(&key), Some(key));
2618 }
2619 for key in 0..600 {
2620 assert_eq!(set.contains(&key), key % 2 == 1, "probe {key}");
2621 }
2622 for key in (1..600).step_by(2) {
2623 assert_eq!(set.remove(&key), Some(key));
2624 }
2625 assert!(set.is_empty());
2626 }
2627
2628 #[test]
2629 fn mixed_structural_and_node_lock_paths_complete_without_deadlock() {
2630 const THREADS: usize = 8;
2631 const OPERATIONS: usize = 1_000;
2632
2633 let set = Arc::new(BTreeSet::<usize>::with_maximum_node_size(8));
2634 for value in 0..256 {
2635 set.insert(value);
2636 }
2637
2638 let start = Arc::new(Barrier::new(THREADS));
2639 let (done_tx, done_rx) = mpsc::channel();
2640 let handles = (0..THREADS)
2641 .map(|worker| {
2642 let set = Arc::clone(&set);
2643 let start = Arc::clone(&start);
2644 let done_tx = done_tx.clone();
2645 thread::spawn(move || {
2646 start.wait();
2647 for operation in 0..OPERATIONS {
2648 let value = (operation * 17 + worker * 31) % 512;
2649 match (operation + worker) % 5 {
2650 0 => {
2651 set.insert(value);
2652 }
2653 1 => {
2654 set.remove(&value);
2655 }
2656 2 => {
2657 let _ = set.contains(&value);
2658 }
2659 3 => {
2660 let _ = set.get_with(&value, Clone::clone);
2661 }
2662 _ => {
2663 set.remove_range(value..=value);
2664 set.insert(value);
2665 }
2666 }
2667 }
2668 done_tx.send(()).unwrap();
2669 })
2670 })
2671 .collect::<Vec<_>>();
2672 drop(done_tx);
2673
2674 for _ in 0..THREADS {
2675 done_rx
2676 .recv_timeout(Duration::from_secs(10))
2677 .expect("mixed structural/node-lock workload did not complete");
2678 }
2679 for handle in handles {
2680 handle.join().unwrap();
2681 }
2682 }
2683
2684 #[test]
2685 fn test_concurrent_insert() {
2686 let set = Arc::new(BTreeSet::<i32>::new());
2687 let num_threads = 128;
2688 let operations_per_thread = 10000;
2689 let mut handles = vec![];
2690
2691 let test_data: Vec<Vec<(i32, i32)>> = (0..num_threads)
2692 .map(|_| {
2693 let mut rng = rand::rng();
2694 (0..operations_per_thread)
2695 .map(|_| {
2696 let value = rng.random_range(0..100000);
2697 let operation = rng.random_range(0..2);
2698 (operation, value)
2699 })
2700 .collect()
2701 })
2702 .collect();
2703
2704 let expected_values = Arc::new(Mutex::new(HashSet::new()));
2705
2706 for thread_idx in 0..num_threads {
2707 let set_clone = Arc::clone(&set);
2708 let expected_values = Arc::clone(&expected_values);
2709 let thread_data = test_data[thread_idx].clone();
2710
2711 let handle = thread::spawn(move || {
2712 for (operation, value) in thread_data {
2713 if operation == 0 {
2714 let _a = set_clone.insert(value);
2715 expected_values.lock().unwrap().insert(value);
2716 }
2717 }
2718 });
2719 handles.push(handle);
2720 }
2721
2722 for handle in handles {
2723 handle.join().unwrap();
2724 }
2725
2726 let expected_values = expected_values.lock().unwrap();
2727 assert_eq!(set.len(), expected_values.len());
2728
2729 for value in expected_values.iter() {
2730 assert!(set.contains(value));
2731 }
2732 }
2733
2734 #[test]
2735 fn test_insert_desc() {
2736 let set = Arc::new(BTreeSet::<i32>::new());
2737
2738 assert!(set.insert(2));
2739 assert!(set.insert(1));
2740 }
2741
2742 #[test]
2743 fn test_insert_st() {
2744 let set = Arc::new(BTreeSet::<i32>::new());
2745 let mut rng = rand::rng();
2746
2747 let n = 2048 * 100;
2748 let range = 0..n;
2749 let mut inserted_values = HashSet::new();
2750 for _ in range {
2751 let value = rng.random_range(0..n);
2752 if inserted_values.insert(value) {
2753 set.insert(value);
2754 }
2755 }
2756
2757 assert_eq!(
2758 set.len(),
2759 inserted_values.len(),
2760 "Length did not match, missing: {:?}",
2761 set.index
2762 .read()
2763 .values()
2764 .flat_map(|node| node.read().iter().cloned().collect::<Vec<_>>())
2765 .collect::<HashSet<_>>()
2766 .symmetric_difference(&inserted_values)
2767 .collect::<Vec<_>>()
2768 );
2769 for i in inserted_values {
2770 assert!(
2771 set.contains(&i),
2772 "Did not find: {} with index: {:?}",
2773 i,
2774 set.index.read().keys().cloned().collect::<Vec<_>>(),
2775 );
2776 }
2777 }
2778
2779 #[test]
2780 fn test_single_element() {
2781 let set = BTreeSet::<i32>::new();
2782 set.insert(1);
2783 let mut iter = set.into_iter();
2784 assert_eq!(iter.next(), Some(1));
2785 assert_eq!(iter.next(), None);
2786 assert_eq!(iter.next_back(), None);
2787 }
2788
2789 #[test]
2790 fn test_multiple_elements() {
2791 let set = BTreeSet::<i32>::new();
2792 set.insert(1);
2793 set.insert(2);
2794 set.insert(3);
2795 let mut iter = set.into_iter();
2796 assert_eq!(iter.next(), Some(1));
2797 assert_eq!(iter.next_back(), Some(3));
2798 assert_eq!(iter.next(), Some(2));
2799 assert_eq!(iter.next(), None);
2800 assert_eq!(iter.next_back(), None);
2801 }
2802
2803 #[test]
2804 fn test_bidirectional_iteration() {
2805 let set = BTreeSet::<i32>::with_maximum_node_size(3);
2806 for i in 1..=20 {
2807 set.insert(i);
2808 }
2809 let mut iter = set.into_iter();
2810 for i in 0..10 {
2811 let tree = set.index.read().keys().cloned().collect::<Vec<_>>();
2813
2814 let expected_next = i + 1;
2815 let actual_next = iter.next();
2816 assert_eq!(actual_next, Some(expected_next), "Tree: {:?}", tree);
2817
2818 let expected_next_back = 20 - i;
2819 let actual_next_back = iter.next_back();
2820 assert_eq!(actual_next_back, Some(expected_next_back), "Tree: {:?}", tree);
2821 }
2822 assert_eq!(iter.next(), None);
2823 assert_eq!(iter.next_back(), None);
2824 }
2825
2826 #[test]
2827 fn test_fused_iterator() {
2828 let set = BTreeSet::<i32>::new();
2829 set.insert(1);
2830 let mut iter = set.into_iter();
2831 assert_eq!(iter.next(), Some(1));
2832 assert_eq!(iter.next(), None);
2833 assert_eq!(iter.next(), None);
2834 }
2835
2836 #[test]
2837 fn test_fused_iterator_back() {
2838 let set = BTreeSet::<i32>::new();
2839 set.insert(1);
2840 let mut iter = set.into_iter();
2841 assert_eq!(iter.next_back(), Some(1));
2842 assert_eq!(iter.next_back(), None);
2843 assert_eq!(iter.next_back(), None);
2844 }
2845
2846 #[test]
2847 fn test_out_of_bounds_range() {
2848 let btree: BTreeSet<usize> = BTreeSet::from_iter(0..10);
2849 assert_eq!(btree.range((Included(5), Included(10))).count(), 5);
2850 assert_eq!(btree.range((Included(5), Included(11))).count(), 5);
2851 assert_eq!(btree.range((Included(5), Included(10 + DEFAULT_INNER_SIZE))).count(), 5);
2852 assert_eq!(btree.range((Included(0), Included(11))).count(), 10);
2853 }
2854
2855 #[test]
2856 fn test_iterating_over_blocks() {
2857 let btree = BTreeSet::from_iter((0..(DEFAULT_INNER_SIZE + 10)).into_iter());
2858 assert_eq!(btree.iter().count(), (0..(DEFAULT_INNER_SIZE + 10)).count());
2859 let start = btree.range(0..DEFAULT_INNER_SIZE).into_iter().collect::<Vec<_>>();
2860
2861 assert_eq!(start, (0..DEFAULT_INNER_SIZE).collect::<Vec<_>>());
2862 assert_eq!(
2863 btree.range(0..=DEFAULT_INNER_SIZE).into_iter().collect::<Vec<_>>(),
2864 (0..=DEFAULT_INNER_SIZE).collect::<Vec<_>>()
2865 );
2866 assert_eq!(
2867 btree.range(0..=DEFAULT_INNER_SIZE + 1).count(),
2868 (0..=DEFAULT_INNER_SIZE + 1).count()
2869 );
2870 assert_eq!(btree.iter().rev().count(), (0..(DEFAULT_INNER_SIZE + 10)).count());
2871 assert_eq!(
2872 btree.range(0..DEFAULT_INNER_SIZE).rev().count(),
2873 (0..DEFAULT_INNER_SIZE).count()
2874 );
2875 assert_eq!(
2876 btree.range(0..=DEFAULT_INNER_SIZE).rev().count(),
2877 (0..=DEFAULT_INNER_SIZE).count()
2878 );
2879 assert_eq!(
2880 btree.range(0..=DEFAULT_INNER_SIZE + 1).rev().count(),
2881 (0..=DEFAULT_INNER_SIZE + 1).count()
2882 );
2883 }
2884
2885 #[test]
2886 fn test_empty_set() {
2887 let btree: BTreeSet<usize> = BTreeSet::new();
2888 assert_eq!(btree.iter().count(), 0);
2889 assert_eq!(btree.range(0..0).count(), 0);
2890 assert_eq!(btree.range(0..).count(), 0);
2891 assert_eq!(btree.range(..0).count(), 0);
2892 assert_eq!(btree.range(..).count(), 0);
2893 assert_eq!(btree.range(0..=0).count(), 0);
2894 assert_eq!(btree.range(..1).count(), 0);
2895
2896 assert_eq!(btree.iter().rev().count(), 0);
2897 assert_eq!(btree.range(0..0).rev().count(), 0);
2898 assert_eq!(btree.range(..).rev().count(), 0);
2899 assert_eq!(btree.range(..1).rev().count(), 0);
2900
2901 assert_eq!(btree.range(..DEFAULT_INNER_SIZE).count(), 0);
2902 assert_eq!(btree.range(DEFAULT_INNER_SIZE..DEFAULT_INNER_SIZE * 2).count(), 0);
2903 }
2904
2905 #[test]
2906 fn test_remove_range() {
2907 let btree = BTreeSet::from_iter(0..(DEFAULT_INNER_SIZE * 2));
2909 let expected_len = DEFAULT_INNER_SIZE * 2;
2910 let actual_len = btree.len();
2911 assert_eq!(expected_len, actual_len);
2912
2913 btree.remove_range(5..15);
2915 let expected_len = expected_len - 10;
2916 let actual_len = btree.len();
2917 assert_eq!(expected_len, actual_len);
2918
2919 btree.remove_range(DEFAULT_INNER_SIZE - 5..DEFAULT_INNER_SIZE + 5);
2921 let expected_len = expected_len - 10;
2922 let actual_len = btree.len();
2923 assert_eq!(expected_len, actual_len);
2924
2925 btree.remove_range(..DEFAULT_INNER_SIZE / 2);
2927 let expected_len = expected_len - (DEFAULT_INNER_SIZE / 2) + 10;
2929 let actual_len = btree.len();
2930 assert_eq!(expected_len, actual_len);
2931
2932 let from = (DEFAULT_INNER_SIZE * 3) / 2;
2935 btree.remove_range(from..);
2936 let expected_len = expected_len - DEFAULT_INNER_SIZE / 2;
2937 let actual_len = btree.len();
2938 assert_eq!(expected_len, actual_len);
2939
2940 btree.remove_range(..);
2942 assert_eq!(btree.len(), 0);
2943
2944 for i in 0..(DEFAULT_INNER_SIZE * 2) {
2946 btree.insert(i);
2947 }
2948 let expected_len = DEFAULT_INNER_SIZE * 2;
2949 let actual_len = btree.len();
2950 assert_eq!(expected_len, actual_len);
2951
2952 btree.remove_range((std::ops::Bound::Excluded(5), std::ops::Bound::Excluded(15)));
2953 let expected_len = expected_len - 9;
2954 let actual_len = btree.len();
2955 assert_eq!(expected_len, actual_len);
2956
2957 btree.remove_range((
2958 std::ops::Bound::Included(DEFAULT_INNER_SIZE),
2959 std::ops::Bound::Excluded(DEFAULT_INNER_SIZE + 10),
2960 ));
2961 let expected_len = expected_len - 10;
2962 let actual_len = btree.len();
2963 assert_eq!(expected_len, actual_len);
2964
2965 btree.remove_range(DEFAULT_INNER_SIZE * 3..DEFAULT_INNER_SIZE * 4);
2967 let expected_len = expected_len;
2968 let actual_len = btree.len();
2969 assert_eq!(expected_len, actual_len);
2970
2971 btree.remove_range(DEFAULT_INNER_SIZE * 2 - 5..DEFAULT_INNER_SIZE * 3);
2973 let expected_len = expected_len - 5;
2974 let actual_len = btree.len();
2975 assert_eq!(expected_len, actual_len);
2976 }
2977
2978 #[test]
2979 fn remove_range_end_bound_regressions() {
2980 let set = BTreeSet::<u64>::with_maximum_node_size(4);
2982 for value in 0..10 {
2983 set.insert(value);
2984 }
2985 set.remove_range(7..);
2986 assert_eq!(set.iter().collect::<Vec<_>>(), (0..7).collect::<Vec<_>>());
2987
2988 let set = BTreeSet::<u64>::with_maximum_node_size(4);
2990 for value in 0..10 {
2991 set.insert(value);
2992 }
2993 set.remove_range(..);
2994 assert_eq!(set.len(), 0);
2995 assert!(set.is_empty());
2996
2997 let set = BTreeSet::<u64>::with_maximum_node_size(4);
2999 for value in 0..10 {
3000 set.insert(value);
3001 }
3002 set.remove_range(3..=5);
3003 assert_eq!(set.iter().collect::<Vec<_>>(), vec![0, 1, 2, 6, 7, 8, 9]);
3004
3005 let set = BTreeSet::<u64>::with_maximum_node_size(4);
3007 for value in 0..10 {
3008 set.insert(value);
3009 }
3010 set.remove_range(2..=2);
3011 assert_eq!(set.iter().collect::<Vec<_>>(), vec![0, 1, 3, 4, 5, 6, 7, 8, 9]);
3012
3013 let set = BTreeSet::<u64>::with_maximum_node_size(3);
3016 for value in 0..9 {
3017 set.insert(value);
3018 }
3019 let boundary = *set.index.read().first_key_value().expect("node must exist").0;
3020 set.remove_range(0..boundary);
3021 let expected = (0..9).filter(|value| *value >= boundary).collect::<Vec<_>>();
3022 assert_eq!(set.iter().collect::<Vec<_>>(), expected);
3023 }
3024
3025 #[test]
3026 fn remove_range_matches_btreeset_oracle() {
3027 use std::ops::Bound;
3028
3029 fn oracle_case(node_size: usize, values: &[u64], start: Bound<u64>, end: Bound<u64>) {
3030 let set = BTreeSet::<u64>::with_maximum_node_size(node_size);
3031 for &value in values {
3032 set.insert(value);
3033 }
3034 let mut oracle = values.iter().copied().collect::<std::collections::BTreeSet<_>>();
3035
3036 let range = (start, end);
3037 oracle.retain(|value| !std::ops::RangeBounds::contains(&range, value));
3038 set.remove_range(range);
3039
3040 assert_eq!(
3041 set.iter().collect::<Vec<_>>(),
3042 oracle.iter().copied().collect::<Vec<_>>(),
3043 "node_size={node_size}, start={start:?}, end={end:?}"
3044 );
3045 assert_eq!(
3046 set.len(),
3047 oracle.len(),
3048 "node_size={node_size}, start={start:?}, end={end:?}"
3049 );
3050 }
3051
3052 let values = (0..15u64).map(|value| value * 2).collect::<Vec<_>>();
3055 let mut bounds = vec![Bound::Unbounded];
3056 for probe in 0..=30u64 {
3057 bounds.push(Bound::Included(probe));
3058 bounds.push(Bound::Excluded(probe));
3059 }
3060
3061 for node_size in [4usize, 7, 64] {
3063 for &start in &bounds {
3064 for &end in &bounds {
3065 oracle_case(node_size, &values, start, end);
3066 }
3067 }
3068 }
3069 }
3070
3071 #[test]
3072 fn remove_range_clears_detached_nodes() {
3073 let set = BTreeSet::<u64>::with_maximum_node_size(4);
3078 for value in 0..32 {
3079 set.insert(value);
3080 }
3081
3082 let detached = set
3083 .index
3084 .read()
3085 .range((Included(&2), Bound::Unbounded))
3086 .nth(1)
3087 .unwrap()
3088 .1
3089 .clone();
3090 let detached_values = detached.read().iter().copied().collect::<Vec<_>>();
3091 assert!(detached_values.iter().all(|value| (2..30).contains(value)));
3092
3093 set.remove_range(2..30);
3094
3095 assert!(detached.read().is_empty());
3096 assert!(detached_values.iter().all(|value| !set.contains(value)));
3097 }
3098
3099 #[test]
3100 fn remove_reaches_value_above_every_index_key() {
3101 let set = BTreeSet::<u64>::new();
3102 for value in [1u64, 2, 3] {
3103 set.insert(value);
3104 }
3105
3106 {
3111 let node = set.index.read().last_key_value().expect("node must exist").1.clone();
3112 let mut guard = node.write();
3113 NodeLike::insert(&mut *guard, 5u64);
3114 }
3115
3116 assert!(set.contains(&5));
3117 assert_eq!(set.remove(&5), Some(5), "value above every index key must be removable");
3118 assert!(!set.contains(&5));
3119 assert_eq!(set.iter().collect::<Vec<_>>(), vec![1, 2, 3]);
3120 }
3121
3122 fn drain_node_with_pending_unlink(set: &BTreeSet<u64>, values: &[u64], stale_key: u64) -> Operation<u64, Vec<u64>> {
3126 let node = set.index.read().last_key_value().expect("node must exist").1.clone();
3127 {
3128 let mut guard = node.write();
3129 for value in values {
3130 NodeLike::delete(&mut *guard, value).expect("seeded value must be present");
3131 }
3132 }
3133 Operation::MakeUnreachable(node, stale_key)
3134 }
3135
3136 #[test]
3137 fn split_commit_against_drained_node_fails_instead_of_dropping_insert() {
3138 let set = BTreeSet::<u64>::new();
3139 for seeded in [10u64, 20, 30] {
3140 set.insert(seeded);
3141 }
3142 let node = set.index.read().last_key_value().expect("node must exist").1.clone();
3143 let pending_split = Operation::Split(node.clone(), 30u64, 15u64);
3145 {
3147 let mut guard = node.write();
3148 for seeded in [10u64, 20, 30] {
3149 NodeLike::delete(&mut *guard, &seeded).expect("seeded value must be present");
3150 }
3151 }
3152
3153 assert!(pending_split
3156 .commit::<false>(&mut set.index.write(), super::no_identity_adoption)
3157 .is_err());
3158 assert!(
3159 set.index.read().get(&30).is_some(),
3160 "drained node must stay linked for the retry"
3161 );
3162
3163 assert!(set.insert(15));
3165 assert!(set.contains(&15));
3166 assert_eq!(set.remove(&15), Some(15));
3167 assert!(set.is_empty());
3168 }
3169
3170 #[cfg(feature = "cdc")]
3171 #[test]
3172 fn split_commit_against_drained_node_does_not_panic_in_cdc_build() {
3173 let set = BTreeSet::<u64>::new();
3174 for seeded in [10u64, 20, 30] {
3175 set.insert(seeded);
3176 }
3177 let node = set.index.read().last_key_value().expect("node must exist").1.clone();
3178 let pending_split = Operation::Split(node.clone(), 30u64, 15u64);
3179 {
3180 let mut guard = node.write();
3181 for seeded in [10u64, 20, 30] {
3182 NodeLike::delete(&mut *guard, &seeded).expect("seeded value must be present");
3183 }
3184 }
3185
3186 assert!(pending_split
3189 .commit::<true>(&mut set.index.write(), super::no_identity_adoption)
3190 .is_err());
3191 assert!(set.index.read().get(&30).is_some());
3192
3193 let (old, _events) = set.put_cdc(15);
3194 assert!(old.is_none());
3195 assert!(set.contains(&15));
3196 }
3197
3198 #[test]
3199 fn insert_into_emptied_node_survives_stale_make_unreachable() {
3200 for value in [5u64, 40u64] {
3202 let set = BTreeSet::<u64>::new();
3203 for seeded in [10u64, 20, 30] {
3204 set.insert(seeded);
3205 }
3206 let pending_unlink = drain_node_with_pending_unlink(&set, &[10, 20, 30], 30);
3207
3208 assert!(set.insert(value));
3211
3212 let _ = pending_unlink.commit::<false>(&mut set.index.write(), super::no_identity_adoption);
3215
3216 assert!(set.contains(&value), "value {value} lost after stale unlink");
3217 assert_eq!(set.iter().collect::<Vec<_>>(), vec![value]);
3218 assert_eq!(set.remove(&value), Some(value));
3219 assert!(!set.contains(&value));
3220 assert_eq!(set.len(), 0);
3221 }
3222 }
3223
3224 #[test]
3225 fn stale_make_unreachable_rekeys_refilled_node_instead_of_unlinking() {
3226 for value in [5u64, 40u64] {
3227 let set = BTreeSet::<u64>::new();
3228 for seeded in [10u64, 20, 30] {
3229 set.insert(seeded);
3230 }
3231 let node = set.index.read().last_key_value().expect("node must exist").1.clone();
3232 let pending_unlink = drain_node_with_pending_unlink(&set, &[10, 20, 30], 30);
3233
3234 {
3238 let mut guard = node.write();
3239 NodeLike::insert(&mut *guard, value);
3240 }
3241 let pending_repair = Operation::UpdateMax(node.clone(), 30u64);
3242
3243 assert!(pending_unlink
3246 .commit::<false>(&mut set.index.write(), super::no_identity_adoption)
3247 .is_ok());
3248 let _ = pending_repair.commit::<false>(&mut set.index.write(), super::no_identity_adoption);
3250
3251 assert!(set.contains(&value), "value {value} lost to stale unlink");
3252 assert_eq!(set.remove(&value), Some(value));
3253 assert!(set.is_empty());
3254 }
3255 }
3256
3257 #[test]
3258 fn published_route_updates_when_a_non_last_boundary_shrinks() {
3259 let set = BTreeSet::<u64>::with_maximum_node_size(8);
3260 set.attach_nodes([vec![1, 10], vec![20, 30]]);
3261
3262 assert_eq!(set.remove(&10), Some(10));
3267 assert!(set.insert(5));
3268 assert!(set.contains(&5));
3269 assert_eq!(set.get(&5).map(|value| *value.get()), Some(5));
3270 }
3271
3272 #[test]
3273 fn attach_repairs_a_stale_last_node_boundary() {
3274 let set = BTreeSet::<u64>::with_maximum_node_size(8);
3275 set.attach_node(vec![1, 10]);
3276
3277 assert_eq!(set.remove(&10), Some(10));
3282 set.attach_node(vec![5, 20]);
3283
3284 assert_eq!(set.get(&5).map(|value| *value.get()), Some(5));
3285 assert!(set.contains(&5));
3286 }
3287
3288 #[test]
3289 fn attach_repairs_a_stale_low_last_node_boundary() {
3290 let set = BTreeSet::<u64>::with_maximum_node_size(8);
3291 set.attach_node(vec![1, 10]);
3292
3293 assert!(set.insert(20));
3297 set.attach_node(vec![25, 30]);
3298
3299 assert_eq!(set.get(&20).map(|value| *value.get()), Some(20));
3300 assert_eq!(set.get(&25).map(|value| *value.get()), Some(25));
3301 }
3302
3303 #[test]
3304 fn attach_recovers_from_a_missing_published_identity() {
3305 let set = BTreeSet::<u64>::with_maximum_node_size(8);
3306 set.attach_node(vec![1, 10]);
3307 let last_identity = {
3308 let index = set.index.read();
3309 super::node_identity(index.last_key_value().unwrap().1)
3310 };
3311 assert!(set.index.published_keys.lock().remove(&last_identity).is_some());
3312
3313 set.attach_node(vec![20, 30]);
3314
3315 for value in [1, 10, 20, 30] {
3316 assert_eq!(set.get(&value).map(|found| *found.get()), Some(value));
3317 }
3318 }
3319
3320 #[test]
3321 fn insert_recovers_from_a_missing_replaced_node_identity() {
3322 let set = BTreeSet::<u64>::with_maximum_node_size(8);
3323 set.attach_node(vec![1, 10]);
3324 let old_node = set.index.read().last_key_value().unwrap().1.clone();
3325 assert!(set
3326 .index
3327 .published_keys
3328 .lock()
3329 .remove(&super::node_identity(&old_node))
3330 .is_some());
3331
3332 {
3333 let mut index = set.index.write();
3334 let replaced = index.insert(10, Arc::new(parking_lot::RwLock::new(vec![5, 10])));
3335 assert!(replaced.is_some_and(|node| Arc::ptr_eq(&node, &old_node)));
3336 }
3337
3338 assert!(!set.contains(&1));
3339 assert!(set.contains(&5));
3340 assert!(set.contains(&10));
3341 }
3342
3343 #[test]
3344 fn remove_recovers_from_a_missing_node_identity() {
3345 let set = BTreeSet::<u64>::with_maximum_node_size(8);
3346 set.attach_node(vec![1, 10]);
3347 let old_node = set.index.read().last_key_value().unwrap().1.clone();
3348 assert!(set
3349 .index
3350 .published_keys
3351 .lock()
3352 .remove(&super::node_identity(&old_node))
3353 .is_some());
3354
3355 {
3356 let mut index = set.index.write();
3357 let removed = index.remove(&10).expect("canonical route exists");
3358 assert!(Arc::ptr_eq(&removed, &old_node));
3359 }
3360
3361 assert!(set.is_empty());
3362 assert!(!set.contains(&1));
3363 }
3364
3365 #[test]
3366 fn missing_published_remove_does_not_clone_a_shared_chunk() {
3367 let mut published = super::PublishedNodeIndex::<u64, Vec<u64>> {
3368 chunks: Vec::new(),
3369 len: 0,
3370 };
3371 for key in 0..16 {
3372 published.insert(key, Arc::new(parking_lot::RwLock::new(vec![key])));
3373 }
3374 let snapshot = published.clone();
3375 assert!(Arc::ptr_eq(&published.chunks[0], &snapshot.chunks[0]));
3376
3377 assert!(published.remove(&100).is_none());
3378
3379 assert!(Arc::ptr_eq(&published.chunks[0], &snapshot.chunks[0]));
3380 }
3381
3382 #[test]
3383 fn published_chunks_have_split_merge_hysteresis() {
3384 let mut published = super::PublishedNodeIndex::<u64, Vec<u64>> {
3385 chunks: Vec::new(),
3386 len: 0,
3387 };
3388 for key in 0..=128 {
3389 published.insert(key, Arc::new(parking_lot::RwLock::new(vec![key])));
3390 }
3391 assert_eq!(published.chunks.len(), 2);
3392
3393 for key in 0..33 {
3394 assert!(published.remove(&key).is_some());
3395 }
3396 assert_eq!(published.chunks.len(), 1);
3397
3398 published.insert(0, Arc::new(parking_lot::RwLock::new(vec![0])));
3399 assert_eq!(
3400 published.chunks.len(),
3401 1,
3402 "one insert after a merge must not split again"
3403 );
3404 assert!(published.remove(&0).is_some());
3405 assert_eq!(
3406 published.chunks.len(),
3407 1,
3408 "one remove after a merge must not change chunking"
3409 );
3410 }
3411
3412 #[test]
3413 fn attach_boundary_repair_is_a_complete_publication_by_itself() {
3414 let set = BTreeSet::<u64>::with_maximum_node_size(8);
3415 set.attach_node(vec![1, 10]);
3416 assert_eq!(set.remove(&10), Some(10));
3417
3418 {
3423 let mut index = set.index.write();
3424 index.repair_last_route_before_insert();
3425 }
3426 set.attach_node(vec![5, 20]);
3427
3428 assert_eq!(set.get(&5).map(|found| *found.get()), Some(5));
3429 }
3430
3431 #[cfg(debug_assertions)]
3432 #[test]
3433 #[should_panic(expected = "generic topology removal requires publication")]
3434 fn publication_must_be_enabled_before_an_opt_out_guard_mutates() {
3435 let set = BTreeSet::<u64>::with_maximum_node_size(8);
3436 set.attach_node(vec![1, 10]);
3437 let mut index = set.index.write_rekey();
3438 index.remove(&10);
3439 }
3440
3441 #[test]
3442 fn published_point_routes_match_a_sequential_oracle_under_churn() {
3443 let set = BTreeSet::<u64>::with_maximum_node_size(4);
3444 let mut oracle = std::collections::BTreeSet::new();
3445 let mut state = 0x8f4d_2a71_c390_6be5u64;
3446
3447 for step in 0..2_000 {
3448 state ^= state << 13;
3451 state ^= state >> 7;
3452 state ^= state << 17;
3453 let key = state % 64;
3454 if state & 1 == 0 {
3455 assert_eq!(set.insert(key), oracle.insert(key), "insert step {step}, key {key}");
3456 } else {
3457 assert_eq!(
3458 set.remove(&key).is_some(),
3459 oracle.remove(&key),
3460 "remove step {step}, key {key}"
3461 );
3462 }
3463
3464 for probe in 0..64 {
3465 assert_eq!(
3466 set.contains(&probe),
3467 oracle.contains(&probe),
3468 "point route diverged at step {step}, probe {probe}"
3469 );
3470 }
3471 assert_eq!(
3472 set.iter().collect::<Vec<_>>(),
3473 oracle.iter().copied().collect::<Vec<_>>()
3474 );
3475 }
3476 }
3477
3478 #[test]
3479 fn concurrent_remove_reinsert_over_emptying_nodes_preserves_all_keys() {
3480 const THREADS: u64 = 4;
3481 const ITERATIONS: u64 = 1_000;
3482
3483 let set = Arc::new(BTreeSet::<u64>::with_maximum_node_size(2));
3486 for key in 0..THREADS {
3487 set.insert(key);
3488 }
3489
3490 let start = Arc::new(Barrier::new(THREADS as usize));
3491 let (done_tx, done_rx) = mpsc::channel();
3492 let handles = (0..THREADS)
3493 .map(|key| {
3494 let set = Arc::clone(&set);
3495 let start = Arc::clone(&start);
3496 let done_tx = done_tx.clone();
3497 thread::spawn(move || {
3498 start.wait();
3499 for _ in 0..ITERATIONS {
3500 let mut attempts = 0;
3507 while set.remove(&key).is_none() {
3508 let index = set.index.read();
3509 let present = index.values().any(|node| node.read().contains(&key));
3510 drop(index);
3511 assert!(present, "acknowledged insert of {key} was lost");
3512 attempts += 1;
3513 assert!(attempts < 10_000, "key {key} present but never became removable");
3514 std::hint::spin_loop();
3515 }
3516 assert!(set.insert(key), "{key} still present after acknowledged remove");
3517 }
3518 done_tx.send(()).unwrap();
3519 })
3520 })
3521 .collect::<Vec<_>>();
3522 drop(done_tx);
3523
3524 for _ in 0..THREADS {
3525 done_rx
3526 .recv_timeout(Duration::from_secs(30))
3527 .expect("remove/reinsert workload did not complete in time");
3528 }
3529 for handle in handles {
3530 handle.join().unwrap();
3531 }
3532
3533 for key in 0..THREADS {
3534 assert!(set.contains(&key), "key {key} lost after churn");
3535 assert_eq!(set.remove(&key), Some(key));
3536 }
3537 assert!(set.is_empty());
3538 }
3539
3540 #[test]
3541 fn last_node_rekey_that_crosses_predecessor_is_published() {
3542 let set = BTreeSet::<u64>::with_maximum_node_size(2);
3543 set.attach_nodes([vec![0], vec![2], vec![3]]);
3544 let node = set.index.read().last_key_value().expect("last node").1.clone();
3545
3546 {
3550 let mut values = node.write();
3551 NodeLike::delete(&mut *values, &3).expect("seeded value");
3552 NodeLike::insert(&mut *values, 1);
3553 }
3554 Operation::UpdateMax(node, 3)
3555 .commit::<false>(&mut set.index.write_rekey(), super::no_identity_adoption)
3556 .expect("boundary repair");
3557
3558 assert_eq!(set.iter().collect::<Vec<_>>(), vec![0, 1, 2]);
3559 for value in 0..=2 {
3560 assert_eq!(set.get(&value).map(|found| *found.get()), Some(value));
3561 }
3562 }
3563
3564 #[test]
3565 fn structural_insert_repairs_a_stale_last_route() {
3566 let set = BTreeSet::<u64>::with_maximum_node_size(2);
3567 set.attach_nodes([vec![0], vec![3]]);
3568 let node = set.index.read().last_key_value().expect("last node").1.clone();
3569
3570 {
3572 let mut values = node.write();
3573 NodeLike::delete(&mut *values, &3).expect("seeded value");
3574 NodeLike::insert(&mut *values, 1);
3575 }
3576 Operation::UpdateMax(node, 3)
3577 .commit::<false>(&mut set.index.write_rekey(), super::no_identity_adoption)
3578 .expect("boundary repair");
3579
3580 set.index.write().insert(2, Arc::new(RwLock::new(vec![2])));
3584
3585 assert_eq!(set.iter().collect::<Vec<_>>(), vec![0, 1, 2]);
3586 for value in 0..=2 {
3587 assert_eq!(set.get(&value).map(|found| *found.get()), Some(value));
3588 }
3589 }
3590
3591 #[test]
3592 fn test_remove_single_element() {
3593 let set = BTreeSet::<i32>::new();
3594 set.insert(5);
3595 assert!(set.contains(&5));
3596 assert!(set.remove(&5).is_some());
3597 assert!(!set.contains(&5));
3598 assert!(!set.remove(&5).is_some());
3599 }
3600
3601 #[test]
3602 fn test_remove_multiple_elements() {
3603 let set = BTreeSet::<i32>::new();
3604 for i in 0..2048 {
3605 set.insert(i);
3606 }
3607 for i in 0..2048 {
3608 assert!(set.remove(&i).is_some());
3609 assert!(!set.contains(&i));
3610 }
3611 assert_eq!(set.len(), 0);
3612 }
3613
3614 #[test]
3615 fn test_remove_non_existent() {
3616 let set = BTreeSet::<i32>::new();
3617 set.insert(5);
3618 assert!(!set.remove(&10).is_some());
3619 assert!(set.contains(&5));
3620 }
3621
3622 #[test]
3623 fn test_remove_stress() {
3624 let set = Arc::new(BTreeSet::<i32>::new());
3625 const NUM_ELEMENTS: i32 = 10000;
3626
3627 for i in 0..NUM_ELEMENTS {
3628 set.insert(i);
3629 }
3630 assert_eq!(set.len(), NUM_ELEMENTS as usize, "Incorrect size after insertion");
3631
3632 let num_threads = 8;
3633 let elements_per_thread = NUM_ELEMENTS / num_threads;
3634 let handles: Vec<_> = (0..num_threads)
3635 .map(|t| {
3636 let set = Arc::clone(&set);
3637 thread::spawn(move || {
3638 for i in (t * elements_per_thread)..((t + 1) * elements_per_thread) {
3639 if i % 2 == 1 {
3640 assert!(set.remove(&i).is_some(), "Failed to remove {}", i);
3641 }
3642 }
3643 })
3644 })
3645 .collect();
3646
3647 for handle in handles {
3648 handle.join().unwrap();
3649 }
3650
3651 assert_eq!(set.len(), NUM_ELEMENTS as usize / 2, "Incorrect size after removal");
3652
3653 for i in 0..NUM_ELEMENTS {
3654 if i % 2 == 0 {
3655 assert!(set.contains(&i), "Even number {} should be in the set", i);
3656 } else {
3657 assert!(!set.contains(&i), "Odd number {} should not be in the set", i);
3658 }
3659 }
3660 }
3661
3662 #[test]
3663 fn test_remove_all_elements() {
3664 let set = BTreeSet::<i32>::new();
3665 let n = 2048;
3666
3667 for i in 0..n {
3668 set.insert(i);
3669 }
3670
3671 for i in 0..n {
3672 assert!(set.remove(&i).is_some(), "Failed to remove {}", i);
3673 }
3674
3675 assert_eq!(set.len(), 0, "Set should be empty");
3676
3677 for i in 0..n {
3678 assert!(!set.contains(&i), "Element {} should not be in the set", i);
3679 }
3680 }
3681
3682 #[test]
3683 fn test_range_edge_cases() {
3684 let set = BTreeSet::<i32>::with_maximum_node_size(10);
3685 for i in 0..20 {
3686 set.insert(i);
3687 }
3688 assert_eq!(set.range(0..=0).collect::<Vec<_>>(), vec![0]);
3695 assert_eq!(set.range(0..1).collect::<Vec<_>>(), vec![0]);
3696
3697 assert_eq!(set.range(5..=5).collect::<Vec<_>>(), vec![5]);
3698 assert_eq!(set.range(5..6).collect::<Vec<_>>(), vec![5]);
3699
3700 assert_eq!(set.range(10..=10).collect::<Vec<_>>(), vec![10]);
3701 assert_eq!(set.range(10..11).collect::<Vec<_>>(), vec![10]);
3702
3703 assert_eq!(set.range(0..=3).collect::<Vec<_>>(), vec![0, 1, 2, 3]);
3705 assert_eq!(set.range(0..3).collect::<Vec<_>>(), vec![0, 1, 2]);
3706
3707 assert_eq!(set.range(5..=8).collect::<Vec<_>>(), vec![5, 6, 7, 8]);
3708 assert_eq!(set.range(5..8).collect::<Vec<_>>(), vec![5, 6, 7]);
3709
3710 assert_eq!(set.range(10..=13).collect::<Vec<_>>(), vec![10, 11, 12, 13]);
3711 assert_eq!(set.range(10..13).collect::<Vec<_>>(), vec![10, 11, 12]);
3712
3713 assert_eq!(set.range(4..=4).collect::<Vec<_>>(), vec![4]);
3715 assert_eq!(set.range(4..5).collect::<Vec<_>>(), vec![4]);
3716
3717 assert_eq!(set.range(9..=9).collect::<Vec<_>>(), vec![9]);
3718 assert_eq!(set.range(9..10).collect::<Vec<_>>(), vec![9]);
3719
3720 assert_eq!(set.range(19..=19).collect::<Vec<_>>(), vec![19]);
3721 assert_eq!(set.range(19..20).collect::<Vec<_>>(), vec![19]);
3722
3723 assert_eq!(set.range(17..=19).collect::<Vec<_>>(), vec![17, 18, 19]);
3725 assert_eq!(set.range(17..20).collect::<Vec<_>>(), vec![17, 18, 19]);
3726
3727 assert_eq!(set.range(7..=9).collect::<Vec<_>>(), vec![7, 8, 9]);
3728 assert_eq!(set.range(7..10).collect::<Vec<_>>(), vec![7, 8, 9]);
3729
3730 assert_eq!(set.range(2..=4).collect::<Vec<_>>(), vec![2, 3, 4]);
3731 assert_eq!(set.range(2..5).collect::<Vec<_>>(), vec![2, 3, 4]);
3732
3733 assert_eq!(set.range(0..=4).collect::<Vec<_>>(), vec![0, 1, 2, 3, 4]);
3735 assert_eq!(set.range(0..5).collect::<Vec<_>>(), vec![0, 1, 2, 3, 4]);
3736
3737 assert_eq!(set.range(5..=9).collect::<Vec<_>>(), vec![5, 6, 7, 8, 9]);
3738 assert_eq!(set.range(5..10).collect::<Vec<_>>(), vec![5, 6, 7, 8, 9]);
3739
3740 assert_eq!(
3741 set.range(10..=19).collect::<Vec<_>>(),
3742 vec![10, 11, 12, 13, 14, 15, 16, 17, 18, 19]
3743 );
3744 assert_eq!(
3745 set.range(10..20).collect::<Vec<_>>(),
3746 vec![10, 11, 12, 13, 14, 15, 16, 17, 18, 19]
3747 );
3748
3749 assert_eq!(set.range(3..=6).collect::<Vec<_>>(), vec![3, 4, 5, 6]);
3751 assert_eq!(set.range(3..7).collect::<Vec<_>>(), vec![3, 4, 5, 6]);
3752
3753 assert_eq!(set.range(8..=11).collect::<Vec<_>>(), vec![8, 9, 10, 11]);
3754 assert_eq!(set.range(8..12).collect::<Vec<_>>(), vec![8, 9, 10, 11]);
3755
3756 assert_eq!(set.range(0..=0).rev().collect::<Vec<_>>(), vec![0]);
3760 assert_eq!(set.range(0..1).rev().collect::<Vec<_>>(), vec![0]);
3761
3762 assert_eq!(set.range(5..=5).rev().collect::<Vec<_>>(), vec![5]);
3763 assert_eq!(set.range(5..6).rev().collect::<Vec<_>>(), vec![5]);
3764
3765 assert_eq!(set.range(10..=10).rev().collect::<Vec<_>>(), vec![10]);
3766 assert_eq!(set.range(10..11).rev().collect::<Vec<_>>(), vec![10]);
3767
3768 assert_eq!(set.range(0..=3).rev().collect::<Vec<_>>(), vec![3, 2, 1, 0]);
3770 assert_eq!(set.range(0..3).rev().collect::<Vec<_>>(), vec![2, 1, 0]);
3771
3772 assert_eq!(set.range(5..=8).rev().collect::<Vec<_>>(), vec![8, 7, 6, 5]);
3773 assert_eq!(set.range(5..8).rev().collect::<Vec<_>>(), vec![7, 6, 5]);
3774
3775 assert_eq!(set.range(10..=13).rev().collect::<Vec<_>>(), vec![13, 12, 11, 10]);
3776 assert_eq!(set.range(10..13).rev().collect::<Vec<_>>(), vec![12, 11, 10]);
3777
3778 assert_eq!(set.range(4..=4).rev().collect::<Vec<_>>(), vec![4]);
3780 assert_eq!(set.range(4..5).rev().collect::<Vec<_>>(), vec![4]);
3781
3782 assert_eq!(set.range(9..=9).rev().collect::<Vec<_>>(), vec![9]);
3783 assert_eq!(set.range(9..10).rev().collect::<Vec<_>>(), vec![9]);
3784
3785 assert_eq!(set.range(19..=19).rev().collect::<Vec<_>>(), vec![19]);
3786 assert_eq!(set.range(19..20).rev().collect::<Vec<_>>(), vec![19]);
3787
3788 assert_eq!(set.range(17..=19).rev().collect::<Vec<_>>(), vec![19, 18, 17]);
3790 assert_eq!(set.range(17..20).rev().collect::<Vec<_>>(), vec![19, 18, 17]);
3791
3792 assert_eq!(set.range(7..=9).rev().collect::<Vec<_>>(), vec![9, 8, 7]);
3793 assert_eq!(set.range(7..10).rev().collect::<Vec<_>>(), vec![9, 8, 7]);
3794
3795 assert_eq!(set.range(2..=4).rev().collect::<Vec<_>>(), vec![4, 3, 2]);
3796 assert_eq!(set.range(2..5).rev().collect::<Vec<_>>(), vec![4, 3, 2]);
3797
3798 assert_eq!(set.range(0..=4).rev().collect::<Vec<_>>(), vec![4, 3, 2, 1, 0]);
3800 assert_eq!(set.range(0..5).rev().collect::<Vec<_>>(), vec![4, 3, 2, 1, 0]);
3801
3802 assert_eq!(set.range(5..=9).rev().collect::<Vec<_>>(), vec![9, 8, 7, 6, 5]);
3803 assert_eq!(set.range(5..10).rev().collect::<Vec<_>>(), vec![9, 8, 7, 6, 5]);
3804
3805 assert_eq!(
3806 set.range(10..=19).rev().collect::<Vec<_>>(),
3807 vec![19, 18, 17, 16, 15, 14, 13, 12, 11, 10]
3808 );
3809 assert_eq!(
3810 set.range(10..20).rev().collect::<Vec<_>>(),
3811 vec![19, 18, 17, 16, 15, 14, 13, 12, 11, 10]
3812 );
3813
3814 assert_eq!(set.range(3..=6).rev().collect::<Vec<_>>(), vec![6, 5, 4, 3]);
3816 assert_eq!(set.range(3..7).rev().collect::<Vec<_>>(), vec![6, 5, 4, 3]);
3817
3818 assert_eq!(set.range(8..=11).rev().collect::<Vec<_>>(), vec![11, 10, 9, 8]);
3819 assert_eq!(set.range(8..12).rev().collect::<Vec<_>>(), vec![11, 10, 9, 8]);
3820
3821 assert!(set.range(20..).collect::<Vec<_>>().is_empty());
3823 assert!(set.range(..0).collect::<Vec<_>>().is_empty());
3824 assert!(set.range(20..).rev().collect::<Vec<_>>().is_empty());
3825 assert!(set.range(..0).rev().collect::<Vec<_>>().is_empty());
3826 }
3827
3828 #[test]
3829 fn concurrent_range_constructions_at_node_boundaries_do_not_deadlock() {
3830 const THREAD_ITERATIONS: usize = 20_000;
3831
3832 let set = Arc::new(BTreeSet::<u64>::with_maximum_node_size(4));
3833 for value in 0..64 {
3834 set.insert(value);
3835 }
3836
3837 let (done_tx, done_rx) = mpsc::channel();
3843 let forward = {
3844 let set = Arc::clone(&set);
3845 let done_tx = done_tx.clone();
3846 thread::spawn(move || {
3847 for iteration in 0..THREAD_ITERATIONS {
3848 let start = (iteration % 64) as u64;
3849 assert_eq!(set.range(start..).next(), Some(start));
3850 }
3851 done_tx.send(()).unwrap();
3852 })
3853 };
3854 let backward = {
3855 let set = Arc::clone(&set);
3856 let done_tx = done_tx.clone();
3857 thread::spawn(move || {
3858 for iteration in 0..THREAD_ITERATIONS {
3859 let end = (iteration % 64) as u64;
3860 assert_eq!(set.range(..=end).next_back(), Some(end));
3861 }
3862 done_tx.send(()).unwrap();
3863 })
3864 };
3865 drop(done_tx);
3866
3867 for _ in 0..2 {
3868 done_rx
3869 .recv_timeout(Duration::from_secs(30))
3870 .expect("concurrent range constructions deadlocked");
3871 }
3872 forward.join().unwrap();
3873 backward.join().unwrap();
3874 }
3875
3876 fn three_node_set() -> BTreeSet<u64> {
3878 let set = BTreeSet::<u64>::with_maximum_node_size(4);
3879 for value in [0u64, 10, 20, 30, 40, 50, 60] {
3880 set.insert(value);
3881 }
3882 assert_eq!(
3883 set.index.read().keys().copied().collect::<Vec<_>>(),
3884 vec![10, 30, 60],
3885 "fixture geometry changed"
3886 );
3887 set
3888 }
3889
3890 #[test]
3891 fn forward_scan_repositions_when_current_node_vanishes() {
3892 let set = three_node_set();
3893
3894 let mut iter = set.iter();
3895 assert_eq!(iter.next(), Some(0));
3896 assert_eq!(iter.next(), Some(10));
3897 assert_eq!(iter.next(), Some(20));
3898
3899 set.index.write().remove(&30).expect("fixture entry");
3903
3904 assert_eq!(iter.next(), Some(30));
3905 assert_eq!(iter.next(), Some(40));
3906 assert_eq!(iter.next(), Some(50));
3907 assert_eq!(iter.next(), Some(60));
3908 assert_eq!(iter.next(), None);
3909 }
3910
3911 #[test]
3912 fn backward_scan_repositions_when_current_node_vanishes() {
3913 let set = three_node_set();
3914
3915 let mut iter = set.iter();
3916 assert_eq!(iter.next_back(), Some(60));
3917 assert_eq!(iter.next_back(), Some(50));
3918
3919 set.index.write().remove(&60).expect("fixture entry");
3920
3921 assert_eq!(iter.next_back(), Some(40));
3922 assert_eq!(iter.next_back(), Some(30));
3923 assert_eq!(iter.next_back(), Some(20));
3924 assert_eq!(iter.next_back(), Some(10));
3925 assert_eq!(iter.next_back(), Some(0));
3926 assert_eq!(iter.next_back(), None);
3927 }
3928
3929 #[test]
3930 fn forward_scan_does_not_re_yield_after_split_of_finished_node() {
3931 let set = BTreeSet::<u64>::with_maximum_node_size(4);
3934 for value in [0u64, 10, 20, 30, 40] {
3935 set.insert(value);
3936 }
3937
3938 let iter = Iter {
3942 tree: &set,
3943 current_front_batch: None,
3944 current_back_batch: None,
3945 exhausted_front_node: Some(set.index.read().first_key_value().expect("fixture node").1.clone()),
3946 exhausted_back_node: None,
3947 front_partial: None,
3948 back_partial: None,
3949 front_batch_limit: INITIAL_BATCH,
3950 back_batch_limit: INITIAL_BATCH,
3951 current_front_value: Some(20),
3952 current_back_value: None,
3953 met: false,
3954 };
3955
3956 assert_eq!(iter.collect::<Vec<_>>(), vec![30, 40]);
3957 }
3958
3959 #[test]
3960 fn backward_scan_does_not_re_yield_values_from_scanned_range() {
3961 let set = BTreeSet::<u64>::new();
3965 set.attach_node(vec![0u64, 10]);
3966 set.attach_node(vec![30u64, 40]);
3967 {
3968 let node = set.index.read().first_key_value().expect("fixture node").1.clone();
3969 let mut guard = node.write();
3970 NodeLike::insert(&mut *guard, 35u64);
3971 }
3972
3973 let mut iter = Iter {
3974 tree: &set,
3975 current_front_batch: None,
3976 current_back_batch: None,
3977 exhausted_front_node: None,
3978 exhausted_back_node: Some(set.index.read().last_key_value().expect("fixture node").1.clone()),
3979 front_partial: None,
3980 back_partial: None,
3981 front_batch_limit: INITIAL_BATCH,
3982 back_batch_limit: INITIAL_BATCH,
3983 current_front_value: None,
3984 current_back_value: Some(30),
3985 met: false,
3986 };
3987
3988 let mut collected = vec![];
3989 while let Some(value) = iter.next_back() {
3990 collected.push(value);
3991 }
3992 assert_eq!(collected, vec![10, 0]);
3993 }
3994
3995 #[test]
3996 fn backward_scan_does_not_skip_values_split_away_after_positioning() {
3997 let set = BTreeSet::<u64>::with_maximum_node_size(4);
4005 for value in [0u64, 10, 20, 30] {
4006 set.insert(value);
4007 }
4008
4009 let mut iter = set.iter();
4011 set.insert(40);
4015 assert!(set.node_count() > 1, "fixture must split");
4016
4017 let mut collected = vec![];
4018 while let Some(value) = iter.next_back() {
4019 collected.push(value);
4020 }
4021
4022 for baseline in [30u64, 20, 10, 0] {
4028 assert!(
4029 collected.iter().any(|value| *value == baseline),
4030 "baseline value {baseline} skipped by backward scan (yielded: {collected:?})"
4031 );
4032 }
4033 assert!(
4034 collected.windows(2).all(|pair| pair[0] > pair[1]),
4035 "backward scan not strictly decreasing: {collected:?}"
4036 );
4037 }
4038
4039 #[test]
4040 fn bidirectional_meet_into_opposite_held_node_does_not_self_deadlock() {
4041 let set = Arc::new(BTreeSet::<u64>::with_maximum_node_size(4));
4043 for value in [0u64, 10, 20, 30, 40] {
4044 set.insert(value);
4045 }
4046
4047 let (done_tx, done_rx) = mpsc::channel();
4048 let handle = {
4049 let set = Arc::clone(&set);
4050 thread::spawn(move || {
4051 let mut finished = set.iter();
4058 assert_eq!(finished.next_back(), Some(40));
4059 assert_eq!(finished.next(), Some(0));
4060 assert_eq!(finished.next(), Some(10));
4061 assert_eq!(finished.next(), Some(20));
4062 assert_eq!(finished.next(), Some(30));
4063 assert_eq!(finished.next(), None);
4064 assert_eq!(finished.next_back(), None);
4065
4066 let mut iter = set.iter();
4071 assert_eq!(iter.next(), Some(0));
4072 assert_eq!(iter.next_back(), Some(40));
4073 assert_eq!(iter.next_back(), Some(30));
4074 assert_eq!(iter.next_back(), Some(20));
4075 assert_eq!(iter.next_back(), Some(10));
4076 assert_eq!(iter.next_back(), None);
4077 assert_eq!(iter.next(), None);
4078 drop(finished);
4079
4080 done_tx.send(()).unwrap();
4081 })
4082 };
4083
4084 done_rx
4085 .recv_timeout(Duration::from_secs(10))
4086 .expect("bidirectional meet-in-the-middle deadlocked");
4087 handle.join().unwrap();
4088 }
4089
4090 #[test]
4091 fn structural_commits_complete_against_paused_scan() {
4092 let set = Arc::new(BTreeSet::<u64>::with_maximum_node_size(2));
4094 for value in 0..8 {
4095 set.insert(value);
4096 }
4097
4098 let scan_holds_guard = Arc::new(Barrier::new(3));
4099 let (done_tx, done_rx) = mpsc::channel();
4100
4101 let scanner = {
4102 let set = Arc::clone(&set);
4103 let scan_holds_guard = Arc::clone(&scan_holds_guard);
4104 let done_tx = done_tx.clone();
4105 thread::spawn(move || {
4106 let mut iter = set.iter();
4107 assert_eq!(iter.next(), Some(0));
4111 scan_holds_guard.wait();
4112 thread::sleep(Duration::from_millis(100));
4115 let mut collected = vec![];
4120 while let Some(value) = iter.next_back() {
4121 collected.push(value);
4122 }
4123 assert_eq!(collected, vec![7, 6, 5, 4, 3, 2, 1]);
4124 done_tx.send(()).unwrap();
4125 })
4126 };
4127
4128 let writer = {
4129 let set = Arc::clone(&set);
4130 let scan_holds_guard = Arc::clone(&scan_holds_guard);
4131 let done_tx = done_tx.clone();
4132 thread::spawn(move || {
4133 scan_holds_guard.wait();
4134 set.remove_range(0..=0);
4137 done_tx.send(()).unwrap();
4138 })
4139 };
4140 drop(done_tx);
4141 scan_holds_guard.wait();
4142
4143 for _ in 0..2 {
4144 done_rx
4145 .recv_timeout(Duration::from_secs(10))
4146 .expect("scan or structural commit deadlocked");
4147 }
4148 scanner.join().unwrap();
4149 writer.join().unwrap();
4150 assert!(!set.contains(&0));
4151 }
4152
4153 #[test]
4154 fn full_scans_do_not_degrade_quadratically_with_node_count() {
4155 use std::time::Instant;
4156
4157 const VALUES: u64 = 30_000;
4159 let set = BTreeSet::<u64>::with_maximum_node_size(2);
4160 for value in 0..VALUES {
4161 set.insert(value);
4162 }
4163 assert!(
4164 set.node_count() >= (VALUES / 4) as usize,
4165 "fixture must be a many-node tree, got {} nodes",
4166 set.node_count()
4167 );
4168
4169 let started = Instant::now();
4170 assert_eq!(set.iter().count(), VALUES as usize);
4171 let forward = started.elapsed();
4172
4173 let started = Instant::now();
4174 assert_eq!(set.iter().rev().count(), VALUES as usize);
4175 let backward = started.elapsed();
4176
4177 let budget = Duration::from_secs(10);
4183 assert!(
4184 forward < budget,
4185 "forward scan took {forward:?}, node advance is not logarithmic"
4186 );
4187 assert!(
4188 backward < budget,
4189 "backward scan took {backward:?}, node advance is not logarithmic"
4190 );
4191 }
4192
4193 #[test]
4194 fn scans_stay_sorted_and_complete_under_monotonic_insert_churn() {
4195 use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};
4196
4197 const BASELINE: u64 = 400;
4198 const EXTRA: u64 = 2_000;
4199 const SCAN_BOUND: usize = 10_000;
4200
4201 let set = Arc::new(BTreeSet::<u64>::with_maximum_node_size(8));
4204 for value in 0..BASELINE {
4205 set.insert(value);
4206 }
4207
4208 let done = Arc::new(AtomicBool::new(false));
4209 let writer = {
4210 let set = Arc::clone(&set);
4211 let done = Arc::clone(&done);
4212 thread::spawn(move || {
4213 for value in BASELINE..BASELINE + EXTRA {
4214 assert!(set.insert(value));
4215 }
4216 done.store(true, AtomicOrdering::Release);
4217 })
4218 };
4219
4220 let mut scans = 0usize;
4221 loop {
4222 let forward = set.iter().collect::<Vec<_>>();
4223 assert!(
4224 forward.windows(2).all(|pair| pair[0] < pair[1]),
4225 "forward scan not strictly increasing (duplicate or unordered yield)"
4226 );
4227 assert_eq!(
4228 forward.iter().filter(|value| **value < BASELINE).count() as u64,
4229 BASELINE,
4230 "forward scan truncated: baseline keys missing"
4231 );
4232
4233 let backward = set.iter().rev().collect::<Vec<_>>();
4234 assert!(
4235 backward.windows(2).all(|pair| pair[0] > pair[1]),
4236 "backward scan not strictly decreasing (duplicate or unordered yield)"
4237 );
4238 assert_eq!(
4239 backward.iter().filter(|value| **value < BASELINE).count() as u64,
4240 BASELINE,
4241 "backward scan truncated: baseline keys missing"
4242 );
4243
4244 scans += 1;
4245 if done.load(AtomicOrdering::Acquire) || scans >= SCAN_BOUND {
4246 break;
4247 }
4248 }
4249
4250 writer.join().unwrap();
4251
4252 let expected = (0..BASELINE + EXTRA).collect::<Vec<_>>();
4253 assert_eq!(set.iter().collect::<Vec<_>>(), expected);
4254 assert_eq!(set.iter().rev().collect::<Vec<_>>(), {
4255 let mut reversed = expected;
4256 reversed.reverse();
4257 reversed
4258 });
4259 }
4260
4261 #[test]
4262 fn collected_owned_values_survive_arbitrary_concurrent_mutation() {
4263 use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};
4264
4265 const BASELINE: u64 = 512;
4266 const CHURN: u64 = 4_000;
4267
4268 let set = Arc::new(BTreeSet::<u64>::with_maximum_node_size(8));
4271 for value in 0..BASELINE {
4272 set.insert(value);
4273 }
4274
4275 let done = Arc::new(AtomicBool::new(false));
4276 let writer = {
4277 let set = Arc::clone(&set);
4278 let done = Arc::clone(&done);
4279 thread::spawn(move || {
4280 for value in BASELINE..BASELINE + CHURN {
4281 assert!(set.insert(value));
4282 assert_eq!(set.remove(&value), Some(value));
4283 }
4284 done.store(true, AtomicOrdering::Release);
4285 })
4286 };
4287
4288 let mut snapshots: Vec<Vec<u64>> = Vec::new();
4293 loop {
4294 let snapshot: Vec<u64> = set.iter().collect();
4295 snapshots.push(snapshot);
4296 if done.load(AtomicOrdering::Acquire) {
4297 break;
4298 }
4299 }
4300 writer.join().unwrap();
4301
4302 set.remove_range(..);
4305 assert!(set.is_empty());
4306
4307 for snapshot in snapshots {
4308 assert!(
4309 snapshot.windows(2).all(|pair| pair[0] < pair[1]),
4310 "snapshot not strictly increasing"
4311 );
4312 assert_eq!(
4313 snapshot.iter().filter(|value| **value < BASELINE).count() as u64,
4314 BASELINE,
4315 "snapshot lost baseline keys"
4316 );
4317 }
4318 }
4319
4320 #[test]
4321 fn parallel_iter_and_mut() {
4322 let set = Arc::new(BTreeSet::<i32>::new());
4323 for i in 0..10_000 {
4324 set.insert(i);
4325 }
4326
4327 let set_clone = Arc::clone(&set);
4328 let handle = thread::spawn(move || {
4329 for _ in 0..1000 {
4330 let mut _sum = 0;
4331 for value in set_clone.iter() {
4332 _sum += value;
4333 }
4334 }
4335 });
4336
4337 for i in 10_000..20_000 {
4338 set.insert(i);
4339 }
4340 handle.join().unwrap();
4341 }
4342
4343 #[test]
4352 fn a_scan_across_several_batch_installs_is_complete_and_ordered() {
4353 let set: BTreeSet<u64> = BTreeSet::new();
4354 let count = (INITIAL_BATCH * 20) as u64;
4357 for i in 0..count {
4358 set.insert(i);
4359 }
4360
4361 let seen: Vec<u64> = set.iter().collect();
4362 let expected: Vec<u64> = (0..count).collect();
4363 assert_eq!(seen, expected, "a partial-batch scan lost or repeated elements");
4364 }
4365
4366 #[test]
4368 fn a_backward_scan_across_several_installs_is_complete_and_ordered() {
4369 let set: BTreeSet<u64> = BTreeSet::new();
4370 let count = (INITIAL_BATCH * 20) as u64;
4371 for i in 0..count {
4372 set.insert(i);
4373 }
4374
4375 let seen: Vec<u64> = set.iter().rev().collect();
4376 let expected: Vec<u64> = (0..count).rev().collect();
4377 assert_eq!(
4378 seen, expected,
4379 "a partial-batch backward scan lost or repeated elements"
4380 );
4381 }
4382
4383 #[test]
4386 fn a_scan_within_a_single_node_resumes_correctly() {
4387 let set: BTreeSet<u64> = BTreeSet::with_maximum_node_size(DEFAULT_INNER_SIZE);
4388 let count = 200u64;
4389 for i in 0..count {
4390 set.insert(i);
4391 }
4392 assert_eq!(set.node_count(), 1, "fixture wants one node");
4393
4394 let seen: Vec<u64> = set.iter().collect();
4395 assert_eq!(seen, (0..count).collect::<Vec<_>>());
4396 }
4397
4398 #[test]
4403 fn a_single_element_range_yields_one_element() {
4404 let set: BTreeSet<u64> = BTreeSet::new();
4405 for i in 0..1_000u64 {
4406 set.insert(i);
4407 }
4408
4409 for probe in [0u64, 1, 499, 998, 999] {
4410 let got: Vec<u64> = set.range(probe..=probe).collect();
4411 assert_eq!(got, vec![probe], "range({probe}..={probe})");
4412 }
4413 assert!(set.range(1_000..=1_000).next().is_none(), "absent key");
4414 }
4415
4416 #[test]
4421 fn ranges_spanning_batch_boundaries_are_exact() {
4422 let set: BTreeSet<u64> = BTreeSet::new();
4423 for i in 0..500u64 {
4424 set.insert(i);
4425 }
4426
4427 for width in 1..=(INITIAL_BATCH * 8) as u64 {
4428 let start = 100u64;
4429 let got: Vec<u64> = set.range(start..start + width).collect();
4430 let expected: Vec<u64> = (start..start + width).collect();
4431 assert_eq!(got, expected, "range width {width}");
4432 }
4433 }
4434
4435 #[test]
4440 fn a_double_ended_scan_meets_without_repeating() {
4441 let set: BTreeSet<u64> = BTreeSet::new();
4442 let count = (INITIAL_BATCH * 10) as u64;
4443 for i in 0..count {
4444 set.insert(i);
4445 }
4446
4447 let mut iter = set.iter();
4448 let mut front = Vec::new();
4449 let mut back = Vec::new();
4450 loop {
4451 match iter.next() {
4452 Some(v) => front.push(v),
4453 None => break,
4454 }
4455 match iter.next_back() {
4456 Some(v) => back.push(v),
4457 None => break,
4458 }
4459 }
4460 back.reverse();
4461 front.extend(back);
4462 front.sort_unstable();
4463 assert_eq!(
4464 front,
4465 (0..count).collect::<Vec<_>>(),
4466 "double-ended scan is not a partition"
4467 );
4468 }
4469
4470 #[test]
4486 fn a_scan_under_concurrent_mutation_terminates() {
4487 use std::sync::atomic::{AtomicBool, Ordering};
4488 use std::sync::Arc;
4489
4490 const SIZE: u64 = 4_000;
4491 const CAP: usize = (SIZE * 20) as usize;
4494
4495 for _ in 0..8 {
4496 let set: Arc<BTreeSet<u64>> = Arc::new(BTreeSet::new());
4497 for i in 0..SIZE {
4498 set.insert(i);
4499 }
4500 let stop = Arc::new(AtomicBool::new(false));
4501
4502 let writers: Vec<_> = (0..3)
4504 .map(|w| {
4505 let (set, stop) = (Arc::clone(&set), Arc::clone(&stop));
4506 std::thread::spawn(move || {
4507 let mut i = SIZE + w * 100_000;
4508 while !stop.load(Ordering::Relaxed) {
4509 set.insert(i);
4510 set.remove(&i);
4511 i += 1;
4512 }
4513 })
4514 })
4515 .collect();
4516
4517 let mut yielded = 0usize;
4518 for _ in set.iter() {
4519 yielded += 1;
4520 if yielded >= CAP {
4521 break;
4522 }
4523 }
4524
4525 stop.store(true, Ordering::Relaxed);
4526 for w in writers {
4527 w.join().expect("writer did not panic");
4528 }
4529
4530 assert!(
4531 yielded < CAP,
4532 "scan did not make progress under concurrent mutation: {yielded} yields"
4533 );
4534 }
4535 }
4536
4537 #[allow(clippy::manual_contains)]
4545 fn yielded(seen: &[u64], value: u64) -> bool {
4546 seen.iter().any(|item| *item == value)
4547 }
4548
4549 #[test]
4562 fn deleting_a_yielded_element_does_not_skip_a_live_one() {
4563 for prefix in 1..12usize {
4564 let set: BTreeSet<u64> = BTreeSet::new();
4565 for i in 0..256u64 {
4566 set.insert(i);
4567 }
4568
4569 let mut seen = Vec::new();
4570 for value in set.iter() {
4571 seen.push(value);
4572 if seen.len() == prefix {
4573 set.remove(&0);
4574 }
4575 }
4576
4577 for expected in 1..256u64 {
4578 assert!(
4579 yielded(&seen, expected),
4580 "prefix {prefix}: {expected} was present for the whole scan but was never yielded"
4581 );
4582 }
4583 }
4584 }
4585
4586 #[test]
4590 fn deleting_a_yielded_element_backwards_does_not_skip_a_live_one() {
4591 for prefix in 1..12usize {
4592 let set: BTreeSet<u64> = BTreeSet::new();
4593 for i in 0..256u64 {
4594 set.insert(i);
4595 }
4596
4597 let mut seen = Vec::new();
4598 for value in set.iter().rev() {
4599 seen.push(value);
4600 if seen.len() == prefix {
4601 set.remove(&255);
4602 }
4603 }
4604
4605 for expected in 0..255u64 {
4606 assert!(
4607 yielded(&seen, expected),
4608 "prefix {prefix}: {expected} was present for the whole scan but was never yielded"
4609 );
4610 }
4611 }
4612 }
4613}