feat: tcp-echo-server uses tasks, test runner uses wasmtime cli, tcp-listener supports v6

The tcp-echo-server example has been rewritten to spawn the echo part of
each accept into a task, which  means it can accept new connections
while other echos are in flight.

The tcp-echo-server test has been rewritten to test that connections can
be accepted while other echoes are in flight.

The tcp-echo-server test has been rewritten to use wasmtime cli as a
process, rather than use wasmtime as a crate. This drops wasmtime from
the dev-dependencies of the workspace, which is good because it was
running a quite out-of-date wasmtime.

Fix up the missing conversions to/from std::net::SocketAddr in
wstd::net::tcp_listener, so that we can send a Display impl of the
listening address from the guest to host, and parse it out in the host
(see get_listening_address)

Author: pchickey

Cargo.toml

@@ -84,9 +84,6 @@ test-programs = { path = "test-programs" }
 test-programs-artifacts = { path = "test-programs/artifacts" }
 ureq = { version = "2.12.1", default-features = false }
 wasi = "0.14.0"
-wasmtime = "26"
-wasmtime-wasi = "26"
-wasmtime-wasi-http = "26"
 wstd = { path = "." }
 wstd-macro = { path = "macro", version = "=0.5.4" }
 

examples/tcp_echo_server.rs

@@ -12,7 +12,10 @@ async fn main() -> io::Result<()> {
     while let Some(stream) = incoming.next().await {
         let stream = stream?;
         println!("Accepted from: {}", stream.peer_addr()?);
-        io::copy(&stream, &stream).await?;
+        wstd::runtime::spawn(async move {
+            let _ = io::copy(&stream, &stream).await;
+        })
+        .detach();
     }
     Ok(())
 }

src/net/tcp_listener.rs

@@ -33,15 +33,8 @@ impl TcpListener {
             wasi::sockets::tcp_create_socket::create_tcp_socket(family).map_err(to_io_err)?;
         let network = wasi::sockets::instance_network::instance_network();
 
-        let local_address = match addr {
-            SocketAddr::V4(addr) => {
-                let ip = addr.ip().octets();
-                let address = (ip[0], ip[1], ip[2], ip[3]);
-                let port = addr.port();
-                IpSocketAddress::Ipv4(Ipv4SocketAddress { port, address })
-            }
-            SocketAddr::V6(_) => todo!("IPv6 not yet supported in `wstd::net::TcpListener`"),
-        };
+        let local_address = sockaddr_to_wasi(addr);
+
         socket
             .start_bind(&network, local_address)
             .map_err(to_io_err)?;
@@ -56,10 +49,11 @@ impl TcpListener {
     }
 
     /// Returns the local socket address of this listener.
-    // TODO: make this return an actual socket addr
-    pub fn local_addr(&self) -> io::Result<String> {
-        let addr = self.socket.local_address().map_err(to_io_err)?;
-        Ok(format!("{addr:?}"))
+    pub fn local_addr(&self) -> io::Result<std::net::SocketAddr> {
+        self.socket
+            .local_address()
+            .map_err(to_io_err)
+            .map(sockaddr_from_wasi)
     }
 
     /// Returns an iterator over the connections being received on this listener.
@@ -105,3 +99,51 @@ pub(super) fn to_io_err(err: ErrorCode) -> io::Error {
         _ => ErrorKind::Other.into(),
     }
 }
+
+fn sockaddr_from_wasi(addr: IpSocketAddress) -> std::net::SocketAddr {
+    use wasi::sockets::network::Ipv6SocketAddress;
+    match addr {
+        IpSocketAddress::Ipv4(Ipv4SocketAddress { address, port }) => {
+            std::net::SocketAddr::V4(std::net::SocketAddrV4::new(
+                std::net::Ipv4Addr::new(address.0, address.1, address.2, address.3),
+                port,
+            ))
+        }
+        IpSocketAddress::Ipv6(Ipv6SocketAddress {
+            address,
+            port,
+            flow_info,
+            scope_id,
+        }) => std::net::SocketAddr::V6(std::net::SocketAddrV6::new(
+            std::net::Ipv6Addr::new(
+                address.0, address.1, address.2, address.3, address.4, address.5, address.6,
+                address.7,
+            ),
+            port,
+            flow_info,
+            scope_id,
+        )),
+    }
+}
+
+fn sockaddr_to_wasi(addr: std::net::SocketAddr) -> IpSocketAddress {
+    use wasi::sockets::network::Ipv6SocketAddress;
+    match addr {
+        std::net::SocketAddr::V4(addr) => {
+            let ip = addr.ip().octets();
+            IpSocketAddress::Ipv4(Ipv4SocketAddress {
+                address: (ip[0], ip[1], ip[2], ip[3]),
+                port: addr.port(),
+            })
+        }
+        std::net::SocketAddr::V6(addr) => {
+            let ip = addr.ip().segments();
+            IpSocketAddress::Ipv6(Ipv6SocketAddress {
+                address: (ip[0], ip[1], ip[2], ip[3], ip[4], ip[5], ip[6], ip[7]),
+                port: addr.port(),
+                flow_info: addr.flowinfo(),
+                scope_id: addr.scope_id(),
+            })
+        }
+    }
+}

src/runtime/mod.rs

@@ -13,6 +13,7 @@
 mod block_on;
 mod reactor;
 
+pub use ::async_task::Task;
 pub use block_on::block_on;
 pub use reactor::{AsyncPollable, Reactor, WaitFor};
 use std::cell::RefCell;
@@ -22,3 +23,14 @@ use std::cell::RefCell;
 std::thread_local! {
 pub(crate) static REACTOR: RefCell<Option<Reactor>> = const { RefCell::new(None) };
 }
+
+/// Spawn a `Future` as a `Task` on the current `Reactor`.
+///
+/// Panics if called from outside `block_on`.
+pub fn spawn<F, T>(fut: F) -> Task<T>
+where
+    F: std::future::Future<Output = T> + 'static,
+    T: 'static,
+{
+    Reactor::current().spawn(fut)
+}

test-programs/artifacts/Cargo.toml

@@ -13,9 +13,6 @@ anyhow.workspace = true
 test-log.workspace = true
 test-programs-artifacts.workspace = true
 ureq.workspace = true
-wasmtime.workspace = true
-wasmtime-wasi.workspace = true
-wasmtime-wasi-http.workspace = true
 
 [build-dependencies]
 cargo_metadata.workspace = true

test-programs/artifacts/tests/tcp_echo_server.rs

@@ -1,123 +1,112 @@
-use anyhow::{anyhow, Context, Result};
-use wasmtime::{
-    component::{Component, Linker, ResourceTable},
-    Config, Engine, Store,
-};
-use wasmtime_wasi::{pipe::MemoryOutputPipe, WasiCtx, WasiView};
-use wasmtime_wasi_http::{WasiHttpCtx, WasiHttpView};
-
-struct Ctx {
-    table: ResourceTable,
-    wasi: WasiCtx,
-    http: WasiHttpCtx,
-}
-
-impl WasiView for Ctx {
-    fn table(&mut self) -> &mut ResourceTable {
-        &mut self.table
-    }
-    fn ctx(&mut self) -> &mut WasiCtx {
-        &mut self.wasi
-    }
-}
-
-impl WasiHttpView for Ctx {
-    fn table(&mut self) -> &mut ResourceTable {
-        &mut self.table
-    }
-    fn ctx(&mut self) -> &mut WasiHttpCtx {
-        &mut self.http
-    }
-}
-
-fn run_in_wasmtime(wasm: &[u8], stdout: Option<MemoryOutputPipe>) -> Result<()> {
-    let config = Config::default();
-    let engine = Engine::new(&config).context("creating engine")?;
-    let component = Component::new(&engine, wasm).context("loading component")?;
-
-    let mut linker: Linker<Ctx> = Linker::new(&engine);
-    wasmtime_wasi::add_to_linker_sync(&mut linker).context("add wasi to linker")?;
-    wasmtime_wasi_http::add_only_http_to_linker_sync(&mut linker)
-        .context("add wasi-http to linker")?;
-
-    let mut builder = WasiCtx::builder();
-    builder.inherit_stderr().inherit_network();
-    let wasi = match stdout {
-        Some(stdout) => builder.stdout(stdout).build(),
-        None => builder.inherit_stdout().build(),
-    };
-    let mut store = Store::new(
-        &engine,
-        Ctx {
-            table: ResourceTable::new(),
-            wasi,
-            http: WasiHttpCtx::new(),
-        },
-    );
-
-    let instance = linker.instantiate(&mut store, &component)?;
-    let run_interface = instance
-        .get_export(&mut store, None, "wasi:cli/run@0.2.0")
-        .ok_or_else(|| anyhow!("wasi:cli/run missing?"))?;
-    let run_func_export = instance
-        .get_export(&mut store, Some(&run_interface), "run")
-        .ok_or_else(|| anyhow!("run export missing?"))?;
-    let run_func = instance
-        .get_typed_func::<(), (Result<(), ()>,)>(&mut store, &run_func_export)
-        .context("run as typed func")?;
-
-    println!("entering wasm...");
-    let (runtime_result,) = run_func.call(&mut store, ())?;
-    runtime_result.map_err(|()| anyhow!("run returned an error"))?;
-    println!("done");
-
-    Ok(())
-}
+use anyhow::{Context, Result};
+use std::process::Command;
 
 #[test_log::test]
 fn tcp_echo_server() -> Result<()> {
     use std::io::{Read, Write};
     use std::net::{Shutdown, TcpStream};
-    use std::thread::sleep;
-    use std::time::Duration;
 
     println!("testing {}", test_programs_artifacts::TCP_ECHO_SERVER);
-    let wasm = std::fs::read(test_programs_artifacts::TCP_ECHO_SERVER).context("read wasm")?;
 
-    let pipe = wasmtime_wasi::pipe::MemoryOutputPipe::new(1024 * 1024);
-    let write_end = pipe.clone();
-    let wasmtime_thread = std::thread::spawn(move || run_in_wasmtime(&wasm, Some(write_end)));
+    // Run the component in wasmtime
+    // -Sinherit-network allows it to accept network connections
+    let mut wasmtime_process = Command::new("wasmtime")
+        .arg("run")
+        .arg("-Sinherit-network")
+        .arg(test_programs_artifacts::TCP_ECHO_SERVER)
+        .stdout(std::process::Stdio::piped())
+        .spawn()?;
 
-    'wait: loop {
-        sleep(Duration::from_millis(100));
-        for line in pipe.contents().split(|c| *c == b'\n') {
-            if line.starts_with(b"Listening on") {
-                break 'wait;
-            }
-        }
-    }
+    let addr = get_listening_address(wasmtime_process.stdout.take().expect("stdout is piped"))?;
 
-    let mut tcpstream =
-        TcpStream::connect("127.0.0.1:8080").context("connect to wasm echo server")?;
-    println!("connected to wasm echo server");
+    println!("tcp echo server is listening on {addr:?}");
 
-    const MESSAGE: &[u8] = b"hello, echoserver!\n";
+    let mut stream1 = TcpStream::connect(&addr).context("connect stream1")?;
+    println!("stream1 connected");
 
-    tcpstream.write_all(MESSAGE).context("write to socket")?;
-    println!("wrote to echo server");
+    let mut stream2 = TcpStream::connect(&addr).context("connect stream2")?;
+    println!("stream3 connected");
 
-    tcpstream.shutdown(Shutdown::Write)?;
+    const MESSAGE1: &[u8] = b"hello, echoserver!\n";
 
-    let mut readback = Vec::new();
-    tcpstream
-        .read_to_end(&mut readback)
-        .context("read from socket")?;
+    stream1.write_all(MESSAGE1).context("write to stream1")?;
+    println!("stream1 wrote to echo server");
 
-    println!("read from wasm server");
-    assert_eq!(MESSAGE, readback);
+    let mut stream3 = TcpStream::connect(&addr).context("connect stream3")?;
+    println!("stream3 connected");
+
+    const MESSAGE2: &[u8] = b"hello, gussie!\n";
+    stream2.write_all(MESSAGE2).context("write to stream1")?;
+    println!("stream2 wrote to echo server");
+
+    stream1.shutdown(Shutdown::Write)?;
+    stream2.shutdown(Shutdown::Write)?;
+
+    let mut readback2 = Vec::new();
+    stream2
+        .read_to_end(&mut readback2)
+        .context("read from stream2")?;
+    println!("read from stream2");
+
+    let mut readback1 = Vec::new();
+    stream1
+        .read_to_end(&mut readback1)
+        .context("read from stream1")?;
+    println!("read from stream1");
+
+    assert_eq!(MESSAGE1, readback1, "readback of stream 1");
+    assert_eq!(MESSAGE2, readback2, "readback of stream 2");
+
+    const MESSAGE3: &[u8] = b"hello, willa!\n";
+    stream3.write_all(MESSAGE3).context("write to stream1")?;
+    println!("stream3 wrote to echo server");
+    stream3.shutdown(Shutdown::Write)?;
+
+    let mut readback3 = Vec::new();
+    stream3
+        .read_to_end(&mut readback3)
+        .context("read from stream3")?;
+    println!("read from stream3");
+    assert_eq!(MESSAGE3, readback3, "readback of stream 3");
+
+    wasmtime_process.kill()?;
 
-    if wasmtime_thread.is_finished() {
-        wasmtime_thread.join().expect("wasmtime panicked")?;
-    }
     Ok(())
 }
+
+fn get_listening_address(
+    mut wasmtime_stdout: std::process::ChildStdout,
+) -> Result<std::net::SocketAddr> {
+    use std::io::Read;
+    use std::thread::sleep;
+    use std::time::Duration;
+
+    // Gather complete contents of stdout here
+    let mut stdout_contents = String::new();
+    loop {
+        // Wait for process to print
+        sleep(Duration::from_millis(100));
+
+        // Read more that the process printed, append to contents
+        let mut buf = vec![0; 4096];
+        let len = wasmtime_stdout
+            .read(&mut buf)
+            .context("reading wasmtime stdout")?;
+        buf.truncate(len);
+        stdout_contents
+            .push_str(std::str::from_utf8(&buf).context("wasmtime stdout should be string")?);
+
+        // Parse out the line where guest program says where it is listening
+        for line in stdout_contents.lines() {
+            if let Some(rest) = line.strip_prefix("Listening on ") {
+                // Forget wasmtime_stdout, rather than drop it, so that any
+                // subsequent stdout from wasmtime doesn't panic on a broken
+                // pipe.
+                std::mem::forget(wasmtime_stdout);
+                return rest
+                    .parse()
+                    .with_context(|| format!("parsing socket addr from line: {line:?}"));
+            }
+        }
+    }
+}