kernel_api/
channel.rs

1//! An async multi-producer, single-consumer channel
2//!
3//! A new channel can be created by calling [`unbounded()`] which returns a [`Sender`] and [`Receiver`].
4//! The [`Sender`] can be [`clone`]d, while the [`Receiver`] cannot.
5//!
6//! New items can be pushed to the back of the queue with [`Sender::push()`], and items can be popped
7//! with [`Receiver::pop()`] or [`Receiver::try_pop()`].
8
9use alloc::sync::Arc;
10use core::fmt::{Debug, Formatter};
11use core::pin::Pin;
12use core::task::{Context, Poll};
13use crossbeam_queue::SegQueue;
14use futures::task::AtomicWaker;
15
16/// The sending end of a channel
17pub struct Sender<T> {
18	inner: Arc<ChannelInner<T>>,
19}
20
21impl<T> Clone for Sender<T> {
22	fn clone(&self) -> Self {
23		Self { inner: Arc::clone(&self.inner) }
24	}
25}
26
27/// The receiving end of a channel
28pub struct Receiver<T> {
29	inner: Arc<ChannelInner<T>>,
30}
31
32struct ChannelInner<T> {
33	queue: SegQueue<T>,
34	waker: AtomicWaker,
35}
36
37/// Creates a new unbounded chanel
38pub fn unbounded<T>() -> (Sender<T>, Receiver<T>) {
39	let inner = ChannelInner {
40		queue: SegQueue::new(),
41		waker: AtomicWaker::new(),
42	};
43	let inner = Arc::new(inner);
44
45	(Sender { inner: Arc::clone(&inner) }, Receiver { inner })
46}
47
48impl<T> Sender<T> {
49	/// Pushes a new item into the back of the channel
50	pub fn push(&self, val: T) {
51		self.inner.queue.push(val);
52		self.inner.waker.wake();
53	}
54}
55
56impl<T> Receiver<T> {
57	/// Pushes a new item into the back of the channel
58	pub fn push(&self, val: T) {
59		self.inner.queue.push(val);
60		self.inner.waker.wake();
61	}
62
63	/// Pops the front item from the channel, blocking until an item is available
64	pub fn pop(&self) -> impl Future<Output = T> + '_ {
65		struct Waiter<'a, T>(&'a ChannelInner<T>);
66
67		impl<T> Future for Waiter<'_, T> {
68			type Output = T;
69
70			fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
71				if let Some(val) = self.0.queue.pop() { return Poll::Ready(val); }
72				self.0.waker.register(cx.waker());
73				match self.0.queue.pop() {
74					Some(val) => Poll::Ready(val),
75					None => Poll::Pending,
76				}
77			}
78		}
79
80		Waiter(&self.inner)
81	}
82
83	/// Pops the front item in the channel, if one exists
84	pub fn try_pop(&self) -> Option<T> {
85		self.inner.queue.pop()
86	}
87
88	/// Creates a new [`Sender`] for this Receiver
89	pub fn sender(&self) -> Sender<T> {
90		Sender {
91			inner: Arc::clone(&self.inner)
92		}
93	}
94
95	/// The number of items available to pop from the channel
96	pub fn len(&self) -> usize { self.inner.queue.len() }
97}
98
99impl<T> Debug for Sender<T> {
100	fn fmt(&self, f: &mut Formatter<'_>) -> core::fmt::Result {
101		write!(f, "Sender {{ .. }}")
102	}
103}
104
105impl<T> Debug for Receiver<T> {
106	fn fmt(&self, f: &mut Formatter<'_>) -> core::fmt::Result {
107		write!(f, "Receiver {{ .. }}")
108	}
109}
kernel_api/channel.rs

kernel_api/
channel.rs
