more scheduler

crates/wasi-c2/src/sched.rs

@@ -1,6 +1,7 @@
+use crate::clocks::WasiSystemClock;
 use crate::file::WasiFile;
 use crate::Error;
-use cap_std::time::{SystemClock, SystemTime};
+use cap_std::time::SystemTime;
 use std::cell::Ref;
 pub mod subscription;
 
@@ -9,7 +10,7 @@ use subscription::{
 };
 
 pub trait WasiSched {
-    fn poll_oneoff(&self, subs: SubscriptionSet) -> Result<(), Error>;
+    fn poll_oneoff<'a>(&self, poll: &mut Poll<'a>) -> Result<(), Error>;
     fn sched_yield(&self) -> Result<(), Error>;
 }
 
@@ -17,7 +18,7 @@ pub trait WasiSched {
 pub struct SyncSched {}
 
 impl WasiSched for SyncSched {
-    fn poll_oneoff(&self, subs: SubscriptionSet) -> Result<(), Error> {
+    fn poll_oneoff<'a>(&self, poll: &mut Poll<'a>) -> Result<(), Error> {
         todo!()
     }
     fn sched_yield(&self) -> Result<(), Error> {
@@ -58,7 +59,7 @@ impl<'a> Poll<'a> {
         self.subs
             .push((Subscription::Read(RwSubscription::new(file)), ud));
     }
-    pub fn results(self, clock: &SystemClock) -> Vec<(SubscriptionResult, Userdata)> {
+    pub fn results(self, clock: &dyn WasiSystemClock) -> Vec<(SubscriptionResult, Userdata)> {
         self.subs
             .into_iter()
             .filter_map(|(s, ud)| SubscriptionResult::from_subscription(s, clock).map(|r| (r, ud)))

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

@@ -1,7 +1,8 @@
+use crate::clocks::WasiSystemClock;
 use crate::file::WasiFile;
 use crate::Error;
 use bitflags::bitflags;
-use cap_std::time::{SystemClock, SystemTime};
+use cap_std::time::{Duration, SystemTime};
 use std::cell::{Cell, Ref};
 
 bitflags! {
@@ -38,8 +39,12 @@ pub struct TimerSubscription {
 }
 
 impl TimerSubscription {
-    pub fn result(&self, clock: &SystemClock) -> Option<Result<(), Error>> {
-        if self.deadline.duration_since(clock.now()).is_ok() {
+    pub fn result(&self, clock: &dyn WasiSystemClock) -> Option<Result<(), Error>> {
+        if self
+            .deadline
+            .duration_since(clock.now(Duration::from_secs(0)))
+            .is_ok()
+        {
             Some(Ok(()))
         } else {
             None
@@ -82,7 +87,10 @@ pub enum SubscriptionResult {
 }
 
 impl SubscriptionResult {
-    pub fn from_subscription(s: Subscription, clock: &SystemClock) -> Option<SubscriptionResult> {
+    pub fn from_subscription(
+        s: Subscription,
+        clock: &dyn WasiSystemClock,
+    ) -> Option<SubscriptionResult> {
         match s {
             Subscription::Read(s) => s.result().map(SubscriptionResult::Read),
             Subscription::Write(s) => s.result().map(SubscriptionResult::Write),

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

@@ -1,6 +1,8 @@
 #![allow(unused_variables)]
 use crate::dir::{DirCaps, DirEntry, DirFdStat, ReaddirCursor, ReaddirEntity, TableDirExt};
 use crate::file::{FdFlags, FdStat, FileCaps, FileEntry, FileType, Filestat, OFlags};
+use crate::sched::subscription::{RwEventFlags, SubscriptionResult};
+use crate::sched::Poll;
 use crate::{Error, WasiCtx};
 use fs_set_times::SystemTimeSpec;
 use std::cell::{Ref, RefMut};
@@ -886,8 +888,98 @@ impl<'a> wasi_snapshot_preview1::WasiSnapshotPreview1 for WasiCtx {
         events: &GuestPtr<types::Event>,
         nsubscriptions: types::Size,
     ) -> Result<types::Size, Error> {
-        self.sched.poll_oneoff(todo!())?;
-        Ok(0)
+        let mut poll = Poll::new();
+
+        let subs = subs.as_array(nsubscriptions);
+        for sub_elem in subs.iter() {
+            let sub_ptr = sub_elem?;
+            let sub = sub_ptr.read()?;
+            match sub.u {
+                types::SubscriptionU::Clock(clock) => {
+                    todo!()
+                }
+                types::SubscriptionU::FdRead(read) => {
+                    todo!()
+                }
+                types::SubscriptionU::FdWrite(write) => {
+                    todo!()
+                }
+            }
+        }
+
+        self.sched.poll_oneoff(&mut poll)?;
+
+        let results = poll.results(self.clocks.system.deref());
+        let num_results = results.len();
+        assert!(
+            num_results <= nsubscriptions as usize,
+            "results exceeds subscriptions"
+        );
+        let events = events.as_array(
+            num_results
+                .try_into()
+                .expect("not greater than nsubscriptions"),
+        );
+        for ((result, userdata), event_elem) in results.into_iter().zip(events.iter()) {
+            let event_ptr = event_elem?;
+            let userdata: types::Userdata = userdata.into();
+            event_ptr.write(match result {
+                SubscriptionResult::Read(r) => {
+                    let type_ = types::Eventtype::FdRead;
+                    match r {
+                        Ok((nbytes, flags)) => types::Event {
+                            userdata,
+                            error: types::Errno::Success,
+                            type_,
+                            fd_readwrite: types::EventFdReadwrite {
+                                nbytes,
+                                flags: types::Eventrwflags::from(&flags),
+                            },
+                        },
+                        Err(e) => types::Event {
+                            userdata,
+                            error: e.try_into().expect("non-trapping"),
+                            type_,
+                            fd_readwrite: fd_readwrite_empty(),
+                        },
+                    }
+                }
+                SubscriptionResult::Write(r) => {
+                    let type_ = types::Eventtype::FdWrite;
+                    match r {
+                        Ok((nbytes, flags)) => types::Event {
+                            userdata,
+                            error: types::Errno::Success,
+                            type_,
+                            fd_readwrite: types::EventFdReadwrite {
+                                nbytes,
+                                flags: types::Eventrwflags::from(&flags),
+                            },
+                        },
+                        Err(e) => types::Event {
+                            userdata,
+                            error: e.try_into().expect("non-trapping"),
+                            type_,
+                            fd_readwrite: fd_readwrite_empty(),
+                        },
+                    }
+                }
+                SubscriptionResult::Timer(r) => {
+                    let type_ = types::Eventtype::Clock;
+                    types::Event {
+                        userdata,
+                        error: match r {
+                            Ok(()) => types::Errno::Success,
+                            Err(e) => e.try_into().expect("non-trapping"),
+                        },
+                        type_,
+                        fd_readwrite: fd_readwrite_empty(),
+                    }
+                }
+            })?;
+        }
+
+        Ok(num_results.try_into().expect("results fit into memory"))
     }
 
     fn proc_exit(&self, status: types::Exitcode) -> wiggle::Trap {
@@ -1319,3 +1411,20 @@ fn dirent_bytes(dirent: types::Dirent) -> Vec<u8> {
     unsafe { ptr.write_unaligned(dirent) };
     bytes
 }
+
+impl From<&RwEventFlags> for types::Eventrwflags {
+    fn from(flags: &RwEventFlags) -> types::Eventrwflags {
+        let mut out = types::Eventrwflags::empty();
+        if flags.contains(RwEventFlags::HANGUP) {
+            out = out | types::Eventrwflags::FD_READWRITE_HANGUP;
+        }
+        out
+    }
+}
+
+fn fd_readwrite_empty() -> types::EventFdReadwrite {
+    types::EventFdReadwrite {
+        nbytes: 0,
+        flags: types::Eventrwflags::empty(),
+    }
+}