Skip to content

fix(wasi): enable all WasiFiles to be pollable #3913

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
pchickey merged 1 commit into from
Mar 10, 2022
Merged

fix(wasi): enable all WasiFiles to be pollable #3913

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions crates/wasi-common/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ bitflags = "1.2"
rustix = "0.33.0"

[target.'cfg(windows)'.dependencies]
io-extras = "0.13.2"
winapi = "0.3"

[badges]
Expand Down
9 changes: 9 additions & 0 deletions crates/wasi-common/cap-std-sync/src/file.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,15 @@ impl WasiFile for File {
fn as_any(&self) -> &dyn Any {
self
}
#[cfg(unix)]
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> {
self.0.sync_data()?;
Ok(())
Expand Down
18 changes: 18 additions & 0 deletions crates/wasi-common/cap-std-sync/src/net.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,15 @@ macro_rules! wasi_listen_write_impl {
fn as_any(&self) -> &dyn Any {
self
}
#[cfg(unix)]
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> {
let (stream, _) = self.0.accept()?;
let mut stream = <$stream>::from_cap_std(stream);
Expand Down Expand Up @@ -170,6 +179,15 @@ macro_rules! wasi_stream_write_impl {
fn as_any(&self) -> &dyn Any {
self
}
#[cfg(unix)]
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> {
Ok(FileType::SocketStream)
}
Expand Down
52 changes: 10 additions & 42 deletions crates/wasi-common/cap-std-sync/src/sched/unix.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,15 +1,8 @@
use cap_std::time::Duration;
use io_lifetimes::{AsFd, BorrowedFd};
use rustix::io::{PollFd, PollFlags};
use std::convert::TryInto;
use wasi_common::{
file::WasiFile,
sched::{
subscription::{RwEventFlags, Subscription},
Poll,
},
Error, ErrorExt,
};
use wasi_common::sched::subscription::{RwEventFlags, Subscription};
use wasi_common::{sched::Poll, Error, ErrorExt};

pub async fn poll_oneoff<'a>(poll: &mut Poll<'a>) -> Result<(), Error> {
if poll.is_empty() {
Expand All @@ -19,16 +12,18 @@ pub async fn poll_oneoff<'a>(poll: &mut Poll<'a>) -> Result<(), Error> {
for s in poll.rw_subscriptions() {
match s {
Subscription::Read(f) => {
let fd = wasi_file_fd(f.file).ok_or(
Error::invalid_argument().context("read subscription fd downcast failed"),
)?;
let fd = f
.file
.pollable()
.ok_or(Error::invalid_argument().context("file is not pollable"))?;
pollfds.push(PollFd::from_borrowed_fd(fd, PollFlags::IN));
}

Subscription::Write(f) => {
let fd = wasi_file_fd(f.file).ok_or(
Error::invalid_argument().context("write subscription fd downcast failed"),
)?;
let fd = f
.file
.pollable()
.ok_or(Error::invalid_argument().context("file is not pollable"))?;
pollfds.push(PollFd::from_borrowed_fd(fd, PollFlags::OUT));
}
Subscription::MonotonicClock { .. } => unreachable!(),
Expand Down Expand Up @@ -85,30 +80,3 @@ pub async fn poll_oneoff<'a>(poll: &mut Poll<'a>) -> Result<(), Error> {
}
Ok(())
}

fn wasi_file_fd(f: &dyn WasiFile) -> Option<BorrowedFd<'_>> {
let a = f.as_any();
if a.is::<crate::file::File>() {
Some(a.downcast_ref::<crate::file::File>().unwrap().as_fd())
} else if a.is::<crate::net::TcpStream>() {
Some(a.downcast_ref::<crate::net::TcpStream>().unwrap().as_fd())
} else if a.is::<crate::net::TcpListener>() {
Some(a.downcast_ref::<crate::net::TcpListener>().unwrap().as_fd())
} else if a.is::<crate::net::UnixStream>() {
Some(a.downcast_ref::<crate::net::UnixStream>().unwrap().as_fd())
} else if a.is::<crate::net::UnixListener>() {
Some(
a.downcast_ref::<crate::net::UnixListener>()
.unwrap()
.as_fd(),
)
} else if a.is::<crate::stdio::Stdin>() {
Some(a.downcast_ref::<crate::stdio::Stdin>().unwrap().as_fd())
} else if a.is::<crate::stdio::Stdout>() {
Some(a.downcast_ref::<crate::stdio::Stdout>().unwrap().as_fd())
} else if a.is::<crate::stdio::Stderr>() {
Some(a.downcast_ref::<crate::stdio::Stderr>().unwrap().as_fd())
} else {
None
}
}
70 changes: 7 additions & 63 deletions crates/wasi-common/cap-std-sync/src/sched/windows.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,30 +9,21 @@
// taken the time to improve it. See bug #2880.

use anyhow::Context;
use io_extras::os::windows::{AsRawHandleOrSocket, RawHandleOrSocket};
use std::ops::Deref;
use std::sync::mpsc::{self, Receiver, RecvTimeoutError, Sender, TryRecvError};
use std::sync::Mutex;
use std::thread;
use std::time::Duration;
use wasi_common::{
file::WasiFile,
sched::{
subscription::{RwEventFlags, Subscription},
Poll,
},
Error, ErrorExt,
};
use wasi_common::sched::subscription::{RwEventFlags, Subscription};
use wasi_common::{file::WasiFile, sched::Poll, Error, ErrorExt};

pub async fn poll_oneoff<'a>(poll: &mut Poll<'a>) -> Result<(), Error> {
poll_oneoff_(poll, wasi_file_is_stdin, wasi_file_raw_handle).await
poll_oneoff_(poll, wasi_file_is_stdin).await
}

// For reuse by wasi-tokio, which has a different WasiFile -> RawHandle translator.
pub async fn poll_oneoff_<'a>(
poll: &mut Poll<'a>,
file_is_stdin: impl Fn(&dyn WasiFile) -> bool,
file_to_handle: impl Fn(&dyn WasiFile) -> Option<RawHandleOrSocket>,
) -> Result<(), Error> {
if poll.is_empty() {
return Ok(());
Expand Down Expand Up @@ -61,21 +52,17 @@ pub async fn poll_oneoff_<'a>(
Subscription::Read(r) => {
if file_is_stdin(r.file.deref()) {
stdin_read_subs.push(r);
} else if file_to_handle(r.file.deref()).is_some() {
} else if r.file.pollable().is_some() {
immediate_reads.push(r);
} else {
return Err(
Error::invalid_argument().context("read subscription fd downcast failed")
);
return Err(Error::invalid_argument().context("file is not pollable"));
}
}
Subscription::Write(w) => {
if file_to_handle(w.file.deref()).is_some() {
if w.file.pollable().is_some() {
immediate_writes.push(w);
} else {
return Err(
Error::invalid_argument().context("write subscription fd downcast failed")
);
return Err(Error::invalid_argument().context("file is not pollable"));
}
}
Subscription::MonotonicClock { .. } => unreachable!(),
Expand Down Expand Up @@ -139,49 +126,6 @@ pub fn wasi_file_is_stdin(f: &dyn WasiFile) -> bool {
f.as_any().is::<crate::stdio::Stdin>()
}

pub fn wasi_file_raw_handle(f: &dyn WasiFile) -> Option<RawHandleOrSocket> {
let a = f.as_any();
if a.is::<crate::file::File>() {
Some(
a.downcast_ref::<crate::file::File>()
.unwrap()
.as_raw_handle_or_socket(),
)
} else if a.is::<crate::net::TcpStream>() {
Some(
a.downcast_ref::<crate::net::TcpStream>()
.unwrap()
.as_raw_handle_or_socket(),
)
} else if a.is::<crate::net::TcpListener>() {
Some(
a.downcast_ref::<crate::net::TcpListener>()
.unwrap()
.as_raw_handle_or_socket(),
)
} else if a.is::<crate::stdio::Stdin>() {
Some(
a.downcast_ref::<crate::stdio::Stdin>()
.unwrap()
.as_raw_handle_or_socket(),
)
} else if a.is::<crate::stdio::Stdout>() {
Some(
a.downcast_ref::<crate::stdio::Stdout>()
.unwrap()
.as_raw_handle_or_socket(),
)
} else if a.is::<crate::stdio::Stderr>() {
Some(
a.downcast_ref::<crate::stdio::Stderr>()
.unwrap()
.as_raw_handle_or_socket(),
)
} else {
None
}
}

enum PollState {
Ready,
NotReady, // Not ready, but did not wait
Expand Down
18 changes: 18 additions & 0 deletions crates/wasi-common/cap-std-sync/src/stdio.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,15 @@ 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())
}

#[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> {
if self.isatty() {
Ok(FileType::CharacterDevice)
Expand Down Expand Up @@ -98,6 +107,15 @@ macro_rules! wasi_file_write_impl {
fn as_any(&self) -> &dyn Any {
self
}
#[cfg(unix)]
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> {
if self.isatty() {
Ok(FileType::CharacterDevice)
Expand Down
10 changes: 10 additions & 0 deletions crates/wasi-common/src/file.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,16 @@ pub trait WasiFile: Send + Sync {
fn as_any(&self) -> &dyn Any;
async fn get_filetype(&mut self) -> Result<FileType, Error>;

#[cfg(unix)]
fn pollable(&self) -> Option<rustix::fd::BorrowedFd> {
None
}

#[cfg(windows)]
fn pollable(&self) -> Option<io_extras::os::windows::RawHandleOrSocket> {
None
}

fn isatty(&mut self) -> bool {
false
}
Expand Down
9 changes: 9 additions & 0 deletions crates/wasi-common/tokio/src/file.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,15 @@ macro_rules! wasi_file_impl {
fn as_any(&self) -> &dyn Any {
self
}
#[cfg(unix)]
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> {
block_on_dummy_executor(|| self.0.datasync())
}
Expand Down
46 changes: 1 addition & 45 deletions crates/wasi-common/tokio/src/sched/windows.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,4 @@
use crate::block_on_dummy_executor;
use io_extras::os::windows::{AsRawHandleOrSocket, RawHandleOrSocket};
use wasi_cap_std_sync::sched::windows::poll_oneoff_;
use wasi_common::{file::WasiFile, sched::Poll, Error};

Expand All @@ -8,52 +7,9 @@ pub async fn poll_oneoff<'a>(poll: &mut Poll<'a>) -> Result<(), Error> {
// we use the blocking poll_oneoff implementation from the wasi-cap-std-crate.
// We provide a function specific to this crate's WasiFile types for downcasting
// to a RawHandle.
block_on_dummy_executor(move || poll_oneoff_(poll, wasi_file_is_stdin, wasi_file_raw_handle))
block_on_dummy_executor(move || poll_oneoff_(poll, wasi_file_is_stdin))
}

pub fn wasi_file_is_stdin(f: &dyn WasiFile) -> bool {
f.as_any().is::<crate::stdio::Stdin>()
}

fn wasi_file_raw_handle(f: &dyn WasiFile) -> Option<RawHandleOrSocket> {
let a = f.as_any();
if a.is::<crate::file::File>() {
Some(
a.downcast_ref::<crate::file::File>()
.unwrap()
.as_raw_handle_or_socket(),
)
} else if a.is::<crate::net::TcpListener>() {
Some(
a.downcast_ref::<crate::net::TcpListener>()
.unwrap()
.as_raw_handle_or_socket(),
)
} else if a.is::<crate::net::TcpStream>() {
Some(
a.downcast_ref::<crate::net::TcpStream>()
.unwrap()
.as_raw_handle_or_socket(),
)
} else if a.is::<crate::stdio::Stdin>() {
Some(
a.downcast_ref::<crate::stdio::Stdin>()
.unwrap()
.as_raw_handle_or_socket(),
)
} else if a.is::<crate::stdio::Stdout>() {
Some(
a.downcast_ref::<crate::stdio::Stdout>()
.unwrap()
.as_raw_handle_or_socket(),
)
} else if a.is::<crate::stdio::Stderr>() {
Some(
a.downcast_ref::<crate::stdio::Stderr>()
.unwrap()
.as_raw_handle_or_socket(),
)
} else {
None
}
}