ferroui_linuxframebuffer/
epoll_dispatcher_impl.rs

1//! The dispatcher implementation of the platform (the port of
2//! `EpollDispatcherImpl.cs`): a loop that waits, with an event poll, on a
3//! pipe that other threads write to in order to wake it and on a timer
4//! descriptor armed for the next timer.
5
6use std::time::Duration;
7
8/// The value the timer descriptor is armed with for a wait of `wait_for`:
9/// seconds (at most 100, after which the loop looks again) and
10/// nanoseconds.
11///
12/// The reference computes the same from a time span. A wait of no time at
13/// all would disarm the descriptor (both fields zero) and the loop would
14/// sleep until something else woke it; it is one nanosecond here.
15pub(crate) fn timer_value(wait_for: Duration) -> (i64, i64) {
16    let tv_sec = wait_for.as_secs().min(100) as i64;
17    let tv_nsec = wait_for.subsec_nanos() as i64;
18    if tv_sec == 0 && tv_nsec == 0 {
19        return (0, 1);
20    }
21    (tv_sec, tv_nsec)
22}
23
24#[cfg(target_os = "linux")]
25pub use imp::EpollDispatcherImpl;
26
27#[cfg(target_os = "linux")]
28mod imp {
29    use super::timer_value;
30    use ferroui_base::platform::IManagedDispatcherInputProvider;
31    use ferroui_base::threading::{
32        CancellationToken, DispatcherImplEvent, IControlledDispatcherImpl, IDispatcherImpl,
33        IDispatcherImplWithExplicitBackgroundProcessing, IDispatcherImplWithPendingInput, IDispatcherSignal,
34    };
35    use std::cell::Cell;
36    use std::rc::Rc;
37    use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
38    use std::thread::{self, ThreadId};
39    use std::time::{Duration, Instant};
40
41    mod event_codes {
42        pub const TIMER: u64 = 1;
43        pub const SIGNAL: u64 = 2;
44    }
45
46    #[derive(Default)]
47    struct State {
48        signaled: bool,
49        wakeup_requested: bool,
50        background_processing_requested: bool,
51    }
52
53    /// The part of the dispatcher implementation other threads reach: the
54    /// flags under their lock and the writing end of the wake-up pipe.
55    struct Signal {
56        lock: Mutex<State>,
57        sigwrite: i32,
58    }
59
60    impl Signal {
61        fn lock(&self) -> MutexGuard<'_, State> {
62            self.lock.lock().unwrap_or_else(PoisonError::into_inner)
63        }
64
65        /// `Wakeup`, with the lock held by the caller.
66        fn wakeup_locked(&self, state: &mut State) {
67            if state.wakeup_requested {
68                return;
69            }
70            state.wakeup_requested = true;
71            let buf: i32 = 0;
72            // SAFETY: writes one byte of a local value to a descriptor this
73            // object owns for the life of the process. A full pipe (the
74            // descriptor does not block) means a wake-up is already pending.
75            unsafe {
76                libc::write(self.sigwrite, (&buf as *const i32).cast(), 1);
77            }
78        }
79
80        fn wakeup(&self) {
81            let mut state = self.lock();
82            self.wakeup_locked(&mut state);
83        }
84    }
85
86    impl IDispatcherSignal for Signal {
87        fn signal(&self) {
88            let mut state = self.lock();
89            if state.signaled {
90                return;
91            }
92            state.signaled = true;
93            self.wakeup_locked(&mut state);
94        }
95    }
96
97    /// The dispatcher implementation of the framebuffer platform.
98    pub struct EpollDispatcherImpl {
99        input_provider: Box<dyn IManagedDispatcherInputProvider>,
100        main_thread: ThreadId,
101        sigread: i32,
102        timerfd: i32,
103        signal: Arc<Signal>,
104        next_timer: Cell<Option<Duration>>,
105        epoll: i32,
106        clock: Instant,
107        signaled: DispatcherImplEvent,
108        timer: DispatcherImplEvent,
109        ready_for_background_processing: DispatcherImplEvent,
110    }
111
112    impl EpollDispatcherImpl {
113        /// # Panics
114        /// Panics when the event poll cannot be created or a descriptor
115        /// cannot be attached to it.
116        pub fn new(input_provider: Box<dyn IManagedDispatcherInputProvider>) -> Rc<Self> {
117            // SAFETY: plain system calls; the event structures and the
118            // array of descriptors are local and valid for the calls.
119            let (epoll, sigread, sigwrite, timerfd) = unsafe {
120                let epoll = libc::epoll_create1(libc::EPOLL_CLOEXEC);
121                if epoll == -1 {
122                    panic!("epoll_create1 failed");
123                }
124
125                let mut fds = [0i32; 2];
126                libc::pipe2(fds.as_mut_ptr(), libc::O_NONBLOCK | libc::O_CLOEXEC);
127                let (sigread, sigwrite) = (fds[0], fds[1]);
128
129                let mut ev = libc::epoll_event { events: libc::EPOLLIN as u32, u64: event_codes::SIGNAL };
130                if libc::epoll_ctl(epoll, libc::EPOLL_CTL_ADD, sigread, &mut ev) == -1 {
131                    panic!("Unable to attach signal pipe to epoll");
132                }
133
134                let timerfd = libc::timerfd_create(libc::CLOCK_MONOTONIC, libc::O_NONBLOCK | libc::O_CLOEXEC);
135                ev.u64 = event_codes::TIMER;
136                if libc::epoll_ctl(epoll, libc::EPOLL_CTL_ADD, timerfd, &mut ev) == -1 {
137                    panic!("Unable to attach timer fd to epoll");
138                }
139                (epoll, sigread, sigwrite, timerfd)
140            };
141
142            Rc::new(Self {
143                input_provider,
144                main_thread: thread::current().id(),
145                sigread,
146                timerfd,
147                signal: Arc::new(Signal { lock: Mutex::new(State::default()), sigwrite }),
148                next_timer: Cell::new(None),
149                epoll,
150                clock: Instant::now(),
151                signaled: DispatcherImplEvent::new(),
152                timer: DispatcherImplEvent::new(),
153                ready_for_background_processing: DispatcherImplEvent::new(),
154            })
155        }
156
157        fn check_signaled(&self) -> bool {
158            {
159                let mut state = self.signal.lock();
160                if !state.signaled {
161                    return false;
162                }
163                state.signaled = false;
164            }
165
166            self.signaled.raise();
167            true
168        }
169
170        fn set_timer(&self, tv_sec: i64, tv_nsec: i64) {
171            // SAFETY: an all-zero `itimerspec` is valid; the structure is
172            // local and the old value is not asked for.
173            unsafe {
174                let mut timer: libc::itimerspec = std::mem::zeroed();
175                timer.it_value.tv_sec = tv_sec as _;
176                timer.it_value.tv_nsec = tv_nsec as _;
177                libc::timerfd_settime(self.timerfd, 0, &timer, std::ptr::null_mut());
178            }
179        }
180
181        fn drain(fd: i32) {
182            let mut buf: i64 = 0;
183            // SAFETY: reads at most eight bytes into a local value from a
184            // descriptor that does not block.
185            while unsafe { libc::read(fd, (&mut buf as *mut i64).cast(), 8) } > 0 {}
186        }
187    }
188
189    impl IDispatcherImpl for EpollDispatcherImpl {
190        fn current_thread_is_loop_thread(&self) -> bool {
191            thread::current().id() == self.main_thread
192        }
193
194        fn signal(&self) {
195            self.signal.signal();
196        }
197
198        fn signal_handle(&self) -> Arc<dyn IDispatcherSignal> {
199            self.signal.clone()
200        }
201
202        fn signaled(&self) -> &DispatcherImplEvent {
203            &self.signaled
204        }
205
206        fn timer(&self) -> &DispatcherImplEvent {
207            &self.timer
208        }
209
210        fn now(&self) -> i64 {
211            self.clock.elapsed().as_millis() as i64
212        }
213
214        fn update_timer(&self, due_time_in_ms: Option<i64>) {
215            self.next_timer.set(due_time_in_ms.map(|ms| Duration::from_millis(ms.max(0) as u64)));
216            if due_time_in_ms.is_some() {
217                self.signal.wakeup();
218            }
219        }
220
221        fn as_pending_input(&self) -> Option<&dyn IDispatcherImplWithPendingInput> {
222            Some(self)
223        }
224
225        fn as_explicit_background_processing(&self) -> Option<&dyn IDispatcherImplWithExplicitBackgroundProcessing> {
226            Some(self)
227        }
228
229        fn as_controlled(&self) -> Option<&dyn IControlledDispatcherImpl> {
230            Some(self)
231        }
232    }
233
234    impl IDispatcherImplWithPendingInput for EpollDispatcherImpl {
235        fn can_query_pending_input(&self) -> bool {
236            true
237        }
238
239        fn has_pending_input(&self) -> bool {
240            self.input_provider.has_input()
241        }
242    }
243
244    impl IDispatcherImplWithExplicitBackgroundProcessing for EpollDispatcherImpl {
245        fn ready_for_background_processing(&self) -> &DispatcherImplEvent {
246            &self.ready_for_background_processing
247        }
248
249        fn request_background_processing(&self) {
250            let mut state = self.signal.lock();
251            state.background_processing_requested = true;
252            self.signal.wakeup_locked(&mut state);
253        }
254    }
255
256    impl IControlledDispatcherImpl for EpollDispatcherImpl {
257        fn run_loop(&self, cancellation_token: CancellationToken) {
258            while !cancellation_token.is_cancellation_requested() {
259                let now = self.clock.elapsed();
260                if self.next_timer.get().is_some_and(|next_timer| now > next_timer) {
261                    self.timer.raise();
262                    continue;
263                }
264
265                if self.check_signaled() {
266                    continue;
267                }
268
269                if self.input_provider.has_input() {
270                    self.input_provider.dispatch_next_input_event();
271                    continue;
272                }
273
274                let trigger_background_processing =
275                    std::mem::take(&mut self.signal.lock().background_processing_requested);
276
277                if trigger_background_processing {
278                    self.ready_for_background_processing.raise();
279                    continue;
280                }
281
282                match self.next_timer.get() {
283                    Some(next_timer) => {
284                        let Some(wait_for) = next_timer.checked_sub(now) else {
285                            continue;
286                        };
287                        let (tv_sec, tv_nsec) = timer_value(wait_for);
288                        self.set_timer(tv_sec, tv_nsec);
289                    }
290                    None => self.set_timer(0, 0),
291                }
292
293                // SAFETY: `ev` is a valid place for one event.
294                unsafe {
295                    let mut ev = libc::epoll_event { events: 0, u64: 0 };
296                    libc::epoll_wait(self.epoll, &mut ev, 1, -1);
297                }
298
299                // Drain the signaled pipe
300                Self::drain(self.sigread);
301
302                // Drain timer fd
303                Self::drain(self.timerfd);
304
305                self.signal.lock().wakeup_requested = false;
306            }
307        }
308    }
309
310    #[cfg(test)]
311    mod tests {
312        // Not from the reference, which has no tests for this file. They
313        // run the loop on a real event poll, on Linux only.
314        use super::*;
315        use ferroui_base::threading::CancellationTokenSource;
316        use std::cell::RefCell;
317        use std::collections::VecDeque;
318
319        #[derive(Default)]
320        struct Input {
321            events: Rc<RefCell<VecDeque<Box<dyn FnOnce()>>>>,
322        }
323
324        impl IManagedDispatcherInputProvider for Input {
325            fn has_input(&self) -> bool {
326                !self.events.borrow().is_empty()
327            }
328
329            fn dispatch_next_input_event(&self) {
330                let next = self.events.borrow_mut().pop_front();
331                if let Some(next) = next {
332                    next();
333                }
334            }
335        }
336
337        #[test]
338        fn the_loop_raises_the_timer_the_signal_input_and_background_processing_and_ends_when_cancelled() {
339            let input = Input::default();
340            let events = input.events.clone();
341            let dispatcher = EpollDispatcherImpl::new(Box::new(input));
342            let log = Rc::new(RefCell::new(Vec::<&'static str>::new()));
343            let cts = Arc::new(CancellationTokenSource::new());
344
345            {
346                let log = log.clone();
347                let weak = Rc::downgrade(&dispatcher);
348                dispatcher.timer().add(Rc::new(move |()| {
349                    log.borrow_mut().push("timer");
350                    let dispatcher = weak.upgrade().unwrap();
351                    dispatcher.update_timer(None);
352                    dispatcher.request_background_processing();
353                }));
354            }
355            {
356                let log = log.clone();
357                dispatcher.signaled().add(Rc::new(move |()| log.borrow_mut().push("signaled")));
358            }
359            {
360                let log = log.clone();
361                let cts = cts.clone();
362                dispatcher.ready_for_background_processing().add(Rc::new(move |()| {
363                    log.borrow_mut().push("background");
364                    cts.cancel();
365                }));
366            }
367            {
368                let log = log.clone();
369                events.borrow_mut().push_back(Box::new(move || log.borrow_mut().push("input")));
370            }
371
372            // The signal comes from another thread, as a post does.
373            let handle = dispatcher.signal_handle();
374            thread::spawn(move || handle.signal()).join().unwrap();
375            dispatcher.update_timer(Some(dispatcher.now() + 30));
376            assert!(dispatcher.has_pending_input());
377
378            let started = Instant::now();
379            dispatcher.run_loop(cts.token());
380
381            assert_eq!(*log.borrow(), ["signaled", "input", "timer", "background"]);
382            assert!(started.elapsed() >= Duration::from_millis(25), "the timer was raised early");
383            assert!(started.elapsed() < Duration::from_secs(5));
384            assert!(!dispatcher.has_pending_input());
385        }
386
387        #[test]
388        fn a_cancelled_token_ends_a_waiting_loop_when_the_loop_is_woken() {
389            let dispatcher = EpollDispatcherImpl::new(Box::new(Input::default()));
390            let cts = Arc::new(CancellationTokenSource::new());
391            let handle = dispatcher.signal_handle();
392            let worker = {
393                let cts = cts.clone();
394                thread::spawn(move || {
395                    thread::sleep(Duration::from_millis(30));
396                    cts.cancel();
397                    handle.signal();
398                })
399            };
400            dispatcher.run_loop(cts.token());
401            worker.join().unwrap();
402            assert!(dispatcher.current_thread_is_loop_thread());
403        }
404    }
405}
406
407#[cfg(test)]
408mod tests {
409    // Not from the reference, which has no tests for this file.
410    use super::*;
411
412    #[test]
413    fn the_timer_is_armed_for_the_wait_with_the_seconds_capped() {
414        assert_eq!(timer_value(Duration::from_millis(1_500)), (1, 500_000_000));
415        assert_eq!(timer_value(Duration::from_micros(250)), (0, 250_000));
416        assert_eq!(timer_value(Duration::from_secs(3_600)), (100, 0));
417        // Both fields zero would disarm the timer.
418        assert_eq!(timer_value(Duration::ZERO), (0, 1));
419    }
420}
ferroui_linuxframebuffer/epoll_dispatcher_impl.rs

ferroui_linuxframebuffer/
epoll_dispatcher_impl.rs
