diff --git a/cyw43/src/ioctl.rs b/cyw43/src/ioctl.rs index f8b2d9aba..af8bb695b 100644 --- a/cyw43/src/ioctl.rs +++ b/cyw43/src/ioctl.rs @@ -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 + '_ { 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 + '_ { + 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) { diff --git a/embassy-executor/src/spawner.rs b/embassy-executor/src/spawner.rs index 41320d4c3..ce24589bf 100644 --- a/embassy-executor/src/spawner.rs +++ b/embassy-executor/src/spawner.rs @@ -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 { 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 { 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. diff --git a/embassy-net-esp-hosted/src/ioctl.rs b/embassy-net-esp-hosted/src/ioctl.rs index e2a6815aa..512023206 100644 --- a/embassy-net-esp-hosted/src/ioctl.rs +++ b/embassy-net-esp-hosted/src/ioctl.rs @@ -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 + '_ { 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 + '_ { poll_fn(|cx| { let mut this = self.0.borrow_mut(); if this.is_init { @@ -118,6 +117,5 @@ impl Shared { Poll::Pending } }) - .await } } diff --git a/embassy-net-nrf91/src/lib.rs b/embassy-net-nrf91/src/lib.rs index 3abe2c766..61fcaea1f 100644 --- a/embassy-net-nrf91/src/lib.rs +++ b/embassy-net-nrf91/src/lib.rs @@ -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 + '_ { 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 { diff --git a/embassy-net/src/raw.rs b/embassy-net/src/raw.rs index a88bcc458..c9f753f13 100644 --- a/embassy-net/src/raw.rs +++ b/embassy-net/src/raw.rs @@ -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 + '_ { + 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 + '_ { + 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 + '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 + '_ { + poll_fn(|cx| { self.with_mut(|s, _| { if s.send_queue() == 0 { Poll::Ready(()) @@ -176,7 +176,6 @@ impl<'a> RawSocket<'a> { } }) }) - .await } } diff --git a/embassy-net/src/tcp.rs b/embassy-net/src/tcp.rs index 32d374064..d0230b581 100644 --- a/embassy-net/src/tcp.rs +++ b/embassy-net/src/tcp.rs @@ -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 + '_ { + 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 + '_ { + 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 { - self.io.write(buf).await + pub fn write<'s>(&'s mut self, buf: &'s [u8]) -> impl Future> + '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> + '_ { + 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 + '_ { + 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 { - self.io.read(buf).await + pub fn read<'s>(&'s mut self, buf: &'s mut [u8]) -> impl Future> + '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 + '_ { + 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 { - self.io.write(buf).await + pub fn write<'s>(&'s mut self, buf: &'s [u8]) -> impl Future> + '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> + '_ { + 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 { - poll_fn(move |cx| { + fn read<'s>(&'s mut self, buf: &'s mut [u8]) -> impl Future> + '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 { - poll_fn(move |cx| { + fn write<'s>(&'s mut self, buf: &'s [u8]) -> impl Future> + '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(&mut self, f: F) -> Result @@ -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> + '_ { + 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 { diff --git a/embassy-net/src/udp.rs b/embassy-net/src/udp.rs index 76602edc2..64a22d45b 100644 --- a/embassy-net/src/udp.rs +++ b/embassy-net/src/udp.rs @@ -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 + '_ { + 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> + '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 + '_ { + 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 + '_ { + 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. diff --git a/embassy-nrf/src/buffered_uarte.rs b/embassy-nrf/src/buffered_uarte.rs index b55e70a36..c3fcfd06e 100644 --- a/embassy-nrf/src/buffered_uarte.rs +++ b/embassy-nrf/src/buffered_uarte.rs @@ -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 { + pub fn write<'a>(&'a mut self, buf: &'a [u8]) -> impl Future> + '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> + '_ { 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> { 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`. diff --git a/embassy-nrf/src/qspi.rs b/embassy-nrf/src/qspi.rs index 255b43c33..17e127700 100755 --- a/embassy-nrf/src/qspi.rs +++ b/embassy-nrf/src/qspi.rs @@ -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 { 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() { diff --git a/embassy-nrf/src/usb/mod.rs b/embassy-nrf/src/usb/mod.rs index a9bf16708..06dae694b 100644 --- a/embassy-nrf/src/usb/mod.rs +++ b/embassy-nrf/src/usb/mod.rs @@ -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 { + 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 { 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 { + self.wait_enabled_state(false) } } diff --git a/embassy-nrf/src/usb/vbus_detect.rs b/embassy-nrf/src/usb/vbus_detect.rs index bdc088dcb..8794beb2d 100644 --- a/embassy-nrf/src/usb/vbus_detect.rs +++ b/embassy-nrf/src/usb/vbus_detect.rs @@ -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> { + 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> { poll_fn(move |cx| { POWER_WAKER.register(cx.waker()); @@ -175,6 +174,5 @@ impl VbusDetect for &SoftwareVbusDetect { Poll::Pending } }) - .await } } diff --git a/embassy-rp/src/adc.rs b/embassy-rp/src/adc.rs index 9582e43c8..19441f194 100644 --- a/embassy-rp/src/adc.rs +++ b/embassy-rp/src/adc.rs @@ -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 { 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. diff --git a/embassy-sync/src/mutex.rs b/embassy-sync/src/mutex.rs index 08f66e374..f25f74336 100644 --- a/embassy-sync/src/mutex.rs +++ b/embassy-sync/src/mutex.rs @@ -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> { 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. diff --git a/embassy-sync/src/once_lock.rs b/embassy-sync/src/once_lock.rs index 55608ba32..cd05b986d 100644 --- a/embassy-sync/src/once_lock.rs +++ b/embassy-sync/src/once_lock.rs @@ -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 OnceLock { /// 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 { poll_fn(|cx| match self.try_get() { Some(data) => Poll::Ready(data), None => { @@ -63,7 +63,6 @@ impl OnceLock { Poll::Pending } }) - .await } /// Try to get a reference to the underlying value if it exists. diff --git a/embassy-sync/src/watch.rs b/embassy-sync/src/watch.rs index 404e31714..e76646c0b 100644 --- a/embassy-sync/src/watch.rs +++ b/embassy-sync/src/watch.rs @@ -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 + ?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 + '_ { + 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. diff --git a/embassy-sync/src/zerocopy_channel.rs b/embassy-sync/src/zerocopy_channel.rs index fabb69bf6..56433cd8a 100644 --- a/embassy-sync/src/zerocopy_channel.rs +++ b/embassy-sync/src/zerocopy_channel.rs @@ -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 { + 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 { + 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. diff --git a/embassy-usb/src/class/cdc_acm.rs b/embassy-usb/src/class/cdc_acm.rs index 2823e522e..c5b1a56fe 100644 --- a/embassy-usb/src/class/cdc_acm.rs +++ b/embassy-usb/src/class/cdc_acm.rs @@ -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 + '_ { poll_fn(|cx| { if self.changed.load(Ordering::Relaxed) { self.changed.store(false, Ordering::Relaxed); @@ -118,7 +118,6 @@ impl ControlShared { Poll::Pending } }) - .await; } } diff --git a/embassy-usb/src/class/uac1/speaker.rs b/embassy-usb/src/class/uac1/speaker.rs index 6c3a4e378..25de25d9c 100644 --- a/embassy-usb/src/class/uac1/speaker.rs +++ b/embassy-usb/src/class/uac1/speaker.rs @@ -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 + '_ { 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; } }