wasi_cap_std_sync/sched/
unix.rs

1use cap_std::time::Duration;
2use rustix::event::{PollFd, PollFlags};
3use std::convert::TryInto;
4use wasi_common::sched::subscription::{RwEventFlags, Subscription};
5use wasi_common::{sched::Poll, Error, ErrorExt};
6
7pub async fn poll_oneoff<'a>(poll: &mut Poll<'a>) -> Result<(), Error> {
8    if poll.is_empty() {
9        return Ok(());
10    }
11    let mut pollfds = Vec::new();
12    for s in poll.rw_subscriptions() {
13        match s {
14            Subscription::Read(f) => {
15                let fd = f
16                    .file
17                    .pollable()
18                    .ok_or(Error::invalid_argument().context("file is not pollable"))?;
19                pollfds.push(PollFd::from_borrowed_fd(fd, PollFlags::IN));
20            }
21
22            Subscription::Write(f) => {
23                let fd = f
24                    .file
25                    .pollable()
26                    .ok_or(Error::invalid_argument().context("file is not pollable"))?;
27                pollfds.push(PollFd::from_borrowed_fd(fd, PollFlags::OUT));
28            }
29            Subscription::MonotonicClock { .. } => unreachable!(),
30        }
31    }
32
33    let ready = loop {
34        let poll_timeout = if let Some(t) = poll.earliest_clock_deadline() {
35            let duration = t.duration_until().unwrap_or(Duration::from_secs(0));
36            (duration.as_millis() + 1) // XXX try always rounding up?
37                .try_into()
38                .map_err(|_| Error::overflow().context("poll timeout"))?
39        } else {
40            std::os::raw::c_int::max_value()
41        };
42        tracing::debug!(
43            poll_timeout = tracing::field::debug(poll_timeout),
44            poll_fds = tracing::field::debug(&pollfds),
45            "poll"
46        );
47        match rustix::event::poll(&mut pollfds, poll_timeout) {
48            Ok(ready) => break ready,
49            Err(rustix::io::Errno::INTR) => continue,
50            Err(err) => return Err(std::io::Error::from(err).into()),
51        }
52    };
53    if ready > 0 {
54        for (rwsub, pollfd) in poll.rw_subscriptions().zip(pollfds.into_iter()) {
55            let revents = pollfd.revents();
56            let (nbytes, rwsub) = match rwsub {
57                Subscription::Read(sub) => {
58                    let ready = sub.file.num_ready_bytes()?;
59                    (std::cmp::max(ready, 1), sub)
60                }
61                Subscription::Write(sub) => (0, sub),
62                _ => unreachable!(),
63            };
64            if revents.contains(PollFlags::NVAL) {
65                rwsub.error(Error::badf());
66            } else if revents.contains(PollFlags::ERR) {
67                rwsub.error(Error::io());
68            } else if revents.contains(PollFlags::HUP) {
69                rwsub.complete(nbytes, RwEventFlags::HANGUP);
70            } else {
71                rwsub.complete(nbytes, RwEventFlags::empty());
72            };
73        }
74    } else {
75        poll.earliest_clock_deadline()
76            .expect("timed out")
77            .result()
78            .expect("timer deadline is past")
79            .unwrap()
80    }
81    Ok(())
82}
wasi_cap_std_sync/sched/unix.rs

wasi_cap_std_sync/sched/
unix.rs
