Asynchronous threading phase2

lib/cli/Cargo.toml

@@ -72,6 +72,7 @@ hex = "0.4.3"
 flate2 = "1.0.25"
 cargo_metadata = "0.15.2"
 tar = "0.4.38"
+bytes = "1"
 thiserror = "1.0.37"
 log = "0.4.17"
 semver = "1.0.14"

lib/cli/src/commands/run.rs

@@ -334,7 +334,7 @@ impl RunWithPathBuf {
                         .instantiate(&mut store, &module, program_name, self.args.clone())
                         .with_context(|| "failed to instantiate WASI module")?;
 
-                    let capable_of_deep_sleep = ctx.data(&store).capable_of_deep_sleep();
+                    let capable_of_deep_sleep = unsafe { ctx.data(&store).capable_of_deep_sleep() };
                     ctx.data_mut(&mut store)
                         .enable_deep_sleep = capable_of_deep_sleep;
 

lib/cli/src/commands/run/wasi.rs

@@ -1,5 +1,6 @@
 use crate::utils::{parse_envvar, parse_mapdir};
 use anyhow::Result;
+use bytes::Bytes;
 use std::{
     collections::{BTreeSet, HashMap},
     path::{Path, PathBuf},
@@ -12,7 +13,7 @@ use wasmer::{
 use wasmer_wasix::{
     default_fs_backing, get_wasi_versions,
     os::{tty_sys::SysTty, TtyBridge},
-    rewind,
+    rewind_ext,
     runners::MappedDirectory,
     runtime::task_manager::tokio::TokioTaskManager,
     types::__WASI_STDIN_FILENO,
@@ -270,40 +271,30 @@ impl Wasi {
         run: RunProperties,
         mut store: Store,
         tx: Sender<Result<i32>>,
-        rewind_state: Option<(RewindState, Result<(), Errno>)>,
+        rewind_state: Option<(RewindState, Bytes)>,
     ) {
         // If we need to rewind then do so
         let ctx = run.ctx;
-        if let Some((mut rewind_state, trigger_res)) = rewind_state {
+        if let Some((rewind_state, rewind_result)) = rewind_state {
             if rewind_state.is_64bit {
-                if let Err(exit_code) =
-                    rewind_state.rewinding_finish::<Memory64>(&ctx, &mut store, trigger_res)
-                {
-                    tx.send(Ok(exit_code.raw())).ok();
-                    return;
-                }
-                let res = rewind::<Memory64>(
+                let res = rewind_ext::<Memory64>(
                     ctx.env.clone().into_mut(&mut store),
                     rewind_state.memory_stack,
                     rewind_state.rewind_stack,
                     rewind_state.store_data,
+                    rewind_result,
                 );
                 if res != Errno::Success {
                     tx.send(Ok(res as i32)).ok();
                     return;
                 }
             } else {
-                if let Err(exit_code) =
-                    rewind_state.rewinding_finish::<Memory32>(&ctx, &mut store, trigger_res)
-                {
-                    tx.send(Ok(exit_code.raw())).ok();
-                    return;
-                }
-                let res = rewind::<Memory32>(
+                let res = rewind_ext::<Memory32>(
                     ctx.env.clone().into_mut(&mut store),
                     rewind_state.memory_stack,
                     rewind_state.rewind_stack,
                     rewind_state.store_data,
+                    rewind_result,
                 );
                 if res != Errno::Success {
                     tx.send(Ok(res as i32)).ok();
@@ -392,9 +383,15 @@ impl Wasi {
                         };
 
                         // Spawns the WASM process after a trigger
-                        tasks
-                            .resume_wasm_after_poller(Box::new(respawn), ctx, store, deep.trigger)
-                            .unwrap();
+                        unsafe {
+                            tasks.resume_wasm_after_poller(
+                                Box::new(respawn),
+                                ctx,
+                                store,
+                                deep.trigger,
+                            )
+                        }
+                        .unwrap();
                         return;
                     }
                     Ok(err) => Err(err.into()),

lib/wasi-types/src/wasi/bindings.rs

@@ -1,3 +1,5 @@
+#[cfg(feature = "enable-serde")]
+use serde::{Deserialize, Serialize};
 use std::mem::MaybeUninit;
 use wasmer::{MemorySize, ValueType};
 // TODO: Remove once bindings generate wai_bindgen_rust::bitflags::bitflags!  (temp hack)
@@ -93,6 +95,7 @@ impl core::fmt::Debug for Clockid {
 #[doc = " merely for alignment with POSIX."]
 #[repr(u16)]
 #[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
+#[cfg_attr(feature = "enable-serde", derive(Serialize, Deserialize))]
 pub enum Errno {
     #[doc = " No error occurred. System call completed successfully."]
     Success,
@@ -875,6 +878,7 @@ pub type Userdata = u64;
 #[doc = " Type of a subscription to an event or its occurrence."]
 #[repr(u8)]
 #[derive(Clone, Copy, PartialEq, Eq)]
+#[cfg_attr(feature = "enable-serde", derive(Serialize, Deserialize))]
 pub enum Eventtype {
     #[doc = " The time value of clock `subscription_clock::id` has"]
     #[doc = " reached timestamp `subscription_clock::timeout`."]
@@ -982,6 +986,7 @@ impl core::fmt::Debug for Preopentype {
 wai_bindgen_rust::bitflags::bitflags! {
     #[doc = " The state of the file descriptor subscribed to with"]
     #[doc = " `eventtype::fd_read` or `eventtype::fd_write`."]
+    #[cfg_attr(feature = "enable-serde", derive(Serialize, Deserialize))]
     pub struct Eventrwflags : u16 {
         #[doc = " The peer of this socket has closed or disconnected."]
         const FD_READWRITE_HANGUP = 1 << 0;
@@ -998,6 +1003,7 @@ impl Eventrwflags {
 #[doc = " `eventtype::fd_write` variants"]
 #[repr(C)]
 #[derive(Copy, Clone)]
+#[cfg_attr(feature = "enable-serde", derive(Serialize, Deserialize))]
 pub struct EventFdReadwrite {
     #[doc = " The number of bytes available for reading or writing."]
     pub nbytes: Filesize,

lib/wasi-types/src/wasi/wasix_manual.rs

@@ -1,3 +1,5 @@
+#[cfg(feature = "enable-serde")]
+use serde::{Deserialize, Serialize};
 use std::mem::MaybeUninit;
 
 use wasmer::{FromToNativeWasmType, MemorySize, ValueType};
@@ -179,7 +181,7 @@ unsafe impl ValueType for StackSnapshot {
 #[repr(C)]
 #[derive(Clone, Copy)]
 pub union JoinStatusUnion {
-    pub nothing_errno: Errno,
+    pub nothing: u8,
     pub exit_normal: Errno,
     pub exit_signal: ErrnoSignal,
     pub stopped: Signal,
@@ -196,7 +198,7 @@ impl core::fmt::Debug for JoinStatus {
         let mut f = binding.field("tag", &self.tag);
         f = unsafe {
             match self.tag {
-                JoinStatusType::Nothing => f.field("nothing_errno", &self.u.nothing_errno),
+                JoinStatusType::Nothing => f.field("nothing", &self.u.nothing),
                 JoinStatusType::ExitNormal => f.field("exit_normal", &self.u.exit_normal),
                 JoinStatusType::ExitSignal => f.field("exit_signal", &self.u.exit_signal),
                 JoinStatusType::Stopped => f.field("stopped", &self.u.stopped),
@@ -242,6 +244,7 @@ unsafe impl<M: MemorySize> ValueType for ThreadStart<M> {
 }
 
 #[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
+#[cfg_attr(feature = "enable-serde", derive(Serialize, Deserialize))]
 pub enum ExitCode {
     Errno(Errno),
     Other(i32),

lib/wasi-web/src/pool.rs

@@ -36,6 +36,7 @@ use wasmer_wasix::{
         },
         SpawnMemoryType,
     },
+    types::wasi::ExitCode,
     wasmer::{AsJs, Memory, MemoryType, Module, Store, WASM_MAX_PAGES},
     wasmer_wasix_types::wasi::Errno,
     InstanceSnapshot, VirtualTaskManager, WasiEnv, WasiFunctionEnv, WasiThreadError,
@@ -79,7 +80,7 @@ struct WasmRunCommand {
     snapshot: Option<InstanceSnapshot>,
     trigger: Option<WasmRunTrigger>,
     update_layout: bool,
-    result: Result<(), Errno>,
+    result: Option<Result<Bytes, ExitCode>>,
 }
 
 trait AssertSendSync: Send + Sync {}
@@ -320,7 +321,7 @@ impl WebThreadPool {
             module_bytes,
             snapshot,
             update_layout,
-            result: Ok(()),
+            result: None,
         });
         let task = Box::into_raw(task);
 
@@ -687,7 +688,7 @@ pub fn schedule_wasm_task(
     wasm_bindgen_futures::spawn_local(async move {
         if let Some(trigger) = trigger {
             let run = trigger.run;
-            task.result = run().await;
+            task.result = Some(run().await);
         }
 
         let task = Box::into_raw(task);

lib/wasi/Cargo.toml

@@ -15,7 +15,7 @@ cfg-if = "1.0"
 thiserror = "1"
 tracing = { version = "0.1" }
 getrandom = "0.2"
-wasmer-wasix-types = { path = "../wasi-types", version = "0.3.0" }
+wasmer-wasix-types = { path = "../wasi-types", version = "0.3.0", features = [ "enable-serde" ] }
 wasmer-types = { path = "../types", version = "=3.2.0", default-features = false }
 wasmer = { path = "../api", version = "=3.2.0", default-features = false, features = ["wat", "js-serializable-module"] }
 virtual-fs = { path = "../vfs", version = "0.2.0", default-features = false, features = ["webc-fs"] }

lib/wasi/src/bin_factory/exec.rs

@@ -3,9 +3,10 @@ use std::{pin::Pin, sync::Arc};
 use crate::{
     os::task::{thread::WasiThreadRunGuard, TaskJoinHandle},
     runtime::task_manager::{TaskWasm, TaskWasmRunProperties},
-    syscalls::rewind,
+    syscalls::rewind_ext,
     RewindState, VirtualBusError, WasiError, WasiRuntimeError,
 };
+use bytes::Bytes;
 use futures::Future;
 use tracing::*;
 use wasmer::{Function, FunctionEnvMut, Memory32, Memory64, Module, Store};
@@ -96,9 +97,7 @@ pub fn spawn_exec_module(
                 // Perform the initialization
                 let ctx = {
                     // If this module exports an _initialize function, run that first.
-                    if let Ok(initialize) = ctx
-                        .data(&store)
-                        .inner()
+                    if let Ok(initialize) = unsafe { ctx.data(&store).inner() }
                         .instance
                         .exports
                         .get_function("_initialize")
@@ -136,8 +135,7 @@ pub fn spawn_exec_module(
 }
 
 fn get_start(ctx: &WasiFunctionEnv, store: &Store) -> Option<Function> {
-    ctx.data(store)
-        .inner()
+    unsafe { ctx.data(store).inner() }
         .instance
         .exports
         .get_function("_start")
@@ -150,44 +148,34 @@ fn call_module(
     ctx: WasiFunctionEnv,
     mut store: Store,
     handle: WasiThreadRunGuard,
-    rewind_state: Option<(RewindState, Result<(), Errno>)>,
+    rewind_state: Option<(RewindState, Bytes)>,
 ) {
     let env = ctx.data(&store);
     let pid = env.pid();
     let tasks = env.tasks().clone();
     handle.thread.set_status_running();
 
     // If we need to rewind then do so
-    if let Some((mut rewind_state, trigger_res)) = rewind_state {
+    if let Some((rewind_state, rewind_result)) = rewind_state {
         if rewind_state.is_64bit {
-            if let Err(exit_code) =
-                rewind_state.rewinding_finish::<Memory64>(&ctx, &mut store, trigger_res)
-            {
-                ctx.data(&store).blocking_cleanup(Some(exit_code));
-                return;
-            }
-            let res = rewind::<Memory64>(
+            let res = rewind_ext::<Memory64>(
                 ctx.env.clone().into_mut(&mut store),
                 rewind_state.memory_stack,
                 rewind_state.rewind_stack,
                 rewind_state.store_data,
+                rewind_result,
             );
             if res != Errno::Success {
                 ctx.data(&store).blocking_cleanup(Some(res.into()));
                 return;
             }
         } else {
-            if let Err(exit_code) =
-                rewind_state.rewinding_finish::<Memory32>(&ctx, &mut store, trigger_res)
-            {
-                ctx.data(&store).blocking_cleanup(Some(exit_code));
-                return;
-            }
-            let res = rewind::<Memory32>(
+            let res = rewind_ext::<Memory32>(
                 ctx.env.clone().into_mut(&mut store),
                 rewind_state.memory_stack,
                 rewind_state.rewind_stack,
                 rewind_state.store_data,
+                rewind_result,
             );
             if res != Errno::Success {
                 ctx.data(&store).blocking_cleanup(Some(res.into()));
@@ -221,16 +209,16 @@ fn call_module(
                     // Create the callback that will be invoked when the thread respawns after a deep sleep
                     let rewind = deep.rewind;
                     let respawn = {
-                        move |ctx, store, trigger_res| {
+                        move |ctx, store, rewind_result| {
                             // Call the thread
-                            call_module(ctx, store, handle, Some((rewind, trigger_res)));
+                            call_module(ctx, store, handle, Some((rewind, rewind_result)));
                         }
                     };
 
                     // Spawns the WASM process after a trigger
-                    if let Err(err) =
+                    if let Err(err) = unsafe {
                         tasks.resume_wasm_after_poller(Box::new(respawn), ctx, store, deep.trigger)
-                    {
+                    } {
                         debug!("failed to go into deep sleep - {}", err);
                     }
                     return;