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            Some(
36                duration
37                    .try_into()
38                    .map_err(|_| Error::overflow().context("poll timeout"))?,
39            )
40        } else {
41            None
42        };
43        tracing::debug!(
44            poll_timeout = tracing::field::debug(poll_timeout),
45            poll_fds = tracing::field::debug(&pollfds),
46            "poll"
47        );
48        match rustix::event::poll(&mut pollfds, poll_timeout.as_ref()) {
49            Ok(ready) => break ready,
50            Err(rustix::io::Errno::INTR) => continue,
51            Err(err) => return Err(std::io::Error::from(err).into()),
52        }
53    };
54    if ready > 0 {
55        for (rwsub, pollfd) in poll.rw_subscriptions().zip(pollfds.into_iter()) {
56            let revents = pollfd.revents();
57            let (nbytes, rwsub) = match rwsub {
58                Subscription::Read(sub) => {
59                    let ready = sub.file.num_ready_bytes()?;
60                    (std::cmp::max(ready, 1), sub)
61                }
62                Subscription::Write(sub) => (0, sub),
63                _ => unreachable!(),
64            };
65            if revents.contains(PollFlags::NVAL) {
66                rwsub.error(Error::badf());
67            } else if revents.contains(PollFlags::ERR) {
68                rwsub.error(Error::io());
69            } else if revents.contains(PollFlags::HUP) {
70                rwsub.complete(nbytes, RwEventFlags::HANGUP);
71            } else {
72                rwsub.complete(nbytes, RwEventFlags::empty());
73            };
74        }
75    } else {
76        poll.earliest_clock_deadline()
77            .expect("timed out")
78            .result()
79            .expect("timer deadline is past")
80            .unwrap()
81    }
82    Ok(())
83}
wasi_common/sync/sched/unix.rs

wasi_common/sync/sched/
unix.rs
