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

wasmer_wasix/syscalls/wasi/
poll_oneoff.rs
