wasi_common/sync/sched/
unix.rs

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

wasi_common/sync/sched/
unix.rs
