T0b1/wasmtime
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

rewrite wasi-tokio as just an task::block_in_place wrapper on cap-std-sync

Browse Source
This commit is contained in:
Pat Hickey
2021-05-04 11:29:02 -07:00
parent 686d8c22f9
commit f76fe8b764
13 changed files with 319 additions and 870 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
Cargo.lock generated
View File

@@ -3039,9 +3039,9 @@ dependencies = [
[[package]]
name = "unsafe-io"
version = "0.6.2"
version = "0.6.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0301dd0f2c21baed606faa2717fbfbb1a68b7e289ea29b40bc21a16f5ae9f5aa"
checksum = "fe39acfe60d3754452ea6881613c3240100b23ffd94a627c138863f8cd314b1b"
dependencies = [
"rustc_version",
"winapi",

2
crates/wasi-common/cap-std-sync/Cargo.toml
View File

@@ -20,7 +20,7 @@ cap-fs-ext = "0.13.7"
cap-time-ext = "0.13.7"
cap-rand = "0.13.2"
fs-set-times = "0.3.1"
unsafe-io = "0.6.2"
unsafe-io = "0.6.5"
system-interface = { version = "0.6.3", features = ["cap_std_impls"] }
tracing = "0.1.19"
bitflags = "1.2"

68
crates/wasi-common/cap-std-sync/src/dir.rs
View File

@@ -15,14 +15,8 @@ impl Dir {
pub fn from_cap_std(dir: cap_std::fs::Dir) -> Self {
Dir(dir)
}
}
#[wiggle::async_trait]
impl WasiDir for Dir {
fn as_any(&self) -> &dyn Any {
self
}
async fn open_file(
pub fn open_file_(
&self,
symlink_follow: bool,
path: &str,
@@ -30,7 +24,7 @@ impl WasiDir for Dir {
read: bool,
write: bool,
fdflags: FdFlags,
) -> Result<Box<dyn WasiFile>, Error> {
) -> Result<File, Error> {
use cap_fs_ext::{FollowSymlinks, OpenOptionsFollowExt};
let mut opts = cap_std::fs::OpenOptions::new();
@@ -82,16 +76,57 @@ impl WasiDir for Dir {
if fdflags.contains(wasi_common::file::FdFlags::NONBLOCK) {
f.set_fd_flags(system_interface::fs::FdFlags::NONBLOCK)?;
}
Ok(Box::new(File::from_cap_std(f)))
Ok(File::from_cap_std(f))
}
async fn open_dir(&self, symlink_follow: bool, path: &str) -> Result<Box<dyn WasiDir>, Error> {
pub fn open_dir_(&self, symlink_follow: bool, path: &str) -> Result<Self, Error> {
let d = if symlink_follow {
self.0.open_dir(Path::new(path))?
} else {
self.0.open_dir_nofollow(Path::new(path))?
};
Ok(Box::new(Dir::from_cap_std(d)))
Ok(Dir::from_cap_std(d))
}
pub fn rename_(&self, src_path: &str, dest_dir: &Self, dest_path: &str) -> Result<(), Error> {
self.0
.rename(Path::new(src_path), &dest_dir.0, Path::new(dest_path))?;
Ok(())
}
pub fn hard_link_(
&self,
src_path: &str,
target_dir: &Self,
target_path: &str,
) -> Result<(), Error> {
let src_path = Path::new(src_path);
let target_path = Path::new(target_path);
self.0.hard_link(src_path, &target_dir.0, target_path)?;
Ok(())
}
}
#[wiggle::async_trait]
impl WasiDir for Dir {
fn as_any(&self) -> &dyn Any {
self
}
async fn open_file(
&self,
symlink_follow: bool,
path: &str,
oflags: OFlags,
read: bool,
write: bool,
fdflags: FdFlags,
) -> Result<Box<dyn WasiFile>, Error> {
let f = self.open_file_(symlink_follow, path, oflags, read, write, fdflags)?;
Ok(Box::new(f))
}
async fn open_dir(&self, symlink_follow: bool, path: &str) -> Result<Box<dyn WasiDir>, Error> {
let d = self.open_dir_(symlink_follow, path)?;
Ok(Box::new(d))
}
async fn create_dir(&self, path: &str) -> Result<(), Error> {
@@ -101,7 +136,7 @@ impl WasiDir for Dir {
async fn readdir(
&self,
cursor: ReaddirCursor,
) -> Result<Box<dyn Iterator<Item = Result<ReaddirEntity, Error>>>, Error> {
) -> Result<Box<dyn Iterator<Item = Result<ReaddirEntity, Error>> + Send>, Error> {
// cap_std's read_dir does not include . and .., we should prepend these.
// Why does the Ok contain a tuple? We can't construct a cap_std::fs::DirEntry, and we don't
// have enough info to make a ReaddirEntity yet.
@@ -208,9 +243,7 @@ impl WasiDir for Dir {
.as_any()
.downcast_ref::<Self>()
.ok_or(Error::badf().context("failed downcast to cap-std Dir"))?;
self.0
.rename(Path::new(src_path), &dest_dir.0, Path::new(dest_path))?;
Ok(())
self.rename_(src_path, dest_dir, dest_path)
}
async fn hard_link(
&self,
@@ -222,10 +255,7 @@ impl WasiDir for Dir {
.as_any()
.downcast_ref::<Self>()
.ok_or(Error::badf().context("failed downcast to cap-std Dir"))?;
let src_path = Path::new(src_path);
let target_path = Path::new(target_path);
self.0.hard_link(src_path, &target_dir.0, target_path)?;
Ok(())
self.hard_link_(src_path, target_dir, target_path)
}
async fn set_times(
&self,

4
crates/wasi-common/src/dir.rs
View File

@@ -7,7 +7,7 @@ use std::ops::Deref;
use std::path::PathBuf;
#[wiggle::async_trait]
pub trait WasiDir {
pub trait WasiDir: Send + Sync {
fn as_any(&self) -> &dyn Any;
async fn open_file(
&self,
@@ -24,7 +24,7 @@ pub trait WasiDir {
async fn readdir(
&self,
cursor: ReaddirCursor,
) -> Result<Box<dyn Iterator<Item = Result<ReaddirEntity, Error>>>, Error>;
) -> Result<Box<dyn Iterator<Item = Result<ReaddirEntity, Error>> + Send>, Error>;
async fn symlink(&self, old_path: &str, new_path: &str) -> Result<(), Error>;
async fn remove_dir(&self, path: &str) -> Result<(), Error>;
async fn unlink_file(&self, path: &str) -> Result<(), Error>;

2
crates/wasi-common/src/file.rs
View File

@@ -5,7 +5,7 @@ use std::cell::{Ref, RefMut};
use std::ops::{Deref, DerefMut};
#[wiggle::async_trait]
pub trait WasiFile {
pub trait WasiFile: Send {
fn as_any(&self) -> &dyn Any;
async fn datasync(&self) -> Result<(), Error>; // write op
async fn sync(&self) -> Result<(), Error>; // file op

4
crates/wasi-common/src/pipe.rs
View File

@@ -106,7 +106,7 @@ impl From<&str> for ReadPipe<io::Cursor<String>> {
}
#[wiggle::async_trait]
impl<R: Read + Any> WasiFile for ReadPipe<R> {
impl<R: Read + Any + Send + Sync> WasiFile for ReadPipe<R> {
fn as_any(&self) -> &dyn Any {
self
}
@@ -265,7 +265,7 @@ impl WritePipe<io::Cursor<Vec<u8>>> {
}
#[wiggle::async_trait]
impl<W: Write + Any> WasiFile for WritePipe<W> {
impl<W: Write + Any + Send + Sync> WasiFile for WritePipe<W> {
fn as_any(&self) -> &dyn Any {
self
}

3
crates/wasi-common/tokio/Cargo.toml
View File

@@ -20,10 +20,11 @@ cap-std = "0.13.7"
cap-fs-ext = "0.13.7"
cap-time-ext = "0.13.7"
fs-set-times = "0.3.1"
unsafe-io = "0.6.2"
unsafe-io = "0.6.5"
system-interface = { version = "0.6.3", features = ["cap_std_impls"] }
tracing = "0.1.19"
bitflags = "1.2"
anyhow = "1"
[target.'cfg(unix)'.dependencies]
libc = "0.2"

215
crates/wasi-common/tokio/src/dir.rs
View File

@@ -1,21 +1,17 @@
use crate::{
asyncify,
file::{filetype_from, File},
};
use cap_fs_ext::{DirEntryExt, DirExt, MetadataExt, SystemTimeSpec};
use crate::{asyncify, file::File};
use std::any::Any;
use std::path::{Path, PathBuf};
use std::path::PathBuf;
use wasi_common::{
dir::{ReaddirCursor, ReaddirEntity, WasiDir},
file::{FdFlags, FileType, Filestat, OFlags, WasiFile},
file::{FdFlags, Filestat, OFlags, WasiFile},
Error, ErrorExt,
};
pub struct Dir(cap_std::fs::Dir);
pub struct Dir(wasi_cap_std_sync::dir::Dir);
impl Dir {
pub fn from_cap_std(dir: cap_std::fs::Dir) -> Self {
Dir(dir)
Dir(wasi_cap_std_sync::dir::Dir::from_cap_std(dir))
}
}
@@ -33,172 +29,59 @@ impl WasiDir for Dir {
write: bool,
fdflags: FdFlags,
) -> Result<Box<dyn WasiFile>, Error> {
use cap_fs_ext::{FollowSymlinks, OpenOptionsFollowExt};
use wasi_common::file::FdFlags;
let mut opts = cap_std::fs::OpenOptions::new();
if oflags.contains(OFlags::CREATE | OFlags::EXCLUSIVE) {
opts.create_new(true);
opts.write(true);
} else if oflags.contains(OFlags::CREATE) {
opts.create(true);
opts.write(true);
}
if oflags.contains(OFlags::TRUNCATE) {
opts.truncate(true);
}
if read {
opts.read(true);
}
if write {
opts.write(true);
} else {
// If not opened write, open read. This way the OS lets us open the file.
// If FileCaps::READ is not set, read calls will be rejected at the
// get_cap check.
opts.read(true);
}
if fdflags.contains(FdFlags::APPEND) {
opts.append(true);
}
if symlink_follow {
opts.follow(FollowSymlinks::Yes);
} else {
opts.follow(FollowSymlinks::No);
}
// the DSYNC, SYNC, and RSYNC flags are ignored! We do not
// have support for them in cap-std yet.
// ideally OpenOptions would just support this though:
// https://github.com/bytecodealliance/cap-std/issues/146
if fdflags.intersects(FdFlags::DSYNC | FdFlags::SYNC | FdFlags::RSYNC) {
return Err(Error::not_supported().context("SYNC family of FdFlags"));
}
let f = asyncify(move || self.0.open_with(Path::new(path), &opts))?;
let mut f = File::from_cap_std(f);
// NONBLOCK does not have an OpenOption either, but we can patch that on with set_fd_flags:
if fdflags.contains(FdFlags::NONBLOCK) {
f.set_fdflags(FdFlags::NONBLOCK).await?;
}
Ok(Box::new(f))
let f = asyncify(move || async move {
self.0
.open_file_(symlink_follow, path, oflags, read, write, fdflags)
})?;
Ok(Box::new(File::from_inner(f)))
}
async fn open_dir(&self, symlink_follow: bool, path: &str) -> Result<Box<dyn WasiDir>, Error> {
let path = unsafe { std::mem::transmute::<_, &'static str>(path) };
let d = if symlink_follow {
asyncify(move || self.0.open_dir(Path::new(path)))?
} else {
asyncify(move || self.0.open_dir_nofollow(Path::new(path)))?
};
Ok(Box::new(Dir::from_cap_std(d)))
let d = asyncify(move || async move { self.0.open_dir_(symlink_follow, path) })?;
Ok(Box::new(Dir(d)))
}
async fn create_dir(&self, path: &str) -> Result<(), Error> {
asyncify(|| self.0.create_dir(Path::new(path)))?;
Ok(())
asyncify(|| self.0.create_dir(path))
}
async fn readdir(
&self,
cursor: ReaddirCursor,
) -> Result<Box<dyn Iterator<Item = Result<ReaddirEntity, Error>>>, Error> {
// cap_std's read_dir does not include . and .., we should prepend these.
// Why does the Ok contain a tuple? We can't construct a cap_std::fs::DirEntry, and we don't
// have enough info to make a ReaddirEntity yet.
let dir_meta = asyncify(|| self.0.dir_metadata())?;
let rd = vec![
{
let name = ".".to_owned();
Ok((FileType::Directory, dir_meta.ino(), name))
},
{
let name = "..".to_owned();
Ok((FileType::Directory, dir_meta.ino(), name))
},
]
.into_iter()
.chain(
// Now process the `DirEntry`s:
self.0.entries()?.map(|entry| {
let entry = entry?;
// XXX full_metadata blocks, but we arent in an async iterator:
let meta = entry.full_metadata()?;
let inode = meta.ino();
let filetype = filetype_from(&meta.file_type());
let name = entry
.file_name()
.into_string()
.map_err(|_| Error::illegal_byte_sequence().context("filename"))?;
Ok((filetype, inode, name))
}),
)
// Enumeration of the iterator makes it possible to define the ReaddirCursor
.enumerate()
.map(|(ix, r)| match r {
Ok((filetype, inode, name)) => Ok(ReaddirEntity {
next: ReaddirCursor::from(ix as u64 + 1),
filetype,
inode,
name,
}),
Err(e) => Err(e),
})
.skip(u64::from(cursor) as usize);
) -> Result<Box<dyn Iterator<Item = Result<ReaddirEntity, Error>> + Send>, Error> {
struct I(Box<dyn Iterator<Item = Result<ReaddirEntity, Error>> + Send>);
impl Iterator for I {
type Item = Result<ReaddirEntity, Error>;
fn next(&mut self) -> Option<Self::Item> {
tokio::task::block_in_place(move || self.0.next())
}
}
Ok(Box::new(rd))
let inner = asyncify(move || self.0.readdir(cursor))?;
Ok(Box::new(I(inner)))
}
async fn symlink(&self, src_path: &str, dest_path: &str) -> Result<(), Error> {
asyncify(|| self.0.symlink(src_path, dest_path))?;
Ok(())
asyncify(move || self.0.symlink(src_path, dest_path))
}
async fn remove_dir(&self, path: &str) -> Result<(), Error> {
asyncify(|| self.0.remove_dir(Path::new(path)))?;
Ok(())
asyncify(move || self.0.remove_dir(path))
}
async fn unlink_file(&self, path: &str) -> Result<(), Error> {
asyncify(|| self.0.remove_file_or_symlink(Path::new(path)))?;
Ok(())
asyncify(move || self.0.unlink_file(path))
}
async fn read_link(&self, path: &str) -> Result<PathBuf, Error> {
let link = asyncify(|| self.0.read_link(Path::new(path)))?;
Ok(link)
asyncify(move || self.0.read_link(path))
}
async fn get_filestat(&self) -> Result<Filestat, Error> {
let meta = asyncify(|| self.0.dir_metadata())?;
Ok(Filestat {
device_id: meta.dev(),
inode: meta.ino(),
filetype: filetype_from(&meta.file_type()),
nlink: meta.nlink(),
size: meta.len(),
atim: meta.accessed().map(|t| Some(t.into_std())).unwrap_or(None),
mtim: meta.modified().map(|t| Some(t.into_std())).unwrap_or(None),
ctim: meta.created().map(|t| Some(t.into_std())).unwrap_or(None),
})
asyncify(|| self.0.get_filestat())
}
async fn get_path_filestat(
&self,
path: &str,
follow_symlinks: bool,
) -> Result<Filestat, Error> {
let meta = if follow_symlinks {
asyncify(|| self.0.metadata(Path::new(path)))?
} else {
asyncify(|| self.0.symlink_metadata(Path::new(path)))?
};
Ok(Filestat {
device_id: meta.dev(),
inode: meta.ino(),
filetype: filetype_from(&meta.file_type()),
nlink: meta.nlink(),
size: meta.len(),
atim: meta.accessed().map(|t| Some(t.into_std())).unwrap_or(None),
mtim: meta.modified().map(|t| Some(t.into_std())).unwrap_or(None),
ctim: meta.created().map(|t| Some(t.into_std())).unwrap_or(None),
})
asyncify(move || self.0.get_path_filestat(path, follow_symlinks))
}
async fn rename(
&self,
@@ -209,12 +92,8 @@ impl WasiDir for Dir {
let dest_dir = dest_dir
.as_any()
.downcast_ref::<Self>()
.ok_or(Error::badf().context("failed downcast to cap-std Dir"))?;
asyncify(|| {
self.0
.rename(Path::new(src_path), &dest_dir.0, Path::new(dest_path))
})?;
Ok(())
.ok_or(Error::badf().context("failed downcast to tokio Dir"))?;
asyncify(move || async move { self.0.rename_(src_path, &dest_dir.0, dest_path) })
}
async fn hard_link(
&self,
@@ -225,11 +104,8 @@ impl WasiDir for Dir {
let target_dir = target_dir
.as_any()
.downcast_ref::<Self>()
.ok_or(Error::badf().context("failed downcast to cap-std Dir"))?;
let src_path = Path::new(src_path);
let target_path = Path::new(target_path);
asyncify(|| self.0.hard_link(src_path, &target_dir.0, target_path))?;
Ok(())
.ok_or(Error::badf().context("failed downcast to tokio Dir"))?;
asyncify(move || async move { self.0.hard_link_(src_path, &target_dir.0, target_path) })
}
async fn set_times(
&self,
@@ -238,30 +114,7 @@ impl WasiDir for Dir {
mtime: Option<wasi_common::SystemTimeSpec>,
follow_symlinks: bool,
) -> Result<(), Error> {
asyncify(|| {
if follow_symlinks {
self.0.set_times(
Path::new(path),
convert_systimespec(atime),
convert_systimespec(mtime),
)
} else {
self.0.set_symlink_times(
Path::new(path),
convert_systimespec(atime),
convert_systimespec(mtime),
)
}
})?;
Ok(())
}
}
fn convert_systimespec(t: Option<wasi_common::SystemTimeSpec>) -> Option<SystemTimeSpec> {
match t {
Some(wasi_common::SystemTimeSpec::Absolute(t)) => Some(SystemTimeSpec::Absolute(t)),
Some(wasi_common::SystemTimeSpec::SymbolicNow) => Some(SystemTimeSpec::SymbolicNow),
None => None,
asyncify(move || self.0.set_times(path, atime, mtime, follow_symlinks))
}
}

271
crates/wasi-common/tokio/src/file.rs
View File

@@ -1,188 +1,114 @@
use crate::asyncify;
use cap_fs_ext::MetadataExt;
use fs_set_times::{SetTimes, SystemTimeSpec};
use std::any::Any;
use std::convert::TryInto;
use std::io;
use system_interface::fs::{FileIoExt, GetSetFdFlags};
use system_interface::io::ReadReady;
use wasi_common::{
file::{Advice, FdFlags, FileType, Filestat, WasiFile},
Error, ErrorExt,
Error,
};
mod internal {
use std::sync::{Mutex, MutexGuard};
#[cfg(not(windows))]
use unsafe_io::os::posish::{AsRawFd, RawFd};
#[cfg(windows)]
use unsafe_io::os::windows::{AsRawHandleOrSocket, RawHandleOrSocket};
use unsafe_io::OwnsRaw;
// This internal type wraps tokio's File so that we can impl the
// `AsUnsafeFile` trait. We impl this on an internal type, rather than on
// super::File, because we don't want consumers of this library to be able
// to use our `AsUnsafeFile`.
// Mutex is required because this type requires internal mutation for the
// tokio AsyncWriteExt methods to work, and must be Send.
pub(super) struct Internal(Mutex<tokio::fs::File>);
impl Internal {
pub fn new(f: tokio::fs::File) -> Self {
Internal(Mutex::new(f))
}
pub fn inner(&self) -> MutexGuard<tokio::fs::File> {
self.0.lock().unwrap()
}
}
#[cfg(not(windows))]
impl AsRawFd for Internal {
fn as_raw_fd(&self) -> RawFd {
self.inner().as_raw_fd()
}
}
#[cfg(windows)]
impl AsRawHandleOrSocket for Internal {
fn as_raw_handle_or_socket(&self) -> RawHandleOrSocket {
self.inner().as_raw_handle_or_socket()
}
}
// Safety: `Internal` owns its handle.
unsafe impl OwnsRaw for Internal {}
}
pub struct File(internal::Internal);
pub struct File(wasi_cap_std_sync::file::File);
impl File {
pub fn from_cap_std(file: cap_std::fs::File) -> Self {
File(internal::Internal::new(tokio::fs::File::from_std(
file.into_std(),
)))
pub(crate) fn from_inner(file: wasi_cap_std_sync::file::File) -> Self {
File(file)
}
async fn metadata(&self) -> Result<cap_std::fs::Metadata, Error> {
use unsafe_io::AsUnsafeFile;
asyncify(|| Ok(cap_std::fs::Metadata::from_file(&self.0.as_file_view())?))
pub fn from_cap_std(file: cap_std::fs::File) -> Self {
Self::from_inner(wasi_cap_std_sync::file::File::from_cap_std(file))
}
}
#[wiggle::async_trait]
impl WasiFile for File {
pub struct Stdin(wasi_cap_std_sync::stdio::Stdin);
pub fn stdin() -> Stdin {
Stdin(wasi_cap_std_sync::stdio::stdin())
}
pub struct Stdout(wasi_cap_std_sync::stdio::Stdout);
pub fn stdout() -> Stdout {
Stdout(wasi_cap_std_sync::stdio::stdout())
}
pub struct Stderr(wasi_cap_std_sync::stdio::Stderr);
pub fn stderr() -> Stderr {
Stderr(wasi_cap_std_sync::stdio::stderr())
}
macro_rules! wasi_file_impl {
($ty:ty) => {
#[wiggle::async_trait]
impl WasiFile for $ty {
fn as_any(&self) -> &dyn Any {
self
}
async fn datasync(&self) -> Result<(), Error> {
self.0.inner().sync_data().await?;
Ok(())
asyncify(|| self.0.datasync())
}
async fn sync(&self) -> Result<(), Error> {
self.0.inner().sync_all().await?;
Ok(())
asyncify(|| self.0.sync())
}
async fn get_filetype(&self) -> Result<FileType, Error> {
let meta = self.metadata().await?;
Ok(filetype_from(&meta.file_type()))
asyncify(|| self.0.get_filetype())
}
async fn get_fdflags(&self) -> Result<FdFlags, Error> {
let fdflags = asyncify(|| self.0.get_fd_flags())?;
Ok(from_sysif_fdflags(fdflags))
asyncify(|| self.0.get_fdflags())
}
async fn set_fdflags(&mut self, fdflags: FdFlags) -> Result<(), Error> {
if fdflags.intersects(
wasi_common::file::FdFlags::DSYNC
| wasi_common::file::FdFlags::SYNC
| wasi_common::file::FdFlags::RSYNC,
) {
return Err(Error::invalid_argument().context("cannot set DSYNC, SYNC, or RSYNC flag"));
}
asyncify(move || self.0.set_fd_flags(to_sysif_fdflags(fdflags)))?;
Ok(())
asyncify(|| self.0.set_fdflags(fdflags))
}
async fn get_filestat(&self) -> Result<Filestat, Error> {
let meta = self.metadata().await?;
Ok(Filestat {
device_id: meta.dev(),
inode: meta.ino(),
filetype: filetype_from(&meta.file_type()),
nlink: meta.nlink(),
size: meta.len(),
atim: meta.accessed().map(|t| Some(t.into_std())).unwrap_or(None),
mtim: meta.modified().map(|t| Some(t.into_std())).unwrap_or(None),
ctim: meta.created().map(|t| Some(t.into_std())).unwrap_or(None),
})
asyncify(|| self.0.get_filestat())
}
async fn set_filestat_size(&self, size: u64) -> Result<(), Error> {
self.0.inner().set_len(size).await?;
Ok(())
asyncify(move || self.0.set_filestat_size(size))
}
async fn advise(&self, offset: u64, len: u64, advice: Advice) -> Result<(), Error> {
asyncify(move || self.0.advise(offset, len, convert_advice(advice)))?;
Ok(())
asyncify(move || self.0.advise(offset, len, advice))
}
async fn allocate(&self, offset: u64, len: u64) -> Result<(), Error> {
asyncify(move || self.0.allocate(offset, len))?;
Ok(())
asyncify(move || self.0.allocate(offset, len))
}
async fn set_times(
async fn read_vectored<'a>(
&self,
atime: Option<wasi_common::SystemTimeSpec>,
mtime: Option<wasi_common::SystemTimeSpec>,
) -> Result<(), Error> {
asyncify(|| {
self.0
.set_times(convert_systimespec(atime), convert_systimespec(mtime))
})?;
Ok(())
}
async fn read_vectored<'a>(&self, bufs: &mut [io::IoSliceMut<'a>]) -> Result<u64, Error> {
use std::ops::DerefMut;
use tokio::io::AsyncReadExt;
let mut nbytes: usize = 0;
for b in bufs.iter_mut() {
let n = self.0.inner().read(b.deref_mut()).await?;
nbytes += n;
if n < b.len() {
break;
}
}
Ok(nbytes.try_into()?)
bufs: &mut [io::IoSliceMut<'a>],
) -> Result<u64, Error> {
asyncify(move || self.0.read_vectored(bufs))
}
async fn read_vectored_at<'a>(
&self,
bufs: &mut [io::IoSliceMut<'a>],
offset: u64,
) -> Result<u64, Error> {
let n = asyncify(move || self.0.read_vectored_at(bufs, offset))?;
Ok(n.try_into()?)
asyncify(move || self.0.read_vectored_at(bufs, offset))
}
async fn write_vectored<'a>(&self, bufs: &[io::IoSlice<'a>]) -> Result<u64, Error> {
use tokio::io::AsyncWriteExt;
let mut n: usize = 0;
n += self.0.inner().write_vectored(bufs).await?;
Ok(n.try_into()?)
asyncify(move || self.0.write_vectored(bufs))
}
async fn write_vectored_at<'a>(
&self,
bufs: &[io::IoSlice<'a>],
offset: u64,
) -> Result<u64, Error> {
let n = asyncify(move || self.0.write_vectored_at(bufs, offset))?;
Ok(n.try_into()?)
asyncify(move || self.0.write_vectored_at(bufs, offset))
}
async fn seek(&self, pos: std::io::SeekFrom) -> Result<u64, Error> {
use tokio::io::AsyncSeekExt;
Ok(self.0.inner().seek(pos).await?)
asyncify(move || self.0.seek(pos))
}
async fn peek(&self, buf: &mut [u8]) -> Result<u64, Error> {
let n = asyncify(move || self.0.peek(buf))?;
Ok(n.try_into()?)
asyncify(move || self.0.peek(buf))
}
async fn set_times(
&self,
atime: Option<wasi_common::SystemTimeSpec>,
mtime: Option<wasi_common::SystemTimeSpec>,
) -> Result<(), Error> {
asyncify(move || self.0.set_times(atime, mtime))
}
async fn num_ready_bytes(&self) -> Result<u64, Error> {
use unsafe_io::AsUnsafeFile;
asyncify(|| self.0.as_file_view().num_ready_bytes())
asyncify(|| self.0.num_ready_bytes())
}
#[cfg(not(windows))]
async fn readable(&mut self) -> Result<(), Error> {
// The Inner impls OwnsRaw, which asserts exclusive use of the handle by the owned object.
@@ -240,86 +166,11 @@ impl WasiFile for File {
// Windows uses a rawfd based scheduler :(
Err(Error::badf())
}
}
};
}
pub fn filetype_from(ft: &cap_std::fs::FileType) -> FileType {
use cap_fs_ext::FileTypeExt;
if ft.is_dir() {
FileType::Directory
} else if ft.is_symlink() {
FileType::SymbolicLink
} else if ft.is_socket() {
if ft.is_block_device() {
FileType::SocketDgram
} else {
FileType::SocketStream
}
} else if ft.is_block_device() {
FileType::BlockDevice
} else if ft.is_char_device() {
FileType::CharacterDevice
} else if ft.is_file() {
FileType::RegularFile
} else {
FileType::Unknown
}
}
pub fn convert_systimespec(t: Option<wasi_common::SystemTimeSpec>) -> Option<SystemTimeSpec> {
match t {
Some(wasi_common::SystemTimeSpec::Absolute(t)) => {
Some(SystemTimeSpec::Absolute(t.into_std()))
}
Some(wasi_common::SystemTimeSpec::SymbolicNow) => Some(SystemTimeSpec::SymbolicNow),
None => None,
}
}
pub fn to_sysif_fdflags(f: wasi_common::file::FdFlags) -> system_interface::fs::FdFlags {
let mut out = system_interface::fs::FdFlags::empty();
if f.contains(wasi_common::file::FdFlags::APPEND) {
out |= system_interface::fs::FdFlags::APPEND;
}
if f.contains(wasi_common::file::FdFlags::DSYNC) {
out |= system_interface::fs::FdFlags::DSYNC;
}
if f.contains(wasi_common::file::FdFlags::NONBLOCK) {
out |= system_interface::fs::FdFlags::NONBLOCK;
}
if f.contains(wasi_common::file::FdFlags::RSYNC) {
out |= system_interface::fs::FdFlags::RSYNC;
}
if f.contains(wasi_common::file::FdFlags::SYNC) {
out |= system_interface::fs::FdFlags::SYNC;
}
out
}
pub fn from_sysif_fdflags(f: system_interface::fs::FdFlags) -> wasi_common::file::FdFlags {
let mut out = wasi_common::file::FdFlags::empty();
if f.contains(system_interface::fs::FdFlags::APPEND) {
out |= wasi_common::file::FdFlags::APPEND;
}
if f.contains(system_interface::fs::FdFlags::DSYNC) {
out |= wasi_common::file::FdFlags::DSYNC;
}
if f.contains(system_interface::fs::FdFlags::NONBLOCK) {
out |= wasi_common::file::FdFlags::NONBLOCK;
}
if f.contains(system_interface::fs::FdFlags::RSYNC) {
out |= wasi_common::file::FdFlags::RSYNC;
}
if f.contains(system_interface::fs::FdFlags::SYNC) {
out |= wasi_common::file::FdFlags::SYNC;
}
out
}
pub fn convert_advice(advice: Advice) -> system_interface::fs::Advice {
match advice {
Advice::Normal => system_interface::fs::Advice::Normal,
Advice::Sequential => system_interface::fs::Advice::Sequential,
Advice::Random => system_interface::fs::Advice::Random,
Advice::WillNeed => system_interface::fs::Advice::WillNeed,
Advice::DontNeed => system_interface::fs::Advice::DontNeed,
Advice::NoReuse => system_interface::fs::Advice::NoReuse,
}
}
wasi_file_impl!(File);
wasi_file_impl!(Stdin);
wasi_file_impl!(Stdout);
wasi_file_impl!(Stderr);

18
crates/wasi-common/tokio/src/lib.rs
View File

@@ -4,6 +4,7 @@ pub mod sched;
pub mod stdio;
use std::cell::RefCell;
use std::future::Future;
use std::path::Path;
use std::rc::Rc;
pub use wasi_cap_std_sync::{clocks_ctx, random_ctx};
@@ -70,15 +71,14 @@ impl WasiCtxBuilder {
pub fn stderr(self, f: Box<dyn wasi_common::WasiFile>) -> Self {
WasiCtxBuilder(self.0.stderr(f))
}
// XXX our crate needs its own stdios
pub fn inherit_stdin(self) -> Self {
self.stdin(Box::new(wasi_cap_std_sync::stdio::stdin()))
self.stdin(Box::new(crate::stdio::stdin()))
}
pub fn inherit_stdout(self) -> Self {
self.stdout(Box::new(wasi_cap_std_sync::stdio::stdout()))
self.stdout(Box::new(crate::stdio::stdout()))
}
pub fn inherit_stderr(self) -> Self {
self.stderr(Box::new(wasi_cap_std_sync::stdio::stderr()))
self.stderr(Box::new(crate::stdio::stderr()))
}
pub fn inherit_stdio(self) -> Self {
self.inherit_stdin().inherit_stdout().inherit_stderr()
@@ -96,10 +96,14 @@ impl WasiCtxBuilder {
}
}
pub(crate) fn asyncify<'a, F, T>(f: F) -> Result<T, Error>
// This function takes the "async" code which is in fact blocking
// but always returns Poll::Ready, and executes it in a dummy executor
// on a blocking thread in the tokio thread pool.
pub(crate) fn asyncify<'a, F, Fut, T>(f: F) -> Result<T, Error>
where
F: FnOnce() -> Result<T, std::io::Error> + Send + 'a,
F: FnOnce() -> Fut + Send + 'a,
Fut: Future<Output = Result<T, Error>>,
T: Send + 'static,
{
tokio::task::block_in_place(f).map_err(Into::into)
tokio::task::block_in_place(move || unsafe { wiggle::run_in_dummy_executor(f()) })
}

5
crates/wasi-common/tokio/src/sched/unix.rs
View File

@@ -78,7 +78,10 @@ pub async fn poll_oneoff<'a>(poll: &mut Poll<'a>) -> Result<(), Error> {
}
}
if let Some(Some(remaining_duration)) = duration {
tokio::time::timeout(remaining_duration, futures).await??;
match tokio::time::timeout(remaining_duration, futures).await {
Ok(r) => r?,
Err(_deadline_elapsed) => {}
}
} else {
futures.await?;
}

77
crates/wasi-common/tokio/src/sched/windows.rs
View File

@@ -9,39 +9,18 @@ use wasi_common::{
file::WasiFile,
sched::{
subscription::{RwEventFlags, Subscription},
Poll,
Poll, WasiSched,
},
Error, ErrorExt,
};
pub async fn poll_oneoff<'a>(poll: &'_ Poll<'a>) -> Result<(), Error> {
pub async fn poll_oneoff<'a>(poll: &mut Poll<'a>) -> Result<(), Error> {
if poll.is_empty() {
return Ok(());
}
let mut ready = false;
let timeout = poll.earliest_clock_deadline();
let mut stdin_read_subs = Vec::new();
let mut immediate_subs = Vec::new();
for s in poll.rw_subscriptions() {
match s {
Subscription::Read(r) if r.file.as_any().is::<crate::stdio::Stdin>() => {
stdin_read_subs.push(r);
}
Subscription::Read(rw) | Subscription::Write(rw) => {
if wasi_file_raw_handle(rw.file.deref()).is_some() {
immediate_subs.push(s);
} else {
return Err(Error::invalid_argument()
.context("read/write subscription fd downcast failed"));
}
}
Subscription::MonotonicClock { .. } => unreachable!(),
}
}
if !stdin_read_subs.is_empty() {
let waitmode = if let Some(t) = timeout {
let waitmode = if let Some(t) = poll.earliest_clock_deadline() {
if let Some(duration) = t.duration_until() {
WaitMode::Timeout(duration)
} else {
@@ -54,6 +33,37 @@ pub async fn poll_oneoff<'a>(poll: &'_ Poll<'a>) -> Result<(), Error> {
WaitMode::Infinite
}
};
let mut stdin_read_subs = Vec::new();
let mut immediate_reads = Vec::new();
let mut immediate_writes = Vec::new();
for s in poll.rw_subscriptions() {
match s {
Subscription::Read(r) => {
if r.file.as_any().is::<crate::stdio::Stdin>() {
stdin_read_subs.push(r);
} else if wasi_file_raw_handle(r.file.deref()).is_some() {
immediate_reads.push(r);
} else {
return Err(
Error::invalid_argument().context("read subscription fd downcast failed")
);
}
}
Subscription::Write(w) => {
if wasi_file_raw_handle(w.file.deref()).is_some() {
immediate_writes.push(w);
} else {
return Err(
Error::invalid_argument().context("write subscription fd downcast failed")
);
}
}
Subscription::MonotonicClock { .. } => unreachable!(),
}
}
if !stdin_read_subs.is_empty() {
let state = STDIN_POLL
.lock()
.map_err(|_| Error::trap("failed to take lock of STDIN_POLL"))?
@@ -79,12 +89,7 @@ pub async fn poll_oneoff<'a>(poll: &'_ Poll<'a>) -> Result<(), Error> {
}
}
}
for sub in immediate_subs {
match sub {
Subscription::Read(r) => {
// XXX This doesnt strictly preserve the behavior in the earlier
// implementation, which would always do complete(0) for reads from
// stdout/err.
for r in immediate_reads {
match r.file.num_ready_bytes().await {
Ok(ready_bytes) => {
r.complete(ready_bytes, RwEventFlags::empty());
@@ -96,22 +101,17 @@ pub async fn poll_oneoff<'a>(poll: &'_ Poll<'a>) -> Result<(), Error> {
}
}
}
Subscription::Write(w) => {
for w in immediate_writes {
// Everything is always ready for writing, apparently?
w.complete(0, RwEventFlags::empty());
ready = true;
}
Subscription::MonotonicClock { .. } => unreachable!(),
}
}
if !ready {
if let Some(t) = timeout {
if let Some(duration) = t.duration_until() {
if let WaitMode::Timeout(duration) = waitmode {
thread::sleep(duration);
}
}
}
Ok(())
}
@@ -154,6 +154,7 @@ enum PollState {
Error(std::io::Error),
}
#[derive(Copy, Clone)]
enum WaitMode {
Timeout(Duration),
Infinite,

296
crates/wasi-common/tokio/src/stdio.rs
View File

@@ -1,295 +1 @@
use crate::file::convert_systimespec;
use fs_set_times::SetTimes;
use std::any::Any;
use std::convert::TryInto;
use std::io;
use std::io::{Read, Write};
use unsafe_io::AsUnsafeFile;
use wasi_common::{
file::{Advice, FdFlags, FileType, Filestat, WasiFile},
Error, ErrorExt,
};
mod internal {
#[cfg(unix)]
use std::os::unix::io::{AsRawFd, RawFd};
#[cfg(windows)]
use std::os::windows::io::{AsRawHandle, RawHandle};
use unsafe_io::OwnsRaw;
pub(super) struct TokioStdin(tokio::io::Stdin);
impl TokioStdin {
pub fn new() -> Self {
TokioStdin(tokio::io::stdin())
}
}
#[cfg(windows)]
impl AsRawHandle for TokioStdin {
fn as_raw_handle(&self) -> RawHandle {
self.0.as_raw_handle()
}
}
#[cfg(unix)]
impl AsRawFd for TokioStdin {
fn as_raw_fd(&self) -> RawFd {
self.0.as_raw_fd()
}
}
unsafe impl OwnsRaw for TokioStdin {}
pub(super) struct TokioStdout(tokio::io::Stdout);
impl TokioStdout {
pub fn new() -> Self {
TokioStdout(tokio::io::stdout())
}
}
#[cfg(windows)]
impl AsRawHandle for TokioStdout {
fn as_raw_handle(&self) -> RawHandle {
self.0.as_raw_handle()
}
}
#[cfg(unix)]
impl AsRawFd for TokioStdout {
fn as_raw_fd(&self) -> RawFd {
self.0.as_raw_fd()
}
}
unsafe impl OwnsRaw for TokioStdout {}
pub(super) struct TokioStderr(tokio::io::Stderr);
impl TokioStderr {
pub fn new() -> Self {
TokioStderr(tokio::io::stderr())
}
}
#[cfg(windows)]
impl AsRawHandle for TokioStderr {
fn as_raw_handle(&self) -> RawHandle {
self.0.as_raw_handle()
}
}
#[cfg(unix)]
impl AsRawFd for TokioStderr {
fn as_raw_fd(&self) -> RawFd {
self.0.as_raw_fd()
}
}
unsafe impl OwnsRaw for TokioStderr {}
}
pub struct Stdin(internal::TokioStdin);
pub fn stdin() -> Stdin {
Stdin(internal::TokioStdin::new())
}
#[wiggle::async_trait]
impl WasiFile for Stdin {
fn as_any(&self) -> &dyn Any {
self
}
async fn datasync(&self) -> Result<(), Error> {
Ok(())
}
async fn sync(&self) -> Result<(), Error> {
Ok(())
}
async fn get_filetype(&self) -> Result<FileType, Error> {
Ok(FileType::Unknown)
}
async fn get_fdflags(&self) -> Result<FdFlags, Error> {
Ok(FdFlags::empty())
}
async fn set_fdflags(&mut self, _fdflags: FdFlags) -> Result<(), Error> {
Err(Error::badf())
}
async fn get_filestat(&self) -> Result<Filestat, Error> {
let meta = self.0.as_file_view().metadata()?;
Ok(Filestat {
device_id: 0,
inode: 0,
filetype: self.get_filetype().await?,
nlink: 0,
size: meta.len(),
atim: meta.accessed().ok(),
mtim: meta.modified().ok(),
ctim: meta.created().ok(),
})
}
async fn set_filestat_size(&self, _size: u64) -> Result<(), Error> {
Err(Error::badf())
}
async fn advise(&self, _offset: u64, _len: u64, _advice: Advice) -> Result<(), Error> {
Err(Error::badf())
}
async fn allocate(&self, _offset: u64, _len: u64) -> Result<(), Error> {
Err(Error::badf())
}
async fn read_vectored<'a>(&self, bufs: &mut [io::IoSliceMut<'a>]) -> Result<u64, Error> {
let n = self.0.as_file_view().read_vectored(bufs)?;
Ok(n.try_into().map_err(|_| Error::range())?)
}
async fn read_vectored_at<'a>(
&self,
_bufs: &mut [io::IoSliceMut<'a>],
_offset: u64,
) -> Result<u64, Error> {
Err(Error::seek_pipe())
}
async fn write_vectored<'a>(&self, _bufs: &[io::IoSlice<'a>]) -> Result<u64, Error> {
Err(Error::badf())
}
async fn write_vectored_at<'a>(
&self,
_bufs: &[io::IoSlice<'a>],
_offset: u64,
) -> Result<u64, Error> {
Err(Error::badf())
}
async fn seek(&self, _pos: std::io::SeekFrom) -> Result<u64, Error> {
Err(Error::seek_pipe())
}
async fn peek(&self, _buf: &mut [u8]) -> Result<u64, Error> {
Err(Error::seek_pipe())
}
async fn set_times(
&self,
atime: Option<wasi_common::SystemTimeSpec>,
mtime: Option<wasi_common::SystemTimeSpec>,
) -> Result<(), Error> {
self.0
.set_times(convert_systimespec(atime), convert_systimespec(mtime))?;
Ok(())
}
#[cfg(not(windows))]
async fn num_ready_bytes(&self) -> Result<u64, Error> {
Ok(posish::io::fionread(&self.0)?)
}
#[cfg(windows)]
async fn num_ready_bytes(&self) -> Result<u64, Error> {
// conservative but correct is the best we can do
Ok(0)
}
async fn readable(&mut self) -> Result<(), Error> {
Err(Error::badf())
}
async fn writable(&mut self) -> Result<(), Error> {
Err(Error::badf())
}
}
macro_rules! wasi_file_write_impl {
($ty:ty) => {
#[wiggle::async_trait]
impl WasiFile for $ty {
fn as_any(&self) -> &dyn Any {
self
}
async fn datasync(&self) -> Result<(), Error> {
Ok(())
}
async fn sync(&self) -> Result<(), Error> {
Ok(())
}
async fn get_filetype(&self) -> Result<FileType, Error> {
Ok(FileType::Unknown)
}
async fn get_fdflags(&self) -> Result<FdFlags, Error> {
Ok(FdFlags::APPEND)
}
async fn set_fdflags(&mut self, _fdflags: FdFlags) -> Result<(), Error> {
Err(Error::badf())
}
async fn get_filestat(&self) -> Result<Filestat, Error> {
let meta = self.0.as_file_view().metadata()?;
Ok(Filestat {
device_id: 0,
inode: 0,
filetype: self.get_filetype().await?,
nlink: 0,
size: meta.len(),
atim: meta.accessed().ok(),
mtim: meta.modified().ok(),
ctim: meta.created().ok(),
})
}
async fn set_filestat_size(&self, _size: u64) -> Result<(), Error> {
Err(Error::badf())
}
async fn advise(&self, _offset: u64, _len: u64, _advice: Advice) -> Result<(), Error> {
Err(Error::badf())
}
async fn allocate(&self, _offset: u64, _len: u64) -> Result<(), Error> {
Err(Error::badf())
}
async fn read_vectored<'a>(
&self,
_bufs: &mut [io::IoSliceMut<'a>],
) -> Result<u64, Error> {
Err(Error::badf())
}
async fn read_vectored_at<'a>(
&self,
_bufs: &mut [io::IoSliceMut<'a>],
_offset: u64,
) -> Result<u64, Error> {
Err(Error::badf())
}
async fn write_vectored<'a>(&self, bufs: &[io::IoSlice<'a>]) -> Result<u64, Error> {
let n = self.0.as_file_view().write_vectored(bufs)?;
Ok(n.try_into().map_err(|c| Error::range().context(c))?)
}
async fn write_vectored_at<'a>(
&self,
_bufs: &[io::IoSlice<'a>],
_offset: u64,
) -> Result<u64, Error> {
Err(Error::seek_pipe())
}
async fn seek(&self, _pos: std::io::SeekFrom) -> Result<u64, Error> {
Err(Error::seek_pipe())
}
async fn peek(&self, _buf: &mut [u8]) -> Result<u64, Error> {
Err(Error::badf())
}
async fn set_times(
&self,
atime: Option<wasi_common::SystemTimeSpec>,
mtime: Option<wasi_common::SystemTimeSpec>,
) -> Result<(), Error> {
self.0
.set_times(convert_systimespec(atime), convert_systimespec(mtime))?;
Ok(())
}
async fn num_ready_bytes(&self) -> Result<u64, Error> {
Ok(0)
}
async fn readable(&mut self) -> Result<(), Error> {
Err(Error::badf())
}
async fn writable(&mut self) -> Result<(), Error> {
Err(Error::badf())
}
}
};
}
pub struct Stdout(internal::TokioStdout);
pub fn stdout() -> Stdout {
Stdout(internal::TokioStdout::new())
}
wasi_file_write_impl!(Stdout);
pub struct Stderr(internal::TokioStderr);
pub fn stderr() -> Stderr {
Stderr(internal::TokioStderr::new())
}
wasi_file_write_impl!(Stderr);
pub use crate::file::{stderr, stdin, stdout, Stderr, Stdin, Stdout};