itsscb/embassy
1
0
Fork 0
mirror of https://github.com/embassy-rs/embassy.git synced 2025-09-26 20:00:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

Desugar some async fns

Browse Source
This commit is contained in:
Dániel Buga 2024-12-30 12:13:13 +01:00
parent a4f8fddd69
commit 44217aa092
No known key found for this signature in database
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;
}
}