fix: use slot instead of trigger for async recv

# Fix async rendezvous channel deadlock (Issue #140) ## Problem Async receivers use `Hook::trigger` which has no slot to store messages. When a sender finds a waiting async receiver on a rendezvous channel (bounded(0)), it: 1. Calls `fire_send()` which returns `(Some(msg), signal)` because trigger has no slot 2. Puts the message into `chan.queue` 3. Fires the signal to wake the receiver 4. Returns `Ok(())` immediately This violates rendezvous semantics where the sender should block until the receiver actually takes the message. The premature return causes race conditions and deadlocks in high-throughput scenarios. ## Root Cause In `src/async.rs`, `RecvFut::poll_inner` creates a trigger hook: ```rust || Hook::trigger(AsyncSignal::new(cx, stream)) ``` `Hook::trigger` has no slot (`self.0 = None`), so when `fire_send` is called: ```rust pub fn fire_send(&self, msg: T) -> (Option<T>, &S) { let ret = match self.lock() { Some(mut lock) => { *lock = Some(msg); None } None => Some(msg), // trigger returns Some(msg) }; (ret, self.signal()) } ``` The message is returned back and placed in the queue instead of being delivered directly to the receiver. ## Fix Change async receivers to use `Hook::slot` instead of `Hook::trigger`: ```rust // Before || Hook::trigger(AsyncSignal::new(cx, stream)) // After || Hook::slot(None, AsyncSignal::new(cx, stream)) ``` And check the slot first when polling: ```rust if let Some(hook) = self.hook.as_ref() { // Check if message was delivered directly to our slot if let Some(msg) = hook.try_take() { return Poll::Ready(Ok(msg)); } // ... rest of the logic } ``` ## Reproduction ```rust use flume::bounded; use std::time::{Duration, Instant}; const MESSAGES: usize = 1_000_000; const THREADS: usize = 4; #[tokio::main] async fn main() { // Run MPMC test repeatedly (simulating benchmark) loop { let (tx, rx) = bounded::<usize>(0); let mut handles = Vec::new(); for _ in 0..THREADS { let tx = tx.clone(); handles.push(tokio::spawn(async move { for i in 1..MESSAGES / THREADS + 1 { tx.send_async(i).await.unwrap(); } })); } for _ in 0..THREADS { let rx = rx.clone(); handles.push(tokio::spawn(async move { for _ in 0..MESSAGES / THREADS { rx.recv_async().await.unwrap(); } })); } for h in handles { h.await.unwrap(); } // Deadlocks after 1-2 iterations without the fix } } ``` ## Testing - All existing tests pass - The reproduction case completes successfully with the fix - Benchmark `flume-async` runs to completion ## Notes - Zero-sized types (ZST) may not trigger this bug due to compiler optimizations - The bug is more likely to manifest in repeated runs or high-throughput scenarios - This is related to Issue #140