fuchsia_async/handle/zircon/
on_interrupt.rs

1// Copyright 2025 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use futures::Stream;
6use std::pin::Pin;
7use std::sync::atomic::{AtomicUsize, Ordering};
8use std::task::Poll;
9use zx::{
10    AsHandleRef, BootTimeline, Instant, Interrupt, InterruptKind, RealInterruptKind, Timeline, sys,
11};
12
13use crate::runtime::{EHandle, PacketReceiver, RawReceiverRegistration};
14use futures::task::{AtomicWaker, Context};
15
16struct OnInterruptReceiver {
17    maybe_timestamp: AtomicUsize,
18    task: AtomicWaker,
19}
20
21impl OnInterruptReceiver {
22    fn get_interrupt(&self, cx: &mut Context<'_>) -> Poll<sys::zx_time_t> {
23        let mut timestamp = self.maybe_timestamp.swap(0, Ordering::Relaxed);
24        if timestamp == 0 {
25            // The interrupt did not fire -- register to receive a wakeup when it does.
26            self.task.register(cx.waker());
27            // Check again for a timestamp after registering for a wakeup in case it fired
28            // between registering and the initial load.
29            // NOTE: We might be able to use a weaker ordering because we use AtomicWaker.
30            timestamp = self.maybe_timestamp.swap(0, Ordering::SeqCst);
31        }
32        if timestamp == 0 { Poll::Pending } else { Poll::Ready(timestamp as i64) }
33    }
34
35    fn set_timestamp(&self, timestamp: sys::zx_time_t) {
36        self.maybe_timestamp.store(timestamp as usize, Ordering::SeqCst);
37        self.task.wake();
38    }
39}
40
41impl PacketReceiver for OnInterruptReceiver {
42    fn receive_packet(&self, packet: zx::Packet) {
43        let zx::PacketContents::Interrupt(interrupt) = packet.contents() else {
44            return;
45        };
46        self.set_timestamp(interrupt.timestamp());
47    }
48}
49
50pin_project_lite::pin_project! {
51/// A stream that returns each time an interrupt fires.
52#[must_use = "future streams do nothing unless polled"]
53pub struct OnInterrupt<K: InterruptKind = RealInterruptKind, T: Timeline = BootTimeline> {
54    interrupt: Interrupt<K, T>,
55    #[pin]
56    registration: RawReceiverRegistration<OnInterruptReceiver>,
57}
58
59impl<K: InterruptKind, T: Timeline> PinnedDrop for OnInterrupt<K, T> {
60        fn drop(mut this: Pin<&mut Self>) {
61        this.unregister()
62    }
63}
64
65}
66
67impl<K: InterruptKind, T: Timeline> OnInterrupt<K, T> {
68    /// Creates a new OnInterrupt object which will notifications when `interrupt` fires.
69    /// NOTE: This will only work on a port that was created with the BIND_TO_INTERRUPT option.
70    pub fn new(interrupt: Interrupt<K, T>) -> Self {
71        Self {
72            interrupt,
73            registration: RawReceiverRegistration::new(OnInterruptReceiver {
74                maybe_timestamp: AtomicUsize::new(0),
75                task: AtomicWaker::new(),
76            }),
77        }
78    }
79
80    fn register(
81        mut registration: Pin<&mut RawReceiverRegistration<OnInterruptReceiver>>,
82        interrupt: &Interrupt<K, T>,
83        cx: Option<&mut Context<'_>>,
84    ) -> Result<(), zx::Status> {
85        registration.as_mut().register(EHandle::local());
86
87        // If a context has been supplied, we must register it now before calling
88        // `bind_port` below to avoid races.
89        if let Some(cx) = cx {
90            registration.receiver().task.register(cx.waker());
91        }
92
93        interrupt.bind_port(registration.port().unwrap(), registration.key().unwrap())?;
94
95        Ok(())
96    }
97
98    fn unregister(self: Pin<&mut Self>) {
99        let mut this = self.project();
100        if let Some((ehandle, key)) = this.registration.as_mut().unregister() {
101            let _ = ehandle.port().cancel(key);
102        }
103    }
104}
105
106impl<K: InterruptKind, T: Timeline> AsHandleRef for OnInterrupt<K, T> {
107    fn as_handle_ref(&self) -> zx::HandleRef<'_> {
108        self.interrupt.as_handle_ref()
109    }
110}
111
112impl<K: InterruptKind, T: Timeline> AsRef<Interrupt<K, T>> for OnInterrupt<K, T> {
113    fn as_ref(&self) -> &Interrupt<K, T> {
114        &self.interrupt
115    }
116}
117
118impl<K: InterruptKind, T: Timeline> Stream for OnInterrupt<K, T> {
119    type Item = Result<Instant<T>, zx::Status>;
120    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
121        if !self.registration.is_registered() {
122            let mut this = self.project();
123            Self::register(this.registration.as_mut(), this.interrupt, Some(cx))?;
124            Poll::Pending
125        } else {
126            match self.registration.receiver().get_interrupt(cx) {
127                Poll::Ready(timestamp) => {
128                    Poll::Ready(Some(Ok(Instant::<T>::from_nanos(timestamp))))
129                }
130                Poll::Pending => Poll::Pending,
131            }
132        }
133    }
134}
135
136#[cfg(test)]
137mod test {
138    use super::*;
139    use futures::future::pending;
140
141    #[test]
142    fn wait_for_event() -> Result<(), zx::Status> {
143        let port = zx::Port::create_with_opts(zx::PortOptions::BIND_TO_INTERRUPT);
144        let mut exec = crate::TestExecutor::builder().port(port).build();
145        let mut deliver_events =
146            || assert!(exec.run_until_stalled(&mut pending::<()>()).is_pending());
147
148        let irq = zx::VirtualInterrupt::create_virtual()?;
149        let mut irq = std::pin::pin!(OnInterrupt::new(irq));
150        let (waker, waker_count) = futures_test::task::new_count_waker();
151        let cx = &mut std::task::Context::from_waker(&waker);
152
153        // Check that `irq` is still pending before the interrupt has fired.
154        assert_eq!(irq.as_mut().poll_next(cx), Poll::Pending);
155        deliver_events();
156        assert_eq!(waker_count, 0);
157        assert_eq!(irq.as_mut().poll_next(cx), Poll::Pending);
158
159        // Trigger the interrupt and check that we receive the same timestamp.
160        let timestamp = zx::BootInstant::from_nanos(10);
161        irq.interrupt.trigger(timestamp)?;
162        deliver_events();
163        assert_eq!(waker_count, 1);
164        let expected: Result<_, zx::Status> = Ok(timestamp);
165        assert_eq!(irq.as_mut().poll_next(cx), Poll::Ready(Some(expected)));
166
167        // Check that we are polling pending now.
168        deliver_events();
169        assert_eq!(irq.as_mut().poll_next(cx), Poll::Pending);
170
171        // Signal a second time to check that the stream works.
172        irq.interrupt.ack()?;
173        let timestamp = zx::BootInstant::from_nanos(20);
174        irq.interrupt.trigger(timestamp)?;
175        deliver_events();
176        let expected: Result<_, zx::Status> = Ok(timestamp);
177        assert_eq!(irq.as_mut().poll_next(cx), Poll::Ready(Some(expected)));
178
179        Ok(())
180    }
181
182    #[test]
183    fn wait_for_event_monotonic() -> Result<(), zx::Status> {
184        let port = zx::Port::create_with_opts(zx::PortOptions::BIND_TO_INTERRUPT);
185        let mut exec = crate::TestExecutor::builder().port(port).build();
186        let mut deliver_events =
187            || assert!(exec.run_until_stalled(&mut pending::<()>()).is_pending());
188
189        let irq =
190            zx::Interrupt::<zx::VirtualInterruptKind, zx::MonotonicTimeline>::create_virtual()?;
191        let mut irq = std::pin::pin!(OnInterrupt::new(irq));
192        let (waker, waker_count) = futures_test::task::new_count_waker();
193        let cx = &mut std::task::Context::from_waker(&waker);
194
195        // Check that `irq` is still pending before the interrupt has fired.
196        assert_eq!(irq.as_mut().poll_next(cx), Poll::Pending);
197        deliver_events();
198        assert_eq!(waker_count, 0);
199        assert_eq!(irq.as_mut().poll_next(cx), Poll::Pending);
200
201        // Trigger the interrupt and check that we receive the same timestamp.
202        let timestamp = zx::MonotonicInstant::from_nanos(10);
203        irq.interrupt.trigger(timestamp)?;
204        deliver_events();
205        assert_eq!(waker_count, 1);
206        let expected: Result<_, zx::Status> = Ok(timestamp);
207        assert_eq!(irq.as_mut().poll_next(cx), Poll::Ready(Some(expected)));
208
209        // Check that we are polling pending now.
210        deliver_events();
211        assert_eq!(irq.as_mut().poll_next(cx), Poll::Pending);
212
213        // Signal a second time to check that the stream works.
214        irq.interrupt.ack()?;
215        let timestamp = zx::MonotonicInstant::from_nanos(20);
216        irq.interrupt.trigger(timestamp)?;
217        deliver_events();
218        let expected: Result<_, zx::Status> = Ok(timestamp);
219        assert_eq!(irq.as_mut().poll_next(cx), Poll::Ready(Some(expected)));
220
221        Ok(())
222    }
223
224    #[test]
225    fn incorrect_port() -> Result<(), zx::Status> {
226        let _exec = crate::TestExecutor::new();
227
228        let irq = zx::VirtualInterrupt::create_virtual()?;
229        let mut irq = std::pin::pin!(OnInterrupt::new(irq));
230        let (waker, _waker_count) = futures_test::task::new_count_waker();
231        let cx = &mut std::task::Context::from_waker(&waker);
232
233        // Polling the interrupt should cause an error.
234        assert_eq!(irq.as_mut().poll_next(cx), Poll::Ready(Some(Err(zx::Status::WRONG_TYPE))));
235
236        Ok(())
237    }
238}
fuchsia_async/handle/zircon/on_interrupt.rs

fuchsia_async/handle/zircon/
on_interrupt.rs
