osal_rs/async_primitives/
queue.rs

1/***************************************************************************
2 *
3 * osal-rs
4 * Copyright (C) 2026 Antonio Salsi <passy.linux@zresa.it>
5 *
6 * This library is free software; you can redistribute it and/or
7 * modify it under the terms of the GNU Lesser General Public
8 * License as published by the Free Software Foundation; either
9 * version 2.1 of the License, or (at your option) any later version.
10 *
11 * This library is distributed in the hope that it will be useful,
12 * but WITHOUT ANY WARRANTY; without even the implied warranty of
13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
14 * Lesser General Public License for more details.
15 *
16 * You should have received a copy of the GNU Lesser General Public
17 * License along with this library; if not, see <https://www.gnu.org/licenses/>.
18 *
19 ***************************************************************************/
20
21//! Async wrapper for OSAL queues.
22//!
23//! [`AsyncQueue`] wraps an [`Arc<Queue>`] and exposes `fetch_async` /
24//! `post_async` methods that return `Future`s.  The synchronous `post` /
25//! `fetch` counterparts on the wrapper also wake any sleeping async task
26//! waiting on the other end.
27
28use core::future::Future;
29use core::pin::Pin;
30use core::task::{Context, Poll};
31use core::time::Duration;
32
33
34use alloc::sync::Arc;
35
36use crate::os::types::{UBaseType, TickType};
37use crate::os::Queue;
38use crate::traits::QueueFn;
39use crate::utils::{Error, Result};
40
41use super::waker_slot::WakerSlot;
42
43/// An async-capable queue wrapping an `Arc<Queue>`.
44///
45/// Both the synchronous and asynchronous APIs are available.  When using the
46/// synchronous API, pending async waiters are woken automatically.
47pub struct AsyncQueue {
48    inner: Arc<Queue>,
49    rx_waker: Arc<WakerSlot>,
50    tx_waker: Arc<WakerSlot>,
51}
52
53impl AsyncQueue {
54    /// Creates an `AsyncQueue` that owns a new OSAL queue.
55    ///
56    /// `size` is the maximum number of items; `message_size` is each item's
57    /// size in bytes.
58    ///
59    /// # Errors
60    ///
61    /// Returns the underlying OSAL error if the queue cannot be created.
62    pub fn new(size: u32, message_size: u32) -> Result<Self> {
63        let inner = Queue::new(size as UBaseType, message_size as UBaseType)?;
64        Ok(Self {
65            inner: Arc::new(inner),
66            rx_waker: Arc::new(WakerSlot::new()),
67            tx_waker: Arc::new(WakerSlot::new()),
68        })
69    }
70
71    /// Posts `item` synchronously and wakes any async task waiting to receive.
72    pub fn post(&self, item: &[u8], timeout_ms: u64) -> Result<()> {
73        let ticks = TickType::try_from(
74            Duration::from_millis(timeout_ms)
75                .as_millis()
76                .try_into()
77                .unwrap_or(TickType::MAX),
78        )
79        .unwrap_or(TickType::MAX);
80        let result = self.inner.post(item, ticks);
81        if result.is_ok() {
82            self.rx_waker.wake();
83        }
84        result
85    }
86
87    /// Fetches an item synchronously and wakes any async task waiting to send.
88    pub fn fetch(&self, buf: &mut [u8], timeout_ms: u64) -> Result<()> {
89        let ticks = TickType::try_from(
90            Duration::from_millis(timeout_ms)
91                .as_millis()
92                .try_into()
93                .unwrap_or(TickType::MAX),
94        )
95        .unwrap_or(TickType::MAX);
96        let result = self.inner.fetch(buf, ticks);
97        if result.is_ok() {
98            self.tx_waker.wake();
99        }
100        result
101    }
102
103    /// Returns a `Future` that resolves when an item has been fetched into `buf`.
104    pub fn fetch_async<'a>(&'a self, buf: &'a mut [u8]) -> FetchFuture<'a> {
105        FetchFuture {
106            queue: self,
107            buf,
108        }
109    }
110
111    /// Returns a `Future` that resolves when `item` has been posted to the queue.
112    pub fn post_async<'a>(&'a self, item: &'a [u8]) -> PostFuture<'a> {
113        PostFuture {
114            queue: self,
115            item,
116        }
117    }
118}
119
120/// Future returned by [`AsyncQueue::fetch_async`].
121pub struct FetchFuture<'a> {
122    queue: &'a AsyncQueue,
123    buf: &'a mut [u8],
124}
125
126impl<'a> Future for FetchFuture<'a> {
127    type Output = Result<()>;
128
129    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
130        let this = self.get_mut();
131
132        // Non-blocking attempt first.
133        match this.queue.inner.fetch(this.buf, 0) {
134            Ok(()) => {
135                this.queue.tx_waker.wake();
136                return Poll::Ready(Ok(()));
137            }
138            Err(Error::Timeout) => {}
139            Err(e) => return Poll::Ready(Err(e)),
140        }
141
142        // Store waker, then try again to close the TOCTOU window.
143        this.queue.rx_waker.store(cx.waker());
144
145        match this.queue.inner.fetch(this.buf, 0) {
146            Ok(()) => {
147                this.queue.tx_waker.wake();
148                Poll::Ready(Ok(()))
149            }
150            Err(Error::Timeout) => Poll::Pending,
151            Err(e) => Poll::Ready(Err(e)),
152        }
153    }
154}
155
156/// Future returned by [`AsyncQueue::post_async`].
157pub struct PostFuture<'a> {
158    queue: &'a AsyncQueue,
159    item: &'a [u8],
160}
161
162impl<'a> Future for PostFuture<'a> {
163    type Output = Result<()>;
164
165    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
166        let this = self.get_mut();
167
168        match this.queue.inner.post(this.item, 0) {
169            Ok(()) => {
170                this.queue.rx_waker.wake();
171                return Poll::Ready(Ok(()));
172            }
173            Err(Error::Timeout) => {}
174            Err(e) => return Poll::Ready(Err(e)),
175        }
176
177        this.queue.tx_waker.store(cx.waker());
178
179        match this.queue.inner.post(this.item, 0) {
180            Ok(()) => {
181                this.queue.rx_waker.wake();
182                Poll::Ready(Ok(()))
183            }
184            Err(Error::Timeout) => Poll::Pending,
185            Err(e) => Poll::Ready(Err(e)),
186        }
187    }
188}
osal_rs/async_primitives/queue.rs

osal_rs/async_primitives/
queue.rs
