wasi-c2: implement a synchronous poll_oneoff in terms of yanix

doesnt work on unix yet! also breaks all the rules about using the
cap-std family instead of rawfds! but this is cool and im happy with it

Cargo.lock

@@ -2472,6 +2472,7 @@ dependencies = [
  "thiserror",
  "tracing",
  "wiggle",
+ "yanix 0.22.0 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
 [[package]]
@@ -2501,7 +2502,7 @@ dependencies = [
  "wiggle",
  "winapi",
  "winx 0.22.0",
- "yanix",
+ "yanix 0.22.0",
 ]
 
 [[package]]
@@ -3111,6 +3112,19 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "yanix"
+version = "0.22.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0504d76a87b9e77f1057d419a51acb4344b9e14eaf37dde22cf1fd0ec28901db"
+dependencies = [
+ "bitflags",
+ "cfg-if 1.0.0",
+ "filetime",
+ "libc",
+ "tracing",
+]
+
 [[package]]
 name = "z3"
 version = "0.7.1"

crates/wasi-c2/Cargo.toml

@@ -32,6 +32,7 @@ cap-rand = "0.9"
 fs-set-times = "0.2.2"
 cfg-if = "1"
 bitflags = "1.2"
+yanix = "0.22"
 
 [badges]
 maintenance = { status = "actively-developed" }

crates/wasi-c2/src/ctx.rs

@@ -75,7 +75,7 @@ impl WasiCtxBuilder {
         self.0.insert_file(
             0,
             f,
-            FileCaps::READ, // XXX fixme: more rights are ok, but this is read-only
+            FileCaps::READ | FileCaps::POLL_READWRITE, // XXX fixme: more rights are ok, but this is read-only
         );
         self
     }
@@ -84,7 +84,7 @@ impl WasiCtxBuilder {
         self.0.insert_file(
             1,
             f,
-            FileCaps::WRITE, // XXX fixme: more rights are ok, but this is append only
+            FileCaps::WRITE | FileCaps::POLL_READWRITE, // XXX fixme: more rights are ok, but this is append only
         );
         self
     }
@@ -93,7 +93,7 @@ impl WasiCtxBuilder {
         self.0.insert_file(
             2,
             f,
-            FileCaps::WRITE, // XXX fixme: more rights are ok, but this is append only
+            FileCaps::WRITE | FileCaps::POLL_READWRITE, // XXX fixme: more rights are ok, but this is append only
         );
         self
     }

crates/wasi-c2/src/sched.rs

@@ -58,7 +58,7 @@ impl<'a> Poll<'a> {
     }
     pub fn subscribe_write(&mut self, file: Ref<'a, dyn WasiFile>, ud: Userdata) {
         self.subs
-            .push((Subscription::Read(RwSubscription::new(file)), ud));
+            .push((Subscription::Write(RwSubscription::new(file)), ud));
     }
     pub fn results(self) -> Vec<(SubscriptionResult, Userdata)> {
         self.subs
@@ -66,7 +66,25 @@ impl<'a> Poll<'a> {
             .filter_map(|(s, ud)| SubscriptionResult::from_subscription(s).map(|r| (r, ud)))
             .collect()
     }
-    pub fn subscriptions(&'a self) -> impl Iterator<Item = &Subscription<'a>> {
+    pub fn is_empty(&self) -> bool {
-        self.subs.iter().map(|(s, _ud)| s)
+        self.subs.is_empty()
+    }
+    pub fn earliest_system_timer(&'a self) -> Option<&SystemTimerSubscription<'a>> {
+        let mut subs = self
+            .subs
+            .iter()
+            .filter_map(|(s, _ud)| match s {
+                Subscription::SystemTimer(t) => Some(t),
+                _ => None,
+            })
+            .collect::<Vec<&SystemTimerSubscription<'a>>>();
+        subs.sort_by(|a, b| a.deadline.cmp(&b.deadline));
+        subs.into_iter().next() // First element is earliest
+    }
+    pub fn rw_subscriptions(&'a self) -> impl Iterator<Item = &Subscription<'a>> {
+        self.subs.iter().filter_map(|(s, _ud)| match s {
+            Subscription::Read { .. } | Subscription::Write { .. } => Some(s),
+            _ => None,
+        })
     }
 }

crates/wasi-c2/src/sched/subscription.rs

@@ -41,12 +41,11 @@ pub struct SystemTimerSubscription<'a> {
 }
 
 impl<'a> SystemTimerSubscription<'a> {
+    pub fn now(&self) -> SystemTime {
+        self.clock.now(self.precision)
+    }
     pub fn result(&self) -> Option<Result<(), Error>> {
-        if self
+        if self.now().duration_since(self.deadline).is_ok() {
-            .deadline
-            .duration_since(self.clock.now(self.precision))
-            .is_ok()
-        {
             Some(Ok(()))
         } else {
             None

crates/wasi-c2/src/sched/sync.rs

@@ -1,27 +1,102 @@
-use crate::file::WasiFile;
-use crate::sched::subscription::{RwSubscription, Subscription, SystemTimerSubscription};
-use crate::sched::{Poll, WasiSched};
-use crate::Error;
-use std::any::Any;
-use std::ops::Deref;
 #[cfg(unix)]
-use std::os::unix::io::{AsRawFd, RawFd};
+pub use unix::*;
 
-#[derive(Default)]
+#[cfg(unix)]
-pub struct SyncSched {}
+mod unix {
+    use crate::file::WasiFile;
+    use crate::sched::subscription::{RwEventFlags, Subscription};
+    use crate::sched::{Poll, WasiSched};
+    use crate::Error;
+    use cap_std::time::Duration;
+    use std::convert::TryInto;
+    use std::ops::Deref;
+    use std::os::unix::io::{AsRawFd, RawFd};
+    use yanix::poll::{PollFd, PollFlags};
 
-impl WasiSched for SyncSched {
+    #[derive(Default)]
-    fn poll_oneoff(&self, poll: &Poll) -> Result<(), Error> {
+    pub struct SyncSched {}
-        for s in poll.subscriptions() {
+
+    impl WasiSched for SyncSched {
+        fn poll_oneoff<'a>(&self, poll: &'a Poll<'a>) -> Result<(), Error> {
+            if poll.is_empty() {
+                return Ok(());
+            }
+            let mut pollfds = Vec::new();
+            let timeout = poll.earliest_system_timer();
+            for s in poll.rw_subscriptions() {
                 match s {
-                Subscription::Read(f) | Subscription::Write(f) => {
+                    Subscription::Read(f) => {
                         let raw_fd = wasi_file_raw_fd(f.file.deref()).ok_or(Error::Inval)?;
-                    todo!()
+                        pollfds.push(unsafe { PollFd::new(raw_fd, PollFlags::POLLIN) });
                     }
-                Subscription::SystemTimer(t) => {
+
-                    todo!()
+                    Subscription::Write(f) => {
+                        let raw_fd = wasi_file_raw_fd(f.file.deref()).ok_or(Error::Inval)?;
+                        pollfds.push(unsafe { PollFd::new(raw_fd, PollFlags::POLLOUT) });
+                    }
+                    Subscription::SystemTimer { .. } => unreachable!(),
                 }
             }
+
+            let ready = loop {
+                let poll_timeout = if let Some(t) = timeout {
+                    let duration = t
+                        .deadline
+                        .duration_since(t.clock.now(t.precision))
+                        .unwrap_or(Duration::from_secs(0));
+                    (duration.as_millis() + 1) // XXX try always rounding up?
+                        .try_into()
+                        .map_err(|_| Error::Overflow)?
+                } else {
+                    libc::c_int::max_value()
+                };
+                tracing::debug!(
+                    poll_timeout = tracing::field::debug(poll_timeout),
+                    poll_fds = tracing::field::debug(&pollfds),
+                    "poll"
+                );
+                match yanix::poll::poll(&mut pollfds, poll_timeout) {
+                    Ok(ready) => break ready,
+                    Err(_) => {
+                        let last_err = std::io::Error::last_os_error();
+                        if last_err.raw_os_error().unwrap() == libc::EINTR {
+                            continue;
+                        } else {
+                            return Err(last_err.into());
+                        }
+                    }
+                }
+            };
+            if ready > 0 {
+                for (rwsub, pollfd) in poll.rw_subscriptions().zip(pollfds.into_iter()) {
+                    if let Some(revents) = pollfd.revents() {
+                        let (nbytes, rwsub) = match rwsub {
+                            Subscription::Read(sub) => {
+                                (1, sub)
+                                // XXX FIXME: query_nbytes in wasi-common/src/sys/poll.rs
+                                // uses metadata.len - tell to calculate for regular files,
+                                // ioctl(fd, FIONREAD) for large files
+                            }
+                            Subscription::Write(sub) => (0, sub),
+                            _ => unreachable!(),
+                        };
+                        if revents.contains(PollFlags::POLLNVAL) {
+                            rwsub.error(Error::Badf);
+                        } else if revents.contains(PollFlags::POLLERR) {
+                            rwsub.error(Error::Io);
+                        } else if revents.contains(PollFlags::POLLHUP) {
+                            rwsub.complete(nbytes, RwEventFlags::HANGUP);
+                        } else {
+                            rwsub.complete(nbytes, RwEventFlags::empty());
+                        };
+                    }
+                }
+            } else {
+                timeout
+                    .expect("timed out")
+                    .result()
+                    .expect("timer deadline is past")
+                    .unwrap()
             }
             Ok(())
         }
@@ -29,9 +104,9 @@ impl WasiSched for SyncSched {
             std::thread::yield_now();
             Ok(())
         }
-}
+    }
 
-fn wasi_file_raw_fd(f: &dyn WasiFile) -> Option<RawFd> {
+    fn wasi_file_raw_fd(f: &dyn WasiFile) -> Option<RawFd> {
         let a = f.as_any();
         if a.is::<cap_std::fs::File>() {
             Some(a.downcast_ref::<cap_std::fs::File>().unwrap().as_raw_fd())
@@ -52,4 +127,5 @@ fn wasi_file_raw_fd(f: &dyn WasiFile) -> Option<RawFd> {
         } else {
             None
         }
+    }
 }

crates/wasi-c2/src/snapshots/preview_1.rs

@@ -891,6 +891,10 @@ impl<'a> wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiCtx {
         events: &GuestPtr<types::Event>,
         nsubscriptions: types::Size,
     ) -> Result<types::Size, Error> {
+        if nsubscriptions == 0 {
+            return Err(Error::Inval);
+        }
+
         use cap_std::time::{Duration, SystemClock};
         let table = self.table();
         let mut poll = Poll::new();

crates/wasi-c2/src/table.rs

@@ -62,7 +62,7 @@ impl Table {
                 Err(Error::Exist) // Does exist, but borrowed
             }
         } else {
-            Err(Error::Exist) // Does not exist
+            Err(Error::Badf) // Does not exist
         }
     }