fix: fix remove and clear returning true when the key is stale

This fixes a bug where `Slab::remove` and `Pool::clear` incorrectly
return `true` (indicating that the value was removed/cleared) when the
value has already been cleared at the current generation.

Now, these APIs and only return `true` when the slot at the given index
was actually marked for removal. When the generation is too old, these
methods now return `false` as expected.

src/lib.rs

@@ -629,9 +629,9 @@ impl<'a, T, C: cfg::Config> Drop for Guard<'a, T, C> {
         if self.inner.release() {
             atomic::fence(atomic::Ordering::Acquire);
             if Tid::<C>::current().as_usize() == self.shard.tid {
-                self.shard.remove_local(self.key);
+                self.shard.take_local(self.key);
             } else {
-                self.shard.remove_remote(self.key);
+                self.shard.take_remote(self.key);
             }
         }
     }

src/page/mod.rs

@@ -233,8 +233,7 @@ where
         self.slab.with(|slab| {
             let slab = unsafe { &*slab }.as_ref();
             if let Some(slot) = slab.and_then(|slab| slab.get(offset)) {
-                slot.try_remove_value(gen, offset, free_list);
-                true
+                slot.try_remove_value(gen, offset, free_list)
             } else {
                 false
             }
@@ -327,6 +326,26 @@ where
             }
         })
     }
+
+    pub(crate) fn clear<F: FreeList<C>>(
+        &self,
+        addr: Addr<C>,
+        gen: slot::Generation<C>,
+        free_list: &F,
+    ) -> bool {
+        let offset = addr.offset() - self.prev_sz;
+
+        test_println!("-> offset {:?}", offset);
+
+        self.slab.with(|slab| {
+            let slab = unsafe { &*slab }.as_ref();
+            if let Some(slot) = slab.and_then(|slab| slab.get(offset)) {
+                slot.clear_storage(gen, offset, free_list)
+            } else {
+                false
+            }
+        })
+    }
 }
 
 impl fmt::Debug for Local {

src/page/slot.rs

@@ -166,12 +166,13 @@ where
         }
     }
 
-    /// Marks this slot for mutation
+    /// Marks this slot to be released, returning `true` if the slot can be
+    /// mutated *now* and `false` otherwise.
     ///
     /// This method checks if there are any references to this slot. If there _are_ valid
     /// references, it just marks them for modification and returns and the next thread calling
     /// either `clear_storage` or `remove_value` will try and modify the storage
-    fn mark_release(&self, gen: Generation<C>) -> bool {
+    fn mark_release(&self, gen: Generation<C>) -> Option<bool> {
         let mut lifecycle = self.lifecycle.load(Ordering::Acquire);
         let mut curr_gen;
 
@@ -187,9 +188,23 @@ where
 
             // Is the slot still at the generation we are trying to remove?
             if gen != curr_gen {
-                return false;
+                return None;
             }
 
+            let state = Lifecycle::<C>::from_packed(lifecycle).state;
+            test_println!("-> mark_release; state={:?};", state);
+            match state {
+                State::Removing => {
+                    test_println!("--> mark_release; cannot release (already removed!)");
+                    return None;
+                }
+                State::Marked => {
+                    test_println!("--> mark_release; already marked;");
+                    break;
+                }
+                State::Present => {}
+            };
+
             // Set the new state to `MARKED`.
             let new_lifecycle = Lifecycle::<C>::MARKED.pack(lifecycle);
             test_println!(
@@ -218,7 +233,7 @@ where
 
         // Are there currently outstanding references to the slot? If so, it
         // will have to be removed when those references are dropped.
-        refs.value > 0
+        Some(refs.value == 0)
     }
 
     /// Mutates this slot.
@@ -358,7 +373,8 @@ where
         Some(gen)
     }
 
-    /// Tries to remove the value in the slot
+    /// Tries to remove the value in the slot, returning `true` if the value was
+    /// removed.
     ///
     /// This method tries to remove the value in the slot. If there are existing references, then
     /// the slot is marked for removal and the next thread calling either this method or
@@ -369,14 +385,32 @@ where
         gen: Generation<C>,
         offset: usize,
         free: &F,
-    ) -> Option<T> {
-        if self.mark_release(gen) {
-            None
-        } else {
-            // Otherwise, we can remove the slot now!
+    ) -> bool {
+        let should_remove = match self.mark_release(gen) {
+            // If `mark_release` returns `Some`, a value exists at this
+            // generation. The bool inside this option indicates whether or not
+            // _we're_ allowed to remove the value.
+            Some(should_remove) => should_remove,
+            // Otherwise, the generation we tried to remove has already expired,
+            // and we did not mark anything for removal.
+            None => {
+                test_println!(
+                    "-> try_remove_value; nothing exists at generation={:?}",
+                    gen
+                );
+                return false;
+            }
+        };
+
+        test_println!("-> try_remove_value; marked!");
+
+        if should_remove {
+            // We're allowed to remove the slot now!
             test_println!("-> try_remove_value; can remove now");
-            self.remove_value(gen, offset, free)
+            self.remove_value(gen, offset, free);
         }
+
+        true
     }
 
     #[inline]
@@ -416,13 +450,31 @@ where
         offset: usize,
         free: &F,
     ) -> bool {
-        if self.mark_release(gen) {
-            return false;
+        let should_clear = match self.mark_release(gen) {
+            // If `mark_release` returns `Some`, a value exists at this
+            // generation. The bool inside this option indicates whether or not
+            // _we're_ allowed to clear the value.
+            Some(should_clear) => should_clear,
+            // Otherwise, the generation we tried to remove has already expired,
+            // and we did not mark anything for removal.
+            None => {
+                test_println!(
+                    "-> try_clear_storage; nothing exists at generation={:?}",
+                    gen
+                );
+                return false;
+            }
+        };
+
+        test_println!("-> try_clear_storage; marked!");
+
+        if should_clear {
+            // We're allowed to remove the slot now!
+            test_println!("-> try_remove_value; can clear now");
+            return self.clear_storage(gen, offset, free);
         }
 
-        // Otherwise, we can remove the slot now!
-        test_println!("-> try_clear_storage; can remove now");
-        self.clear_storage(gen, offset, free)
+        true
     }
 
     /// Clear this slot's storage

src/pool.rs

@@ -388,9 +388,9 @@ where
         if self.inner.release() {
             atomic::fence(atomic::Ordering::Acquire);
             if Tid::<C>::current().as_usize() == self.shard.tid {
-                self.shard.mark_clear_local(self.key);
+                self.shard.clear_local(self.key);
             } else {
-                self.shard.mark_clear_remote(self.key);
+                self.shard.clear_remote(self.key);
             }
         }
     }
@@ -530,9 +530,9 @@ where
         if should_clear {
             atomic::fence(atomic::Ordering::Acquire);
             if Tid::<C>::current().as_usize() == self.shard.tid {
-                self.shard.mark_clear_local(self.key);
+                self.shard.clear_local(self.key);
             } else {
-                self.shard.mark_clear_remote(self.key);
+                self.shard.clear_remote(self.key);
             }
         }
     }

src/shard.rs

@@ -202,6 +202,29 @@ where
         shared.mark_clear(addr, C::unpack_gen(idx), shared.free_list())
     }
 
+    pub(crate) fn clear_local(&self, idx: usize) -> bool {
+        debug_assert_eq!(Tid::<C>::from_packed(idx).as_usize(), self.tid);
+        let (addr, page_index) = page::indices::<C>(idx);
+
+        if page_index > self.shared.len() {
+            return false;
+        }
+
+        self.shared[page_index].clear(addr, C::unpack_gen(idx), self.local(page_index))
+    }
+
+    pub(crate) fn clear_remote(&self, idx: usize) -> bool {
+        debug_assert_eq!(Tid::<C>::from_packed(idx).as_usize(), self.tid);
+        let (addr, page_index) = page::indices::<C>(idx);
+
+        if page_index > self.shared.len() {
+            return false;
+        }
+
+        let shared = &self.shared[page_index];
+        shared.clear(addr, C::unpack_gen(idx), shared.free_list())
+    }
+
     #[inline(always)]
     fn local(&self, i: usize) -> &page::Local {
         #[cfg(debug_assertions)]

src/tests/loom_pool.rs

@@ -117,8 +117,9 @@ fn concurrent_create_with_clear() {
         while next.is_none() {
             next = cvar.wait(next).unwrap();
         }
-        assert!(!pool.clear(idx1));
-
+        // the item should be marked (clear returns true)...
+        assert!(pool.clear(idx1));
+        // ...but the value shouldn't be removed yet.
         item1.assert_not_clear();
 
         t1.join().expect("thread 1 unable to join");