diff --git a/src/channel.rs b/src/channel.rs index 0d7e4b0..8d59348 100644 --- a/src/channel.rs +++ b/src/channel.rs @@ -69,7 +69,13 @@ impl Drop for Sender { let unpark = { let mut g = self.inner.lock().unwrap(); g.senders -= 1; - if g.senders == 0 && g.queue.is_empty() { + // Wake the parked receiver on the last sender drop regardless of + // whether the queue is empty. A plain `recv` only ever parks on an + // empty queue (so this is unchanged for it), but a selective + // `recv_match` may be parked on a *non-empty* queue holding only + // non-matching messages — it must wake to observe closure and + // return Err rather than sleep forever. + if g.senders == 0 { g.parked_receiver.take() } else { None @@ -134,6 +140,64 @@ impl Receiver { } } + /// Selective receive: remove and return the first queued message for which + /// `pred` holds, leaving the rest in arrival order. If no queued message + /// matches, parks and re-scans on every send (a selective receiver may park + /// on a *non-empty* queue). Returns `Err(RecvError)` only once the channel + /// is closed and no queued message matches. + /// + /// `pred` is run while the channel lock is held: keep it cheap and pure, + /// and do not call back into this channel from inside it. It is modelled as + /// `Fn` (not `FnMut`) deliberately — it is re-run from scratch on every + /// scan, so a stateful predicate would observe surprising re-counting. + pub fn recv_match(&self, pred: F) -> Result + where + F: Fn(&T) -> bool, + { + loop { + { + let mut g = self.inner.lock().unwrap(); + if let Some(i) = g.queue.iter().position(|v| pred(v)) { + // position() found it, so remove() returns Some. + return Ok(g.queue.remove(i).unwrap()); + } + if g.senders == 0 { + // Closed and nothing queued can ever match. + return Err(RecvError); + } + let me = crate::actor::current_pid() + .expect("recv_match() called outside an actor"); + debug_assert!( + g.parked_receiver.is_none(), + "channel has more than one receiver" + ); + g.parked_receiver = Some(me); + crate::te!(crate::trace::Event::RecvPark(me)); + } + // Release the lock before parking — the unparker will need it. + crate::scheduler::park_current(); + crate::te!(crate::trace::Event::RecvWake(crate::actor::current_pid().unwrap())); + } + } + + /// Non-blocking selective receive. `Ok(Some(v))` if a queued message + /// matched `pred` (removed, rest left in order), `Ok(None)` if the channel + /// is open but nothing matched, `Err(RecvError)` if closed and nothing + /// matched. Same predicate contract as [`recv_match`](Self::recv_match). + pub fn try_recv_match(&self, pred: F) -> Result, RecvError> + where + F: Fn(&T) -> bool, + { + let mut g = self.inner.lock().unwrap(); + if let Some(i) = g.queue.iter().position(|v| pred(v)) { + return Ok(Some(g.queue.remove(i).unwrap())); + } + if g.senders == 0 { + return Err(RecvError); + } + Ok(None) + } + /// Non-blocking. `Ok(Some(v))` if a message was available, `Ok(None)` if /// the channel is empty but open, `Err(RecvError)` if closed and drained. pub fn try_recv(&self) -> Result, RecvError> { diff --git a/tests/selective_recv.rs b/tests/selective_recv.rs new file mode 100644 index 0000000..a60f4fd --- /dev/null +++ b/tests/selective_recv.rs @@ -0,0 +1,193 @@ +//! Selective-receive tests: `Receiver::recv_match` and `try_recv_match`. +//! +//! `recv_match(pred)` scans the queued messages, removes and returns the first +//! one for which `pred` holds, and leaves the rest in arrival order. With no +//! match it parks and re-scans on every send (a selective receiver may park on +//! a *non-empty* queue), returning `Err` only once the channel is closed and no +//! queued message matches. `try_recv_match` is the non-blocking variant. + +use smarm::{channel, run, spawn}; +use std::sync::{Arc, Mutex}; + +// --------------------------------------------------------------------------- +// recv_match — same-actor scan semantics (no parking) +// --------------------------------------------------------------------------- + +// First matching message is pulled out of arrival order; non-matches stay +// queued in their original order for later plain recv(). +#[test] +fn match_pulled_first_non_matches_remain_in_order() { + let got = Arc::new(Mutex::new(Vec::::new())); + let got2 = got.clone(); + + run(move || { + let (tx, rx) = channel::(); + for v in [1, 2, 3] { + tx.send(v).unwrap(); + } + // 2 is the only even; it's pulled despite arriving second. + let m = rx.recv_match(|&v| v % 2 == 0).unwrap(); + // The non-matches remain, in order: 1 then 3. + let a = rx.recv().unwrap(); + let b = rx.recv().unwrap(); + let mut g = got2.lock().unwrap(); + g.push(m); + g.push(a); + g.push(b); + }); + + assert_eq!(*got.lock().unwrap(), vec![2, 1, 3]); +} + +// A read-only captured value (request-id style) is a plain `Fn` predicate. +#[test] +fn matches_on_captured_value() { + let got = Arc::new(Mutex::new(0i64)); + let got2 = got.clone(); + + run(move || { + let (tx, rx) = channel::(); + for v in [10, 20, 30] { + tx.send(v).unwrap(); + } + let target = 20; + let m = rx.recv_match(move |&v| v == target).unwrap(); + *got2.lock().unwrap() = m; + }); + + assert_eq!(*got.lock().unwrap(), 20); +} + +// --------------------------------------------------------------------------- +// recv_match — parking on a non-empty, no-match queue +// --------------------------------------------------------------------------- + +// The receiver parks while the queue holds only non-matching messages, must +// wake on a send that does NOT empty->fill the queue, re-scan, re-park on a +// still-non-matching send, and finally return when a match arrives. +#[test] +fn parks_on_non_matching_queue_then_wakes_on_match() { + let got = Arc::new(Mutex::new(0i64)); + let got2 = got.clone(); + + run(move || { + let (tx, rx) = channel::(); + tx.send(1).unwrap(); // odd: queued, will not match + let h = spawn(move || { + // Scans [1], no even -> parks on a non-empty queue. + let v = rx.recv_match(|&v| v % 2 == 0).unwrap(); + *got2.lock().unwrap() = v; + }); + smarm::yield_now(); // let the child park + tx.send(3).unwrap(); // odd: child wakes, re-scans [1,3], re-parks + tx.send(4).unwrap(); // even: child wakes, scans [1,3,4], returns 4 + h.join().unwrap(); + }); + + assert_eq!(*got.lock().unwrap(), 4); +} + +// --------------------------------------------------------------------------- +// recv_match — close semantics +// --------------------------------------------------------------------------- + +// Channel closes (all senders dropped) while only non-matching messages are +// queued: recv_match must return Err rather than block forever. +#[test] +fn closed_with_only_non_matches_returns_err() { + let saw_err = Arc::new(Mutex::new(false)); + let saw_err2 = saw_err.clone(); + + run(move || { + let (tx, rx) = channel::(); + let h = spawn(move || { + // Wants an even; only odds will ever exist. + let r = rx.recv_match(|&v| v % 2 == 0); + *saw_err2.lock().unwrap() = r.is_err(); + }); + smarm::yield_now(); // child parks on empty queue + tx.send(1).unwrap(); // odd: wakes, re-scans, re-parks (no match) + drop(tx); // last sender gone, still no match -> Err + h.join().unwrap(); + }); + + assert!(*saw_err.lock().unwrap()); +} + +// A match present on a closed channel is still returned (closure doesn't +// pre-empt an available match). +#[test] +fn closed_but_match_present_returns_match() { + let got = Arc::new(Mutex::new(0i64)); + let got2 = got.clone(); + + run(move || { + let (tx, rx) = channel::(); + tx.send(1).unwrap(); + tx.send(2).unwrap(); + drop(tx); // closed, but a match (2) is queued + let m = rx.recv_match(|&v| v % 2 == 0).unwrap(); + *got2.lock().unwrap() = m; + }); + + assert_eq!(*got.lock().unwrap(), 2); +} + +// --------------------------------------------------------------------------- +// try_recv_match — non-blocking +// --------------------------------------------------------------------------- + +#[test] +fn try_recv_match_states() { + let out = Arc::new(Mutex::new(Vec::::new())); + let out2 = out.clone(); + + run(move || { + let (tx, rx) = channel::(); + let log = |s: &str| out2.lock().unwrap().push(s.to_string()); + + // Empty + open -> Ok(None). + log(match rx.try_recv_match(|&v| v % 2 == 0) { + Ok(None) => "open_empty_none", + _ => "WRONG_1", + }); + + // Non-matching present + open -> Ok(None), message left in place. + tx.send(1).unwrap(); + log(match rx.try_recv_match(|&v| v % 2 == 0) { + Ok(None) => "open_nomatch_none", + _ => "WRONG_2", + }); + + // Match present -> Ok(Some(v)). + tx.send(2).unwrap(); + log(match rx.try_recv_match(|&v| v % 2 == 0) { + Ok(Some(2)) => "got_match", + _ => "WRONG_3", + }); + + // Drain the leftover non-match with a plain recv to confirm it stayed. + log(match rx.recv() { + Ok(1) => "leftover_one", + _ => "WRONG_4", + }); + + // Closed + no match -> Err. + drop(tx); + log(match rx.try_recv_match(|&v| v % 2 == 0) { + Err(_) => "closed_err", + _ => "WRONG_5", + }); + }); + + assert_eq!( + *out.lock().unwrap(), + vec![ + "open_empty_none", + "open_nomatch_none", + "got_match", + "leftover_one", + "closed_err", + ] + ); +}