wasmer_wasix/syscalls/wasi/
poll_oneoff.rs

1use serde::{Deserialize, Serialize};
2use wasmer_wasix_types::wasi::{Subclockflags, SubscriptionClock, Userdata};
3
4use super::*;
5use crate::{
6    WasiInodes,
7    fs::{InodeValFilePollGuard, InodeValFilePollGuardJoin},
8    state::PollEventSet,
9    syscalls::*,
10};
11
12/// An event that occurred.
13#[derive(Serialize, Deserialize)]
14pub enum EventResultType {
15    Clock(u8),
16    Fd(EventFdReadwrite),
17}
18
19/// An event that occurred.
20#[derive(Serialize, Deserialize)]
21pub struct EventResult {
22    /// User-provided value that got attached to `subscription::userdata`.
23    pub userdata: Userdata,
24    /// If non-zero, an error that occurred while processing the subscription request.
25    pub error: Errno,
26    /// Type of event that was triggered
27    pub type_: Eventtype,
28    /// The type of the event that occurred, and the contents of the event
29    pub inner: EventResultType,
30}
31impl EventResult {
32    pub fn into_event(self) -> Event {
33        Event {
34            userdata: self.userdata,
35            error: self.error,
36            type_: self.type_,
37            u: match self.inner {
38                EventResultType::Clock(id) => EventUnion { clock: id },
39                EventResultType::Fd(fd) => EventUnion { fd_readwrite: fd },
40            },
41        }
42    }
43}
44
45/// ### `poll_oneoff()`
46/// Concurrently poll for a set of events
47///
48/// Inputs:
49/// - `const __wasi_subscription_t *in`
50///     The events to subscribe to
51/// - `__wasi_event_t *out`
52///     The events that have occured
53/// - `u32 nsubscriptions`
54///     The number of subscriptions and the number of events
55///
56/// Output:
57/// - `u32 nevents`
58///     The number of events seen
59#[instrument(level = "trace", skip_all, fields(timeout_ms = field::Empty, fd_guards = field::Empty, seen = field::Empty), ret)]
60pub fn poll_oneoff<M: MemorySize + 'static>(
61    mut ctx: FunctionEnvMut<'_, WasiEnv>,
62    in_: WasmPtr<Subscription, M>,
63    out_: WasmPtr<Event, M>,
64    nsubscriptions: M::Offset,
65    nevents: WasmPtr<M::Offset, M>,
66) -> Result<Errno, WasiError> {
67    WasiEnv::do_pending_operations(&mut ctx)?;
68
69    // An empty subscription list would otherwise block forever in the poll loop.
70    if nsubscriptions == M::ZERO {
71        return Ok(Errno::Inval);
72    }
73
74    ctx = wasi_try_ok!(maybe_backoff::<M>(ctx)?);
75    ctx = wasi_try_ok!(maybe_snapshot::<M>(ctx)?);
76
77    ctx.data_mut().poll_seed += 1;
78    let mut env = ctx.data();
79    let mut memory = unsafe { env.memory_view(&ctx) };
80
81    let subscription_array = wasi_try_mem_ok!(in_.slice(&memory, nsubscriptions));
82    let mut subscriptions = Vec::with_capacity(subscription_array.len() as usize);
83    for n in 0..subscription_array.len() {
84        let n = (n + env.poll_seed) % subscription_array.len();
85        let sub = subscription_array.index(n);
86        let s = wasi_try_mem_ok!(sub.read());
87        subscriptions.push((None, PollEventSet::default(), s));
88    }
89
90    // We clear the number of events
91    wasi_try_mem_ok!(nevents.write(&memory, M::ZERO));
92
93    // Function to invoke once the poll is finished
94    let process_events = |ctx: &FunctionEnvMut<'_, WasiEnv>, triggered_events: Vec<Event>| {
95        let mut env = ctx.data();
96        let mut memory = unsafe { env.memory_view(&ctx) };
97
98        // Process all the events that were triggered
99        let mut events_seen: u32 = 0;
100        let event_array = wasi_try_mem!(out_.slice(&memory, nsubscriptions));
101        for event in triggered_events {
102            wasi_try_mem!(event_array.index(events_seen as u64).write(event));
103            events_seen += 1;
104        }
105        let events_seen: M::Offset = events_seen.into();
106        let out_ptr = nevents.deref(&memory);
107        wasi_try_mem!(out_ptr.write(events_seen));
108        Errno::Success
109    };
110
111    // Poll and receive all the events that triggered
112    poll_oneoff_internal::<M, _>(ctx, subscriptions, process_events)
113}
114
115struct PollBatch {
116    pid: WasiProcessId,
117    tid: WasiThreadId,
118    evts: Vec<Event>,
119    joins: Vec<InodeValFilePollGuardJoin>,
120}
121impl PollBatch {
122    fn new(pid: WasiProcessId, tid: WasiThreadId, fds: Vec<InodeValFilePollGuard>) -> Self {
123        Self {
124            pid,
125            tid,
126            evts: Vec::new(),
127            joins: fds
128                .into_iter()
129                .map(InodeValFilePollGuardJoin::new)
130                .collect(),
131        }
132    }
133}
134impl Future for PollBatch {
135    type Output = Result<Vec<EventResult>, Errno>;
136    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
137        let pid = self.pid;
138        let tid = self.tid;
139        let mut done = false;
140
141        let mut evts = Vec::new();
142        for mut join in self.joins.iter_mut() {
143            let fd = join.fd();
144            let peb = join.peb();
145            let mut guard = Pin::new(join);
146            match guard.poll(cx) {
147                Poll::Pending => {}
148                Poll::Ready(e) => {
149                    for (evt, readiness) in e {
150                        tracing::trace!(
151                            fd,
152                            readiness = ?readiness,
153                            userdata = evt.userdata,
154                            ty = evt.type_ as u8,
155                            peb,
156                            "triggered"
157                        );
158                        evts.push(evt);
159                    }
160                }
161            }
162        }
163
164        if !evts.is_empty() {
165            return Poll::Ready(Ok(evts));
166        }
167
168        Poll::Pending
169    }
170}
171
172pub(crate) fn poll_fd_guard(
173    state: &Arc<WasiState>,
174    peb: PollEventSet,
175    fd: WasiFd,
176    s: Subscription,
177) -> Result<InodeValFilePollGuard, Errno> {
178    Ok(match fd {
179        __WASI_STDERR_FILENO => WasiInodes::stderr(&state.fs.fd_map)
180            .map(|g| g.into_poll_guard(fd, peb, s))
181            .map_err(fs_error_into_wasi_err)?,
182        __WASI_STDOUT_FILENO => WasiInodes::stdout(&state.fs.fd_map)
183            .map(|g| g.into_poll_guard(fd, peb, s))
184            .map_err(fs_error_into_wasi_err)?,
185        _ => {
186            let fd_entry = state.fs.get_fd(fd)?;
187            let requires_access = match s.type_ {
188                Eventtype::FdRead => Rights::FD_READ,
189                Eventtype::FdWrite => Rights::FD_WRITE,
190                _ => Rights::empty(),
191            };
192
193            if !(fd_entry.inner.rights.contains(Rights::POLL_FD_READWRITE)
194                && fd_entry.inner.rights.contains(requires_access))
195            {
196                return Err(Errno::Access);
197            }
198            let inode = fd_entry.inode;
199
200            {
201                let guard = inode.read();
202                if let Some(guard) =
203                    crate::fs::InodeValFilePollGuard::new(fd, peb, s, guard.deref())
204                {
205                    guard
206                } else {
207                    return Err(Errno::Badf);
208                }
209            }
210        }
211    })
212}
213
214/// ### `poll_oneoff()`
215/// Concurrently poll for a set of events
216///
217/// Inputs:
218/// - `const __wasi_subscription_t *in`
219///   The events to subscribe to
220/// - `__wasi_event_t *out`
221///   The events that have occured
222/// - `u32 nsubscriptions`
223///   The number of subscriptions and the number of events
224///
225/// Output:
226/// - `u32 nevents`
227///   The number of events seen
228pub(crate) fn poll_oneoff_internal<'a, M: MemorySize, After>(
229    mut ctx: FunctionEnvMut<'a, WasiEnv>,
230    mut subs: Vec<(Option<WasiFd>, PollEventSet, Subscription)>,
231    process_events: After,
232) -> Result<Errno, WasiError>
233where
234    After: FnOnce(&FunctionEnvMut<'a, WasiEnv>, Vec<Event>) -> Errno,
235{
236    wasi_try_ok!(WasiEnv::process_signals_and_exit(&mut ctx)?);
237
238    let pid = ctx.data().pid();
239    let tid = ctx.data().tid();
240    let subs_len = subs.len();
241
242    // Determine if we are in silent polling mode
243    let mut env = ctx.data();
244    let state = ctx.data().state.deref();
245    let memory = unsafe { env.memory_view(&ctx) };
246
247    // These are used when we capture what clocks (timeouts) are being
248    // subscribed too
249    let clock_cnt = subs
250        .iter()
251        .filter(|a| a.2.type_ == Eventtype::Clock)
252        .count();
253    let mut clock_subs: Vec<(SubscriptionClock, u64)> = Vec::with_capacity(subs.len());
254    let mut time_to_sleep = Duration::MAX;
255
256    // First we extract all the subscriptions into an array so that they
257    // can be processed
258    let mut env = ctx.data();
259    let state = ctx.data().state.deref();
260    let mut memory = unsafe { env.memory_view(&ctx) };
261    for (fd, peb, s) in subs.iter_mut() {
262        let fd = match s.type_ {
263            Eventtype::FdRead => {
264                let file_descriptor = unsafe { s.data.fd_readwrite.file_descriptor };
265                *fd = Some(file_descriptor);
266                *peb |= (PollEvent::PollIn as PollEventSet);
267                file_descriptor
268            }
269            Eventtype::FdWrite => {
270                let file_descriptor = unsafe { s.data.fd_readwrite.file_descriptor };
271                *fd = Some(file_descriptor);
272                *peb |= (PollEvent::PollOut as PollEventSet);
273                file_descriptor
274            }
275            Eventtype::Clock => {
276                let clock_info = unsafe { s.data.clock };
277                if clock_info.clock_id == Clockid::Realtime
278                    || clock_info.clock_id == Clockid::Monotonic
279                {
280                    // Ignore duplicates
281                    if clock_subs
282                        .iter()
283                        .any(|c| c.0.clock_id == clock_info.clock_id && c.1 == s.userdata)
284                    {
285                        continue;
286                    }
287
288                    // If the timeout duration is zero then this is an immediate check rather than
289                    // a sleep itself
290                    if clock_info.timeout == 0 {
291                        time_to_sleep = Duration::MAX;
292                    } else if clock_info.timeout == 1 {
293                        time_to_sleep = Duration::ZERO;
294                        clock_subs.push((clock_info, s.userdata));
295                    } else {
296                        // if the timeout is specified as an absolute time in the future,
297                        // we should calculate the duration we need to sleep
298                        time_to_sleep = if clock_info
299                            .flags
300                            .contains(Subclockflags::SUBSCRIPTION_CLOCK_ABSTIME)
301                        {
302                            let now = wasi_try_ok!(platform_clock_time_get(
303                                Snapshot0Clockid::Monotonic,
304                                1
305                            )) as u64;
306
307                            if clock_info.timeout <= now {
308                                Duration::ZERO
309                            } else {
310                                Duration::from_nanos(clock_info.timeout) - Duration::from_nanos(now)
311                            }
312                        } else {
313                            // if the timeout is not absolute, just use it as duration
314                            Duration::from_nanos(clock_info.timeout)
315                        };
316
317                        clock_subs.push((clock_info, s.userdata));
318                    }
319                    continue;
320                } else {
321                    error!("polling not implemented for these clocks yet");
322                    return Ok(Errno::Inval);
323                }
324            }
325            _ => {
326                continue;
327            }
328        };
329    }
330
331    let mut events_seen: u32 = 0;
332
333    let batch = {
334        // Build the batch of things we are going to poll
335        let state = ctx.data().state.clone();
336        let tasks = ctx.data().tasks().clone();
337        let mut guards = {
338            // We start by building a list of files we are going to poll
339            // and open a read lock on them all
340            let mut fd_guards = Vec::with_capacity(subs.len());
341
342            #[allow(clippy::significant_drop_in_scrutinee)]
343            for (fd, peb, s) in subs {
344                if let Some(fd) = fd {
345                    let wasi_file_ref = wasi_try_ok!(poll_fd_guard(&state, peb, fd, s));
346                    fd_guards.push(wasi_file_ref);
347                }
348            }
349
350            if fd_guards.len() > 10 {
351                let small_list: Vec<_> = fd_guards.iter().take(10).collect();
352                tracing::Span::current().record("fd_guards", format!("{small_list:?}..."));
353            } else {
354                tracing::Span::current().record("fd_guards", format!("{fd_guards:?}"));
355            }
356
357            fd_guards
358        };
359
360        // Block polling the file descriptors
361        PollBatch::new(pid, tid, guards)
362    };
363
364    // If the time is infinite then we omit the time_to_sleep parameter
365    let timeout = match time_to_sleep {
366        Duration::ZERO => {
367            Span::current().record("timeout_ns", "nonblocking");
368            Some(Duration::ZERO)
369        }
370        Duration::MAX => {
371            Span::current().record("timeout_ns", "infinite");
372            None
373        }
374        time => {
375            Span::current().record("timeout_ns", time.as_millis());
376            Some(time)
377        }
378    };
379
380    // Function to process a timeout
381    let process_timeout = {
382        let clock_subs = clock_subs.clone();
383        |ctx: &FunctionEnvMut<'a, WasiEnv>| {
384            // The timeout has triggered so lets add that event
385            if clock_subs.is_empty() {
386                tracing::warn!("triggered_timeout (without any clock subscriptions)");
387            }
388            let mut evts = Vec::new();
389            for (clock_info, userdata) in clock_subs {
390                let evt = Event {
391                    userdata,
392                    error: Errno::Success,
393                    type_: Eventtype::Clock,
394                    u: EventUnion { clock: 0 },
395                };
396                Span::current().record(
397                    "seen",
398                    format!(
399                        "clock(id={},userdata={})",
400                        clock_info.clock_id as u32, evt.userdata
401                    ),
402                );
403                evts.push(evt);
404            }
405            evts
406        }
407    };
408
409    #[cfg(feature = "sys")]
410    if env.capabilities.threading.enable_blocking_sleep && subs_len == 1 {
411        // Here, `poll_oneoff` is merely in a sleeping state
412        // due to a single relative timer event. This particular scenario was
413        // added following experimental findings indicating that std::thread::sleep
414        // yields more consistent sleep durations, allowing wasmer to meet
415        // real-time demands with greater precision.
416        if let Some(timeout) = timeout {
417            std::thread::sleep(timeout);
418            process_events(&ctx, process_timeout(&ctx));
419            return Ok(Errno::Success);
420        }
421    }
422
423    let tasks = env.tasks().clone();
424    let timeout = async move {
425        if let Some(timeout) = timeout {
426            tasks.sleep_now(timeout).await;
427        } else {
428            InfiniteSleep::default().await
429        }
430    };
431
432    // Build the trigger using the timeout
433    let trigger = async move {
434        tokio::select! {
435            res = batch => res,
436            _ = timeout => Err(Errno::Timedout)
437        }
438    };
439
440    // We replace the process events callback with another callback
441    // which will interpret the error codes
442    let process_events = {
443        let clock_subs = clock_subs.clone();
444        |ctx: &FunctionEnvMut<'a, WasiEnv>, events: Result<Vec<Event>, Errno>| {
445            // Process the result
446            match events {
447                Ok(evts) => {
448                    // If its a timeout then return an event for it
449                    if evts.len() == 1 {
450                        Span::current().record("seen", format!("{:?}", evts.first().unwrap()));
451                    } else {
452                        Span::current().record("seen", format!("trigger_cnt=({})", evts.len()));
453                    }
454
455                    // Process the events
456                    process_events(ctx, evts)
457                }
458                Err(Errno::Timedout) => process_events(ctx, process_timeout(ctx)),
459                // If nonblocking the Errno::Again needs to be turned into an empty list
460                Err(Errno::Again) => process_events(ctx, Default::default()),
461                // Otherwise process the error
462                Err(err) => {
463                    tracing::warn!("failed to poll during deep sleep - {}", err);
464                    err
465                }
466            }
467        }
468    };
469
470    // If we are rewound then its time to process them
471    if let Some(events) = unsafe { handle_rewind::<M, Result<Vec<EventResult>, Errno>>(&mut ctx) } {
472        let events = events.map(|events| events.into_iter().map(EventResult::into_event).collect());
473        process_events(&ctx, events);
474        return Ok(Errno::Success);
475    }
476
477    // We use asyncify with a deep sleep to wait on new IO events
478    let res = __asyncify_with_deep_sleep::<M, Result<Vec<EventResult>, Errno>, _>(
479        ctx,
480        Box::pin(trigger),
481    )?;
482    if let AsyncifyAction::Finish(mut ctx, events) = res {
483        let events = events.map(|events| events.into_iter().map(EventResult::into_event).collect());
484        process_events(&ctx, events);
485    }
486    Ok(Errno::Success)
487}
wasmer_wasix/syscalls/wasi/poll_oneoff.rs

wasmer_wasix/syscalls/wasi/
poll_oneoff.rs
