wasi-threads: an initial implementation (#5484)

This commit includes a set of changes that add initial support for `wasi-threads` to Wasmtime:

* feat: remove mutability from the WasiCtx Table

This patch adds interior mutability to the WasiCtx Table and the Table elements.

Major pain points:
* `File` only needs `RwLock<cap_std::fs::File>` to implement
  `File::set_fdflags()` on Windows, because of [1]
* Because `File` needs a `RwLock` and `RwLock*Guard` cannot
  be hold across an `.await`, The `async` from
  `async fn num_ready_bytes(&self)` had to be removed
* Because `File` needs a `RwLock` and `RwLock*Guard` cannot
  be dereferenced in `pollable`, the signature of
  `fn pollable(&self) -> Option<rustix::fd::BorrowedFd>`
  changed to `fn pollable(&self) -> Option<Arc<dyn AsFd + '_>>`

[1] da238e324e/src/fs/fd_flags.rs (L210-L217)

* wasi-threads: add an initial implementation

This change is a first step toward implementing `wasi-threads` in
Wasmtime. We may find that it has some missing pieces, but the core
functionality is there: when `wasi::thread_spawn` is called by a running
WebAssembly module, a function named `wasi_thread_start` is found in the
module's exports and called in a new instance. The shared memory of the
original instance is reused in the new instance.

This new WASI proposal is in its early stages and details are still
being hashed out in the [spec] and [wasi-libc] repositories. Due to its
experimental state, the `wasi-threads` functionality is hidden behind
both a compile-time and runtime flag: one must build with `--features
wasi-threads` but also run the Wasmtime CLI with `--wasm-features
threads` and `--wasi-modules experimental-wasi-threads`. One can
experiment with `wasi-threads` by running:

```console
$ cargo run --features wasi-threads -- \
    --wasm-features threads --wasi-modules experimental-wasi-threads \
    <a threads-enabled module>
```

Threads-enabled Wasm modules are not yet easy to build. Hopefully this
is resolved soon, but in the meantime see the use of
`THREAD_MODEL=posix` in the [wasi-libc] repository for some clues on
what is necessary. Wiggle complicates things by requiring the Wasm
memory to be exported with a certain name and `wasi-threads` also
expects that memory to be imported; this build-time obstacle can be
overcome with the `--import-memory --export-memory` flags only available
in the latest Clang tree. Due to all of this, the included tests are
written directly in WAT--run these with:

```console
$ cargo test --features wasi-threads -p wasmtime-cli -- cli_tests
```

[spec]: https://github.com/WebAssembly/wasi-threads
[wasi-libc]: https://github.com/WebAssembly/wasi-libc

This change does not protect the WASI implementations themselves from
concurrent access. This is already complete in previous commits or left
for future commits in certain cases (e.g., wasi-nn).

* wasi-threads: factor out process exit logic

As is being discussed [elsewhere], either calling `proc_exit` or
trapping in any thread should halt execution of all threads. The
Wasmtime CLI already has logic for adapting a WebAssembly error code to
a code expected in each OS. This change factors out this logic to a new
function, `maybe_exit_on_error`, for use within the `wasi-threads`
implementation.

This will work reasonably well for CLI users of Wasmtime +
`wasi-threads`, but embedders will want something better in the future:
when a `wasi-threads` threads fails, they may not want their application
to exit. Handling this is tricky, because it will require cancelling the
threads spawned by the `wasi-threads` implementation, something that is
not trivial to do in Rust. With this change, we defer that work until
later in order to provide a working implementation of `wasi-threads` for
experimentation.

[elsewhere]: https://github.com/WebAssembly/wasi-threads/pull/17

* review: work around `fd_fdstat_set_flags`

In order to make progress with wasi-threads, this change temporarily
works around limitations induced by `wasi-common`'s
`fd_fdstat_set_flags` to allow `&mut self` use in the implementation.
Eventual resolution is tracked in
https://github.com/bytecodealliance/wasmtime/issues/5643. This change
makes several related helper functions (e.g., `set_fdflags`) take `&mut
self` as well.

* test: use `wait`/`notify` to improve `threads.wat` test

Previously, the test simply executed in a loop for some hardcoded number
of iterations. This changes uses `wait` and `notify` and atomic
operations to keep track of when the spawned threads are done and join
on the main thread appropriately.

* various fixes and tweaks due to the PR review

---------

Signed-off-by: Harald Hoyer <harald@profian.com>
Co-authored-by: Harald Hoyer <harald@profian.com>
Co-authored-by: Alex Crichton <alex@alexcrichton.com>

crates/wasi-common/cap-std-sync/src/file.rs

@@ -31,24 +31,23 @@ impl WasiFile for File {
     fn pollable(&self) -> Option<rustix::fd::BorrowedFd> {
         Some(self.0.as_fd())
     }
-
     #[cfg(windows)]
     fn pollable(&self) -> Option<io_extras::os::windows::RawHandleOrSocket> {
         Some(self.0.as_raw_handle_or_socket())
     }
-    async fn datasync(&mut self) -> Result<(), Error> {
+    async fn datasync(&self) -> Result<(), Error> {
         self.0.sync_data()?;
         Ok(())
     }
-    async fn sync(&mut self) -> Result<(), Error> {
+    async fn sync(&self) -> Result<(), Error> {
         self.0.sync_all()?;
         Ok(())
     }
-    async fn get_filetype(&mut self) -> Result<FileType, Error> {
+    async fn get_filetype(&self) -> Result<FileType, Error> {
         let meta = self.0.metadata()?;
         Ok(filetype_from(&meta.file_type()))
     }
-    async fn get_fdflags(&mut self) -> Result<FdFlags, Error> {
+    async fn get_fdflags(&self) -> Result<FdFlags, Error> {
         let fdflags = get_fd_flags(&self.0)?;
         Ok(fdflags)
     }
@@ -64,7 +63,7 @@ impl WasiFile for File {
         self.0.set_fd_flags(set_fd_flags)?;
         Ok(())
     }
-    async fn get_filestat(&mut self) -> Result<Filestat, Error> {
+    async fn get_filestat(&self) -> Result<Filestat, Error> {
         let meta = self.0.metadata()?;
         Ok(Filestat {
             device_id: meta.dev(),
@@ -77,20 +76,20 @@ impl WasiFile for File {
             ctim: meta.created().map(|t| Some(t.into_std())).unwrap_or(None),
         })
     }
-    async fn set_filestat_size(&mut self, size: u64) -> Result<(), Error> {
+    async fn set_filestat_size(&self, size: u64) -> Result<(), Error> {
         self.0.set_len(size)?;
         Ok(())
     }
-    async fn advise(&mut self, offset: u64, len: u64, advice: Advice) -> Result<(), Error> {
+    async fn advise(&self, offset: u64, len: u64, advice: Advice) -> Result<(), Error> {
         self.0.advise(offset, len, convert_advice(advice))?;
         Ok(())
     }
-    async fn allocate(&mut self, offset: u64, len: u64) -> Result<(), Error> {
+    async fn allocate(&self, offset: u64, len: u64) -> Result<(), Error> {
         self.0.allocate(offset, len)?;
         Ok(())
     }
     async fn set_times(
-        &mut self,
+        &self,
         atime: Option<wasi_common::SystemTimeSpec>,
         mtime: Option<wasi_common::SystemTimeSpec>,
     ) -> Result<(), Error> {
@@ -98,41 +97,41 @@ impl WasiFile for File {
             .set_times(convert_systimespec(atime), convert_systimespec(mtime))?;
         Ok(())
     }
-    async fn read_vectored<'a>(&mut self, bufs: &mut [io::IoSliceMut<'a>]) -> Result<u64, Error> {
+    async fn read_vectored<'a>(&self, bufs: &mut [io::IoSliceMut<'a>]) -> Result<u64, Error> {
         let n = self.0.read_vectored(bufs)?;
         Ok(n.try_into()?)
     }
     async fn read_vectored_at<'a>(
-        &mut self,
+        &self,
         bufs: &mut [io::IoSliceMut<'a>],
         offset: u64,
     ) -> Result<u64, Error> {
         let n = self.0.read_vectored_at(bufs, offset)?;
         Ok(n.try_into()?)
     }
-    async fn write_vectored<'a>(&mut self, bufs: &[io::IoSlice<'a>]) -> Result<u64, Error> {
+    async fn write_vectored<'a>(&self, bufs: &[io::IoSlice<'a>]) -> Result<u64, Error> {
         let n = self.0.write_vectored(bufs)?;
         Ok(n.try_into()?)
     }
     async fn write_vectored_at<'a>(
-        &mut self,
+        &self,
         bufs: &[io::IoSlice<'a>],
         offset: u64,
     ) -> Result<u64, Error> {
         let n = self.0.write_vectored_at(bufs, offset)?;
         Ok(n.try_into()?)
     }
-    async fn seek(&mut self, pos: std::io::SeekFrom) -> Result<u64, Error> {
+    async fn seek(&self, pos: std::io::SeekFrom) -> Result<u64, Error> {
         Ok(self.0.seek(pos)?)
     }
-    async fn peek(&mut self, buf: &mut [u8]) -> Result<u64, Error> {
+    async fn peek(&self, buf: &mut [u8]) -> Result<u64, Error> {
         let n = self.0.peek(buf)?;
         Ok(n.try_into()?)
     }
-    async fn num_ready_bytes(&self) -> Result<u64, Error> {
+    fn num_ready_bytes(&self) -> Result<u64, Error> {
         Ok(self.0.num_ready_bytes()?)
     }
-    fn isatty(&mut self) -> bool {
+    fn isatty(&self) -> bool {
         self.0.is_terminal()
     }
 }

crates/wasi-common/cap-std-sync/src/lib.rs

@@ -94,15 +94,15 @@ impl WasiCtxBuilder {
         }
         Ok(self)
     }
-    pub fn stdin(mut self, f: Box<dyn WasiFile>) -> Self {
+    pub fn stdin(self, f: Box<dyn WasiFile>) -> Self {
         self.0.set_stdin(f);
         self
     }
-    pub fn stdout(mut self, f: Box<dyn WasiFile>) -> Self {
+    pub fn stdout(self, f: Box<dyn WasiFile>) -> Self {
         self.0.set_stdout(f);
         self
     }
-    pub fn stderr(mut self, f: Box<dyn WasiFile>) -> Self {
+    pub fn stderr(self, f: Box<dyn WasiFile>) -> Self {
         self.0.set_stderr(f);
         self
     }
@@ -118,12 +118,12 @@ impl WasiCtxBuilder {
     pub fn inherit_stdio(self) -> Self {
         self.inherit_stdin().inherit_stdout().inherit_stderr()
     }
-    pub fn preopened_dir(mut self, dir: Dir, guest_path: impl AsRef<Path>) -> Result<Self, Error> {
+    pub fn preopened_dir(self, dir: Dir, guest_path: impl AsRef<Path>) -> Result<Self, Error> {
         let dir = Box::new(crate::dir::Dir::from_cap_std(dir));
         self.0.push_preopened_dir(dir, guest_path)?;
         Ok(self)
     }
-    pub fn preopened_socket(mut self, fd: u32, socket: impl Into<Socket>) -> Result<Self, Error> {
+    pub fn preopened_socket(self, fd: u32, socket: impl Into<Socket>) -> Result<Self, Error> {
         let socket: Socket = socket.into();
         let file: Box<dyn WasiFile> = socket.into();
 

crates/wasi-common/cap-std-sync/src/net.rs

@@ -86,22 +86,21 @@ macro_rules! wasi_listen_write_impl {
             fn pollable(&self) -> Option<rustix::fd::BorrowedFd> {
                 Some(self.0.as_fd())
             }
-
             #[cfg(windows)]
             fn pollable(&self) -> Option<io_extras::os::windows::RawHandleOrSocket> {
                 Some(self.0.as_raw_handle_or_socket())
             }
-            async fn sock_accept(&mut self, fdflags: FdFlags) -> Result<Box<dyn WasiFile>, Error> {
+            async fn sock_accept(&self, fdflags: FdFlags) -> Result<Box<dyn WasiFile>, Error> {
                 let (stream, _) = self.0.accept()?;
                 let mut stream = <$stream>::from_cap_std(stream);
                 stream.set_fdflags(fdflags).await?;
                 Ok(Box::new(stream))
             }
-            async fn get_filetype(&mut self) -> Result<FileType, Error> {
+            async fn get_filetype(&self) -> Result<FileType, Error> {
                 Ok(FileType::SocketStream)
             }
             #[cfg(unix)]
-            async fn get_fdflags(&mut self) -> Result<FdFlags, Error> {
+            async fn get_fdflags(&self) -> Result<FdFlags, Error> {
                 let fdflags = get_fd_flags(&self.0)?;
                 Ok(fdflags)
             }
@@ -117,7 +116,7 @@ macro_rules! wasi_listen_write_impl {
                 }
                 Ok(())
             }
-            async fn num_ready_bytes(&self) -> Result<u64, Error> {
+            fn num_ready_bytes(&self) -> Result<u64, Error> {
                 Ok(1)
             }
         }
@@ -180,16 +179,15 @@ macro_rules! wasi_stream_write_impl {
             fn pollable(&self) -> Option<rustix::fd::BorrowedFd> {
                 Some(self.0.as_fd())
             }
-
             #[cfg(windows)]
             fn pollable(&self) -> Option<io_extras::os::windows::RawHandleOrSocket> {
                 Some(self.0.as_raw_handle_or_socket())
             }
-            async fn get_filetype(&mut self) -> Result<FileType, Error> {
+            async fn get_filetype(&self) -> Result<FileType, Error> {
                 Ok(FileType::SocketStream)
             }
             #[cfg(unix)]
-            async fn get_fdflags(&mut self) -> Result<FdFlags, Error> {
+            async fn get_fdflags(&self) -> Result<FdFlags, Error> {
                 let fdflags = get_fd_flags(&self.0)?;
                 Ok(fdflags)
             }
@@ -206,23 +204,23 @@ macro_rules! wasi_stream_write_impl {
                 Ok(())
             }
             async fn read_vectored<'a>(
-                &mut self,
+                &self,
                 bufs: &mut [io::IoSliceMut<'a>],
             ) -> Result<u64, Error> {
                 use std::io::Read;
                 let n = Read::read_vectored(&mut &*self.as_socketlike_view::<$std_ty>(), bufs)?;
                 Ok(n.try_into()?)
             }
-            async fn write_vectored<'a>(&mut self, bufs: &[io::IoSlice<'a>]) -> Result<u64, Error> {
+            async fn write_vectored<'a>(&self, bufs: &[io::IoSlice<'a>]) -> Result<u64, Error> {
                 use std::io::Write;
                 let n = Write::write_vectored(&mut &*self.as_socketlike_view::<$std_ty>(), bufs)?;
                 Ok(n.try_into()?)
             }
-            async fn peek(&mut self, buf: &mut [u8]) -> Result<u64, Error> {
+            async fn peek(&self, buf: &mut [u8]) -> Result<u64, Error> {
                 let n = self.0.peek(buf)?;
                 Ok(n.try_into()?)
             }
-            async fn num_ready_bytes(&self) -> Result<u64, Error> {
+            fn num_ready_bytes(&self) -> Result<u64, Error> {
                 let val = self.as_socketlike_view::<$std_ty>().num_ready_bytes()?;
                 Ok(val)
             }
@@ -244,7 +242,7 @@ macro_rules! wasi_stream_write_impl {
             }
 
             async fn sock_recv<'a>(
-                &mut self,
+                &self,
                 ri_data: &mut [std::io::IoSliceMut<'a>],
                 ri_flags: RiFlags,
             ) -> Result<(u64, RoFlags), Error> {
@@ -272,7 +270,7 @@ macro_rules! wasi_stream_write_impl {
             }
 
             async fn sock_send<'a>(
-                &mut self,
+                &self,
                 si_data: &[std::io::IoSlice<'a>],
                 si_flags: SiFlags,
             ) -> Result<u64, Error> {
@@ -284,7 +282,7 @@ macro_rules! wasi_stream_write_impl {
                 Ok(n as u64)
             }
 
-            async fn sock_shutdown(&mut self, how: SdFlags) -> Result<(), Error> {
+            async fn sock_shutdown(&self, how: SdFlags) -> Result<(), Error> {
                 let how = if how == SdFlags::RD | SdFlags::WR {
                     cap_std::net::Shutdown::Both
                 } else if how == SdFlags::RD {

crates/wasi-common/cap-std-sync/src/sched/unix.rs

@@ -55,7 +55,7 @@ pub async fn poll_oneoff<'a>(poll: &mut Poll<'a>) -> Result<(), Error> {
             let revents = pollfd.revents();
             let (nbytes, rwsub) = match rwsub {
                 Subscription::Read(sub) => {
-                    let ready = sub.file.num_ready_bytes().await?;
+                    let ready = sub.file.num_ready_bytes()?;
                     (std::cmp::max(ready, 1), sub)
                 }
                 Subscription::Write(sub) => (0, sub),

crates/wasi-common/cap-std-sync/src/sched/windows.rs

@@ -96,7 +96,7 @@ pub async fn poll_oneoff_<'a>(
         }
     }
     for r in immediate_reads {
-        match r.file.num_ready_bytes().await {
+        match r.file.num_ready_bytes() {
             Ok(ready_bytes) => {
                 r.complete(ready_bytes, RwEventFlags::empty());
                 ready = true;

crates/wasi-common/cap-std-sync/src/stdio.rs

@@ -31,6 +31,7 @@ impl WasiFile for Stdin {
     fn as_any(&self) -> &dyn Any {
         self
     }
+
     #[cfg(unix)]
     fn pollable(&self) -> Option<rustix::fd::BorrowedFd> {
         Some(self.0.as_fd())
@@ -40,32 +41,33 @@ impl WasiFile for Stdin {
     fn pollable(&self) -> Option<io_extras::os::windows::RawHandleOrSocket> {
         Some(self.0.as_raw_handle_or_socket())
     }
-    async fn get_filetype(&mut self) -> Result<FileType, Error> {
+
+    async fn get_filetype(&self) -> Result<FileType, Error> {
         if self.isatty() {
             Ok(FileType::CharacterDevice)
         } else {
             Ok(FileType::Unknown)
         }
     }
-    async fn read_vectored<'a>(&mut self, bufs: &mut [io::IoSliceMut<'a>]) -> Result<u64, Error> {
+    async fn read_vectored<'a>(&self, bufs: &mut [io::IoSliceMut<'a>]) -> Result<u64, Error> {
         let n = (&*self.0.as_filelike_view::<File>()).read_vectored(bufs)?;
         Ok(n.try_into().map_err(|_| Error::range())?)
     }
     async fn read_vectored_at<'a>(
-        &mut self,
+        &self,
         _bufs: &mut [io::IoSliceMut<'a>],
         _offset: u64,
     ) -> Result<u64, Error> {
         Err(Error::seek_pipe())
     }
-    async fn seek(&mut self, _pos: std::io::SeekFrom) -> Result<u64, Error> {
+    async fn seek(&self, _pos: std::io::SeekFrom) -> Result<u64, Error> {
         Err(Error::seek_pipe())
     }
-    async fn peek(&mut self, _buf: &mut [u8]) -> Result<u64, Error> {
+    async fn peek(&self, _buf: &mut [u8]) -> Result<u64, Error> {
         Err(Error::seek_pipe())
     }
     async fn set_times(
-        &mut self,
+        &self,
         atime: Option<wasi_common::SystemTimeSpec>,
         mtime: Option<wasi_common::SystemTimeSpec>,
     ) -> Result<(), Error> {
@@ -73,10 +75,10 @@ impl WasiFile for Stdin {
             .set_times(convert_systimespec(atime), convert_systimespec(mtime))?;
         Ok(())
     }
-    async fn num_ready_bytes(&self) -> Result<u64, Error> {
+    fn num_ready_bytes(&self) -> Result<u64, Error> {
         Ok(self.0.num_ready_bytes()?)
     }
-    fn isatty(&mut self) -> bool {
+    fn isatty(&self) -> bool {
         self.0.is_terminal()
     }
 }
@@ -111,39 +113,38 @@ macro_rules! wasi_file_write_impl {
             fn pollable(&self) -> Option<rustix::fd::BorrowedFd> {
                 Some(self.0.as_fd())
             }
-
             #[cfg(windows)]
             fn pollable(&self) -> Option<io_extras::os::windows::RawHandleOrSocket> {
                 Some(self.0.as_raw_handle_or_socket())
             }
-            async fn get_filetype(&mut self) -> Result<FileType, Error> {
+            async fn get_filetype(&self) -> Result<FileType, Error> {
                 if self.isatty() {
                     Ok(FileType::CharacterDevice)
                 } else {
                     Ok(FileType::Unknown)
                 }
             }
-            async fn get_fdflags(&mut self) -> Result<FdFlags, Error> {
+            async fn get_fdflags(&self) -> Result<FdFlags, Error> {
                 Ok(FdFlags::APPEND)
             }
-            async fn write_vectored<'a>(&mut self, bufs: &[io::IoSlice<'a>]) -> Result<u64, Error> {
+            async fn write_vectored<'a>(&self, bufs: &[io::IoSlice<'a>]) -> Result<u64, Error> {
                 let n = (&*self.0.as_filelike_view::<File>()).write_vectored(bufs)?;
                 Ok(n.try_into().map_err(|_| {
                     Error::range().context("converting write_vectored total length")
                 })?)
             }
             async fn write_vectored_at<'a>(
-                &mut self,
+                &self,
                 _bufs: &[io::IoSlice<'a>],
                 _offset: u64,
             ) -> Result<u64, Error> {
                 Err(Error::seek_pipe())
             }
-            async fn seek(&mut self, _pos: std::io::SeekFrom) -> Result<u64, Error> {
+            async fn seek(&self, _pos: std::io::SeekFrom) -> Result<u64, Error> {
                 Err(Error::seek_pipe())
             }
             async fn set_times(
-                &mut self,
+                &self,
                 atime: Option<wasi_common::SystemTimeSpec>,
                 mtime: Option<wasi_common::SystemTimeSpec>,
             ) -> Result<(), Error> {
@@ -151,7 +152,7 @@ macro_rules! wasi_file_write_impl {
                     .set_times(convert_systimespec(atime), convert_systimespec(mtime))?;
                 Ok(())
             }
-            fn isatty(&mut self) -> bool {
+            fn isatty(&self) -> bool {
                 self.0.is_terminal()
             }
         }