Merge #895

895: Implement embedded-nal-async traits for embassy-net r=lulf a=lulf



Co-authored-by: Ulf Lilleengen <[email protected]>

4 files changed, 299 insertions, 1 deletions

diff --git a/embassy-net/Cargo.toml b/embassy-net/Cargo.toml
index 64cb5bd8f..fface207b 100644
--- a/embassy-net/Cargo.toml
+++ b/embassy-net/Cargo.toml
@@ -31,6 +31,7 @@ pool-16 = []
 pool-32 = []
 pool-64 = []
 pool-128 = []
+unstable-traits = []
 
 [dependencies]
 
@@ -48,6 +49,8 @@ generic-array       = { version = "0.14.4", default-features = false }
 stable_deref_trait  = { version = "1.2.0", default-features = false }
 futures             = { version = "0.3.17", default-features = false, features = [ "async-await" ] }
 atomic-pool = "0.2.1"
+atomic-polyfill = "0.1.5"
+embedded-nal-async = "0.2.0"
 
 [dependencies.smoltcp]
 version = "0.8.0"
diff --git a/embassy-net/src/tcp.rs b/embassy-net/src/tcp.rs
index c18391ace..2e276ecd3 100644
--- a/embassy-net/src/tcp.rs
+++ b/embassy-net/src/tcp.rs
@@ -328,3 +328,172 @@ impl<'d> embedded_io::asynch::Write for TcpWriter<'d> {
         self.io.flush()
     }
 }
+
+#[cfg(feature = "unstable-traits")]
+pub mod client {
+    use core::mem::MaybeUninit;
+    use core::ptr::NonNull;
+
+    use atomic_polyfill::{AtomicBool, Ordering};
+    use embedded_nal_async::IpAddr;
+
+    use super::*;
+
+    /// TCP client capable of creating up to N multiple connections with tx and rx buffers according to TX_SZ and RX_SZ.
+    pub struct TcpClient<'d, D: Device, const N: usize, const TX_SZ: usize = 1024, const RX_SZ: usize = 1024> {
+        stack: &'d Stack<D>,
+        state: &'d TcpClientState<N, TX_SZ, RX_SZ>,
+    }
+
+    impl<'d, D: Device, const N: usize, const TX_SZ: usize, const RX_SZ: usize> TcpClient<'d, D, N, TX_SZ, RX_SZ> {
+        /// Create a new TcpClient
+        pub fn new(stack: &'d Stack<D>, state: &'d TcpClientState<N, TX_SZ, RX_SZ>) -> Self {
+            Self { stack, state }
+        }
+    }
+
+    impl<'d, D: Device, const N: usize, const TX_SZ: usize, const RX_SZ: usize> embedded_nal_async::TcpConnect
+        for TcpClient<'d, D, N, TX_SZ, RX_SZ>
+    {
+        type Error = Error;
+        type Connection<'m> = TcpConnection<'m, N, TX_SZ, RX_SZ> where Self: 'm;
+        type ConnectFuture<'m> = impl Future<Output = Result<Self::Connection<'m>, Self::Error>> + 'm
+    where
+        Self: 'm;
+
+        fn connect<'m>(&'m self, remote: embedded_nal_async::SocketAddr) -> Self::ConnectFuture<'m> {
+            async move {
+                let addr: crate::IpAddress = match remote.ip() {
+                    IpAddr::V4(addr) => crate::IpAddress::Ipv4(crate::Ipv4Address::from_bytes(&addr.octets())),
+                    #[cfg(feature = "proto-ipv6")]
+                    IpAddr::V6(addr) => crate::IpAddress::Ipv6(crate::Ipv6Address::from_bytes(&addr.octets())),
+                    #[cfg(not(feature = "proto-ipv6"))]
+                    IpAddr::V6(_) => panic!("ipv6 support not enabled"),
+                };
+                let remote_endpoint = (addr, remote.port());
+                let mut socket = TcpConnection::new(&self.stack, self.state)?;
+                socket
+                    .socket
+                    .connect(remote_endpoint)
+                    .await
+                    .map_err(|_| Error::ConnectionReset)?;
+                Ok(socket)
+            }
+        }
+    }
+
+    pub struct TcpConnection<'d, const N: usize, const TX_SZ: usize, const RX_SZ: usize> {
+        socket: TcpSocket<'d>,
+        state: &'d TcpClientState<N, TX_SZ, RX_SZ>,
+        bufs: NonNull<([u8; TX_SZ], [u8; RX_SZ])>,
+    }
+
+    impl<'d, const N: usize, const TX_SZ: usize, const RX_SZ: usize> TcpConnection<'d, N, TX_SZ, RX_SZ> {
+        fn new<D: Device>(stack: &'d Stack<D>, state: &'d TcpClientState<N, TX_SZ, RX_SZ>) -> Result<Self, Error> {
+            let mut bufs = state.pool.alloc().ok_or(Error::ConnectionReset)?;
+            Ok(Self {
+                socket: unsafe { TcpSocket::new(stack, &mut bufs.as_mut().0, &mut bufs.as_mut().1) },
+                state,
+                bufs,
+            })
+        }
+    }
+
+    impl<'d, const N: usize, const TX_SZ: usize, const RX_SZ: usize> Drop for TcpConnection<'d, N, TX_SZ, RX_SZ> {
+        fn drop(&mut self) {
+            unsafe {
+                self.socket.close();
+                self.state.pool.free(self.bufs);
+            }
+        }
+    }
+
+    impl<'d, const N: usize, const TX_SZ: usize, const RX_SZ: usize> embedded_io::Io
+        for TcpConnection<'d, N, TX_SZ, RX_SZ>
+    {
+        type Error = Error;
+    }
+
+    impl<'d, const N: usize, const TX_SZ: usize, const RX_SZ: usize> embedded_io::asynch::Read
+        for TcpConnection<'d, N, TX_SZ, RX_SZ>
+    {
+        type ReadFuture<'a> = impl Future<Output = Result<usize, Self::Error>>
+    where
+        Self: 'a;
+
+        fn read<'a>(&'a mut self, buf: &'a mut [u8]) -> Self::ReadFuture<'a> {
+            self.socket.read(buf)
+        }
+    }
+
+    impl<'d, const N: usize, const TX_SZ: usize, const RX_SZ: usize> embedded_io::asynch::Write
+        for TcpConnection<'d, N, TX_SZ, RX_SZ>
+    {
+        type WriteFuture<'a> = impl Future<Output = Result<usize, Self::Error>>
+    where
+        Self: 'a;
+
+        fn write<'a>(&'a mut self, buf: &'a [u8]) -> Self::WriteFuture<'a> {
+            self.socket.write(buf)
+        }
+
+        type FlushFuture<'a> = impl Future<Output = Result<(), Self::Error>>
+    where
+        Self: 'a;
+
+        fn flush<'a>(&'a mut self) -> Self::FlushFuture<'a> {
+            self.socket.flush()
+        }
+    }
+
+    /// State for TcpClient
+    pub struct TcpClientState<const N: usize, const TX_SZ: usize, const RX_SZ: usize> {
+        pool: Pool<([u8; TX_SZ], [u8; RX_SZ]), N>,
+    }
+
+    impl<const N: usize, const TX_SZ: usize, const RX_SZ: usize> TcpClientState<N, TX_SZ, RX_SZ> {
+        pub const fn new() -> Self {
+            Self { pool: Pool::new() }
+        }
+    }
+
+    unsafe impl<const N: usize, const TX_SZ: usize, const RX_SZ: usize> Sync for TcpClientState<N, TX_SZ, RX_SZ> {}
+
+    struct Pool<T, const N: usize> {
+        used: [AtomicBool; N],
+        data: [UnsafeCell<MaybeUninit<T>>; N],
+    }
+
+    impl<T, const N: usize> Pool<T, N> {
+        const VALUE: AtomicBool = AtomicBool::new(false);
+        const UNINIT: UnsafeCell<MaybeUninit<T>> = UnsafeCell::new(MaybeUninit::uninit());
+
+        const fn new() -> Self {
+            Self {
+                used: [Self::VALUE; N],
+                data: [Self::UNINIT; N],
+            }
+        }
+    }
+
+    impl<T, const N: usize> Pool<T, N> {
+        fn alloc(&self) -> Option<NonNull<T>> {
+            for n in 0..N {
+                if self.used[n].swap(true, Ordering::SeqCst) == false {
+                    let p = self.data[n].get() as *mut T;
+                    return Some(unsafe { NonNull::new_unchecked(p) });
+                }
+            }
+            None
+        }
+
+        /// safety: p must be a pointer obtained from self.alloc that hasn't been freed yet.
+        unsafe fn free(&self, p: NonNull<T>) {
+            let origin = self.data.as_ptr() as *mut T;
+            let n = p.as_ptr().offset_from(origin);
+            assert!(n >= 0);
+            assert!((n as usize) < N);
+            self.used[n as usize].store(false, Ordering::SeqCst);
+        }
+    }
+}
diff --git a/examples/stm32h7/Cargo.toml b/examples/stm32h7/Cargo.toml
index 8b1999b30..896046759 100644
--- a/examples/stm32h7/Cargo.toml
+++ b/examples/stm32h7/Cargo.toml
@@ -7,7 +7,7 @@ version = "0.1.0"
 embassy-util = { version = "0.1.0", path = "../../embassy-util", features = ["defmt"] }
 embassy-executor = { version = "0.1.0", path = "../../embassy-executor", features = ["defmt", "defmt-timestamp-uptime", "unstable-traits", "time-tick-32768hz"] }
 embassy-stm32 = { version = "0.1.0", path = "../../embassy-stm32", features = ["nightly", "defmt", "stm32h743bi", "net", "time-driver-any", "exti", "unstable-pac", "unstable-traits"] }
-embassy-net = { path = "../../embassy-net", features = ["defmt", "tcp", "dhcpv4", "medium-ethernet", "pool-16"] }
+embassy-net = { path = "../../embassy-net", features = ["defmt", "tcp", "dhcpv4", "medium-ethernet", "pool-16", "unstable-traits"] }
 embedded-io = { version = "0.3.0", features = ["async"] }
 
 defmt = "0.3"
@@ -18,6 +18,7 @@ cortex-m-rt = "0.7.0"
 embedded-hal = "0.2.6"
 embedded-hal-1 = { package = "embedded-hal", version = "1.0.0-alpha.8" }
 embedded-hal-async = { version = "0.1.0-alpha.1" }
+embedded-nal-async = "0.2.0"
 panic-probe = { version = "0.3", features = ["print-defmt"] }
 futures = { version = "0.3.17", default-features = false, features = ["async-await"] }
 heapless = { version = "0.7.5", default-features = false }
diff --git a/examples/stm32h7/src/bin/eth_client.rs b/examples/stm32h7/src/bin/eth_client.rs
new file mode 100644
index 000000000..a66c6f196
--- /dev/null
+++ b/examples/stm32h7/src/bin/eth_client.rs
@@ -0,0 +1,125 @@
+#![no_std]
+#![no_main]
+#![feature(type_alias_impl_trait)]
+
+use defmt::*;
+use embassy_executor::executor::Spawner;
+use embassy_executor::time::{Duration, Timer};
+use embassy_net::tcp::client::{TcpClient, TcpClientState};
+use embassy_net::{Stack, StackResources};
+use embassy_stm32::eth::generic_smi::GenericSMI;
+use embassy_stm32::eth::{Ethernet, State};
+use embassy_stm32::peripherals::ETH;
+use embassy_stm32::rng::Rng;
+use embassy_stm32::time::mhz;
+use embassy_stm32::{interrupt, Config, Peripherals};
+use embassy_util::Forever;
+use embedded_io::asynch::Write;
+use embedded_nal_async::{Ipv4Addr, SocketAddr, SocketAddrV4, TcpConnect};
+use rand_core::RngCore;
+use {defmt_rtt as _, panic_probe as _};
+
+macro_rules! forever {
+    ($val:expr) => {{
+        type T = impl Sized;
+        static FOREVER: Forever<T> = Forever::new();
+        FOREVER.put_with(move || $val)
+    }};
+}
+
+type Device = Ethernet<'static, ETH, GenericSMI, 4, 4>;
+
+#[embassy_executor::task]
+async fn net_task(stack: &'static Stack<Device>) -> ! {
+    stack.run().await
+}
+
+pub fn config() -> Config {
+    let mut config = Config::default();
+    config.rcc.sys_ck = Some(mhz(400));
+    config.rcc.hclk = Some(mhz(200));
+    config.rcc.pll1.q_ck = Some(mhz(100));
+    config
+}
+
+#[embassy_executor::main(config = "config()")]
+async fn main(spawner: Spawner, p: Peripherals) -> ! {
+    info!("Hello World!");
+
+    // Generate random seed.
+    let mut rng = Rng::new(p.RNG);
+    let mut seed = [0; 8];
+    rng.fill_bytes(&mut seed);
+    let seed = u64::from_le_bytes(seed);
+
+    let eth_int = interrupt::take!(ETH);
+    let mac_addr = [0x00, 0x00, 0xDE, 0xAD, 0xBE, 0xEF];
+
+    let device = unsafe {
+        Ethernet::new(
+            forever!(State::new()),
+            p.ETH,
+            eth_int,
+            p.PA1,
+            p.PA2,
+            p.PC1,
+            p.PA7,
+            p.PC4,
+            p.PC5,
+            p.PG13,
+            p.PB13,
+            p.PG11,
+            GenericSMI,
+            mac_addr,
+            0,
+        )
+    };
+
+    let config = embassy_net::ConfigStrategy::Dhcp;
+    //let config = embassy_net::ConfigStrategy::Static(embassy_net::Config {
+    //    address: Ipv4Cidr::new(Ipv4Address::new(10, 42, 0, 61), 24),
+    //    dns_servers: Vec::new(),
+    //    gateway: Some(Ipv4Address::new(10, 42, 0, 1)),
+    //});
+
+    // Init network stack
+    let stack = &*forever!(Stack::new(
+        device,
+        config,
+        forever!(StackResources::<1, 2, 8>::new()),
+        seed
+    ));
+
+    // Launch network task
+    unwrap!(spawner.spawn(net_task(&stack)));
+
+    info!("Network task initialized");
+
+    // To ensure DHCP configuration before trying connect
+    Timer::after(Duration::from_secs(20)).await;
+
+    static STATE: TcpClientState<1, 1024, 1024> = TcpClientState::new();
+    let client = TcpClient::new(&stack, &STATE);
+
+    loop {
+        let addr = SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(10, 42, 0, 1), 8000));
+
+        info!("connecting...");
+        let r = client.connect(addr).await;
+        if let Err(e) = r {
+            info!("connect error: {:?}", e);
+            Timer::after(Duration::from_secs(1)).await;
+            continue;
+        }
+        let mut connection = r.unwrap();
+        info!("connected!");
+        loop {
+            let r = connection.write_all(b"Hello\n").await;
+            if let Err(e) = r {
+                info!("write error: {:?}", e);
+                return;
+            }
+            Timer::after(Duration::from_secs(1)).await;
+        }
+    }
+}