LailaTheElf/embassy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #3698 from bugadani/desugar

Browse Source 
Desugar some async fns
This commit is contained in:
Dario Nieuwenhuis 2024-12-30 11:50:48 +00:00 committed by GitHub
commit 9e34058c2b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
18 changed files with 103 additions and 124 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
cyw43/src/ioctl.rs
View File

@ -1,5 +1,5 @@
use core::cell::{Cell, RefCell};
use core::future::poll_fn;
use core::future::{poll_fn, Future};
use core::task::{Poll, Waker};
use embassy_sync::waitqueue::WakerRegistration;
@ -71,7 +71,7 @@ impl IoctlState {
self.wakers.borrow_mut().runner.register(waker);
}
pub async fn wait_complete(&self) -> usize {
pub fn wait_complete(&self) -> impl Future<Output = usize> + '_ {
poll_fn(|cx| {
if let IoctlStateInner::Done { resp_len } = self.state.get() {
Poll::Ready(resp_len)
@ -80,22 +80,18 @@ impl IoctlState {
Poll::Pending
}
})
.await
}
pub async fn wait_pending(&self) -> PendingIoctl {
let pending = poll_fn(|cx| {
pub fn wait_pending(&self) -> impl Future<Output = PendingIoctl> + '_ {
poll_fn(|cx| {
if let IoctlStateInner::Pending(pending) = self.state.get() {
self.state.set(IoctlStateInner::Sent { buf: pending.buf });
Poll::Ready(pending)
} else {
self.register_runner(cx.waker());
Poll::Pending
}
})
.await;
self.state.set(IoctlStateInner::Sent { buf: pending.buf });
pending
}
pub fn cancel_ioctl(&self) {

8
embassy-executor/src/spawner.rs
View File

@ -1,4 +1,4 @@
use core::future::poll_fn;
use core::future::{poll_fn, Future};
use core::marker::PhantomData;
use core::mem;
use core::sync::atomic::Ordering;
@ -100,7 +100,7 @@ impl Spawner {
/// # Panics
///
/// Panics if the current executor is not an Embassy executor.
pub async fn for_current_executor() -> Self {
pub fn for_current_executor() -> impl Future<Output = Self> {
poll_fn(|cx| {
let task = raw::task_from_waker(cx.waker());
let executor = unsafe {
@ -113,7 +113,6 @@ impl Spawner {
let executor = unsafe { raw::Executor::wrap(executor) };
Poll::Ready(Self::new(executor))
})
.await
}
/// Spawn a task into an executor.
@ -178,7 +177,7 @@ impl SendSpawner {
/// # Panics
///
/// Panics if the current executor is not an Embassy executor.
pub async fn for_current_executor() -> Self {
pub fn for_current_executor() -> impl Future<Output = Self> {
poll_fn(|cx| {
let task = raw::task_from_waker(cx.waker());
let executor = unsafe {
@ -190,7 +189,6 @@ impl SendSpawner {
};
Poll::Ready(Self::new(executor))
})
.await
}
/// Spawn a task into an executor.

8
embassy-net-esp-hosted/src/ioctl.rs
View File

@ -1,5 +1,5 @@
use core::cell::RefCell;
use core::future::poll_fn;
use core::future::{poll_fn, Future};
use core::task::Poll;
use embassy_sync::waitqueue::WakerRegistration;
@ -38,7 +38,7 @@ impl Shared {
}))
}
pub async fn ioctl_wait_complete(&self) -> usize {
pub fn ioctl_wait_complete(&self) -> impl Future<Output = usize> + '_ {
poll_fn(|cx| {
let mut this = self.0.borrow_mut();
if let IoctlState::Done { resp_len } = this.ioctl {
@ -48,7 +48,6 @@ impl Shared {
Poll::Pending
}
})
.await
}
pub async fn ioctl_wait_pending(&self) -> PendingIoctl {
@ -108,7 +107,7 @@ impl Shared {
this.control_waker.wake();
}
pub async fn init_wait(&self) {
pub fn init_wait(&self) -> impl Future<Output = ()> + '_ {
poll_fn(|cx| {
let mut this = self.0.borrow_mut();
if this.is_init {
@ -118,6 +117,5 @@ impl Shared {
Poll::Pending
}
})
.await
}
}

5
embassy-net-nrf91/src/lib.rs
View File

@ -9,7 +9,7 @@ mod fmt;
pub mod context;
use core::cell::RefCell;
use core::future::poll_fn;
use core::future::{poll_fn, Future};
use core::marker::PhantomData;
use core::mem::{self, MaybeUninit};
use core::ptr::{self, addr_of, addr_of_mut, copy_nonoverlapping};
@ -737,7 +737,7 @@ pub struct Control<'a> {
impl<'a> Control<'a> {
/// Wait for modem IPC to be initialized.
pub async fn wait_init(&self) {
pub fn wait_init(&self) -> impl Future<Output = ()> + '_ {
poll_fn(|cx| {
let mut state = self.state.borrow_mut();
if state.init {
@ -746,7 +746,6 @@ impl<'a> Control<'a> {
state.init_waker.register(cx.waker());
Poll::Pending
})
.await
}
async fn request(&self, msg: &mut Message, req_data: &[u8], resp_data: &mut [u8]) -> usize {

19
embassy-net/src/raw.rs
View File

@ -1,6 +1,6 @@
//! Raw sockets.
use core::future::poll_fn;
use core::future::{poll_fn, Future};
use core::mem;
use core::task::{Context, Poll};
@ -66,8 +66,8 @@ impl<'a> RawSocket<'a> {
///
/// A socket is readable when a packet has been received, or when there are queued packets in
/// the buffer.
pub async fn wait_recv_ready(&self) {
poll_fn(move |cx| self.poll_recv_ready(cx)).await
pub fn wait_recv_ready(&self) -> impl Future<Output = ()> + '_ {
poll_fn(move |cx| self.poll_recv_ready(cx))
}
/// Receive a datagram.
@ -115,8 +115,8 @@ impl<'a> RawSocket<'a> {
///
/// A socket becomes writable when there is space in the buffer, from initial memory or after
/// dispatching datagrams on a full buffer.
pub async fn wait_send_ready(&self) {
poll_fn(move |cx| self.poll_send_ready(cx)).await
pub fn wait_send_ready(&self) -> impl Future<Output = ()> + '_ {
poll_fn(move |cx| self.poll_send_ready(cx))
}
/// Wait until a datagram can be sent.
@ -141,8 +141,8 @@ impl<'a> RawSocket<'a> {
/// Send a datagram.
///
/// This method will wait until the datagram has been sent.`
pub async fn send(&self, buf: &[u8]) {
poll_fn(move |cx| self.poll_send(buf, cx)).await
pub fn send<'s>(&'s self, buf: &'s [u8]) -> impl Future<Output = ()> + 's {
poll_fn(|cx| self.poll_send(buf, cx))
}
/// Send a datagram.
@ -165,8 +165,8 @@ impl<'a> RawSocket<'a> {
/// Flush the socket.
///
/// This method will wait until the socket is flushed.
pub async fn flush(&mut self) {
poll_fn(move |cx| {
pub fn flush(&mut self) -> impl Future<Output = ()> + '_ {
poll_fn(|cx| {
self.with_mut(|s, _| {
if s.send_queue() == 0 {
Poll::Ready(())
@ -176,7 +176,6 @@ impl<'a> RawSocket<'a> {
}
})
})
.await
}
}

53
embassy-net/src/tcp.rs
View File

@ -8,7 +8,7 @@
//! Incoming connections when no socket is listening are rejected. To accept many incoming
//! connections, create many sockets and put them all into listening mode.
use core::future::poll_fn;
use core::future::{poll_fn, Future};
use core::mem;
use core::task::{Context, Poll};
@ -79,8 +79,8 @@ impl<'a> TcpReader<'a> {
/// (see [`may_recv()`](TcpSocket::may_recv)), and there is some pending data in the receive buffer.
///
/// This is the equivalent of [read](#method.read), without buffering any data.
pub async fn wait_read_ready(&self) {
poll_fn(move |cx| self.io.poll_read_ready(cx)).await
pub fn wait_read_ready(&self) -> impl Future<Output = ()> + '_ {
poll_fn(move |cx| self.io.poll_read_ready(cx))
}
/// Read data from the socket.
@ -131,24 +131,24 @@ impl<'a> TcpWriter<'a> {
/// (see [`may_send()`](TcpSocket::may_send)), and the transmit buffer is not full.
///
/// This is the equivalent of [write](#method.write), without sending any data.
pub async fn wait_write_ready(&self) {
poll_fn(move |cx| self.io.poll_write_ready(cx)).await
pub fn wait_write_ready(&self) -> impl Future<Output = ()> + '_ {
poll_fn(move |cx| self.io.poll_write_ready(cx))
}
/// Write data to the socket.
///
/// Returns how many bytes were written, or an error. If the socket is not ready to
/// accept data, it waits until it is.
pub async fn write(&mut self, buf: &[u8]) -> Result<usize, Error> {
self.io.write(buf).await
pub fn write<'s>(&'s mut self, buf: &'s [u8]) -> impl Future<Output = Result<usize, Error>> + 's {
self.io.write(buf)
}
/// Flushes the written data to the socket.
///
/// This waits until all data has been sent, and ACKed by the remote host. For a connection
/// closed with [`abort()`](TcpSocket::abort) it will wait for the TCP RST packet to be sent.
pub async fn flush(&mut self) -> Result<(), Error> {
self.io.flush().await
pub fn flush(&mut self) -> impl Future<Output = Result<(), Error>> + '_ {
self.io.flush()
}
/// Call `f` with the largest contiguous slice of octets in the transmit buffer,
@ -300,8 +300,8 @@ impl<'a> TcpSocket<'a> {
/// (see [may_recv](#method.may_recv)), and there is some pending data in the receive buffer.
///
/// This is the equivalent of [read](#method.read), without buffering any data.
pub async fn wait_read_ready(&self) {
poll_fn(move |cx| self.io.poll_read_ready(cx)).await
pub fn wait_read_ready(&self) -> impl Future<Output = ()> + '_ {
poll_fn(move |cx| self.io.poll_read_ready(cx))
}
/// Read data from the socket.
@ -311,8 +311,8 @@ impl<'a> TcpSocket<'a> {
///
/// A return value of Ok(0) means that the socket was closed and is longer
/// able to receive any data.
pub async fn read(&mut self, buf: &mut [u8]) -> Result<usize, Error> {
self.io.read(buf).await
pub fn read<'s>(&'s mut self, buf: &'s mut [u8]) -> impl Future<Output = Result<usize, Error>> + 's {
self.io.read(buf)
}
/// Wait until the socket becomes writable.
@ -321,24 +321,24 @@ impl<'a> TcpSocket<'a> {
/// (see [may_send](#method.may_send)), and the transmit buffer is not full.
///
/// This is the equivalent of [write](#method.write), without sending any data.
pub async fn wait_write_ready(&self) {
poll_fn(move |cx| self.io.poll_write_ready(cx)).await
pub fn wait_write_ready(&self) -> impl Future<Output = ()> + '_ {
poll_fn(move |cx| self.io.poll_write_ready(cx))
}
/// Write data to the socket.
///
/// Returns how many bytes were written, or an error. If the socket is not ready to
/// accept data, it waits until it is.
pub async fn write(&mut self, buf: &[u8]) -> Result<usize, Error> {
self.io.write(buf).await
pub fn write<'s>(&'s mut self, buf: &'s [u8]) -> impl Future<Output = Result<usize, Error>> + 's {
self.io.write(buf)
}
/// Flushes the written data to the socket.
///
/// This waits until all data has been sent, and ACKed by the remote host. For a connection
/// closed with [`abort()`](TcpSocket::abort) it will wait for the TCP RST packet to be sent.
pub async fn flush(&mut self) -> Result<(), Error> {
self.io.flush().await
pub fn flush(&mut self) -> impl Future<Output = Result<(), Error>> + '_ {
self.io.flush()
}
/// Set the timeout for the socket.
@ -501,8 +501,8 @@ impl<'d> TcpIo<'d> {
})
}
async fn read(&mut self, buf: &mut [u8]) -> Result<usize, Error> {
poll_fn(move |cx| {
fn read<'s>(&'s mut self, buf: &'s mut [u8]) -> impl Future<Output = Result<usize, Error>> + 's {
poll_fn(|cx| {
// CAUTION: smoltcp semantics around EOF are different to what you'd expect
// from posix-like IO, so we have to tweak things here.
self.with_mut(|s, _| match s.recv_slice(buf) {
@ -526,7 +526,6 @@ impl<'d> TcpIo<'d> {
Err(tcp::RecvError::InvalidState) => Poll::Ready(Err(Error::ConnectionReset)),
})
})
.await
}
fn poll_write_ready(&self, cx: &mut Context<'_>) -> Poll<()> {
@ -540,8 +539,8 @@ impl<'d> TcpIo<'d> {
})
}
async fn write(&mut self, buf: &[u8]) -> Result<usize, Error> {
poll_fn(move |cx| {
fn write<'s>(&'s mut self, buf: &'s [u8]) -> impl Future<Output = Result<usize, Error>> + 's {
poll_fn(|cx| {
self.with_mut(|s, _| match s.send_slice(buf) {
// Not ready to send (no space in the tx buffer)
Ok(0) => {
@ -554,7 +553,6 @@ impl<'d> TcpIo<'d> {
Err(tcp::SendError::InvalidState) => Poll::Ready(Err(Error::ConnectionReset)),
})
})
.await
}
async fn write_with<F, R>(&mut self, f: F) -> Result<R, Error>
@ -615,8 +613,8 @@ impl<'d> TcpIo<'d> {
.await
}
async fn flush(&mut self) -> Result<(), Error> {
poll_fn(move |cx| {
fn flush(&mut self) -> impl Future<Output = Result<(), Error>> + '_ {
poll_fn(|cx| {
self.with_mut(|s, _| {
let data_pending = (s.send_queue() > 0) && s.state() != tcp::State::Closed;
let fin_pending = matches!(
@ -636,7 +634,6 @@ impl<'d> TcpIo<'d> {
}
})
})
.await
}
fn recv_capacity(&self) -> usize {

22
embassy-net/src/udp.rs
View File

@ -1,6 +1,6 @@
//! UDP sockets.
use core::future::poll_fn;
use core::future::{poll_fn, Future};
use core::mem;
use core::task::{Context, Poll};
@ -107,8 +107,8 @@ impl<'a> UdpSocket<'a> {
///
/// A socket is readable when a packet has been received, or when there are queued packets in
/// the buffer.
pub async fn wait_recv_ready(&self) {
poll_fn(move |cx| self.poll_recv_ready(cx)).await
pub fn wait_recv_ready(&self) -> impl Future<Output = ()> + '_ {
poll_fn(move |cx| self.poll_recv_ready(cx))
}
/// Wait until a datagram can be read.
@ -134,8 +134,11 @@ impl<'a> UdpSocket<'a> {
/// This method will wait until a datagram is received.
///
/// Returns the number of bytes received and the remote endpoint.
pub async fn recv_from(&self, buf: &mut [u8]) -> Result<(usize, UdpMetadata), RecvError> {
poll_fn(move |cx| self.poll_recv_from(buf, cx)).await
pub fn recv_from<'s>(
&'s self,
buf: &'s mut [u8],
) -> impl Future<Output = Result<(usize, UdpMetadata), RecvError>> + 's {
poll_fn(|cx| self.poll_recv_from(buf, cx))
}
/// Receive a datagram.
@ -194,8 +197,8 @@ impl<'a> UdpSocket<'a> {
///
/// A socket becomes writable when there is space in the buffer, from initial memory or after
/// dispatching datagrams on a full buffer.
pub async fn wait_send_ready(&self) {
poll_fn(move |cx| self.poll_send_ready(cx)).await
pub fn wait_send_ready(&self) -> impl Future<Output = ()> + '_ {
poll_fn(|cx| self.poll_send_ready(cx))
}
/// Wait until a datagram can be sent.
@ -297,8 +300,8 @@ impl<'a> UdpSocket<'a> {
/// Flush the socket.
///
/// This method will wait until the socket is flushed.
pub async fn flush(&mut self) {
poll_fn(move |cx| {
pub fn flush(&mut self) -> impl Future<Output = ()> + '_ {
poll_fn(|cx| {
self.with_mut(|s, _| {
if s.send_queue() == 0 {
Poll::Ready(())
@ -308,7 +311,6 @@ impl<'a> UdpSocket<'a> {
}
})
})
.await
}
/// Returns the local endpoint of the socket.

11
embassy-nrf/src/buffered_uarte.rs
View File

@ -9,7 +9,7 @@
//! Please also see [crate::uarte] to understand when [BufferedUarte] should be used.
use core::cmp::min;
use core::future::poll_fn;
use core::future::{poll_fn, Future};
use core::marker::PhantomData;
use core::slice;
use core::sync::atomic::{compiler_fence, AtomicBool, AtomicU8, AtomicUsize, Ordering};
@ -452,7 +452,7 @@ impl<'d, U: UarteInstance> BufferedUarteTx<'d, U> {
}
/// Write a buffer into this writer, returning how many bytes were written.
pub async fn write(&mut self, buf: &[u8]) -> Result<usize, Error> {
pub fn write<'a>(&'a mut self, buf: &'a [u8]) -> impl Future<Output = Result<usize, Error>> + 'a {
poll_fn(move |cx| {
//trace!("poll_write: {:?}", buf.len());
let ss = U::state();
@ -477,7 +477,6 @@ impl<'d, U: UarteInstance> BufferedUarteTx<'d, U> {
Poll::Ready(Ok(n))
})
.await
}
/// Try writing a buffer without waiting, returning how many bytes were written.
@ -504,7 +503,7 @@ impl<'d, U: UarteInstance> BufferedUarteTx<'d, U> {
}
/// Flush this output stream, ensuring that all intermediately buffered contents reach their destination.
pub async fn flush(&mut self) -> Result<(), Error> {
pub fn flush(&mut self) -> impl Future<Output = Result<(), Error>> + '_ {
poll_fn(move |cx| {
//trace!("poll_flush");
let ss = U::state();
@ -517,7 +516,6 @@ impl<'d, U: UarteInstance> BufferedUarteTx<'d, U> {
Poll::Ready(Ok(()))
})
.await
}
}
@ -721,7 +719,7 @@ impl<'d, U: UarteInstance, T: TimerInstance> BufferedUarteRx<'d, U, T> {
}
/// Return the contents of the internal buffer, filling it with more data from the inner reader if it is empty.
pub async fn fill_buf(&mut self) -> Result<&[u8], Error> {
pub fn fill_buf(&mut self) -> impl Future<Output = Result<&'_ [u8], Error>> {
poll_fn(move |cx| {
compiler_fence(Ordering::SeqCst);
//trace!("poll_read");
@ -771,7 +769,6 @@ impl<'d, U: UarteInstance, T: TimerInstance> BufferedUarteRx<'d, U, T> {
let buf = s.rx_buf.buf.load(Ordering::Relaxed);
Poll::Ready(Ok(unsafe { slice::from_raw_parts(buf.add(start), n) }))
})
.await
}
/// Tell this buffer that `amt` bytes have been consumed from the buffer, so they should no longer be returned in calls to `fill_buf`.

5
embassy-nrf/src/qspi.rs
View File

@ -2,7 +2,7 @@
#![macro_use]
use core::future::poll_fn;
use core::future::{poll_fn, Future};
use core::marker::PhantomData;
use core::ptr;
use core::task::Poll;
@ -314,7 +314,7 @@ impl<'d, T: Instance> Qspi<'d, T> {
Ok(())
}
async fn wait_ready(&mut self) {
fn wait_ready(&mut self) -> impl Future<Output = ()> {
poll_fn(move |cx| {
let r = T::regs();
let s = T::state();
@ -324,7 +324,6 @@ impl<'d, T: Instance> Qspi<'d, T> {
}
Poll::Pending
})
.await
}
fn blocking_wait_ready() {

14
embassy-nrf/src/usb/mod.rs
View File

@ -4,7 +4,7 @@
pub mod vbus_detect;
use core::future::poll_fn;
use core::future::{poll_fn, Future};
use core::marker::PhantomData;
use core::mem::MaybeUninit;
use core::sync::atomic::{compiler_fence, AtomicU32, Ordering};
@ -219,8 +219,8 @@ impl<'d, T: Instance, V: VbusDetect> driver::Bus for Bus<'d, T, V> {
regs.enable().write(|x| x.set_enable(false));
}
async fn poll(&mut self) -> Event {
poll_fn(move |cx| {
fn poll(&mut self) -> impl Future<Output = Event> {
poll_fn(|cx| {
BUS_WAKER.register(cx.waker());
let regs = T::regs();
@ -277,7 +277,6 @@ impl<'d, T: Instance, V: VbusDetect> driver::Bus for Bus<'d, T, V> {
Poll::Pending
})
.await
}
fn endpoint_set_stalled(&mut self, ep_addr: EndpointAddress, stalled: bool) {
@ -468,7 +467,7 @@ impl<'d, T: Instance, Dir: EndpointDir> driver::Endpoint for Endpoint<'d, T, Dir
#[allow(private_bounds)]
impl<'d, T: Instance, Dir: EndpointDir> Endpoint<'d, T, Dir> {
async fn wait_enabled_state(&mut self, state: bool) {
fn wait_enabled_state(&mut self, state: bool) -> impl Future<Output = ()> {
let i = self.info.addr.index();
assert!(i != 0);
@ -480,12 +479,11 @@ impl<'d, T: Instance, Dir: EndpointDir> Endpoint<'d, T, Dir> {
Poll::Pending
}
})
.await
}
/// Wait for the endpoint to be disabled
pub async fn wait_disabled(&mut self) {
self.wait_enabled_state(false).await
pub fn wait_disabled(&mut self) -> impl Future<Output = ()> {
self.wait_enabled_state(false)
}
}

10
embassy-nrf/src/usb/vbus_detect.rs
View File

@ -1,6 +1,6 @@
//! Trait and implementations for performing VBUS detection.
use core::future::poll_fn;
use core::future::{poll_fn, Future};
use core::sync::atomic::{AtomicBool, Ordering};
use core::task::Poll;
@ -99,8 +99,8 @@ impl VbusDetect for HardwareVbusDetect {
regs.usbregstatus().read().vbusdetect()
}
async fn wait_power_ready(&mut self) -> Result<(), ()> {
poll_fn(move |cx| {
fn wait_power_ready(&mut self) -> impl Future<Output = Result<(), ()>> {
poll_fn(|cx| {
POWER_WAKER.register(cx.waker());
let regs = USB_REG_PERI;
@ -112,7 +112,6 @@ impl VbusDetect for HardwareVbusDetect {
Poll::Pending
}
})
.await
}
}
@ -163,7 +162,7 @@ impl VbusDetect for &SoftwareVbusDetect {
self.usb_detected.load(Ordering::Relaxed)
}
async fn wait_power_ready(&mut self) -> Result<(), ()> {
fn wait_power_ready(&mut self) -> impl Future<Output = Result<(), ()>> {
poll_fn(move |cx| {
POWER_WAKER.register(cx.waker());
@ -175,6 +174,5 @@ impl VbusDetect for &SoftwareVbusDetect {
Poll::Pending
}
})
.await
}
}

8
embassy-rp/src/adc.rs
View File

@ -1,5 +1,5 @@
//! ADC driver.
use core::future::poll_fn;
use core::future::{poll_fn, Future};
use core::marker::PhantomData;
use core::mem;
use core::sync::atomic::{compiler_fence, Ordering};
@ -193,18 +193,18 @@ impl<'d> Adc<'d, Async> {
Self { phantom: PhantomData }
}
async fn wait_for_ready() {
fn wait_for_ready() -> impl Future<Output = ()> {
let r = Self::regs();
r.inte().write(|w| w.set_fifo(true));
compiler_fence(Ordering::SeqCst);
poll_fn(|cx| {
poll_fn(move |cx| {
WAKER.register(cx.waker());
if r.cs().read().ready() {
return Poll::Ready(());
}
Poll::Pending
})
.await;
}
/// Sample a value from a channel until completed.

5
embassy-sync/src/mutex.rs
View File

@ -2,7 +2,7 @@
//!
//! This module provides a mutex that can be used to synchronize data between asynchronous tasks.
use core::cell::{RefCell, UnsafeCell};
use core::future::poll_fn;
use core::future::{poll_fn, Future};
use core::ops::{Deref, DerefMut};
use core::task::Poll;
use core::{fmt, mem};
@ -73,7 +73,7 @@ where
/// Lock the mutex.
///
/// This will wait for the mutex to be unlocked if it's already locked.
pub async fn lock(&self) -> MutexGuard<'_, M, T> {
pub fn lock(&self) -> impl Future<Output = MutexGuard<'_, M, T>> {
poll_fn(|cx| {
let ready = self.state.lock(|s| {
let mut s = s.borrow_mut();
@ -92,7 +92,6 @@ where
Poll::Pending
}
})
.await
}
/// Attempt to immediately lock the mutex.

5
embassy-sync/src/once_lock.rs
View File

@ -1,7 +1,7 @@
//! Synchronization primitive for initializing a value once, allowing others to await a reference to the value.
use core::cell::Cell;
use core::future::poll_fn;
use core::future::{poll_fn, Future};
use core::mem::MaybeUninit;
use core::sync::atomic::{AtomicBool, Ordering};
use core::task::Poll;
@ -55,7 +55,7 @@ impl<T> OnceLock<T> {
/// Get a reference to the underlying value, waiting for it to be set.
/// If the value is already set, this will return immediately.
pub async fn get(&self) -> &T {
pub fn get(&self) -> impl Future<Output = &T> {
poll_fn(|cx| match self.try_get() {
Some(data) => Poll::Ready(data),
None => {
@ -63,7 +63,6 @@ impl<T> OnceLock<T> {
Poll::Pending
}
})
.await
}
/// Try to get a reference to the underlying value if it exists.

6
embassy-sync/src/watch.rs
View File

@ -1,7 +1,7 @@
//! A synchronization primitive for passing the latest value to **multiple** receivers.
use core::cell::RefCell;
use core::future::poll_fn;
use core::future::{poll_fn, Future};
use core::marker::PhantomData;
use core::ops::{Deref, DerefMut};
use core::task::{Context, Poll};
@ -547,8 +547,8 @@ impl<'a, T: Clone, W: WatchBehavior<T> + ?Sized> Rcv<'a, T, W> {
/// Returns the current value of the `Watch` once it is initialized, marking it as seen.
///
/// **Note**: Futures do nothing unless you `.await` or poll them.
pub async fn get(&mut self) -> T {
poll_fn(|cx| self.watch.poll_get(&mut self.at_id, cx)).await
pub fn get(&mut self) -> impl Future<Output = T> + '_ {
poll_fn(|cx| self.watch.poll_get(&mut self.at_id, cx))
}
/// Tries to get the current value of the `Watch` without waiting, marking it as seen.

24
embassy-sync/src/zerocopy_channel.rs
View File

@ -15,7 +15,7 @@
//! another message will result in an error being returned.
use core::cell::RefCell;
use core::future::poll_fn;
use core::future::{poll_fn, Future};
use core::marker::PhantomData;
use core::task::{Context, Poll};
@ -131,12 +131,15 @@ impl<'a, M: RawMutex, T> Sender<'a, M, T> {
}
/// Asynchronously send a value over the channel.
pub async fn send(&mut self) -> &mut T {
let i = poll_fn(|cx| {
pub fn send(&mut self) -> impl Future<Output = &mut T> {
poll_fn(|cx| {
self.channel.state.lock(|s| {
let s = &mut *s.borrow_mut();
match s.push_index() {
Some(i) => Poll::Ready(i),
Some(i) => {
let r = unsafe { &mut *self.channel.buf.add(i) };
Poll::Ready(r)
}
None => {
s.receive_waker.register(cx.waker());
Poll::Pending
@ -144,8 +147,6 @@ impl<'a, M: RawMutex, T> Sender<'a, M, T> {
}
})
})
.await;
unsafe { &mut *self.channel.buf.add(i) }
}
/// Notify the channel that the sending of the value has been finalized.
@ -213,12 +214,15 @@ impl<'a, M: RawMutex, T> Receiver<'a, M, T> {
}
/// Asynchronously receive a value over the channel.
pub async fn receive(&mut self) -> &mut T {
let i = poll_fn(|cx| {
pub fn receive(&mut self) -> impl Future<Output = &mut T> {
poll_fn(|cx| {
self.channel.state.lock(|s| {
let s = &mut *s.borrow_mut();
match s.pop_index() {
Some(i) => Poll::Ready(i),
Some(i) => {
let r = unsafe { &mut *self.channel.buf.add(i) };
Poll::Ready(r)
}
None => {
s.send_waker.register(cx.waker());
Poll::Pending
@ -226,8 +230,6 @@ impl<'a, M: RawMutex, T> Receiver<'a, M, T> {
}
})
})
.await;
unsafe { &mut *self.channel.buf.add(i) }
}
/// Notify the channel that the receiving of the value has been finalized.

5
embassy-usb/src/class/cdc_acm.rs
View File

@ -1,7 +1,7 @@
//! CDC-ACM class implementation, aka Serial over USB.
use core::cell::{Cell, RefCell};
use core::future::poll_fn;
use core::future::{poll_fn, Future};
use core::mem::{self, MaybeUninit};
use core::sync::atomic::{AtomicBool, Ordering};
use core::task::Poll;
@ -108,7 +108,7 @@ impl Default for ControlShared {
}
impl ControlShared {
async fn changed(&self) {
fn changed(&self) -> impl Future<Output = ()> + '_ {
poll_fn(|cx| {
if self.changed.load(Ordering::Relaxed) {
self.changed.store(false, Ordering::Relaxed);
@ -118,7 +118,6 @@ impl ControlShared {
Poll::Pending
}
})
.await;
}
}

5
embassy-usb/src/class/uac1/speaker.rs
View File

@ -11,7 +11,7 @@
//! The class provides volume and mute controls for each channel.
use core::cell::{Cell, RefCell};
use core::future::poll_fn;
use core::future::{poll_fn, Future};
use core::marker::PhantomData;
use core::sync::atomic::{AtomicBool, AtomicU32, Ordering};
use core::task::Poll;
@ -389,7 +389,7 @@ impl<'d> Default for SharedControl<'d> {
}
impl<'d> SharedControl<'d> {
async fn changed(&self) {
fn changed(&self) -> impl Future<Output = ()> + '_ {
poll_fn(|context| {
if self.changed.load(Ordering::Relaxed) {
self.changed.store(false, Ordering::Relaxed);
@ -399,7 +399,6 @@ impl<'d> SharedControl<'d> {
Poll::Pending
}
})
.await;
}
}