lib.rs raw
1 //! pickle - CSP/Actor building blocks for Rust
2 //!
3 //! Generic channel types that eliminate per-actor boilerplate. Actors own
4 //! their state inside a spawned thread; callers hold typed channel handles
5 //! that are Clone + Send. No `Arc<Mutex<T>>`, no lifetime annotations.
6 //!
7 //! # Channel types
8 //!
9 //! | Type | Pattern | Caller | Actor |
10 //! |------|---------|--------|-------|
11 //! | [`Func<Req, Resp>`] | function call | `call(req) -> resp` | `msg.req`, `msg.reply(resp)` |
12 //! | [`Query<Resp>`] | getter | `call() -> resp` | `msg.reply(resp)` |
13 //! | [`Proc<Req>`] | command | `call(req)` blocks | `msg.req`, `msg.done()` |
14 //! | [`Signal`] | trigger | `call()` blocks | `msg.done()` |
15 //! | [`Inbox<T>`] | fire-and-forget | `send(v)` / `try_send(v)` | recv from `receiver()` |
16 //! | [`Lifecycle`] | shutdown | `stop()` | `stopping()` in select |
17
18 pub use crossbeam_channel::{self, Receiver, Sender, select};
19 use crossbeam_channel::bounded;
20 use std::thread::JoinHandle;
21 use std::time::Duration;
22
23 // ---------------------------------------------------------------------------
24 // CallError
25 // ---------------------------------------------------------------------------
26
27 /// Error returned by `call_timeout` methods.
28 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
29 pub enum CallError {
30 /// The operation timed out.
31 Timeout,
32 /// The receiving end was disconnected.
33 Disconnected,
34 }
35
36 // ---------------------------------------------------------------------------
37 // Func<Req, Resp>
38 // ---------------------------------------------------------------------------
39
40 /// Message sent through a [`Func`] channel.
41 pub struct FuncMsg<Req, Resp> {
42 pub req: Req,
43 resp_tx: Sender<Resp>,
44 }
45
46 impl<Req, Resp> FuncMsg<Req, Resp> {
47 /// Reply to the caller. Consumes self - double-reply is a compile error.
48 pub fn reply(self, resp: Resp) {
49 let _ = self.resp_tx.send(resp);
50 }
51 }
52
53 /// Synchronous request-response channel. Rendezvous (zero-capacity) on the
54 /// request side; the caller blocks until the actor picks up the message and
55 /// replies.
56 #[derive(Clone)]
57 pub struct Func<Req, Resp> {
58 tx: Sender<FuncMsg<Req, Resp>>,
59 rx: Receiver<FuncMsg<Req, Resp>>,
60 }
61
62 impl<Req: Send + 'static, Resp: Send + 'static> Func<Req, Resp> {
63 pub fn new() -> Self {
64 let (tx, rx) = bounded(0);
65 Func { tx, rx }
66 }
67
68 /// Call the actor with `req` and block until a response arrives.
69 pub fn call(&self, req: Req) -> Resp {
70 let (resp_tx, resp_rx) = bounded(1);
71 let msg = FuncMsg { req, resp_tx };
72 self.tx.send(msg).expect("actor disconnected");
73 resp_rx.recv().expect("actor dropped response channel")
74 }
75
76 /// Call with a timeout on both send and receive phases.
77 pub fn call_timeout(&self, req: Req, timeout: Duration) -> Result<Resp, CallError> {
78 let (resp_tx, resp_rx) = bounded(1);
79 let msg = FuncMsg { req, resp_tx };
80 self.tx
81 .send_timeout(msg, timeout)
82 .map_err(|e| match e {
83 crossbeam_channel::SendTimeoutError::Timeout(_) => CallError::Timeout,
84 crossbeam_channel::SendTimeoutError::Disconnected(_) => CallError::Disconnected,
85 })?;
86 resp_rx.recv_timeout(timeout).map_err(|e| match e {
87 crossbeam_channel::RecvTimeoutError::Timeout => CallError::Timeout,
88 crossbeam_channel::RecvTimeoutError::Disconnected => CallError::Disconnected,
89 })
90 }
91
92 /// Clone of the receive end for the actor's `select!`.
93 pub fn receiver(&self) -> Receiver<FuncMsg<Req, Resp>> {
94 self.rx.clone()
95 }
96 }
97
98 impl<Req: Send + 'static, Resp: Send + 'static> Default for Func<Req, Resp> {
99 fn default() -> Self {
100 Self::new()
101 }
102 }
103
104 // ---------------------------------------------------------------------------
105 // Query<Resp>
106 // ---------------------------------------------------------------------------
107
108 /// Message sent through a [`Query`] channel.
109 pub struct QueryMsg<Resp> {
110 resp_tx: Sender<Resp>,
111 }
112
113 impl<Resp> QueryMsg<Resp> {
114 /// Reply to the caller. Consumes self.
115 pub fn reply(self, resp: Resp) {
116 let _ = self.resp_tx.send(resp);
117 }
118 }
119
120 /// Synchronous getter channel. Like [`Func`] but with no request payload.
121 #[derive(Clone)]
122 pub struct Query<Resp> {
123 tx: Sender<QueryMsg<Resp>>,
124 rx: Receiver<QueryMsg<Resp>>,
125 }
126
127 impl<Resp: Send + 'static> Query<Resp> {
128 pub fn new() -> Self {
129 let (tx, rx) = bounded(0);
130 Query { tx, rx }
131 }
132
133 /// Block until the actor replies.
134 pub fn call(&self) -> Resp {
135 let (resp_tx, resp_rx) = bounded(1);
136 let msg = QueryMsg { resp_tx };
137 self.tx.send(msg).expect("actor disconnected");
138 resp_rx.recv().expect("actor dropped response channel")
139 }
140
141 /// Call with a timeout.
142 pub fn call_timeout(&self, timeout: Duration) -> Result<Resp, CallError> {
143 let (resp_tx, resp_rx) = bounded(1);
144 let msg = QueryMsg { resp_tx };
145 self.tx
146 .send_timeout(msg, timeout)
147 .map_err(|e| match e {
148 crossbeam_channel::SendTimeoutError::Timeout(_) => CallError::Timeout,
149 crossbeam_channel::SendTimeoutError::Disconnected(_) => CallError::Disconnected,
150 })?;
151 resp_rx.recv_timeout(timeout).map_err(|e| match e {
152 crossbeam_channel::RecvTimeoutError::Timeout => CallError::Timeout,
153 crossbeam_channel::RecvTimeoutError::Disconnected => CallError::Disconnected,
154 })
155 }
156
157 pub fn receiver(&self) -> Receiver<QueryMsg<Resp>> {
158 self.rx.clone()
159 }
160 }
161
162 impl<Resp: Send + 'static> Default for Query<Resp> {
163 fn default() -> Self {
164 Self::new()
165 }
166 }
167
168 // ---------------------------------------------------------------------------
169 // Proc<Req>
170 // ---------------------------------------------------------------------------
171
172 /// Message sent through a [`Proc`] channel.
173 pub struct ProcMsg<Req> {
174 pub req: Req,
175 done_tx: Sender<()>,
176 }
177
178 impl<Req> ProcMsg<Req> {
179 /// Signal that processing is complete. Consumes self.
180 pub fn done(self) {
181 let _ = self.done_tx.send(());
182 }
183 }
184
185 /// Synchronous command channel. Caller sends a request and blocks until the
186 /// actor calls `done()`.
187 #[derive(Clone)]
188 pub struct Proc<Req> {
189 tx: Sender<ProcMsg<Req>>,
190 rx: Receiver<ProcMsg<Req>>,
191 }
192
193 impl<Req: Send + 'static> Proc<Req> {
194 pub fn new() -> Self {
195 let (tx, rx) = bounded(0);
196 Proc { tx, rx }
197 }
198
199 /// Send `req` and block until the actor calls `done()`.
200 pub fn call(&self, req: Req) {
201 let (done_tx, done_rx) = bounded(1);
202 let msg = ProcMsg { req, done_tx };
203 self.tx.send(msg).expect("actor disconnected");
204 done_rx.recv().expect("actor dropped done channel");
205 }
206
207 /// Call with a timeout.
208 pub fn call_timeout(&self, req: Req, timeout: Duration) -> Result<(), CallError> {
209 let (done_tx, done_rx) = bounded(1);
210 let msg = ProcMsg { req, done_tx };
211 self.tx
212 .send_timeout(msg, timeout)
213 .map_err(|e| match e {
214 crossbeam_channel::SendTimeoutError::Timeout(_) => CallError::Timeout,
215 crossbeam_channel::SendTimeoutError::Disconnected(_) => CallError::Disconnected,
216 })?;
217 done_rx.recv_timeout(timeout).map_err(|e| match e {
218 crossbeam_channel::RecvTimeoutError::Timeout => CallError::Timeout,
219 crossbeam_channel::RecvTimeoutError::Disconnected => CallError::Disconnected,
220 })
221 }
222
223 pub fn receiver(&self) -> Receiver<ProcMsg<Req>> {
224 self.rx.clone()
225 }
226 }
227
228 impl<Req: Send + 'static> Default for Proc<Req> {
229 fn default() -> Self {
230 Self::new()
231 }
232 }
233
234 // ---------------------------------------------------------------------------
235 // Signal
236 // ---------------------------------------------------------------------------
237
238 /// Message sent through a [`Signal`] channel.
239 pub struct SignalMsg {
240 done_tx: Sender<()>,
241 }
242
243 impl SignalMsg {
244 /// Signal that processing is complete. Consumes self.
245 pub fn done(self) {
246 let _ = self.done_tx.send(());
247 }
248 }
249
250 /// Synchronous trigger channel. Like [`Proc`] with no payload.
251 #[derive(Clone)]
252 pub struct Signal {
253 tx: Sender<SignalMsg>,
254 rx: Receiver<SignalMsg>,
255 }
256
257 impl Signal {
258 pub fn new() -> Self {
259 let (tx, rx) = bounded(0);
260 Signal { tx, rx }
261 }
262
263 /// Block until the actor calls `done()`.
264 pub fn call(&self) {
265 let (done_tx, done_rx) = bounded(1);
266 let msg = SignalMsg { done_tx };
267 self.tx.send(msg).expect("actor disconnected");
268 done_rx.recv().expect("actor dropped done channel");
269 }
270
271 /// Call with a timeout.
272 pub fn call_timeout(&self, timeout: Duration) -> Result<(), CallError> {
273 let (done_tx, done_rx) = bounded(1);
274 let msg = SignalMsg { done_tx };
275 self.tx
276 .send_timeout(msg, timeout)
277 .map_err(|e| match e {
278 crossbeam_channel::SendTimeoutError::Timeout(_) => CallError::Timeout,
279 crossbeam_channel::SendTimeoutError::Disconnected(_) => CallError::Disconnected,
280 })?;
281 done_rx.recv_timeout(timeout).map_err(|e| match e {
282 crossbeam_channel::RecvTimeoutError::Timeout => CallError::Timeout,
283 crossbeam_channel::RecvTimeoutError::Disconnected => CallError::Disconnected,
284 })
285 }
286
287 pub fn receiver(&self) -> Receiver<SignalMsg> {
288 self.rx.clone()
289 }
290 }
291
292 impl Default for Signal {
293 fn default() -> Self {
294 Self::new()
295 }
296 }
297
298 // ---------------------------------------------------------------------------
299 // Inbox<T>
300 // ---------------------------------------------------------------------------
301
302 /// Buffered fire-and-forget channel.
303 #[derive(Clone)]
304 pub struct Inbox<T> {
305 tx: Sender<T>,
306 rx: Receiver<T>,
307 }
308
309 impl<T: Send + 'static> Inbox<T> {
310 /// Create an inbox with the given buffer capacity.
311 pub fn new(buffer: usize) -> Self {
312 let (tx, rx) = bounded(buffer);
313 Inbox { tx, rx }
314 }
315
316 /// Blocking send.
317 pub fn send(&self, v: T) {
318 self.tx.send(v).expect("actor disconnected");
319 }
320
321 /// Non-blocking send. Returns `false` if the buffer is full.
322 pub fn try_send(&self, v: T) -> bool {
323 self.tx.try_send(v).is_ok()
324 }
325
326 /// Clone of the receive end for the actor's `select!`.
327 pub fn receiver(&self) -> Receiver<T> {
328 self.rx.clone()
329 }
330 }
331
332 // ---------------------------------------------------------------------------
333 // Lifecycle
334 // ---------------------------------------------------------------------------
335
336 #[allow(dead_code)] // held for Drop side-effect only
337 struct DoneGuard(Option<Sender<()>>);
338
339 impl Drop for DoneGuard {
340 fn drop(&mut self) {
341 // Sender drops, done_rx sees Disconnected, stop() unblocks.
342 }
343 }
344
345 /// Manages actor startup and shutdown.
346 ///
347 /// The pattern:
348 /// 1. Create a `Lifecycle` with `new()`.
349 /// 2. Clone `stopping()` for the actor's select loop.
350 /// 3. Call `spawn()` to start the actor thread.
351 /// 4. Call `stop()` to shut down - drops the stop sender (actor's `stopping()`
352 /// receiver fires on disconnect), then blocks until the actor thread exits.
353 ///
354 /// `stop()` is idempotent after the first call (the second `take()` is a no-op
355 /// on the already-consumed Option). `spawn()` panics if called twice.
356 pub struct Lifecycle {
357 stop_tx: Option<Sender<()>>,
358 stop_rx: Receiver<()>,
359 done_tx: Option<Sender<()>>,
360 done_rx: Receiver<()>,
361 }
362
363 impl Lifecycle {
364 pub fn new() -> Self {
365 let (stop_tx, stop_rx) = bounded(0);
366 let (done_tx, done_rx) = bounded(0);
367 Lifecycle {
368 stop_tx: Some(stop_tx),
369 stop_rx,
370 done_tx: Some(done_tx),
371 done_rx,
372 }
373 }
374
375 /// Returns a receiver that fires (with `Err(Disconnected)`) when `stop()`
376 /// is called. Use in `select!` with `-> _` to catch the disconnect:
377 ///
378 /// ```ignore
379 /// select! {
380 /// recv(stopping) -> _ => return,
381 /// // ...
382 /// }
383 /// ```
384 pub fn stopping(&self) -> Receiver<()> {
385 self.stop_rx.clone()
386 }
387
388 /// Returns a receiver that fires when the actor thread exits (the
389 /// `DoneGuard` is dropped). For external observers that need to wait
390 /// on actor completion without calling `stop()`.
391 pub fn stopped(&self) -> Receiver<()> {
392 self.done_rx.clone()
393 }
394
395 /// Signal the actor to stop and block until it exits.
396 ///
397 /// Drops the stop sender (causing the actor's `stopping()` receiver to
398 /// disconnect), then waits for the done channel to disconnect (which
399 /// happens when the `DoneGuard` inside the actor thread is dropped).
400 pub fn stop(&mut self) {
401 debug_assert!(
402 self.done_tx.is_none(),
403 "stop() called before spawn()"
404 );
405 self.stop_tx.take(); // drop sends disconnect to stopping()
406 let _ = self.done_rx.recv(); // blocks until DoneGuard drops
407 }
408 }
409
410 impl Default for Lifecycle {
411 fn default() -> Self {
412 Self::new()
413 }
414 }
415
416 /// Spawn an actor thread with lifecycle management.
417 ///
418 /// Takes the done sender out of the lifecycle and wraps it in a `DoneGuard`
419 /// whose `Drop` impl closes the done channel - whether the closure returns
420 /// normally or panics. Panics if called twice on the same lifecycle.
421 pub fn spawn<F>(lc: &mut Lifecycle, f: F) -> JoinHandle<()>
422 where
423 F: FnOnce() + Send + 'static,
424 {
425 let done_tx = lc.done_tx.take().expect("spawn() called twice on same Lifecycle");
426 let guard = DoneGuard(Some(done_tx));
427 std::thread::spawn(move || {
428 let _guard = guard;
429 f();
430 })
431 }
432
433 // ---------------------------------------------------------------------------
434 // Tests
435 // ---------------------------------------------------------------------------
436
437 #[cfg(test)]
438 mod tests {
439 use super::*;
440 use std::collections::HashMap;
441 use std::time::Duration;
442
443 // Counter: Signal + Query
444 #[test]
445 fn counter() {
446 let inc = Signal::new();
447 let get = Query::<i32>::new();
448 let mut lc = Lifecycle::new();
449
450 let inc_rx = inc.receiver();
451 let get_rx = get.receiver();
452 let stopping = lc.stopping();
453
454 spawn(&mut lc, move || {
455 let mut n: i32 = 0;
456 loop {
457 select! {
458 recv(stopping) -> _ => return,
459 recv(inc_rx) -> msg => {
460 msg.unwrap().done();
461 n += 1;
462 },
463 recv(get_rx) -> msg => {
464 msg.unwrap().reply(n);
465 },
466 }
467 }
468 });
469
470 inc.call();
471 inc.call();
472 inc.call();
473 assert_eq!(get.call(), 3);
474 lc.stop();
475 }
476
477 // KV store: Func + Proc
478 #[test]
479 fn kv_store() {
480 let set = Proc::<(String, String)>::new();
481 let get = Func::<String, Option<String>>::new();
482 let mut lc = Lifecycle::new();
483
484 let set_rx = set.receiver();
485 let get_rx = get.receiver();
486 let stopping = lc.stopping();
487
488 spawn(&mut lc, move || {
489 let mut store = HashMap::<String, String>::new();
490 loop {
491 select! {
492 recv(stopping) -> _ => return,
493 recv(set_rx) -> msg => {
494 let msg = msg.unwrap();
495 store.insert(msg.req.0.clone(), msg.req.1.clone());
496 msg.done();
497 },
498 recv(get_rx) -> msg => {
499 let msg = msg.unwrap();
500 let val = store.get(&msg.req).cloned();
501 msg.reply(val);
502 },
503 }
504 }
505 });
506
507 set.call(("a".into(), "1".into()));
508 set.call(("b".into(), "2".into()));
509 assert_eq!(get.call("a".into()), Some("1".into()));
510 assert_eq!(get.call("b".into()), Some("2".into()));
511 assert_eq!(get.call("c".into()), None);
512 lc.stop();
513 }
514
515 // Logger: Inbox + Query
516 #[test]
517 fn logger() {
518 let log = Inbox::<String>::new(64);
519 let count = Query::<usize>::new();
520 let mut lc = Lifecycle::new();
521
522 let log_rx = log.receiver();
523 let count_rx = count.receiver();
524 let stopping = lc.stopping();
525
526 spawn(&mut lc, move || {
527 let mut n: usize = 0;
528 loop {
529 select! {
530 recv(stopping) -> _ => return,
531 recv(log_rx) -> msg => {
532 let _ = msg.unwrap();
533 n += 1;
534 },
535 recv(count_rx) -> msg => {
536 msg.unwrap().reply(n);
537 },
538 }
539 }
540 });
541
542 log.send("one".into());
543 log.send("two".into());
544 log.send("three".into());
545 // sync on count to ensure all three are processed
546 assert_eq!(count.call(), 3);
547 lc.stop();
548 }
549
550 // Inbox try_send: buffer full returns false
551 #[test]
552 fn inbox_try_send() {
553 let inbox = Inbox::<i32>::new(2);
554 assert!(inbox.try_send(1));
555 assert!(inbox.try_send(2));
556 assert!(!inbox.try_send(3)); // buffer full
557 }
558
559 // Lifecycle stopped: observable from outside
560 #[test]
561 fn lifecycle_stopped() {
562 let mut lc = Lifecycle::new();
563 let stopped = lc.stopped();
564 let stopping = lc.stopping();
565
566 spawn(&mut lc, move || {
567 // wait for stop signal then exit
568 let _ = stopping.recv();
569 });
570
571 // not stopped yet
572 assert!(stopped.recv_timeout(Duration::from_millis(50)).is_err());
573
574 lc.stop();
575
576 // now the done channel is disconnected - recv returns Err
577 // (it already returned from the recv in stop(), so this should
578 // also see Disconnected)
579 assert!(stopped.recv_timeout(Duration::from_millis(50)).is_err());
580 }
581
582 // call_timeout: unlistened channel times out
583 #[test]
584 fn call_timeout_expires() {
585 let sig = Signal::new();
586 // nobody is listening on the receiver
587 let result = sig.call_timeout(Duration::from_millis(50));
588 assert_eq!(result, Err(CallError::Timeout));
589 }
590
591 // call_timeout on Func
592 #[test]
593 fn func_call_timeout() {
594 let f = Func::<i32, i32>::new();
595 let result = f.call_timeout(42, Duration::from_millis(50));
596 assert_eq!(result, Err(CallError::Timeout));
597 }
598
599 // call_timeout on Query
600 #[test]
601 fn query_call_timeout() {
602 let q = Query::<i32>::new();
603 let result = q.call_timeout(Duration::from_millis(50));
604 assert_eq!(result, Err(CallError::Timeout));
605 }
606
607 // call_timeout on Proc
608 #[test]
609 fn proc_call_timeout() {
610 let p = Proc::<i32>::new();
611 let result = p.call_timeout(42, Duration::from_millis(50));
612 assert_eq!(result, Err(CallError::Timeout));
613 }
614
615 // Concurrent callers: 100 threads incrementing, final count correct
616 #[test]
617 fn concurrent_callers() {
618 let inc = Signal::new();
619 let get = Query::<i32>::new();
620 let mut lc = Lifecycle::new();
621
622 let inc_rx = inc.receiver();
623 let get_rx = get.receiver();
624 let stopping = lc.stopping();
625
626 spawn(&mut lc, move || {
627 let mut n: i32 = 0;
628 loop {
629 select! {
630 recv(stopping) -> _ => return,
631 recv(inc_rx) -> msg => {
632 msg.unwrap().done();
633 n += 1;
634 },
635 recv(get_rx) -> msg => {
636 msg.unwrap().reply(n);
637 },
638 }
639 }
640 });
641
642 let mut handles = Vec::new();
643 for _ in 0..100 {
644 let inc = inc.clone();
645 handles.push(std::thread::spawn(move || {
646 inc.call();
647 }));
648 }
649 for h in handles {
650 h.join().unwrap();
651 }
652 assert_eq!(get.call(), 100);
653 lc.stop();
654 }
655
656 // Clone callers: clone a Func handle, call from both
657 #[test]
658 fn clone_callers() {
659 let f = Func::<i32, i32>::new();
660 let mut lc = Lifecycle::new();
661
662 let f_rx = f.receiver();
663 let stopping = lc.stopping();
664
665 spawn(&mut lc, move || {
666 loop {
667 select! {
668 recv(stopping) -> _ => return,
669 recv(f_rx) -> msg => {
670 let msg = msg.unwrap();
671 let result = msg.req * 2;
672 msg.reply(result);
673 },
674 }
675 }
676 });
677
678 let f2 = f.clone();
679 assert_eq!(f.call(5), 10);
680 assert_eq!(f2.call(7), 14);
681 lc.stop();
682 }
683 }
684