diff --git a/src/arcache/ll.rs b/src/arcache/ll.rs index 32ec25a..58fd542 100644 --- a/src/arcache/ll.rs +++ b/src/arcache/ll.rs @@ -22,14 +22,12 @@ fn alloc_nid() -> usize { { LL_ALLOC_LIST.with(|llist| llist.lock().unwrap().insert(nid)); } - eprintln!("Allocate -> {:?}", nid); + // eprintln!("LL Allocate -> {:?}", nid); nid } -#[cfg(all(test, not(miri)))] +#[cfg(all(test, not(miri), not(feature = "dhat-heap")))] fn release_nid(nid: usize) { - println!("Release -> {:?}", nid); - #[cfg(not(feature = "dhat-heap"))] { let r = LL_ALLOC_LIST.with(|llist| llist.lock().unwrap().remove(&nid)); assert!(r); @@ -53,15 +51,6 @@ pub trait LLWeight { fn ll_weight(&self) -> usize; } -/* -impl LLWeight for T { - #[inline] - default fn ll_weight(&self) -> usize { - 1 - } -} -*/ - #[derive(Clone, Debug)] pub(crate) struct LL where @@ -138,8 +127,7 @@ impl LLNodeOwned where K: LLWeight + Clone + Debug, { - #[allow(clippy::wrong_self_convention)] - fn into_inner(&mut self) -> *mut LLNode { + fn into_inner(mut self) -> *mut LLNode { let x = self.inner; self.inner = ptr::null_mut(); x @@ -187,7 +175,7 @@ where debug_assert!((*self.inner).next.is_null()); debug_assert!((*self.inner).prev.is_null()); } - panic!("dropping LLNodeOwned"); + panic!("dropping LLNodeOwned which has valid content, this should never happen!"); } } } @@ -233,19 +221,29 @@ where self.append_n(n) } - // Append an arbitrary node into this set. - pub(crate) fn append_n(&mut self, mut owned: LLNodeOwned) -> LLNodeRef { + // Append an arbitrary node into this set, at the tail end. + pub(crate) fn append_n(&mut self, owned: LLNodeOwned) -> LLNodeRef { // Who is to the left of tail? let n = owned.into_inner(); unsafe { self.size += (*(*n).k.as_ptr()).ll_weight(); // must be untagged // assert!((*n).tag == 0); + // Tail has no trailing nodes. debug_assert!((*self.tail).next.is_null()); + // Tail must have a previous node. debug_assert!(!(*self.tail).prev.is_null()); + // What is that predecessor? let pred = (*self.tail).prev; debug_assert!(!pred.is_null()); + // Assert that the predecessor points at tail correctly. + // We know now that: + // + // pred <-> tail debug_assert!((*pred).next == self.tail); + + // Tell our node where we are. + // pred <-> n <-> tail (*n).prev = pred; (*n).next = self.tail; // (*n).tag = self.tag; @@ -259,6 +257,7 @@ where debug_assert!(!(*(*n).next).prev.is_null()); debug_assert!((*(*n).prev).next == n); debug_assert!((*(*n).next).prev == n); + // We have asserted the list is sane. }; LLNodeRef { inner: n } } @@ -269,30 +268,57 @@ where if n.inner == unsafe { (*self.tail).prev } { // Done, no-op } else { + let previous_size = self.size; + let owned = self.extract(n); self.append_n(owned); + + let after_size = self.size; + + // Ensure that we didn't actually alter the size of the list during the operation, + // we only re-arranged the content. + debug_assert_eq!(previous_size, after_size); } } // remove this node from the ll, and return it's ptr. - pub(crate) fn pop(&mut self) -> LLNodeOwned { - let n = unsafe { (*self.head).next }; - let owned = self.extract(LLNodeRef { inner: n }); - debug_assert!(!owned.is_null()); - debug_assert!(owned.inner != self.head); - debug_assert!(owned.inner != self.tail); - owned + pub(crate) fn pop(&mut self) -> Option> { + let next = unsafe { (*self.head).next }; + if next == self.tail { + None + } else { + let n = unsafe { (*self.head).next }; + let owned = self.extract(LLNodeRef { inner: n }); + debug_assert!(!owned.is_null()); + debug_assert!(owned.inner != self.head); + debug_assert!(owned.inner != self.tail); + Some(owned) + } + } + + pub(crate) fn pop_n_free(&mut self) -> Option { + if let Some(owned) = self.pop() { + let ll_node = owned.into_inner(); + let k = LLNode::into_inner(ll_node); + Some(k) + } else { + None + } } // Cut a node out from this list from any location. pub(crate) fn extract(&mut self, n: LLNodeRef) -> LLNodeOwned { - assert!(self.size > 0); assert!(!n.is_null()); + assert!(self.size > 0); unsafe { // We should have a prev and next debug_assert!(!(*n.inner).prev.is_null()); debug_assert!(!(*n.inner).next.is_null()); // And that prev's next is us, and next's prev is us. + // This is asserting that we have a proper construction + // + // prev <-> n <-> next + // debug_assert!(!(*(*n.inner).prev).next.is_null()); debug_assert!(!(*(*n.inner).next).prev.is_null()); debug_assert!((*(*n.inner).prev).next == n.inner); @@ -305,14 +331,24 @@ where unsafe { let prev = (*n.inner).prev; let next = (*n.inner).next; + // Currently is: + // // prev <-> n <-> next (*next).prev = prev; (*prev).next = next; + // Now is + // prev <-> next + // Null things for paranoia. if cfg!(test) || cfg!(debug_assertions) { (*n.inner).prev = ptr::null_mut(); (*n.inner).next = ptr::null_mut(); } + // Finally, we have + // + // null <- n -> null + // prev <-> next + // (*n).tag = 0; } @@ -324,10 +360,7 @@ where } pub(crate) fn drop_head(&mut self) { - assert!(self.size > 0); - let next = unsafe { (*self.head).next }; - if next != self.tail { - let mut owned = self.pop(); + if let Some(owned) = self.pop() { let n = owned.into_inner(); LLNode::free(n); } @@ -361,6 +394,62 @@ where Some(l) } } + + #[cfg(test)] + pub(crate) fn verify(&self) { + // Walk the list to ensure it's sane. + + let head = self.head; + let tail = self.tail; + + let expect_size = self.size; + let mut size = 0; + + // Establish that the sentinel nodes exist, and that they + // have the correct out bound null/non-null pointers. + assert_ne!(head, tail); + + unsafe { + assert!((*head).prev.is_null()); + assert!(!(*head).next.is_null()); + + assert!((*tail).next.is_null()); + assert!(!(*tail).prev.is_null()); + } + + // Now we walk each item to assert that they have the correct structure. + let mut n = unsafe { (*head).next }; + // We have to manually check head here. + unsafe { + assert_eq!((*n).prev, head); + } + + while n != tail { + unsafe { + // Setup the pointer for the next iteration. + let next = (*n).next; + + // Add the size. + size += (*(*n).k.as_ptr()).ll_weight(); + + // assert we have outbound pointers. + assert!(!(*n).prev.is_null()); + assert!(!(*n).next.is_null()); + + // Assert that the previous node points to us. + assert!(!(*(*n).prev).next.is_null()); + assert!((*(*n).prev).next == n); + // Assert that the next node points back to us. + // NOTE: This accounts for tail pointing back to us. + assert!(!(*(*n).next).prev.is_null()); + assert!((*(*n).next).prev == n); + + n = next; + } + } + + assert_eq!(expect_size, size); + } } impl Drop for LL @@ -446,13 +535,20 @@ where } #[inline] - fn free(v: *mut Self) { + fn into_inner(v: *mut Self) -> K { debug_assert!(!v.is_null()); let llnode = unsafe { Box::from_raw(v) }; + let k = unsafe { llnode.k.assume_init() }; + #[cfg(all(test, not(miri), not(feature = "dhat-heap")))] + release_nid(llnode.nid); + + k + } + + #[inline] + fn free(v: *mut Self) { // drop the inner k. - let _ = unsafe { llnode.k.assume_init() }; - #[cfg(all(test, not(miri)))] - release_nid(llnode.nid) + let _ = Self::into_inner(v); } #[inline] @@ -460,7 +556,7 @@ where debug_assert!(!v.is_null()); let _llnode = unsafe { Box::from_raw(v) }; // Markers never have a k to drop. - #[cfg(all(test, not(miri)))] + #[cfg(all(test, not(miri), not(feature = "dhat-heap")))] release_nid(_llnode.nid) } } @@ -552,7 +648,7 @@ mod tests { assert!(ll.peek_tail().unwrap().as_ref() == &1); // pop from head - let n3 = ll.pop(); + let n3 = ll.pop().unwrap(); assert!(ll.len() == 3); assert!(ll.peek_head().unwrap().as_ref() == &4); assert!(ll.peek_tail().unwrap().as_ref() == &1); @@ -574,7 +670,7 @@ mod tests { assert!(ll.peek_head().unwrap().as_ref() == &4); assert!(ll.peek_tail().unwrap().as_ref() == &4); // Remove last - let n4 = ll.pop(); + let n4 = ll.pop().unwrap(); assert!(ll.len() == 0); assert!(ll.peek_head().is_none()); assert!(ll.peek_tail().is_none()); @@ -610,9 +706,9 @@ mod tests { assert!(ll.len() == 8); let _n2 = ll.append_k(Weighted { _i: 2 }); assert!(ll.len() == 16); - let n1 = ll.pop(); + let n1 = ll.pop().unwrap(); assert!(ll.len() == 8); - let n2 = ll.pop(); + let n2 = ll.pop().unwrap(); assert!(ll.len() == 0); // Add back so they drop ll.append_n(n1); diff --git a/src/arcache/mod.rs b/src/arcache/mod.rs index ff70bdb..e5795ef 100644 --- a/src/arcache/mod.rs +++ b/src/arcache/mod.rs @@ -22,7 +22,7 @@ use crate::hashmap::{ HashMap as DataMap, HashMapReadTxn as DataMapReadTxn, HashMapWriteTxn as DataMapWriteTxn, }; -#[cfg(feature = "arcache-is-hashtrie")] +#[cfg(all(feature = "arcache-is-hashtrie", not(feature = "arcache-is-hashmap")))] use crate::hashtrie::{ HashTrie as DataMap, HashTrieReadTxn as DataMapReadTxn, HashTrieWriteTxn as DataMapWriteTxn, }; @@ -59,6 +59,8 @@ const WATERMARK_DISABLE_MIN: usize = 128; const WATERMARK_DISABLE_DIVISOR: usize = 20; const WATERMARK_DISABLE_RATIO: usize = 18; +const HAUNTED_SIZE: usize = 1; + enum ThreadCacheItem { Present(V, bool, usize), Removed(bool), @@ -748,6 +750,15 @@ impl< { // drain tlocal into the main cache. tlocal.into_iter().for_each(|(k, tcio)| { + #[cfg(test)] + { + inner.rec.verify(); + inner.freq.verify(); + inner.ghost_rec.verify(); + inner.ghost_freq.verify(); + inner.haunted.verify(); + } + let r = cache.get_mut(&k); match (r, tcio) { (None, ThreadCacheItem::Present(tci, clean, size)) => { @@ -759,7 +770,12 @@ impl< }); // stats.write_includes += 1; stats.include(&k); - cache.insert(k, CacheItem::Rec(llp, tci)); + // The key MUST NOT exist in the cache already. + let existing = cache.insert(k, CacheItem::Rec(llp, tci)); + assert!( + existing.is_none(), + "Impossible state! Key must not already exist in cache!" + ); } (None, ThreadCacheItem::Removed(clean)) => { assert!(clean); @@ -767,9 +783,16 @@ impl< let llp = inner.haunted.append_k(CacheItemInner { k: k.clone(), txid: commit_txid, - size: 1, + size: HAUNTED_SIZE, }); - cache.insert(k, CacheItem::Haunted(llp)); + // The key MUST NOT exist in the cache already. + let existing = cache.insert(k, CacheItem::Haunted(llp)); + assert!( + existing.is_none(), + "Impossible state! Key must not already exist in cache!" + ); + // Must be now in haunted! + debug_assert!(inner.haunted.len() > 0); } (Some(ref mut ci), ThreadCacheItem::Removed(clean)) => { assert!(clean); @@ -778,30 +801,39 @@ impl< CacheItem::Freq(llp, _v) => { let mut owned = inner.freq.extract(llp.clone()); owned.as_mut().txid = commit_txid; + owned.as_mut().size = HAUNTED_SIZE; let pointer = inner.haunted.append_n(owned); + debug_assert!(inner.haunted.len() > 0); CacheItem::Haunted(pointer) } CacheItem::Rec(llp, _v) => { // Remove the node and put it into freq. let mut owned = inner.rec.extract(llp.clone()); owned.as_mut().txid = commit_txid; + owned.as_mut().size = HAUNTED_SIZE; let pointer = inner.haunted.append_n(owned); + debug_assert!(inner.haunted.len() > 0); CacheItem::Haunted(pointer) } CacheItem::GhostFreq(llp) => { let mut owned = inner.ghost_freq.extract(llp.clone()); owned.as_mut().txid = commit_txid; + owned.as_mut().size = HAUNTED_SIZE; let pointer = inner.haunted.append_n(owned); + debug_assert!(inner.haunted.len() > 0); CacheItem::Haunted(pointer) } CacheItem::GhostRec(llp) => { let mut owned = inner.ghost_rec.extract(llp.clone()); owned.as_mut().txid = commit_txid; + owned.as_mut().size = HAUNTED_SIZE; let pointer = inner.haunted.append_n(owned); + debug_assert!(inner.haunted.len() > 0); CacheItem::Haunted(pointer) } CacheItem::Haunted(llp) => { unsafe { llp.make_mut().txid = commit_txid }; + debug_assert!(inner.haunted.len() > 0); CacheItem::Haunted(llp.clone()) } }; @@ -866,7 +898,15 @@ impl< CacheItem::Rec(pointer, tci) } CacheItem::Haunted(llp) => { + // Moving from haunted to recent. + // How can an item be Haunted, but then not be in the haunted set? + + // eprintln!("{:?}", inner.haunted.len()); + let before_len = inner.haunted.len(); + debug_assert!(inner.haunted.len() > 0); let mut owned = inner.haunted.extract(llp.clone()); + debug_assert!(before_len - HAUNTED_SIZE == inner.haunted.len()); + owned.as_mut().txid = commit_txid; owned.as_mut().size = size; stats.include_haunted(&owned.as_ref().k); @@ -878,6 +918,15 @@ impl< mem::swap(*ci, &mut next_state); } } + + #[cfg(test)] + { + inner.rec.verify(); + inner.freq.verify(); + inner.ghost_rec.verify(); + inner.ghost_freq.verify(); + inner.haunted.verify(); + } }); } @@ -918,6 +967,7 @@ impl< // We can't do anything about this ... // Don't touch or rearrange the haunted list, it should be // in commit txid order. + debug_assert!(inner.haunted.len() > 0); CacheItem::Haunted(llp.to_owned()) } }; @@ -1043,12 +1093,25 @@ impl< } } CacheItem::Haunted(llp) => { + // if + // the haunted item is newer + // OR + // inclusion is older than our minimum + // then the item is skipped. if llp.as_ref().txid > txid || inner.min_txid > txid { None } else { + // ELSE we need to update the txid of the haunted item + // to ensure that it's at the list head. + debug_assert!(inner.haunted.len() > 0); + let before_len = inner.haunted.len(); + let mut owned = inner.haunted.extract(llp.to_owned()); + + debug_assert!(before_len - HAUNTED_SIZE == inner.haunted.len()); + owned.as_mut().txid = txid; - owned.as_mut().size = size; + debug_assert!(owned.as_mut().size == HAUNTED_SIZE); stats.include_haunted(&owned.as_mut().k); let pointer = inner.rec.append_n(owned); Some(CacheItem::Rec(pointer, iv)) @@ -1069,7 +1132,12 @@ impl< size, }); stats.include(&k); - cache.insert(k, CacheItem::Rec(llp, iv)); + // The key MUST NOT exist in the cache already. + let existing = cache.insert(k, CacheItem::Rec(llp, iv)); + assert!( + existing.is_none(), + "Impossible state! Key must not already exist in cache!" + ); } } }; @@ -1156,27 +1224,63 @@ impl< size: usize, txid: u64, ) { + let to_ll_before = to_ll.len(); + let ll_before = ll.len(); + let mut added = 0; + let mut removed = 0; + while ll.len() > size { - let mut owned = ll.pop(); - debug_assert!(!owned.is_null()); + #[cfg(test)] + { + ll.verify(); + to_ll.verify(); + } - // Set the item's evict txid. - owned.as_mut().txid = txid; + if let Some(mut owned) = ll.pop() { + debug_assert!(!owned.is_null()); - let pointer = to_ll.append_n(owned); - let mut r = cache.get_mut(&pointer.as_ref().k); + // Track the sizes. + removed += owned.as_mut().size; - match r { - Some(ref mut ci) => { - // Now change the state. - let mut next_state = CacheItem::Haunted(pointer); - mem::swap(*ci, &mut next_state); - } - None => { - // Impossible state! - unreachable!(); - } - }; + assert_eq!(ll.len(), ll_before - removed); + + // Set the item's evict txid. + owned.as_mut().txid = txid; + // Trim the haunted size as needed. + owned.as_mut().size = HAUNTED_SIZE; + added += HAUNTED_SIZE; + + let pointer = to_ll.append_n(owned); + + assert_eq!( + to_ll.len(), + to_ll_before + added, + "Impossible State! List lengths are no longer consistent!" + ); + + let mut r = cache.get_mut(&pointer.as_ref().k); + + match r { + Some(ref mut ci) => { + // Now change the state. + let mut next_state = CacheItem::Haunted(pointer); + mem::swap(*ci, &mut next_state); + } + None => { + // Impossible state! + unreachable!(); + } + }; + } else { + // Impossible state! + unreachable!(); + } + + #[cfg(test)] + { + ll.verify(); + to_ll.verify(); + } } } @@ -1193,44 +1297,66 @@ impl< debug_assert!(ll.len() >= size); while ll.len() > size { - let mut owned = ll.pop(); - debug_assert!(!owned.is_null()); - let mut r = cache.get_mut(&owned.as_ref().k); - // Set the item's evict txid. - owned.as_mut().txid = txid; - match r { - Some(ref mut ci) => { - let mut next_state = match &ci { - CacheItem::Freq(llp, _v) => { - debug_assert!(llp == &owned); - // No need to extract, already popped! - // $ll.extract(*llp); - stats.evict_from_frequent(&owned.as_ref().k); - let pointer = to_ll.append_n(owned); - CacheItem::GhostFreq(pointer) - } - CacheItem::Rec(llp, _v) => { - debug_assert!(llp == &owned); - // No need to extract, already popped! - // $ll.extract(*llp); - stats.evict_from_recent(&owned.as_mut().k); - let pointer = to_ll.append_n(owned); - CacheItem::GhostRec(pointer) - } - _ => { - // Impossible state! - unreachable!(); - } - }; - // Now change the state. - mem::swap(*ci, &mut next_state); - } - None => { - // Impossible state! - unreachable!(); - } + #[cfg(test)] + { + ll.verify(); + to_ll.verify(); } - } + + if let Some(mut owned) = ll.pop() { + debug_assert!(!owned.is_null()); + let mut r = cache.get_mut(&owned.as_ref().k); + // Set the item's evict txid. + owned.as_mut().txid = txid; + match r { + Some(ref mut ci) => { + let mut next_state = match &ci { + CacheItem::Freq(llp, _v) => { + // The pointer from any key MUST be unique! + assert!(llp == &owned, "Impossible State! Pointer in map does not match the pointer from the list!"); + // No need to extract, already popped! + // $ll.extract(*llp); + stats.evict_from_frequent(&owned.as_ref().k); + let pointer = to_ll.append_n(owned); + CacheItem::GhostFreq(pointer) + } + CacheItem::Rec(llp, _v) => { + // The pointer from any key MUST be unique! + assert!(llp == &owned, "Impossible State! Pointer in map does not match the pointer from the list!"); + // No need to extract, already popped! + // $ll.extract(*llp); + stats.evict_from_recent(&owned.as_mut().k); + let pointer = to_ll.append_n(owned); + CacheItem::GhostRec(pointer) + } + _ => { + // Impossible state! All members of the from-ll, must be + // in either the frequent or recent state. + unreachable!(); + } + }; + // Now change the state. + mem::swap(*ci, &mut next_state); + } + None => { + // Impossible state! This indicates that the key was already + // removed. Only one key -> linked-list-pointer should exist at + // anytime. If we already removed this, that indicates there were + // two llp's with the same key! + unreachable!(); + } + }; + } else { + // Impossible state! + unreachable!(); + } + + #[cfg(test)] + { + ll.verify(); + to_ll.verify(); + } + } // end while } #[allow(clippy::cognitive_complexity)] @@ -1378,8 +1504,14 @@ impl< ) where S: ARCacheWriteStat, { - while ll.len() > 0 { - let mut owned = ll.pop(); + while let Some(mut owned) = ll.pop() { + #[cfg(test)] + { + ll.verify(); + gf.verify(); + gr.verify(); + } + debug_assert!(!owned.is_null()); // Set the item's eviction txid. @@ -1410,10 +1542,20 @@ impl< mem::swap(*ci, &mut next_state); } None => { - // Impossible state! + // Impossible state! This indicates that the key was already + // removed. Only one key -> linked-list-pointer should exist at + // anytime. If we already removed this, that indicates there were + // two llp's with the same key! unreachable!(); } } + + #[cfg(test)] + { + ll.verify(); + gf.verify(); + gr.verify(); + } } // end while } @@ -1423,17 +1565,33 @@ impl< min_txid: u64, ) { while let Some(node) = ll.peek_head() { - if min_txid > node.txid { + #[cfg(test)] + { + ll.verify(); + } + + // if the node is older than our min txid. + if node.txid < min_txid { + let before_len = ll.len(); + debug_assert!(ll.len() > 0); + // Need to free from the cache. cache.remove(&node.k); // Okay, this node can be trimmed. ll.drop_head(); + + debug_assert!(before_len - HAUNTED_SIZE == ll.len()); } else { // We are done with this loop, everything else // is newer. break; } + + #[cfg(test)] + { + ll.verify(); + } } } @@ -1562,10 +1720,13 @@ impl< // // If we drop below this again, they'll go back to just insert/remove content only mode. if init_above_watermark { + // We were above, now we are below the limit, go back to just insert/remove only mode. if (inner.freq.len() + inner.rec.len()) < shared.watermark { self.above_watermark.store(false, Ordering::Relaxed); } } else if (inner.freq.len() + inner.rec.len()) >= shared.watermark { + // we were not above above the watermark but now the cache is large enough + // to demand that we should be tracking data. self.above_watermark.store(true, Ordering::Relaxed); } @@ -2054,7 +2215,6 @@ impl< /// Insert an item to the cache, with an associated weight/size factor. See also `insert` pub fn insert_sized(&mut self, k: K, v: V, size: NonZeroUsize) { - let mut v = v; let size = size.get(); // Send a copy forward through time and space. // let _ = self.tx.try_send( @@ -2077,24 +2237,27 @@ impl< // We have a cache, so lets update it. if let Some(ref mut cache) = self.tlocal { self.stats.local_include(); - let n = if cache.tlru.len() >= cache.read_size { - let mut owned = cache.tlru.pop(); - // swap the old_key/old_val out - let mut k_clone = k.clone(); - mem::swap(&mut k_clone, &mut owned.as_mut().k); - mem::swap(&mut v, &mut owned.as_mut().v); - // remove old K from the tree: - cache.set.remove(&k_clone); - // Return the owned node into the lru - cache.tlru.append_n(owned) - } else { - // Just add it! - cache.tlru.append_k(ReadCacheItem { - k: k.clone(), - v, - size, - }) - }; + while cache.tlru.len() >= cache.read_size { + if let Some(owned_inner) = cache.tlru.pop_n_free() { + let existing = cache.set.remove(&owned_inner.k); + // Must have been present. + assert!( + existing.is_some(), + "Impossible state! Key was NOT present in cache!" + ); + } else { + // Somehow the list is empty, but we still are oversize? + debug_assert!(false); + break; + } + } + + // Now add it, as we have enough space. + let n = cache.tlru.append_k(ReadCacheItem { + k: k.clone(), + v, + size, + }); let r = cache.set.insert(k, n); // There should never be a previous value. assert!(r.is_none()); @@ -3155,13 +3318,25 @@ mod tests { #[allow(dead_code)] pub static RUNNING: AtomicBool = AtomicBool::new(false); + #[allow(dead_code)] + pub static READ_OPERATIONS: u32 = 1024; + #[allow(dead_code)] + pub static WRITE_OPERATIONS: u32 = 10240; + + #[allow(dead_code)] + pub static CACHE_SIZE: u32 = 64; + pub static VALUE_MAX_RANGE: u32 = CACHE_SIZE * 8; + #[cfg(test)] fn multi_thread_worker(arc: Arc, Box>>) { while RUNNING.load(Ordering::Relaxed) { let mut rd_txn = arc.read(); - for _i in 0..128 { - let x = rand::random::(); + use rand::Rng; + let mut rng = rand::rng(); + + for _i in 0..VALUE_MAX_RANGE { + let x = rng.random_range(0..VALUE_MAX_RANGE); if rd_txn.get(&x).is_none() { rd_txn.insert(Box::new(x), Box::new(x)) @@ -3173,22 +3348,41 @@ mod tests { #[allow(dead_code)] #[cfg_attr(miri, ignore)] #[cfg_attr(feature = "dhat-heap", test)] - #[cfg(test)] fn test_cache_stress_1() { #[cfg(feature = "dhat-heap")] let _profiler = dhat::Profiler::builder().trim_backtraces(None).build(); + use rand::Rng; + let mut rng = rand::rng(); + let arc: Arc, Box>> = Arc::new( ARCacheBuilder::default() - .set_size(64, 4) + .set_size(CACHE_SIZE as usize, 4) .build() .expect("Invalid cache parameters!"), ); - let thread_count = 4; + // Do some writes ... + for _i in 0..WRITE_OPERATIONS { + let mut wr_txn = arc.write(); + + let x = rng.random_range(0..VALUE_MAX_RANGE); + + if wr_txn.get(&x).is_none() { + wr_txn.insert(Box::new(x), Box::new(x)) + } + + // Can corrupt here no issue. + wr_txn.commit(); + } + + eprintln!("writes pass"); + + let thread_count = 8; RUNNING.store(true, Ordering::Relaxed); + // Now do writes and reads concurrently let handles: Vec<_> = (0..thread_count) .map(|_| { // Build the threads. @@ -3197,9 +3391,15 @@ mod tests { }) .collect(); - for x in 0..1024 { + std::thread::sleep(std::time::Duration::from_secs(5)); + + // Everything is fine until we write. + + for _i in 0..WRITE_OPERATIONS { let mut wr_txn = arc.write(); + let x = rng.random_range(0..VALUE_MAX_RANGE); + if wr_txn.get(&x).is_none() { wr_txn.insert(Box::new(x), Box::new(x)) } @@ -3210,10 +3410,10 @@ mod tests { RUNNING.store(false, Ordering::Relaxed); for handle in handles { - handle.join().unwrap(); + if let Err(err) = handle.join() { + std::panic::resume_unwind(err) + } } - - drop(arc); } #[test] diff --git a/src/internals/hashtrie/cursor.rs b/src/internals/hashtrie/cursor.rs index 78791b7..47a1bb8 100644 --- a/src/internals/hashtrie/cursor.rs +++ b/src/internals/hashtrie/cursor.rs @@ -76,15 +76,15 @@ macro_rules! hash_key { }}; } -#[cfg(all(test, not(miri)))] +#[cfg(all(test, not(miri), not(feature = "dhat-heap")))] thread_local!(static ALLOC_LIST: Mutex> = const { Mutex::new(BTreeSet::new()) }); -#[cfg(all(test, not(miri)))] +#[cfg(all(test, not(miri), not(feature = "dhat-heap")))] thread_local!(static WRITE_LIST: Mutex> = const { Mutex::new(BTreeSet::new()) }); #[cfg(test)] fn assert_released() { - #[cfg(not(miri))] + #[cfg(all(test, not(miri), not(feature = "dhat-heap")))] { let is_empty = ALLOC_LIST.with(|llist| { let x = llist.lock().unwrap(); @@ -129,6 +129,7 @@ impl Debug for Ptr { fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { f.debug_struct("Ptr") .field("p", &self.p) + .field("untagged", &self.p.map_addr(|a| a & UNTAG)) .field("bucket", &self.is_bucket()) .field("dirty", &self.is_dirty()) .field("null", &self.is_null()) @@ -163,7 +164,7 @@ impl Ptr { self.p.addr() & FLAG_DIRTY == FLAG_DIRTY } - #[cfg(all(test, not(miri)))] + #[cfg(all(test, not(miri), not(feature = "dhat-heap")))] fn untagged(&self) -> Self { let p = self.p.map_addr(|a| a & UNTAG); Ptr { p } @@ -187,7 +188,11 @@ impl Ptr { pub(crate) fn as_bucket(&self) -> &Bucket { debug_assert!(self.is_bucket()); #[cfg(all(test, not(miri), not(feature = "dhat-heap")))] - ALLOC_LIST.with(|llist| assert!(llist.lock().unwrap().contains(&self.untagged()))); + { + let contains = + ALLOC_LIST.with(|llist| llist.lock().unwrap().contains(&self.untagged())); + assert!(contains, "as_bucket -> {:?} MISSING", self); + } unsafe { &*(self.p.map_addr(|a| a & UNTAG) as *const Bucket) } } @@ -218,7 +223,11 @@ impl Ptr { pub(crate) fn as_branch(&self) -> &Branch { debug_assert!(self.is_branch()); #[cfg(all(test, not(miri), not(feature = "dhat-heap")))] - ALLOC_LIST.with(|llist| assert!(llist.lock().unwrap().contains(&self.untagged()))); + { + let contains = + ALLOC_LIST.with(|llist| llist.lock().unwrap().contains(&self.untagged())); + assert!(contains, "as_branch -> {:?} MISSING", self); + } unsafe { &*(self.p.map_addr(|a| a & UNTAG) as *const Branch) } } @@ -257,6 +266,7 @@ impl Ptr { } fn free(&self) { + // eprintln!("free -> {:?}", self); // We MUST have allocated this, else it's a double free #[cfg(all(test, not(miri), not(feature = "dhat-heap")))] ALLOC_LIST.with(|llist| assert!(llist.lock().unwrap().contains(&self.untagged()))); @@ -283,11 +293,20 @@ impl Ptr { impl From>> for Ptr { fn from(b: Box>) -> Self { let rptr: *mut Branch = Box::into_raw(b); + + // Assert that the tag bits are not present + assert_eq!( + rptr, + rptr.map_addr(|a| { a & UNTAG }), + "Impossible State! rptr tag bits are set incorrectly!" + ); + #[allow(clippy::let_and_return)] let r = Self { p: rptr.map_addr(|a| a | FLAG_BRANCH) as *mut i32, }; - #[cfg(all(test, not(miri)))] + + #[cfg(all(test, not(miri), not(feature = "dhat-heap")))] ALLOC_LIST.with(|llist| llist.lock().unwrap().insert(r.untagged())); r } @@ -296,11 +315,20 @@ impl From>> for Ptr { impl From>> for Ptr { fn from(b: Box>) -> Self { let rptr: *mut Bucket = Box::into_raw(b); + + // Assert that the tag bits are not present + assert_eq!( + rptr, + rptr.map_addr(|a| { a & UNTAG }), + "Impossible State! rptr tag bits are set incorrectly!" + ); + #[allow(clippy::let_and_return)] let r = Self { p: rptr.map_addr(|a| a | FLAG_BUCKET) as *mut i32, }; - #[cfg(all(test, not(miri)))] + + #[cfg(all(test, not(miri), not(feature = "dhat-heap")))] ALLOC_LIST.with(|llist| llist.lock().unwrap().insert(r.untagged())); r }