drop for slot/receiver

distaff/src/queue.rs

@@ -2,7 +2,7 @@ use std::{
     cell::UnsafeCell,
     collections::{HashMap, HashSet},
     marker::{PhantomData, PhantomPinned},
-    mem::MaybeUninit,
+    mem::{self, MaybeUninit},
     pin::Pin,
     sync::{
         Arc,
@@ -48,6 +48,26 @@ struct Slot<T> {
     state: AtomicU8,
 }
 
+impl<T> Slot<T> {
+    fn new() -> Self {
+        Self {
+            value: UnsafeCell::new(MaybeUninit::uninit()),
+            state: AtomicU8::new(0), // 0 means empty
+        }
+    }
+
+    fn set(&self, value: T) {}
+}
+
+impl<T> Drop for Slot<T> {
+    fn drop(&mut self) {
+        // SAFETY: The value is only initialized when the state is set to 1.
+        if mem::needs_drop::<T>() && self.state.load(Ordering::Acquire) == 1 {
+            unsafe { self.value.as_mut_unchecked().assume_init_drop() };
+        }
+    }
+}
+
 // const BLOCK_SIZE: usize = 8;
 // struct Block<T> {
 //     next: AtomicPtr<Block<T>>,
@@ -139,6 +159,19 @@ impl<T> Receiver<T> {
     }
 }
 
+impl<T> Drop for Receiver<T> {
+    fn drop(&mut self) {
+        if mem::needs_drop::<T>() {
+            // lock the queue
+            let _guard = self.queue.lock();
+            let queue = self.queue.inner();
+
+            // remove the receiver from the queue
+            _ = queue.owned.remove(&self.get_token());
+        }
+    }
+}
+
 impl<T: Send> Receiver<T> {
     pub fn recv(&self) -> T {
         let token = self.get_token();
@@ -150,6 +183,7 @@ impl<T: Send> Receiver<T> {
 
             // check if someone has sent a message to this receiver
             if let Some(t) = queue.poll(token) {
+                queue.parked.remove(&token);
                 return t;
             }