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.
38#[must_use]
39pub fn unbounded<T>() -> (Sender<T>, Receiver<T>) {
40	let inner = ChannelInner {
41		queue: SegQueue::new(),
42		waker: AtomicWaker::new(),
43	};
44	let inner = Arc::new(inner);
45
46	(Sender { inner: Arc::clone(&inner) }, Receiver { inner })
47}
48
49impl<T> Sender<T> {
50	/// Pushes a new item into the back of the channel.
51	pub fn push(&self, val: T) {
52		self.inner.queue.push(val);
53		self.inner.waker.wake();
54	}
55}
56
57impl<T> Receiver<T> {
58	/// Pushes a new item into the back of the channel.
59	pub fn push(&self, val: T) {
60		self.inner.queue.push(val);
61		self.inner.waker.wake();
62	}
63
64	/// Pops the front item from the channel, blocking until an item is available.
65	pub fn pop(&self) -> impl Future<Output = T> + '_ {
66		struct Waiter<'a, T>(&'a ChannelInner<T>);
67
68		impl<T> Future for Waiter<'_, T> {
69			type Output = T;
70
71			fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
72				if let Some(val) = self.0.queue.pop() { return Poll::Ready(val); }
73				self.0.waker.register(cx.waker());
74				self.0.queue.pop()
75					.map_or(Poll::Pending, Poll::Ready)
76			}
77		}
78
79		Waiter(&self.inner)
80	}
81
82	/// Pops the front item in the channel, if one exists.
83	#[must_use]
84	pub fn try_pop(&self) -> Option<T> {
85		self.inner.queue.pop()
86	}
87
88	/// Creates a new [`Sender`] for this Receiver.
89	#[must_use]
90	pub fn sender(&self) -> Sender<T> {
91		Sender {
92			inner: Arc::clone(&self.inner)
93		}
94	}
95
96	/// The number of items available to pop from the channel.
97	#[must_use]
98	pub fn len(&self) -> usize { self.inner.queue.len() }
99
100	/// Returns `true` if there are no items in the channel.
101	#[must_use]
102	pub fn is_empty(&self) -> bool { self.inner.queue.is_empty() }
103}
104
105impl<T> Debug for Sender<T> {
106	fn fmt(&self, f: &mut Formatter<'_>) -> core::fmt::Result {
107		write!(f, "Sender {{ .. }}")
108	}
109}
110
111impl<T> Debug for Receiver<T> {
112	fn fmt(&self, f: &mut Formatter<'_>) -> core::fmt::Result {
113		write!(f, "Receiver {{ .. }}")
114	}
115}
kernel_api/channel.rs

kernel_api/
channel.rs
