Use timerfd on linux to drive Timer

kchibisov · vberger · commit 0bcc2b8b0046 · 2022-02-18T10:58:47.000+01:00
Previously Timer was using thread for all platforms, however platforms
tend to provide efficient API to create timers. This commit makes
`Timer` use timerfd system interface on Linux.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -14,6 +14,7 @@
   associated `Error` type on the `EventSource` trait.
 - **Breaking:** Many API functions now use Calloop's own error type (`calloop::Error`) instead of
   `std::io::Error` as the error variants of their returned results.
+- On Linux `Timer<T>` is now driven by `timerfd`.
 
 ## 0.9.2 -- 2021-12-27
 
diff --git a/src/sources/timer/mod.rs b/src/sources/timer/mod.rs
@@ -9,15 +9,39 @@
 
 use std::cell::RefCell;
 use std::collections::BinaryHeap;
-use std::sync::{
-    atomic::{AtomicBool, Ordering},
-    Arc, Mutex,
-};
+use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
 
-use super::ping::{make_ping, PingError, PingSource};
 use crate::{EventSource, Poll, PostAction, Readiness, Token, TokenFactory};
 
+#[cfg(target_os = "linux")]
+mod timerfd;
+#[cfg(target_os = "linux")]
+use timerfd::{TimerScheduler, TimerSource};
+
+#[cfg(any(
+    target_os = "dragonfly",
+    target_os = "freebsd",
+    target_os = "netbsd",
+    target_os = "openbsd",
+    target_os = "macos"
+))]
+mod threaded;
+
+#[cfg(any(
+    target_os = "dragonfly",
+    target_os = "freebsd",
+    target_os = "netbsd",
+    target_os = "openbsd",
+    target_os = "macos"
+))]
+use threaded::{TimerScheduler, TimerSource};
+
+/// An error arising from processing events for a timer.
+#[derive(thiserror::Error, Debug)]
+#[error(transparent)]
+pub struct TimerError(Box<dyn std::error::Error + Sync + Send>);
+
 /// A Timer event source
 ///
 /// It generates events of type `(T, TimerHandle<T>)`, providing you
@@ -122,23 +146,21 @@ impl<T> EventSource for Timer<T> {
             inner: self.inner.clone(),
         };
         let inner = &self.inner;
-        self.source
-            .process_events(readiness, token, |(), &mut ()| {
-                loop {
-                    let next_expired: Option<T> = {
-                        let mut guard = inner.lock().unwrap();
-                        guard.next_expired()
-                    };
-                    if let Some(val) = next_expired {
-                        callback(val, &mut handle);
-                    } else {
-                        break;
-                    }
+        self.source.process_events(readiness, token, |(), &mut ()| {
+            loop {
+                let next_expired: Option<T> = {
+                    let mut guard = inner.lock().unwrap();
+                    guard.next_expired()
+                };
+                if let Some(val) = next_expired {
+                    callback(val, &mut handle);
+                } else {
+                    break;
                 }
-                // now compute the next timeout and signal if necessary
-                inner.lock().unwrap().reschedule();
-            })
-            .map_err(TimerError)
+            }
+            // now compute the next timeout and signal if necessary
+            inner.lock().unwrap().reschedule();
+        })
     }
 
     fn register(&mut self, poll: &mut Poll, token_factory: &mut TokenFactory) -> crate::Result<()> {
@@ -272,97 +294,6 @@ impl<T> std::cmp::PartialEq for TimeoutData<T> {
 
 impl<T> std::cmp::Eq for TimeoutData<T> {}
 
-/*
- * Scheduling
- */
-
-#[derive(Debug)]
-struct TimerScheduler {
-    current_deadline: Arc<Mutex<Option<Instant>>>,
-    kill_switch: Arc<AtomicBool>,
-    thread: Option<std::thread::JoinHandle<()>>,
-}
-
-type TimerSource = PingSource;
-
-impl TimerScheduler {
-    fn new() -> crate::Result<(TimerScheduler, TimerSource)> {
-        let current_deadline = Arc::new(Mutex::new(None::<Instant>));
-        let thread_deadline = current_deadline.clone();
-
-        let kill_switch = Arc::new(AtomicBool::new(false));
-        let thread_kill = kill_switch.clone();
-
-        let (ping, ping_source) = make_ping()?;
-
-        let thread = std::thread::Builder::new()
-            .name("calloop timer".into())
-            .spawn(move || loop {
-                // stop if requested
-                if thread_kill.load(Ordering::Acquire) {
-                    return;
-                }
-                // otherwise check the timeout
-                let opt_deadline: Option<Instant> = {
-                    // subscope to ensure the mutex does not remain locked while the thread is parked
-                    let guard = thread_deadline.lock().unwrap();
-                    *guard
-                };
-                if let Some(deadline) = opt_deadline {
-                    if let Some(remaining) = deadline.checked_duration_since(Instant::now()) {
-                        // it is not yet expired, go to sleep until it
-                        std::thread::park_timeout(remaining);
-                    } else {
-                        // it is expired, wake the event loop and go to sleep
-                        ping.ping();
-                        std::thread::park();
-                    }
-                } else {
-                    // there is none, got to sleep
-                    std::thread::park();
-                }
-            })?;
-
-        let scheduler = TimerScheduler {
-            current_deadline,
-            kill_switch,
-            thread: Some(thread),
-        };
-        Ok((scheduler, ping_source))
-    }
-
-    fn reschedule(&mut self, new_deadline: Instant) {
-        let mut deadline_guard = self.current_deadline.lock().unwrap();
-        if let Some(current_deadline) = *deadline_guard {
-            if new_deadline < current_deadline || current_deadline <= Instant::now() {
-                *deadline_guard = Some(new_deadline);
-                self.thread.as_ref().unwrap().thread().unpark();
-            }
-        } else {
-            *deadline_guard = Some(new_deadline);
-            self.thread.as_ref().unwrap().thread().unpark();
-        }
-    }
-
-    fn deschedule(&mut self) {
-        *(self.current_deadline.lock().unwrap()) = None;
-    }
-}
-
-impl Drop for TimerScheduler {
-    fn drop(&mut self) {
-        self.kill_switch.store(true, Ordering::Release);
-        let thread = self.thread.take().unwrap();
-        thread.thread().unpark();
-        let _ = thread.join();
-    }
-}
-
-/// An error arising from processing events for a timer.
-#[derive(thiserror::Error, Debug)]
-#[error(transparent)]
-pub struct TimerError(PingError);
-
 /*
  * Tests
  */
diff --git a/src/sources/timer/threaded.rs b/src/sources/timer/threaded.rs
@@ -0,0 +1,141 @@
+//! Timer scheduler which is using thread to schedule timers.
+
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex};
+use std::time::Instant;
+
+use crate::ping::{make_ping, Ping, PingSource};
+use crate::{EventSource, Poll, PostAction, Readiness, Token, TokenFactory};
+
+use super::TimerError;
+
+#[derive(Debug)]
+pub struct TimerSource {
+    source: PingSource,
+}
+
+impl TimerSource {
+    fn new() -> std::io::Result<(Ping, Self)> {
+        let (ping, source) = make_ping()?;
+        Ok((ping, Self { source }))
+    }
+}
+
+impl EventSource for TimerSource {
+    type Event = ();
+    type Metadata = ();
+    type Ret = ();
+    type Error = TimerError;
+
+    fn process_events<C>(
+        &mut self,
+        readiness: Readiness,
+        token: Token,
+        mut callback: C,
+    ) -> Result<PostAction, Self::Error>
+    where
+        C: FnMut(Self::Event, &mut Self::Metadata) -> Self::Ret,
+    {
+        self.source
+            .process_events(readiness, token, |_, &mut _| {
+                callback((), &mut ());
+            })
+            .map_err(|err| TimerError(err.into()))
+    }
+
+    fn register(&mut self, poll: &mut Poll, token_factory: &mut TokenFactory) -> crate::Result<()> {
+        self.source.register(poll, token_factory)
+    }
+
+    fn reregister(
+        &mut self,
+        poll: &mut Poll,
+        token_factory: &mut TokenFactory,
+    ) -> crate::Result<()> {
+        self.source.reregister(poll, token_factory)
+    }
+
+    fn unregister(&mut self, poll: &mut Poll) -> crate::Result<()> {
+        self.source.unregister(poll)
+    }
+}
+
+#[derive(Debug)]
+pub struct TimerScheduler {
+    current_deadline: Arc<Mutex<Option<Instant>>>,
+    kill_switch: Arc<AtomicBool>,
+    thread: Option<std::thread::JoinHandle<()>>,
+}
+
+impl TimerScheduler {
+    pub fn new() -> crate::Result<(TimerScheduler, TimerSource)> {
+        let current_deadline = Arc::new(Mutex::new(None::<Instant>));
+        let thread_deadline = current_deadline.clone();
+
+        let kill_switch = Arc::new(AtomicBool::new(false));
+        let thread_kill = kill_switch.clone();
+
+        let (ping, source) = TimerSource::new()?;
+
+        let thread = std::thread::Builder::new()
+            .name("calloop timer".into())
+            .spawn(move || loop {
+                // stop if requested
+                if thread_kill.load(Ordering::Acquire) {
+                    return;
+                }
+                // otherwise check the timeout
+                let opt_deadline: Option<Instant> = {
+                    // subscope to ensure the mutex does not remain locked while the thread is parked
+                    let guard = thread_deadline.lock().unwrap();
+                    *guard
+                };
+                if let Some(deadline) = opt_deadline {
+                    if let Some(remaining) = deadline.checked_duration_since(Instant::now()) {
+                        // it is not yet expired, go to sleep until it
+                        std::thread::park_timeout(remaining);
+                    } else {
+                        // it is expired, wake the event loop and go to sleep
+                        ping.ping();
+                        std::thread::park();
+                    }
+                } else {
+                    // there is none, got to sleep
+                    std::thread::park();
+                }
+            })?;
+
+        let scheduler = TimerScheduler {
+            current_deadline,
+            kill_switch,
+            thread: Some(thread),
+        };
+        Ok((scheduler, source))
+    }
+
+    pub fn reschedule(&mut self, new_deadline: Instant) {
+        let mut deadline_guard = self.current_deadline.lock().unwrap();
+        if let Some(current_deadline) = *deadline_guard {
+            if new_deadline < current_deadline || current_deadline <= Instant::now() {
+                *deadline_guard = Some(new_deadline);
+                self.thread.as_ref().unwrap().thread().unpark();
+            }
+        } else {
+            *deadline_guard = Some(new_deadline);
+            self.thread.as_ref().unwrap().thread().unpark();
+        }
+    }
+
+    pub fn deschedule(&mut self) {
+        *(self.current_deadline.lock().unwrap()) = None;
+    }
+}
+
+impl Drop for TimerScheduler {
+    fn drop(&mut self) {
+        self.kill_switch.store(true, Ordering::Release);
+        let thread = self.thread.take().unwrap();
+        thread.thread().unpark();
+        let _ = thread.join();
+    }
+}
diff --git a/src/sources/timer/timerfd.rs b/src/sources/timer/timerfd.rs
