Enable async operation

#63

Cargo.toml

@@ -46,7 +46,14 @@ coap-handler-0-1 = { package = "coap-handler", version = "^0.1.4" }
 coap-handler-0-2 = { package = "coap-handler", version = "^0.2.0" }
 embedded-nal = { version = "0.6.0", optional = true }
 embedded-nal-tcpextensions = { version = "0.1", optional = true }
+embedded-nal-async = { version = "0.6", optional = true }
+embedded-io-async = { version = "0.6", optional = true }
 pin-utils = "0.1"
+pin-project = "1.0.11"
+
+embedded-hal-async = { version = "1", optional = true }
+
+critical-section = { version = "1.0", optional = true }
 
 [build-dependencies]
 shlex = "0.1.1"
@@ -70,9 +77,16 @@ panic_handler_format = []
 # only affects that single thread.
 panic_handler_crash = []
 
+# Provide an implementation of critical-section 1.0 using irq_disable()/_restore().
+provide_critical_section_1_0 = ["critical-section/restore-state-u32"]
+
 with_coap_message = []
 with_coap_handler = []
+
 with_embedded_nal = ["embedded-nal", "embedded-nal-tcpextensions"]
+with_embedded_nal_async = [ "embedded-nal-async", "embedded-io-async" ]
+
+with_embedded_hal_async = [ "embedded-hal-async" ]
 
 # See msg::v2 documentation. Enabling this exposes components not under semver
 # guarantees.

src/async_helpers.rs

@@ -0,0 +1,82 @@
+//! Tools used internally to create futures more easily
+
+/// A trait similar to Future that is practical to implement for the typical RIOT situations where
+/// a waker needs to be converted into a function and argument pointer.
+///
+/// Wrapped in a [RiotStylePollStruct], it implements [Future], and the conversion between the arg
+/// pointer and the full struct is taken care of. That wrapper may also do any optimizations such
+/// as not really storing the waker if it can be compressed to a single word instead.
+///
+/// ## Implementing
+///
+/// While this can legally be implemented without unsafe, practical use will require unsafe, and
+/// that requires sticking to the rules:
+///
+/// * Whenever [poll()] is called, do whatever the future needs to do after having been awoken. If
+///   this returns [core::task::Poll::Pending] (and the future wants to be polled ever again), it
+///   must then pass on the `arg` to some RIOT callback setter together with a static function of a
+///   suitable signature. Conventionally, that function is called `Self::callback()`.
+///
+/// * When that callback function is called (and has any arguments), it may inspect the arguments
+///   to decide to return early (for example, if it receives "chatter" that is unrelated to the
+///   completion of the future). If it decides that this is now the callback that should make
+///   progress, it must call [`RiotStylePollStruct::<Self>::callback(arg)`], with `arg` being the
+///   value that was passed around through RIOT from the poll function.
+///
+/// * To the author's knowledge, the mechanism itself has no requirements of not shuffling any
+///   items in and out of any `&mut` that are involved (otherwise, they would be pinned). However,
+///   the callback mechanism itself may require no such shuffling to occur, in which case it is the
+///   implementor's responsibility to not just move its data around.
+pub(crate) trait RiotStyleFuture {
+    type Output;
+    fn poll(&mut self, arg: *mut riot_sys::libc::c_void) -> core::task::Poll<Self::Output>;
+}
+
+/// Wrapper that makes a [core::future::Future] out of a [RiotStyleFuture] (see there for usage)
+// FIXME: I'm not sure the presence and absence of #[pin] is right about these ones, but anyway,
+// given they're not pub, and this module is what captures the unsafety guarantees (assisted by the
+// requirements on RiotStyleFuture), this should be no worse than manually safe-declaring any
+// access to args and waker.
+#[pin_project::pin_project]
+pub(crate) struct RiotStylePollStruct<A: RiotStyleFuture> {
+    // The order of these is important: args is dropped first, thereby unregistering any callbacks.
+    // Only then, the waker too can be dropped.
+    args: A,
+    // We can probably save that one if we rely on the waker pointing to a task, but let's not
+    // force this on the system yet. (The TaskRef is short enough we could store it in the argument
+    // of the callback).
+    waker: Option<core::task::Waker>,
+}
+impl<A: RiotStyleFuture> RiotStylePollStruct<A> {
+    pub(crate) fn new(args: A) -> Self {
+        Self { args, waker: None }
+    }
+
+    /// Reconstruct a Self and run its waker (if one is present)
+    pub(crate) unsafe fn callback(arg: *mut riot_sys::libc::c_void) {
+        // Actually Pin<>, but we just promise not to move it.
+        let f: &mut Self = &mut *(arg as *mut _);
+        // If it fires multiple times, we ignore it (the waker has been taken) -- unless the future
+        // has been polled again, there is no use in waking for it multiple times. (We could also
+        // remove the callback, but who knows how costly that might be).
+        f.waker.take().map(|w| w.wake());
+    }
+}
+impl<A: RiotStyleFuture> core::future::Future for RiotStylePollStruct<A> {
+    type Output = A::Output;
+    fn poll(
+        mut self: core::pin::Pin<&mut Self>,
+        context: &mut core::task::Context<'_>,
+    ) -> core::task::Poll<Self::Output> {
+        let arg = unsafe { self.as_mut().get_unchecked_mut() } as *mut Self as *mut _;
+        match self.args.poll(arg) {
+            core::task::Poll::Pending => {
+                // Actually we only need to do that if we're returning Pending
+                self.waker = Some(context.waker().clone());
+
+                core::task::Poll::Pending
+            }
+            ready => ready,
+        }
+    }
+}

src/error.rs

@@ -23,7 +23,7 @@ pub trait NegativeErrorExt {
 /// represent `Result<positive_usize, NumericError>` as just the isize it originally was. For the
 /// time being, this works well enough, and performance evaluation can later be done against a
 /// manually implemented newtype around isize that'd be used to represent the Result.
-#[derive(Debug, PartialEq)]
+#[derive(Debug, PartialEq, Eq)]
 pub struct NumericError {
     #[deprecated(note = "Use the .number() method")]
     pub number: isize,
@@ -112,4 +112,5 @@ macro_rules! E {
 // See module level comment
 E!(EAGAIN);
 E!(ENOMEM);
+E!(ENOSPC);
 E!(EOVERFLOW);

src/impl_critical_section.rs

@@ -0,0 +1,21 @@
+//! This module implements critical_section using RIOT's irq_disable()/_restore()
+#![cfg(feature = "provide_critical_section_1_0")]
+
+use critical_section::RawRestoreState;
+
+struct CriticalSection(usize);
+critical_section::set_impl!(CriticalSection);
+
+unsafe impl critical_section::Impl for CriticalSection {
+    unsafe fn acquire() -> RawRestoreState {
+        // If this fails to compile (because Rust-on-RIOT has gained support for non-32bit
+        // architectures), by that time hopefully critical-section > 1.1.2 has been released, which
+        // has restore-state-usize. Just increment the dependency version and set its feature from
+        // restore-state-u32 to restore-state-usize.
+        unsafe { riot_sys::irq_disable() }
+    }
+
+    unsafe fn release(token: RawRestoreState) {
+        unsafe { riot_sys::irq_restore(token) };
+    }
+}

src/lib.rs

@@ -152,6 +152,12 @@ pub mod coap_message;
 pub mod socket;
 #[cfg(all(riot_module_sock_udp, feature = "with_embedded_nal"))]
 pub mod socket_embedded_nal;
+#[cfg(all(
+    riot_module_sock_udp,
+    riot_module_sock_aux_local,
+    feature = "with_embedded_nal_async"
+))]
+pub mod socket_embedded_nal_async_udp;
 #[cfg(all(riot_module_sock_tcp, feature = "with_embedded_nal"))]
 pub mod socket_embedded_nal_tcp;
 
@@ -175,6 +181,7 @@ pub mod microbit;
 #[cfg(riot_module_vfs)]
 pub mod vfs;
 
+mod impl_critical_section;
 pub mod interrupt;
 #[path = "main_module.rs"]
 pub mod main;
@@ -183,3 +190,5 @@ pub mod led;
 
 #[cfg(riot_module_auto_init)]
 pub mod auto_init;
+
+mod async_helpers;

src/socket.rs

@@ -134,3 +134,75 @@ mod nal_impls {
         }
     }
 }
+
+// FIXME: This is literally copied from above, just with another module
+#[cfg(feature = "with_embedded_nal_async")]
+mod nal_impls_async {
+    use super::*;
+    use embedded_nal_async::SocketAddr;
+
+    impl Into<UdpEp> for &SocketAddr {
+        fn into(self) -> UdpEp {
+            use SocketAddr::*;
+
+            // Constructing via default avoids using the volatile names of the union types
+            let mut ep: riot_sys::sock_udp_ep_t = Default::default();
+
+            ep.family = match self {
+                V4(_) => riot_sys::AF_INET as _,
+                V6(_) => riot_sys::AF_INET6 as _,
+            };
+            ep.netif = match self {
+                V4(_) => 0,
+                V6(a) => a.scope_id() as _,
+            };
+            ep.port = self.port();
+            match self {
+                V4(a) => {
+                    ep.addr.ipv4 = a.ip().octets();
+                }
+                V6(a) => {
+                    ep.addr.ipv6 = a.ip().octets();
+                }
+            }
+
+            UdpEp(ep)
+        }
+    }
+
+    impl Into<UdpEp> for SocketAddr {
+        fn into(self) -> UdpEp {
+            (&self).into()
+        }
+    }
+
+    impl Into<SocketAddr> for &UdpEp {
+        fn into(self) -> SocketAddr {
+            match self.0.family as _ {
+                riot_sys::AF_INET6 => embedded_nal_async::SocketAddrV6::new(
+                    // unsafe: Access to C union whose type was just checked
+                    unsafe { self.0.addr.ipv6.into() },
+                    self.0.port,
+                    0,
+                    self.0.netif.into(),
+                )
+                .into(),
+
+                riot_sys::AF_INET => embedded_nal_async::SocketAddrV4::new(
+                    // unsafe: Access to C union whose type was just checked
+                    unsafe { self.0.addr.ipv4.into() },
+                    self.0.port,
+                )
+                .into(),
+
+                _ => panic!("Endpoint not expressible in embedded_nal_async"),
+            }
+        }
+    }
+
+    impl Into<SocketAddr> for UdpEp {
+        fn into(self) -> SocketAddr {
+            (&self).into()
+        }
+    }
+}