WASI Preview 2: rewrite streams and pollable implementation

Cargo.toml

@@ -242,6 +242,9 @@ tempfile = "3.1.0"
 filecheck = "0.5.0"
 libc = "0.2.60"
 file-per-thread-logger = "0.2.0"
+tokio = { version = "1.26.0" }
+bytes = "1.4"
+futures = { version = "0.3.27", default-features = false }
 indexmap = "2.0.0"
 pretty_env_logger = "0.5.0"
 syn = "2.0.25"

crates/test-programs/reactor-tests/Cargo.toml

@@ -8,4 +8,4 @@ publish = false
 crate-type=["cdylib"]
 
 [dependencies]
-wit-bindgen = { workspace = true }
+wit-bindgen = { workspace = true, features = ["macros", "realloc"] }

crates/test-programs/tests/command.rs

@@ -1,19 +1,15 @@
 use anyhow::Result;
 use cap_std::{ambient_authority, fs::Dir, time::Duration};
-use std::{
-    io::{Cursor, Write},
-    sync::Mutex,
-};
+use std::{io::Write, sync::Mutex};
 use wasmtime::{
     component::{Component, Linker},
     Config, Engine, Store,
 };
 use wasmtime_wasi::preview2::{
-    clocks::{HostMonotonicClock, HostWallClock},
-    pipe::ReadPipe,
-    wasi::command::add_to_linker,
-    wasi::command::Command,
-    DirPerms, FilePerms, Table, WasiCtx, WasiCtxBuilder, WasiView,
+    command::{add_to_linker, Command},
+    pipe::MemoryInputPipe,
+    DirPerms, FilePerms, HostMonotonicClock, HostWallClock, Table, WasiCtx, WasiCtxBuilder,
+    WasiView,
 };
 
 lazy_static::lazy_static! {
@@ -63,7 +59,7 @@ async fn instantiate(
     Ok((store, command))
 }
 
-#[test_log::test(tokio::test)]
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
 async fn hello_stdout() -> Result<()> {
     let mut table = Table::new();
     let wasi = WasiCtxBuilder::new()
@@ -77,7 +73,7 @@ async fn hello_stdout() -> Result<()> {
         .map_err(|()| anyhow::anyhow!("command returned with failing exit status"))
 }
 
-#[test_log::test(tokio::test)]
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
 async fn panic() -> Result<()> {
     let mut table = Table::new();
     let wasi = WasiCtxBuilder::new()
@@ -100,7 +96,7 @@ async fn panic() -> Result<()> {
     Ok(())
 }
 
-#[test_log::test(tokio::test)]
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
 async fn args() -> Result<()> {
     let mut table = Table::new();
     let wasi = WasiCtxBuilder::new()
@@ -114,7 +110,7 @@ async fn args() -> Result<()> {
         .map_err(|()| anyhow::anyhow!("command returned with failing exit status"))
 }
 
-#[test_log::test(tokio::test)]
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
 async fn random() -> Result<()> {
     let mut table = Table::new();
     let wasi = WasiCtxBuilder::new().build(&mut table)?;
@@ -127,7 +123,7 @@ async fn random() -> Result<()> {
         .map_err(|()| anyhow::anyhow!("command returned with failing exit status"))
 }
 
-#[test_log::test(tokio::test)]
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
 async fn time() -> Result<()> {
     struct FakeWallClock;
 
@@ -173,13 +169,13 @@ async fn time() -> Result<()> {
         .map_err(|()| anyhow::anyhow!("command returned with failing exit status"))
 }
 
-#[test_log::test(tokio::test)]
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
 async fn stdin() -> Result<()> {
     let mut table = Table::new();
     let wasi = WasiCtxBuilder::new()
-        .set_stdin(ReadPipe::new(Cursor::new(
-            "So rested he by the Tumtum tree",
-        )))
+        .set_stdin(MemoryInputPipe::new(
+            "So rested he by the Tumtum tree".into(),
+        ))
         .build(&mut table)?;
 
     let (mut store, command) =
@@ -191,13 +187,13 @@ async fn stdin() -> Result<()> {
         .map_err(|()| anyhow::anyhow!("command returned with failing exit status"))
 }
 
-#[test_log::test(tokio::test)]
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
 async fn poll_stdin() -> Result<()> {
     let mut table = Table::new();
     let wasi = WasiCtxBuilder::new()
-        .set_stdin(ReadPipe::new(Cursor::new(
-            "So rested he by the Tumtum tree",
-        )))
+        .set_stdin(MemoryInputPipe::new(
+            "So rested he by the Tumtum tree".into(),
+        ))
         .build(&mut table)?;
 
     let (mut store, command) =
@@ -209,7 +205,7 @@ async fn poll_stdin() -> Result<()> {
         .map_err(|()| anyhow::anyhow!("command returned with failing exit status"))
 }
 
-#[test_log::test(tokio::test)]
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
 async fn env() -> Result<()> {
     let mut table = Table::new();
     let wasi = WasiCtxBuilder::new()
@@ -226,7 +222,7 @@ async fn env() -> Result<()> {
         .map_err(|()| anyhow::anyhow!("command returned with failing exit status"))
 }
 
-#[test_log::test(tokio::test)]
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
 async fn file_read() -> Result<()> {
     let dir = tempfile::tempdir()?;
 
@@ -248,7 +244,7 @@ async fn file_read() -> Result<()> {
         .map_err(|()| anyhow::anyhow!("command returned with failing exit status"))
 }
 
-#[test_log::test(tokio::test)]
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
 async fn file_append() -> Result<()> {
     let dir = tempfile::tempdir()?;
 
@@ -280,7 +276,7 @@ async fn file_append() -> Result<()> {
     Ok(())
 }
 
-#[test_log::test(tokio::test)]
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
 async fn file_dir_sync() -> Result<()> {
     let dir = tempfile::tempdir()?;
 
@@ -303,7 +299,7 @@ async fn file_dir_sync() -> Result<()> {
         .map_err(|()| anyhow::anyhow!("command returned with failing exit status"))
 }
 
-#[test_log::test(tokio::test)]
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
 async fn exit_success() -> Result<()> {
     let mut table = Table::new();
     let wasi = WasiCtxBuilder::new().build(&mut table)?;
@@ -320,7 +316,7 @@ async fn exit_success() -> Result<()> {
     Ok(())
 }
 
-#[test_log::test(tokio::test)]
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
 async fn exit_default() -> Result<()> {
     let mut table = Table::new();
     let wasi = WasiCtxBuilder::new().build(&mut table)?;
@@ -333,7 +329,7 @@ async fn exit_default() -> Result<()> {
     Ok(())
 }
 
-#[test_log::test(tokio::test)]
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
 async fn exit_failure() -> Result<()> {
     let mut table = Table::new();
     let wasi = WasiCtxBuilder::new().build(&mut table)?;
@@ -350,7 +346,7 @@ async fn exit_failure() -> Result<()> {
     Ok(())
 }
 
-#[test_log::test(tokio::test)]
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
 async fn exit_panic() -> Result<()> {
     let mut table = Table::new();
     let wasi = WasiCtxBuilder::new().build(&mut table)?;
@@ -367,7 +363,7 @@ async fn exit_panic() -> Result<()> {
     Ok(())
 }
 
-#[test_log::test(tokio::test)]
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
 async fn directory_list() -> Result<()> {
     let dir = tempfile::tempdir()?;
 
@@ -382,7 +378,8 @@ async fn directory_list() -> Result<()> {
 
     let mut table = Table::new();
     let wasi = WasiCtxBuilder::new()
-        .inherit_stdio()
+        .inherit_stdout()
+        .inherit_stderr()
         .push_preopened_dir(open_dir, DirPerms::all(), FilePerms::all(), "/")
         .build(&mut table)?;
 
@@ -395,7 +392,7 @@ async fn directory_list() -> Result<()> {
         .map_err(|()| anyhow::anyhow!("command returned with failing exit status"))
 }
 
-#[test_log::test(tokio::test)]
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
 async fn default_clocks() -> Result<()> {
     let mut table = Table::new();
     let wasi = WasiCtxBuilder::new().build(&mut table)?;
@@ -409,7 +406,7 @@ async fn default_clocks() -> Result<()> {
         .map_err(|()| anyhow::anyhow!("command returned with failing exit status"))
 }
 
-#[test_log::test(tokio::test)]
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
 async fn export_cabi_realloc() -> Result<()> {
     let mut table = Table::new();
     let wasi = WasiCtxBuilder::new().build(&mut table)?;
@@ -425,7 +422,7 @@ async fn export_cabi_realloc() -> Result<()> {
         .map_err(|()| anyhow::anyhow!("command returned with failing exit status"))
 }
 
-#[test_log::test(tokio::test)]
+#[test_log::test(tokio::test(flavor = "multi_thread"))]
 async fn read_only() -> Result<()> {
     let dir = tempfile::tempdir()?;