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

Refactor poll_oneoff on *nix (#137)

Browse Source 
* Fixes CraneStation/wasmtime#440

This commit introduces a couple of changes/fixes:
* it annotates `log::debug!` messages with "host" to differentiate
  between file descriptors stored on the WASI side (aka the wrappers)
  and those managed by the host (aka the wrapped)
* it fixes CraneStation/wasmtime#440, i.e., incorrect passing of
  file descriptor to `poll_oneoff` where currently errenously we
  pass in the wrapper instead of the wrapped value
* it adds a couple more `log::debug!` macros calls for easier future
  debugging

* Add partial refactorting to poll_oneoff

This commit lays the groundwork for more clean up to come in
subsequent commits.

* Finalise refactoring of `poll_oneoff`

* Fix compilation error on Windows

* Address majority of suggestions and refactor

Co-authored-by: Marcin Mielniczuk <marmistrz.dev@zoho.eu>

* Add poll_oneoff test case

* Leave timeout in nanoseconds in ClockEventData

Instead of converting the timeout value from nanoseconds to
milliseconds in the host-independent impl, move the conversion
to *nix-specific impl as the conversion is currently only warranted
by the POSIX `poll` syscall.

* Don't fail immediately on bad descriptor

If the user specifies an invalid descriptor inside a subscription,
don't fail immediately but rather generate an event with the thrown
WASI error code, and continue with the remaining, potentially
correct subscriptions.
This commit is contained in:
Jakub Konka
2019-10-30 09:57:59 +01:00
committed by GitHub
parent aecec187d7
commit f3a5186230
8 changed files with 209 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/ctx.rs
View File

@@ -134,7 +134,9 @@ impl WasiCtxBuilder {
}
let mut fe = FdEntry::from(dir)?;
fe.preopen_path = Some(guest_path);
log::debug!("WasiCtx inserting ({:?}, {:?})", preopen_fd, fe);
self.fds.insert(preopen_fd, fe);
log::debug!("WasiCtx fds = {:?}", self.fds);
preopen_fd = preopen_fd.checked_add(1).ok_or(Error::ENFILE)?;
}

128
src/hostcalls_impl/misc.rs
View File

@@ -1,8 +1,9 @@
#![allow(non_camel_case_types)]
use crate::ctx::WasiCtx;
use crate::fdentry::Descriptor;
use crate::memory::*;
use crate::sys::hostcalls_impl;
use crate::{wasm32, Error, Result};
use crate::{host, wasm32, Error, Result};
use log::trace;
use std::convert::TryFrom;
@@ -182,8 +183,16 @@ pub(crate) fn clock_time_get(
enc_timestamp_byref(memory, time_ptr, time)
}
pub(crate) fn sched_yield() -> Result<()> {
trace!("sched_yield()");
std::thread::yield_now();
Ok(())
}
pub(crate) fn poll_oneoff(
_wasi_ctx: &WasiCtx,
wasi_ctx: &WasiCtx,
memory: &mut [u8],
input: wasm32::uintptr_t,
output: wasm32::uintptr_t,
@@ -205,19 +214,120 @@ pub(crate) fn poll_oneoff(
enc_pointee(memory, nevents, 0)?;
let input_slice = dec_slice_of::<wasm32::__wasi_subscription_t>(memory, input, nsubscriptions)?;
let input: Vec<_> = input_slice.iter().map(dec_subscription).collect();
let subscriptions = input_slice
.into_iter()
.map(dec_subscription)
.collect::<Result<Vec<_>>>()?;
let output_slice = dec_slice_of_mut::<wasm32::__wasi_event_t>(memory, output, nsubscriptions)?;
let events_count = hostcalls_impl::poll_oneoff(input, output_slice)?;
let mut output_slice_iter = output_slice.iter_mut();
let mut events_count = 0;
let mut timeout: Option<ClockEventData> = None;
let mut fd_events = Vec::new();
for subscription in subscriptions {
match subscription.type_ {
host::__WASI_EVENTTYPE_CLOCK => {
let clock = unsafe { subscription.u.clock };
let delay = wasi_clock_to_relative_ns_delay(clock)?;
log::debug!("poll_oneoff event.u.clock = {:?}", clock);
log::debug!("poll_oneoff delay = {:?}ns", delay);
let current = ClockEventData {
delay,
userdata: subscription.userdata,
};
let timeout = timeout.get_or_insert(current);
if current.delay < timeout.delay {
*timeout = current;
}
}
type_
if type_ == host::__WASI_EVENTTYPE_FD_READ
|| type_ == host::__WASI_EVENTTYPE_FD_WRITE =>
{
let wasi_fd = unsafe { subscription.u.fd_readwrite.fd };
let rights = if type_ == host::__WASI_EVENTTYPE_FD_READ {
host::__WASI_RIGHT_FD_READ
} else {
host::__WASI_RIGHT_FD_WRITE
};
match unsafe {
wasi_ctx
.get_fd_entry(wasi_fd, rights, 0)
.map(|fe| &fe.fd_object.descriptor)
} {
Ok(descriptor) => fd_events.push(FdEventData {
descriptor,
type_: subscription.type_,
userdata: subscription.userdata,
}),
Err(err) => {
let event = host::__wasi_event_t {
userdata: subscription.userdata,
type_,
error: err.as_wasi_errno(),
u: host::__wasi_event_t___wasi_event_u {
fd_readwrite: host::__wasi_event_t___wasi_event_u___wasi_event_u_fd_readwrite_t {
nbytes: 0,
flags: 0,
}
}
};
*output_slice_iter
.next()
.expect("number of subscriptions has to match number of events") =
enc_event(event);
events_count += 1;
}
}
}
_ => unreachable!(),
}
}
log::debug!("poll_oneoff timeout = {:?}", timeout);
log::debug!("poll_oneoff fd_events = {:?}", fd_events);
let events = hostcalls_impl::poll_oneoff(timeout, fd_events)?;
events_count += events.len();
for event in events {
*output_slice_iter
.next()
.expect("number of subscriptions has to match number of events") = enc_event(event);
}
trace!(" | *nevents={:?}", events_count);
enc_pointee(memory, nevents, events_count)
}
pub(crate) fn sched_yield() -> Result<()> {
trace!("sched_yield()");
fn wasi_clock_to_relative_ns_delay(
wasi_clock: host::__wasi_subscription_t___wasi_subscription_u___wasi_subscription_u_clock_t,
) -> Result<u128> {
use std::time::SystemTime;
std::thread::yield_now();
Ok(())
if wasi_clock.flags != host::__WASI_SUBSCRIPTION_CLOCK_ABSTIME {
return Ok(u128::from(wasi_clock.timeout));
}
let now: u128 = SystemTime::now()
.duration_since(SystemTime::UNIX_EPOCH)
.map_err(|_| Error::ENOTCAPABLE)?
.as_nanos();
let deadline = u128::from(wasi_clock.timeout);
Ok(deadline.saturating_sub(now))
}
#[derive(Debug, Copy, Clone)]
pub(crate) struct ClockEventData {
pub(crate) delay: u128, // delay is expressed in nanoseconds
pub(crate) userdata: host::__wasi_userdata_t,
}
#[derive(Debug)]
pub(crate) struct FdEventData<'a> {
pub(crate) descriptor: &'a Descriptor,
pub(crate) type_: host::__wasi_eventtype_t,
pub(crate) userdata: host::__wasi_userdata_t,
}

14
src/sys/unix/fdentry_impl.rs
View File

@@ -67,14 +67,14 @@ pub(crate) unsafe fn determine_type_rights<Fd: AsRawFd>(
let file = std::mem::ManuallyDrop::new(std::fs::File::from_raw_fd(fd.as_raw_fd()));
let ft = file.metadata()?.file_type();
if ft.is_block_device() {
log::debug!("Fd {:?} is a block device", fd.as_raw_fd());
log::debug!("Host fd {:?} is a block device", fd.as_raw_fd());
(
host::__WASI_FILETYPE_BLOCK_DEVICE,
host::RIGHTS_BLOCK_DEVICE_BASE,
host::RIGHTS_BLOCK_DEVICE_INHERITING,
)
} else if ft.is_char_device() {
log::debug!("Fd {:?} is a char device", fd.as_raw_fd());
log::debug!("Host fd {:?} is a char device", fd.as_raw_fd());
if isatty(fd)? {
(
host::__WASI_FILETYPE_CHARACTER_DEVICE,
@@ -89,21 +89,21 @@ pub(crate) unsafe fn determine_type_rights<Fd: AsRawFd>(
)
}
} else if ft.is_dir() {
log::debug!("Fd {:?} is a directory", fd.as_raw_fd());
log::debug!("Host fd {:?} is a directory", fd.as_raw_fd());
(
host::__WASI_FILETYPE_DIRECTORY,
host::RIGHTS_DIRECTORY_BASE,
host::RIGHTS_DIRECTORY_INHERITING,
)
} else if ft.is_file() {
log::debug!("Fd {:?} is a file", fd.as_raw_fd());
log::debug!("Host fd {:?} is a file", fd.as_raw_fd());
(
host::__WASI_FILETYPE_REGULAR_FILE,
host::RIGHTS_REGULAR_FILE_BASE,
host::RIGHTS_REGULAR_FILE_INHERITING,
)
} else if ft.is_socket() {
log::debug!("Fd {:?} is a socket", fd.as_raw_fd());
log::debug!("Host fd {:?} is a socket", fd.as_raw_fd());
use nix::sys::socket;
match socket::getsockopt(fd.as_raw_fd(), socket::sockopt::SockType)? {
socket::SockType::Datagram => (
@@ -119,14 +119,14 @@ pub(crate) unsafe fn determine_type_rights<Fd: AsRawFd>(
_ => return Err(Error::EINVAL),
}
} else if ft.is_fifo() {
log::debug!("Fd {:?} is a fifo", fd.as_raw_fd());
log::debug!("Host fd {:?} is a fifo", fd.as_raw_fd());
(
host::__WASI_FILETYPE_UNKNOWN,
host::RIGHTS_REGULAR_FILE_BASE,
host::RIGHTS_REGULAR_FILE_INHERITING,
)
} else {
log::debug!("Fd {:?} is unknown", fd.as_raw_fd());
log::debug!("Host fd {:?} is unknown", fd.as_raw_fd());
return Err(Error::EINVAL);
}
};

2
src/sys/unix/hostcalls_impl/fs.rs
View File

@@ -175,7 +175,7 @@ pub(crate) fn path_open(
}
};
log::debug!("path_open new_fd = {:?}", new_fd);
log::debug!("path_open (host) new_fd = {:?}", new_fd);
// Determine the type of the new file descriptor and which rights contradict with this type
Ok(unsafe { File::from_raw_fd(new_fd) })

198
src/sys/unix/hostcalls_impl/misc.rs
View File

@@ -1,12 +1,10 @@
#![allow(non_camel_case_types)]
#![allow(unused_unsafe)]
use crate::memory::*;
use crate::hostcalls_impl::{ClockEventData, FdEventData};
use crate::sys::host_impl;
use crate::{host, wasm32, Error, Result};
use crate::{host, Error, Result};
use nix::libc::{self, c_int};
use std::cmp;
use std::mem::MaybeUninit;
use std::time::SystemTime;
pub(crate) fn clock_res_get(clock_id: host::__wasi_clockid_t) -> Result<host::__wasi_timestamp_t> {
// convert the supported clocks to the libc types, or return EINVAL
@@ -70,198 +68,149 @@ pub(crate) fn clock_time_get(clock_id: host::__wasi_clockid_t) -> Result<host::_
}
pub(crate) fn poll_oneoff(
input: Vec<Result<host::__wasi_subscription_t>>,
output_slice: &mut [wasm32::__wasi_event_t],
) -> Result<wasm32::size_t> {
let timeout = input
.iter()
.filter_map(|event| match event {
Ok(event) if event.type_ == wasm32::__WASI_EVENTTYPE_CLOCK => Some(ClockEventData {
delay: wasi_clock_to_relative_ns_delay(unsafe { event.u.clock }).ok()? / 1_000_000,
userdata: event.userdata,
}),
_ => None,
})
.min_by_key(|event| event.delay);
timeout: Option<ClockEventData>,
fd_events: Vec<FdEventData>,
) -> Result<Vec<host::__wasi_event_t>> {
use nix::{
errno::Errno,
poll::{poll, PollFd, PollFlags},
};
use std::{convert::TryInto, os::unix::prelude::AsRawFd};
let fd_events: Vec<_> = input
.iter()
.filter_map(|event| match event {
Ok(event)
if event.type_ == wasm32::__WASI_EVENTTYPE_FD_READ
|| event.type_ == wasm32::__WASI_EVENTTYPE_FD_WRITE =>
{
Some(FdEventData {
fd: unsafe { event.u.fd_readwrite.fd } as c_int,
type_: event.type_,
userdata: event.userdata,
})
}
_ => None,
})
.collect();
if fd_events.is_empty() && timeout.is_none() {
return Ok(0);
return Ok(vec![]);
}
let mut poll_fds: Vec<_> = fd_events
.iter()
.map(|event| {
let mut flags = nix::poll::PollFlags::empty();
let mut flags = PollFlags::empty();
match event.type_ {
wasm32::__WASI_EVENTTYPE_FD_READ => flags.insert(nix::poll::PollFlags::POLLIN),
wasm32::__WASI_EVENTTYPE_FD_WRITE => flags.insert(nix::poll::PollFlags::POLLOUT),
host::__WASI_EVENTTYPE_FD_READ => flags.insert(PollFlags::POLLIN),
host::__WASI_EVENTTYPE_FD_WRITE => flags.insert(PollFlags::POLLOUT),
// An event on a file descriptor can currently only be of type FD_READ or FD_WRITE
// Nothing else has been defined in the specification, and these are also the only two
// events we filtered before. If we get something else here, the code has a serious bug.
_ => unreachable!(),
};
nix::poll::PollFd::new(event.fd, flags)
PollFd::new(event.descriptor.as_raw_fd(), flags)
})
.collect();
let timeout = timeout.map(|ClockEventData { delay, userdata }| ClockEventData {
delay: cmp::min(delay, c_int::max_value() as u128),
userdata,
let poll_timeout = timeout.map_or(-1, |timeout| {
let delay = timeout.delay / 1_000_000; // poll syscall requires delay to expressed in milliseconds
delay.try_into().unwrap_or(c_int::max_value())
});
let poll_timeout = timeout.map_or(-1, |timeout| timeout.delay as c_int);
log::debug!("poll_oneoff poll_timeout = {:?}", poll_timeout);
let ready = loop {
match nix::poll::poll(&mut poll_fds, poll_timeout) {
match poll(&mut poll_fds, poll_timeout) {
Err(_) => {
if nix::errno::Errno::last() == nix::errno::Errno::EINTR {
if Errno::last() == Errno::EINTR {
continue;
}
return Err(host_impl::errno_from_nix(nix::errno::Errno::last()));
return Err(host_impl::errno_from_nix(Errno::last()));
}
Ok(ready) => break ready as usize,
}
};
let events_count = if ready == 0 {
poll_oneoff_handle_timeout_event(output_slice, timeout)
} else {
let events = fd_events.iter().zip(poll_fds.iter()).take(ready);
poll_oneoff_handle_fd_event(output_slice, events)
};
Ok(events_count)
Ok(if ready == 0 {
poll_oneoff_handle_timeout_event(timeout.expect("timeout should not be None"))
} else {
let events = fd_events.into_iter().zip(poll_fds.into_iter()).take(ready);
poll_oneoff_handle_fd_event(events)?
})
}
// define the `fionread()` function, equivalent to `ioctl(fd, FIONREAD, *bytes)`
nix::ioctl_read_bad!(fionread, nix::libc::FIONREAD, c_int);
fn wasi_clock_to_relative_ns_delay(
wasi_clock: host::__wasi_subscription_t___wasi_subscription_u___wasi_subscription_u_clock_t,
) -> Result<u128> {
if wasi_clock.flags != wasm32::__WASI_SUBSCRIPTION_CLOCK_ABSTIME {
return Ok(u128::from(wasi_clock.timeout));
}
let now: u128 = SystemTime::now()
.duration_since(SystemTime::UNIX_EPOCH)
.map_err(|_| Error::ENOTCAPABLE)?
.as_nanos();
let deadline = u128::from(wasi_clock.timeout);
Ok(deadline.saturating_sub(now))
}
#[derive(Debug, Copy, Clone)]
struct ClockEventData {
delay: u128,
userdata: host::__wasi_userdata_t,
}
#[derive(Debug, Copy, Clone)]
struct FdEventData {
fd: c_int,
type_: host::__wasi_eventtype_t,
userdata: host::__wasi_userdata_t,
}
fn poll_oneoff_handle_timeout_event(
output_slice: &mut [wasm32::__wasi_event_t],
timeout: Option<ClockEventData>,
) -> wasm32::size_t {
if let Some(ClockEventData { userdata, .. }) = timeout {
let output_event = host::__wasi_event_t {
userdata,
type_: wasm32::__WASI_EVENTTYPE_CLOCK,
error: wasm32::__WASI_ESUCCESS,
u: host::__wasi_event_t___wasi_event_u {
fd_readwrite: host::__wasi_event_t___wasi_event_u___wasi_event_u_fd_readwrite_t {
nbytes: 0,
flags: 0,
},
fn poll_oneoff_handle_timeout_event(timeout: ClockEventData) -> Vec<host::__wasi_event_t> {
vec![host::__wasi_event_t {
userdata: timeout.userdata,
type_: host::__WASI_EVENTTYPE_CLOCK,
error: host::__WASI_ESUCCESS,
u: host::__wasi_event_t___wasi_event_u {
fd_readwrite: host::__wasi_event_t___wasi_event_u___wasi_event_u_fd_readwrite_t {
nbytes: 0,
flags: 0,
},
};
output_slice[0] = enc_event(output_event);
1
} else {
// shouldn't happen
0
}
},
}]
}
fn poll_oneoff_handle_fd_event<'t>(
output_slice: &mut [wasm32::__wasi_event_t],
events: impl Iterator<Item = (&'t FdEventData, &'t nix::poll::PollFd)>,
) -> wasm32::size_t {
let mut output_slice_cur = output_slice.iter_mut();
let mut revents_count = 0;
fn poll_oneoff_handle_fd_event<'a>(
events: impl Iterator<Item = (FdEventData<'a>, nix::poll::PollFd)>,
) -> Result<Vec<host::__wasi_event_t>> {
use nix::poll::PollFlags;
use std::{convert::TryInto, os::unix::prelude::AsRawFd};
let mut output_events = Vec::new();
for (fd_event, poll_fd) in events {
log::debug!("poll_oneoff_handle_fd_event fd_event = {:?}", fd_event);
log::debug!("poll_oneoff_handle_fd_event poll_fd = {:?}", poll_fd);
let revents = match poll_fd.revents() {
Some(revents) => revents,
None => continue,
};
log::debug!("poll_oneoff_handle_fd_event revents = {:?}", revents);
let mut nbytes = 0;
if fd_event.type_ == wasm32::__WASI_EVENTTYPE_FD_READ {
let _ = unsafe { fionread(fd_event.fd, &mut nbytes) };
if fd_event.type_ == host::__WASI_EVENTTYPE_FD_READ {
let _ = unsafe { fionread(fd_event.descriptor.as_raw_fd(), &mut nbytes) };
}
let output_event = if revents.contains(nix::poll::PollFlags::POLLNVAL) {
let output_event = if revents.contains(PollFlags::POLLNVAL) {
host::__wasi_event_t {
userdata: fd_event.userdata,
type_: fd_event.type_,
error: wasm32::__WASI_EBADF,
error: host::__WASI_EBADF,
u: host::__wasi_event_t___wasi_event_u {
fd_readwrite:
host::__wasi_event_t___wasi_event_u___wasi_event_u_fd_readwrite_t {
nbytes: 0,
flags: wasm32::__WASI_EVENT_FD_READWRITE_HANGUP,
flags: host::__WASI_EVENT_FD_READWRITE_HANGUP,
},
},
}
} else if revents.contains(nix::poll::PollFlags::POLLERR) {
} else if revents.contains(PollFlags::POLLERR) {
host::__wasi_event_t {
userdata: fd_event.userdata,
type_: fd_event.type_,
error: wasm32::__WASI_EIO,
error: host::__WASI_EIO,
u: host::__wasi_event_t___wasi_event_u {
fd_readwrite:
host::__wasi_event_t___wasi_event_u___wasi_event_u_fd_readwrite_t {
nbytes: 0,
flags: wasm32::__WASI_EVENT_FD_READWRITE_HANGUP,
flags: host::__WASI_EVENT_FD_READWRITE_HANGUP,
},
},
}
} else if revents.contains(nix::poll::PollFlags::POLLHUP) {
} else if revents.contains(PollFlags::POLLHUP) {
host::__wasi_event_t {
userdata: fd_event.userdata,
type_: fd_event.type_,
error: wasm32::__WASI_ESUCCESS,
error: host::__WASI_ESUCCESS,
u: host::__wasi_event_t___wasi_event_u {
fd_readwrite:
host::__wasi_event_t___wasi_event_u___wasi_event_u_fd_readwrite_t {
nbytes: 0,
flags: wasm32::__WASI_EVENT_FD_READWRITE_HANGUP,
flags: host::__WASI_EVENT_FD_READWRITE_HANGUP,
},
},
}
} else if revents.contains(nix::poll::PollFlags::POLLIN)
| revents.contains(nix::poll::PollFlags::POLLOUT)
{
} else if revents.contains(PollFlags::POLLIN) | revents.contains(PollFlags::POLLOUT) {
host::__wasi_event_t {
userdata: fd_event.userdata,
type_: fd_event.type_,
error: wasm32::__WASI_ESUCCESS,
error: host::__WASI_ESUCCESS,
u: host::__wasi_event_t___wasi_event_u {
fd_readwrite:
host::__wasi_event_t___wasi_event_u___wasi_event_u_fd_readwrite_t {
nbytes: nbytes as host::__wasi_filesize_t,
nbytes: nbytes.try_into()?,
flags: 0,
},
},
@@ -269,8 +218,9 @@ fn poll_oneoff_handle_fd_event<'t>(
} else {
continue;
};
*output_slice_cur.next().unwrap() = enc_event(output_event);
revents_count += 1;
output_events.push(output_event);
}
revents_count
Ok(output_events)
}

7
src/sys/windows/hostcalls_impl/misc.rs
View File

@@ -2,6 +2,7 @@
#![allow(unused_unsafe)]
#![allow(unused)]
use crate::helpers::systemtime_to_timestamp;
use crate::hostcalls_impl::{ClockEventData, FdEventData};
use crate::memory::*;
use crate::sys::host_impl;
use crate::{host, wasm32, Error, Result};
@@ -30,9 +31,9 @@ pub(crate) fn clock_time_get(clock_id: host::__wasi_clockid_t) -> Result<host::_
}
pub(crate) fn poll_oneoff(
input: Vec<Result<host::__wasi_subscription_t>>,
output_slice: &mut [wasm32::__wasi_event_t],
) -> Result<wasm32::size_t> {
timeout: Option<ClockEventData>,
fd_events: Vec<FdEventData>,
) -> Result<Vec<host::__wasi_event_t>> {
unimplemented!("poll_oneoff")
}