epoll with fs fixes

Cargo.toml

@@ -27,6 +27,7 @@ wasmer-cache = { version = "=4.0.0", path = "lib/cache", optional = true }
 wasmer-types = { version = "=4.0.0", path = "lib/types" }
 wasmer-middlewares = { version = "=4.0.0", path = "lib/middlewares", optional = true }
 cfg-if = "1.0"
+tokio = { version = "1", features = [ "rt", "rt-multi-thread", "macros" ], optional = true }
 
 [workspace]
 members = [
@@ -49,6 +50,7 @@ members = [
     "lib/registry",
     "lib/sys-utils",
     "lib/types",
+    "lib/virtual-io",
     "lib/virtual-fs",
     "lib/virtual-net",
     "lib/vm",
@@ -235,12 +237,12 @@ required-features = ["backend", "wasi"]
 [[example]]
 name = "wasi-manual-setup"
 path = "examples/wasi_manual_setup.rs"
-required-features = ["backend", "wasi"]
+required-features = ["tokio", "backend", "wasi"]
 
 [[example]]
 name = "wasi-pipes"
 path = "examples/wasi_pipes.rs"
-required-features = ["backend", "wasi"]
+required-features = ["tokio", "backend", "wasi"]
 
 [[example]]
 name = "table"

examples/wasi_manual_setup.rs

@@ -33,6 +33,13 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Let's compile the Wasm module.
     let module = Module::new(&store, wasm_bytes)?;
 
+    println!("Starting `tokio` runtime...");
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+        .enable_all()
+        .build()
+        .unwrap();
+    let _guard = runtime.enter();
+
     println!("Creating `WasiEnv`...");
     // First, we create the `WasiEnv`
     let mut wasi_env = WasiEnv::builder("hello")

lib/c-api/Cargo.toml

@@ -42,6 +42,7 @@ libc = { version = "^0.2", default-features = false }
 thiserror = "1"
 typetag = { version = "0.1", optional = true }
 paste = "1.0"
+tokio = { version = "1", features = [ "rt", "rt-multi-thread", "io-util", "sync", "macros"], default_features = false }
 
 [dev-dependencies]
 field-offset = "0.3.3"
@@ -97,7 +98,7 @@ wasmer-artifact-load = ["wasmer-compiler/wasmer-artifact-load"]
 wasmer-artifact-create = ["wasmer-compiler/wasmer-artifact-create"]
 static-artifact-load = ["wasmer-compiler/static-artifact-load"]
 static-artifact-create = ["wasmer-compiler/static-artifact-create"]
-webc_runner = ["wasmer-wasix/webc_runner", "virtual-fs", "webc"]
+webc_runner = ["virtual-fs", "webc"]
 # Deprecated features.
 jit = ["compiler"]
 

lib/c-api/src/wasm_c_api/wasi/mod.rs

@@ -11,16 +11,19 @@ use super::{
     types::wasm_byte_vec_t,
 };
 use crate::error::update_last_error;
-use lazy_static::__Deref;
 use std::convert::TryFrom;
 use std::ffi::CStr;
 use std::os::raw::c_char;
 use std::slice;
+use std::sync::Arc;
 #[cfg(feature = "webc_runner")]
 use wasmer_api::{AsStoreMut, Imports, Module};
 use wasmer_wasix::{
-    default_fs_backing, get_wasi_version, virtual_fs::AsyncReadExt, virtual_fs::VirtualFile, Pipe,
-    VirtualTaskManager, WasiEnv, WasiEnvBuilder, WasiFunctionEnv, WasiVersion,
+    default_fs_backing, get_wasi_version,
+    runtime::task_manager::{tokio::TokioTaskManager, InlineWaker},
+    virtual_fs::AsyncReadExt,
+    virtual_fs::VirtualFile,
+    Pipe, PluggableRuntime, WasiEnv, WasiEnvBuilder, WasiFunctionEnv, WasiVersion,
 };
 
 #[derive(Debug)]
@@ -30,6 +33,7 @@ pub struct wasi_config_t {
     inherit_stderr: bool,
     inherit_stdin: bool,
     builder: WasiEnvBuilder,
+    runtime: Option<tokio::runtime::Runtime>,
 }
 
 #[no_mangle]
@@ -41,11 +45,18 @@ pub unsafe extern "C" fn wasi_config_new(
     let name_c_str = CStr::from_ptr(program_name);
     let prog_name = c_try!(name_c_str.to_str());
 
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+        .enable_all()
+        .build()
+        .unwrap();
+    let _guard = runtime.enter();
+
     Some(Box::new(wasi_config_t {
         inherit_stdout: true,
         inherit_stderr: true,
         inherit_stdin: true,
         builder: WasiEnv::builder(prog_name).fs(default_fs_backing()),
+        runtime: Some(runtime),
     }))
 }
 
@@ -222,7 +233,7 @@ unsafe fn wasi_env_with_filesystem_inner(
     let module = &module.as_ref()?.inner;
     let imports = imports?;
 
-    let (wasi_env, import_object) = prepare_webc_env(
+    let (wasi_env, import_object, runtime) = prepare_webc_env(
         config,
         &mut store.store_mut(),
         module,
@@ -236,21 +247,37 @@ unsafe fn wasi_env_with_filesystem_inner(
     Some(Box::new(wasi_env_t {
         inner: wasi_env,
         store: store.clone(),
+        _runtime: runtime,
     }))
 }
 
 #[cfg(feature = "webc_runner")]
 fn prepare_webc_env(
-    config: Box<wasi_config_t>,
+    mut config: Box<wasi_config_t>,
     store: &mut impl AsStoreMut,
     module: &Module,
     bytes: &'static u8,
     len: usize,
     package_name: &str,
-) -> Option<(WasiFunctionEnv, Imports)> {
+) -> Option<(WasiFunctionEnv, Imports, tokio::runtime::Runtime)> {
     use virtual_fs::static_fs::StaticFileSystem;
     use webc::v1::{FsEntryType, WebC};
 
+    let store_mut = store.as_store_mut();
+    let runtime = config.runtime.take();
+
+    let runtime = runtime.unwrap_or_else(|| {
+        tokio::runtime::Builder::new_multi_thread()
+            .enable_all()
+            .build()
+            .unwrap()
+    });
+
+    let handle = runtime.handle().clone();
+    let _guard = handle.enter();
+    let mut rt = PluggableRuntime::new(Arc::new(TokioTaskManager::new(handle)));
+    rt.set_engine(Some(store_mut.engine().clone()));
+
     let slice = unsafe { std::slice::from_raw_parts(bytes, len) };
     let volumes = WebC::parse_volumes_from_fileblock(slice).ok()?;
     let top_level_dirs = volumes
@@ -269,7 +296,7 @@ fn prepare_webc_env(
         .collect::<Vec<_>>();
 
     let filesystem = Box::new(StaticFileSystem::init(slice, package_name)?);
-    let mut builder = config.builder;
+    let mut builder = config.builder.runtime(Arc::new(rt));
 
     if !config.inherit_stdout {
         builder.set_stdout(Box::new(Pipe::channel().0));
@@ -289,14 +316,15 @@ fn prepare_webc_env(
     let env = builder.finalize(store).ok()?;
 
     let import_object = env.import_object(store, module).ok()?;
-    Some((env, import_object))
+    Some((env, import_object, runtime))
 }
 
 #[allow(non_camel_case_types)]
 pub struct wasi_env_t {
     /// cbindgen:ignore
     pub(super) inner: WasiFunctionEnv,
     pub(super) store: StoreRef,
+    pub(super) _runtime: tokio::runtime::Runtime,
 }
 
 /// Create a new WASI environment.
@@ -309,6 +337,21 @@ pub unsafe extern "C" fn wasi_env_new(
 ) -> Option<Box<wasi_env_t>> {
     let store = &mut store?.inner;
     let mut store_mut = store.store_mut();
+
+    let runtime = config.runtime.take();
+
+    let runtime = runtime.unwrap_or_else(|| {
+        tokio::runtime::Builder::new_multi_thread()
+            .enable_all()
+            .build()
+            .unwrap()
+    });
+
+    let handle = runtime.handle().clone();
+    let _guard = handle.enter();
+    let mut rt = PluggableRuntime::new(Arc::new(TokioTaskManager::new(handle)));
+    rt.set_engine(Some(store_mut.engine().clone()));
+
     if !config.inherit_stdout {
         config.builder.set_stdout(Box::new(Pipe::channel().0));
     }
@@ -319,11 +362,15 @@ pub unsafe extern "C" fn wasi_env_new(
 
     // TODO: impl capturer for stdin
 
-    let env = c_try!(config.builder.finalize(&mut store_mut));
+    let env = c_try!(config
+        .builder
+        .runtime(Arc::new(rt))
+        .finalize(&mut store_mut));
 
     Some(Box::new(wasi_env_t {
         inner: env,
         store: store.clone(),
+        _runtime: runtime,
     }))
 }
 
@@ -355,14 +402,14 @@ pub unsafe extern "C" fn wasi_env_read_stdout(
     let inner_buffer = slice::from_raw_parts_mut(buffer as *mut _, buffer_len);
     let store = env.store.store();
 
-    let (stdout, tasks) = {
+    let stdout = {
         let data = env.inner.data(&store);
-        (data.stdout(), data.tasks().clone())
+        data.stdout()
     };
 
     if let Ok(mut stdout) = stdout {
         if let Some(stdout) = stdout.as_mut() {
-            read_inner(tasks.deref(), stdout, inner_buffer)
+            read_inner(stdout, inner_buffer)
         } else {
             update_last_error("could not find a file handle for `stdout`");
             -1
@@ -381,13 +428,13 @@ pub unsafe extern "C" fn wasi_env_read_stderr(
 ) -> isize {
     let inner_buffer = slice::from_raw_parts_mut(buffer as *mut _, buffer_len);
     let store = env.store.store();
-    let (stderr, tasks) = {
+    let stderr = {
         let data = env.inner.data(&store);
-        (data.stderr(), data.tasks().clone())
+        data.stderr()
     };
     if let Ok(mut stderr) = stderr {
         if let Some(stderr) = stderr.as_mut() {
-            read_inner(tasks.deref(), stderr, inner_buffer)
+            read_inner(stderr, inner_buffer)
         } else {
             update_last_error("could not find a file handle for `stderr`");
             -1
@@ -399,11 +446,10 @@ pub unsafe extern "C" fn wasi_env_read_stderr(
 }
 
 fn read_inner(
-    tasks: &dyn VirtualTaskManager,
     wasi_file: &mut Box<dyn VirtualFile + Send + Sync + 'static>,
     inner_buffer: &mut [u8],
 ) -> isize {
-    tasks.block_on(async {
+    InlineWaker::block_on(async {
         match wasi_file.read(inner_buffer).await {
             Ok(a) => a as isize,
             Err(err) => {

lib/cli/Cargo.toml

@@ -34,7 +34,7 @@ wasmer-compiler-singlepass = { version = "=4.0.0", path = "../compiler-singlepas
 wasmer-compiler-llvm = { version = "=4.0.0", path = "../compiler-llvm", optional = true }
 wasmer-emscripten = { version = "=4.0.0", path = "../emscripten" }
 wasmer-vm = { version = "=4.0.0", path = "../vm", optional = true }
-wasmer-wasix = { version = "0.9.0", path = "../wasix", features = ["logging", "webc_runner", "webc_runner_rt_wcgi", "webc_runner_rt_wasi", "webc_runner_rt_emscripten", "host-fs"] }
+wasmer-wasix = { version = "0.9.0", path = "../wasix", features = ["logging", "webc_runner_rt_wcgi", "webc_runner_rt_emscripten", "host-fs"] }
 wasmer-wasix-experimental-io-devices = { version = "0.9.0", path = "../wasi-experimental-io-devices", optional = true, features = ["link_external_libs"] }
 wasmer-wast = { version = "=4.0.0", path = "../../tests/lib/wast", optional = true }
 wasmer-cache = { version = "=4.0.0", path = "../cache", features = ["blake3-pure"] }