Merge #1471

1471: embassy-net: Make TcpSocket::abort() async r=Dirbaio a=mkj

This lets callers ensure that the reset packet is sent to the remote host. Otherwise there isn't a way to wait for the smoltcp stack to send the reset.

Requires changes to smoltcp to wake after the reset has been sent, see https://github.com/smoltcp-rs/smoltcp/compare/master...mkj:smoltcp:abort-wake

This commit also adds a "TCP accept" demo of the problem. Without the `.await` for abort() it gets dropped before the RST packet is emitted.

Co-authored-by: Matt Johnston <matt@ucc.asn.au>

embassy-net/src/tcp.rs

@@ -95,7 +95,8 @@ impl<'a> TcpWriter<'a> {
 
     /// Flushes the written data to the socket.
     ///
-    /// This waits until all data has been sent, and ACKed by the remote host.
+    /// 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
     }
@@ -198,7 +199,8 @@ impl<'a> TcpSocket<'a> {
 
     /// Flushes the written data to the socket.
     ///
-    /// This waits until all data has been sent, and ACKed by the remote host.
+    /// 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
     }
@@ -262,6 +264,11 @@ impl<'a> TcpSocket<'a> {
     ///
     /// This instantly closes both the read and write halves of the socket. Any pending data
     /// that has not been sent will be lost.
+    ///
+    /// Note that the TCP RST packet is not sent immediately - if the `TcpSocket` is dropped too soon
+    /// the remote host may not know the connection has been closed.
+    /// `abort()` callers should wait for a [`flush()`](TcpSocket::flush) call to complete before
+    /// dropping or reusing the socket.
     pub fn abort(&mut self) {
         self.io.with_mut(|s, _| s.abort())
     }
@@ -347,9 +354,10 @@ impl<'d> TcpIo<'d> {
     async fn flush(&mut self) -> Result<(), Error> {
         poll_fn(move |cx| {
             self.with_mut(|s, _| {
+                let waiting_close = s.state() == tcp::State::Closed && s.remote_endpoint().is_some();
                 // If there are outstanding send operations, register for wake up and wait
                 // smoltcp issues wake-ups when octets are dequeued from the send buffer
-                if s.send_queue() > 0 {
+                if s.send_queue() > 0 || waiting_close {
                     s.register_send_waker(cx.waker());
                     Poll::Pending
                 // No outstanding sends, socket is flushed

examples/std/src/bin/tcp_accept.rs

@@ -0,0 +1,133 @@
+#![feature(type_alias_impl_trait)]
+
+use core::fmt::Write as _;
+use std::default::Default;
+
+use clap::Parser;
+use embassy_executor::{Executor, Spawner};
+use embassy_net::tcp::TcpSocket;
+use embassy_net::{Config, Ipv4Address, Ipv4Cidr, Stack, StackResources};
+use embassy_time::{Duration, Timer};
+use embedded_io::asynch::Write as _;
+use heapless::Vec;
+use log::*;
+use rand_core::{OsRng, RngCore};
+use static_cell::StaticCell;
+
+#[path = "../tuntap.rs"]
+mod tuntap;
+
+use crate::tuntap::TunTapDevice;
+
+macro_rules! singleton {
+    ($val:expr) => {{
+        type T = impl Sized;
+        static STATIC_CELL: StaticCell<T> = StaticCell::new();
+        STATIC_CELL.init_with(move || $val)
+    }};
+}
+
+#[derive(Parser)]
+#[clap(version = "1.0")]
+struct Opts {
+    /// TAP device name
+    #[clap(long, default_value = "tap0")]
+    tap: String,
+    /// use a static IP instead of DHCP
+    #[clap(long)]
+    static_ip: bool,
+}
+
+#[embassy_executor::task]
+async fn net_task(stack: &'static Stack<TunTapDevice>) -> ! {
+    stack.run().await
+}
+
+#[derive(Default)]
+struct StrWrite(pub heapless::Vec<u8, 30>);
+
+impl core::fmt::Write for StrWrite {
+    fn write_str(&mut self, s: &str) -> Result<(), core::fmt::Error> {
+        self.0.extend_from_slice(s.as_bytes()).unwrap();
+        Ok(())
+    }
+}
+
+#[embassy_executor::task]
+async fn main_task(spawner: Spawner) {
+    let opts: Opts = Opts::parse();
+
+    // Init network device
+    let device = TunTapDevice::new(&opts.tap).unwrap();
+
+    // Choose between dhcp or static ip
+    let config = if opts.static_ip {
+        Config::Static(embassy_net::StaticConfig {
+            address: Ipv4Cidr::new(Ipv4Address::new(192, 168, 69, 2), 24),
+            dns_servers: Vec::new(),
+            gateway: Some(Ipv4Address::new(192, 168, 69, 1)),
+        })
+    } else {
+        Config::Dhcp(Default::default())
+    };
+
+    // Generate random seed
+    let mut seed = [0; 8];
+    OsRng.fill_bytes(&mut seed);
+    let seed = u64::from_le_bytes(seed);
+
+    // Init network stack
+    let stack = &*singleton!(Stack::new(device, config, singleton!(StackResources::<3>::new()), seed));
+
+    // Launch network task
+    spawner.spawn(net_task(stack)).unwrap();
+
+    // Then we can use it!
+    let mut rx_buffer = [0; 4096];
+    let mut tx_buffer = [0; 4096];
+
+    loop {
+        let mut socket = TcpSocket::new(stack, &mut rx_buffer, &mut tx_buffer);
+        socket.set_timeout(Some(Duration::from_secs(10)));
+        info!("Listening on TCP:9999...");
+        if let Err(_) = socket.accept(9999).await {
+            warn!("accept error");
+            continue;
+        }
+
+        info!("Accepted a connection");
+
+        // Write some quick output
+        for i in 1..=5 {
+            let mut w = StrWrite::default();
+            write!(w, "{}!  ", i).unwrap();
+            let r = socket.write_all(&w.0).await;
+            if let Err(e) = r {
+                warn!("write error: {:?}", e);
+                return;
+            }
+
+            Timer::after(Duration::from_millis(500)).await;
+        }
+        info!("Closing the connection");
+        socket.abort();
+        info!("Flushing the RST out...");
+        socket.flush().await;
+        info!("Finished with the socket");
+    }
+}
+
+static EXECUTOR: StaticCell<Executor> = StaticCell::new();
+
+fn main() {
+    env_logger::builder()
+        .filter_level(log::LevelFilter::Debug)
+        .filter_module("async_io", log::LevelFilter::Info)
+        .format_timestamp_nanos()
+        .init();
+
+    let executor = EXECUTOR.init(Executor::new());
+    executor.run(|spawner| {
+        spawner.spawn(main_task(spawner)).unwrap();
+    });
+}