Add brokered timerfd support to ulitebox

litebox/src/broker/mod.rs

@@ -18,6 +18,7 @@ use litebox_broker_protocol::socket::{
     SendFlags as BrokerSendFlags, ShutdownMode, SocketAddressV4, SocketConnectionStatus,
     SocketOutcome, SocketStatusResponse,
 };
+use litebox_broker_protocol::timer::TimerSpec;
 use litebox_broker_transport::channel::LocalCallChannel;
 
 use crate::event::{Events, polling::Pollee};
@@ -97,6 +98,26 @@ pub(crate) trait BrokerControl: Send + Sync {
         mode: EventConsumeMode,
     ) -> core::result::Result<ConsumeEventResponse, BrokerControlError>;
 
+    fn create_timer(&self, clock_id: i32)
+    -> core::result::Result<ObjectHandle, BrokerControlError>;
+
+    fn set_timer(
+        &self,
+        handle: ObjectHandle,
+        specification: TimerSpec,
+        flags: u32,
+    ) -> core::result::Result<(TimerSpec, ReadinessFlags), BrokerControlError>;
+
+    fn get_timer(
+        &self,
+        handle: ObjectHandle,
+    ) -> core::result::Result<TimerSpec, BrokerControlError>;
+
+    fn read_timer(
+        &self,
+        handle: ObjectHandle,
+    ) -> core::result::Result<(u64, bool, ReadinessFlags), BrokerControlError>;
+
     fn create_pipe(
         &self,
         capacity: u64,
@@ -370,6 +391,36 @@ where
         self.request(|local| local.consume_event(handle, mode))
     }
 
+    fn create_timer(
+        &self,
+        clock_id: i32,
+    ) -> core::result::Result<ObjectHandle, BrokerControlError> {
+        self.request(|local| local.create_timer(clock_id))
+    }
+
+    fn set_timer(
+        &self,
+        handle: ObjectHandle,
+        specification: TimerSpec,
+        flags: u32,
+    ) -> core::result::Result<(TimerSpec, ReadinessFlags), BrokerControlError> {
+        self.request(|local| local.set_timer(handle, specification, flags))
+    }
+
+    fn get_timer(
+        &self,
+        handle: ObjectHandle,
+    ) -> core::result::Result<TimerSpec, BrokerControlError> {
+        self.request(|local| local.get_timer(handle))
+    }
+
+    fn read_timer(
+        &self,
+        handle: ObjectHandle,
+    ) -> core::result::Result<(u64, bool, ReadinessFlags), BrokerControlError> {
+        self.request(|local| local.read_timer(handle))
+    }
+
     fn create_pipe(
         &self,
         capacity: u64,

litebox/src/event/counter.rs

@@ -487,7 +487,8 @@ mod tests {
                 }
                 request @ (BrokerOperation::Event(_)
                 | BrokerOperation::Pipe(_)
-                | BrokerOperation::Socket(_)) => {
+                | BrokerOperation::Socket(_)
+                | BrokerOperation::Timer(_)) => {
                     panic!("unexpected broker request: {request:?}")
                 }
             };

litebox/src/event/mod.rs

@@ -6,6 +6,7 @@
 pub mod counter;
 pub mod observer;
 pub mod polling;
+pub mod timer;
 pub mod wait;
 
 bitflags::bitflags! {

litebox/src/event/timer.rs

@@ -0,0 +1,187 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+use alloc::sync::Arc;
+
+use litebox_broker_protocol::ObjectHandle;
+use litebox_broker_protocol::readiness::ReadinessFlags;
+pub use litebox_broker_protocol::timer::TimerSpec;
+use thiserror::Error;
+
+use crate::{
+    LiteBox,
+    broker::{
+        BrokerControl, BrokerPollableRegistry,
+        error::{BrokerControlError, BrokerObjectError},
+        readiness_events,
+    },
+    event::{
+        Events, IOPollable, observer::Observer, polling::Pollee, polling::TryOpError,
+        wait::WaitContext,
+    },
+    platform::TimeProvider,
+    sync::RawSyncPrimitivesProvider,
+};
+
+/// Errors returned by local-core timers.
+#[derive(Clone, Copy, Debug, Error, PartialEq, Eq)]
+#[non_exhaustive]
+pub enum TimerError {
+    #[error("timer operation would block")]
+    WouldBlock,
+    #[error("timer resource exhausted")]
+    ResourceExhausted,
+    #[error("timer permission denied")]
+    PermissionDenied,
+    #[error("timer was cancelled by a clock change")]
+    Cancelled,
+    #[error("timer I/O failed")]
+    Io,
+    #[error("timer backing authority unavailable")]
+    Unavailable,
+}
+
+impl From<BrokerObjectError> for TryOpError<TimerError> {
+    fn from(error: BrokerObjectError) -> Self {
+        match error {
+            BrokerObjectError::WouldBlock => Self::TryAgain,
+            error => Self::Other(error.into()),
+        }
+    }
+}
+
+impl From<BrokerObjectError> for TimerError {
+    fn from(error: BrokerObjectError) -> Self {
+        match error {
+            BrokerObjectError::WouldBlock => Self::WouldBlock,
+            BrokerObjectError::ResourceExhausted | BrokerObjectError::OutOfMemory => {
+                Self::ResourceExhausted
+            }
+            BrokerObjectError::PermissionDenied => Self::PermissionDenied,
+            BrokerObjectError::Control
+            | BrokerObjectError::InvalidObject
+            | BrokerObjectError::PeerClosed
+            | BrokerObjectError::UnsupportedOperation => Self::Io,
+        }
+    }
+}
+
+/// A local-core timer backed by a broker-owned host timerfd.
+pub struct Timer<Platform: RawSyncPrimitivesProvider + TimeProvider> {
+    broker: Arc<dyn BrokerControl>,
+    handle: ObjectHandle,
+    pollable_registry: Arc<BrokerPollableRegistry<Platform>>,
+    pollee: Arc<Pollee<Platform>>,
+}
+
+impl<Platform> Timer<Platform>
+where
+    Platform: RawSyncPrimitivesProvider + TimeProvider,
+{
+    /// Creates a local-core timer for the given clock.
+    ///
+    /// # Panics
+    ///
+    /// Panics if the broker reports an unrecoverable error or returns a protocol
+    /// response that does not match the issued timer request.
+    pub fn new(litebox: &LiteBox<Platform>, clock_id: i32) -> Result<Self, TimerError> {
+        let Some(broker) = litebox.broker_control() else {
+            return Err(TimerError::Unavailable);
+        };
+        let handle = broker
+            .create_timer(clock_id)
+            .map_err(BrokerObjectError::from)
+            .map_err(TimerError::from)?;
+        let pollable_registry = litebox.broker_pollable_registry();
+        let pollee = Arc::new(Pollee::new());
+        pollable_registry.register_pollable(handle, &pollee);
+        Ok(Self {
+            broker,
+            handle,
+            pollable_registry,
+            pollee,
+        })
+    }
+
+    /// Arms or disarms the timer, returning the setting previously in effect.
+    pub fn set_time(&self, specification: TimerSpec, flags: u32) -> Result<TimerSpec, TimerError> {
+        let (previous, readiness) = self
+            .broker
+            .set_timer(self.handle, specification, flags)
+            .map_err(|error| self.broker_request_error(error))?;
+        // Re-arming clears any prior pending expiration; wake observers so a
+        // level-triggered poller re-checks the newly authoritative readiness.
+        self.pollee.notify_observers(readiness_events(readiness));
+        Ok(previous)
+    }
+
+    /// Returns the time remaining until the next expiration and the interval.
+    pub fn get_time(&self) -> Result<TimerSpec, TimerError> {
+        self.broker
+            .get_timer(self.handle)
+            .map_err(|error| self.broker_request_error(error))
+            .map_err(TimerError::from)
+    }
+
+    /// Reads the accumulated expiration count, blocking until the timer expires
+    /// unless `nonblock` is set.
+    pub fn read(
+        &self,
+        cx: &WaitContext<'_, Platform>,
+        nonblock: bool,
+    ) -> Result<u64, TryOpError<TimerError>> {
+        self.pollee.wait(cx, nonblock, Events::IN, || {
+            let (expirations, cancelled, _readiness) = self.drain()?;
+            if cancelled {
+                return Err(TryOpError::Other(TimerError::Cancelled));
+            }
+            Ok(expirations)
+        })
+    }
+
+    fn drain(&self) -> Result<(u64, bool, ReadinessFlags), BrokerObjectError> {
+        self.broker
+            .read_timer(self.handle)
+            .map_err(|error| self.broker_request_error(error))
+    }
+
+    fn broker_request_error(&self, error: BrokerControlError) -> BrokerObjectError {
+        let error = error.into();
+        if error != BrokerObjectError::WouldBlock {
+            self.pollee.notify_observers(Events::ERR);
+        }
+        error
+    }
+}
+
+impl<Platform> Drop for Timer<Platform>
+where
+    Platform: RawSyncPrimitivesProvider + TimeProvider,
+{
+    fn drop(&mut self) {
+        self.pollable_registry.unregister_pollable(self.handle);
+        let _ = self.broker.close_object(self.handle);
+    }
+}
+
+impl<Platform> IOPollable for Timer<Platform>
+where
+    Platform: RawSyncPrimitivesProvider + TimeProvider,
+{
+    fn register_observer(&self, observer: alloc::sync::Weak<dyn Observer<Events>>, mask: Events) {
+        self.pollee.register_observer(observer, mask);
+    }
+
+    fn check_io_events(&self) -> Events {
+        let readiness = match self
+            .broker
+            .check_readiness(self.handle)
+            .map_err(|error| self.broker_request_error(error))
+        {
+            Ok(readiness) => readiness,
+            Err(BrokerObjectError::WouldBlock) => return Events::empty(),
+            Err(_) => return Events::ERR,
+        };
+        readiness_events(readiness)
+    }
+}

litebox/src/pipes.rs

@@ -1199,7 +1199,8 @@ mod tests {
                 }
                 request @ (BrokerOperation::Pipe(_)
                 | BrokerOperation::Event(_)
-                | BrokerOperation::Socket(_)) => {
+                | BrokerOperation::Socket(_)
+                | BrokerOperation::Timer(_)) => {
                     panic!("unexpected broker request: {request:?}")
                 }
             };

litebox_broker_core/src/event.rs

@@ -26,7 +26,9 @@ pub fn add(session: &BrokerSession, handle: ObjectHandle, value: u64) -> Result<
     let mut object = object.write();
     match &mut *object {
         ObjectEntry::Event(event) => event.add(value),
-        ObjectEntry::Pipe(_) | ObjectEntry::Socket(_) => Err(BrokerError::InvalidRights),
+        ObjectEntry::Pipe(_) | ObjectEntry::Socket(_) | ObjectEntry::Timer(_) => {
+            Err(BrokerError::InvalidRights)
+        }
         ObjectEntry::Reserved => Err(BrokerError::Internal),
     }
 }
@@ -41,7 +43,9 @@ pub fn consume(
     let mut object = object.write();
     match &mut *object {
         ObjectEntry::Event(event) => event.consume(mode),
-        ObjectEntry::Pipe(_) | ObjectEntry::Socket(_) => Err(BrokerError::InvalidRights),
+        ObjectEntry::Pipe(_) | ObjectEntry::Socket(_) | ObjectEntry::Timer(_) => {
+            Err(BrokerError::InvalidRights)
+        }
         ObjectEntry::Reserved => Err(BrokerError::Internal),
     }
 }

litebox_broker_core/src/lib.rs

@@ -25,6 +25,7 @@ mod policy;
 pub mod readiness;
 mod session;
 pub mod socket;
+pub mod timer;
 
 use alloc::sync::Arc;
 use core::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
@@ -41,6 +42,7 @@ pub use policy::{
 use session::ObjectReference;
 pub use session::{BrokerSession, CallerCredential, ObjectRights, SessionId};
 use socket::SocketProvider;
+use timer::{TimerProvider, UnsupportedTimerProvider};
 
 /// BrokerCore result type.
 pub type Result<T> = core::result::Result<T, BrokerError>;
@@ -117,6 +119,7 @@ pub struct BrokerCore {
     pub(crate) reserved_pipe_capacity: Arc<AtomicUsize>,
     pub(crate) reserved_sockets: Arc<AtomicUsize>,
     pub(crate) socket_provider: Arc<dyn SocketProvider>,
+    pub(crate) timer_provider: Arc<dyn TimerProvider>,
 }
 
 static BROKER_CORE_CREATED: AtomicBool = AtomicBool::new(false);
@@ -147,9 +150,21 @@ impl BrokerCore {
             reserved_pipe_capacity: Arc::new(AtomicUsize::new(0)),
             reserved_sockets: Arc::new(AtomicUsize::new(0)),
             socket_provider,
+            timer_provider: Arc::new(UnsupportedTimerProvider),
         })
     }
 
+    /// Installs a broker-wide timerfd provider.
+    ///
+    /// The core defaults to [`UnsupportedTimerProvider`]; deployments that
+    /// support timers install a host provider (e.g. the Linux-userland one)
+    /// with this builder after construction.
+    #[must_use]
+    pub fn with_timer_provider(mut self, timer_provider: Arc<dyn TimerProvider>) -> Self {
+        self.timer_provider = timer_provider;
+        self
+    }
+
     /// Returns the configured authority-state limits.
     #[must_use]
     pub const fn limits(&self) -> BrokerCoreLimits {

litebox_broker_core/src/pipe.rs

@@ -62,7 +62,9 @@ pub fn read(session: &BrokerSession, handle: ObjectHandle, length: u32) -> Resul
     let object = object.read();
     match &*object {
         ObjectEntry::Pipe(pipe) => pipe.read(length as usize),
-        ObjectEntry::Event(_) | ObjectEntry::Socket(_) => Err(BrokerError::InvalidRights),
+        ObjectEntry::Event(_) | ObjectEntry::Socket(_) | ObjectEntry::Timer(_) => {
+            Err(BrokerError::InvalidRights)
+        }
         ObjectEntry::Reserved => Err(BrokerError::Internal),
     }
 }
@@ -76,7 +78,9 @@ pub fn write(session: &BrokerSession, handle: ObjectHandle, data: &[u8]) -> Resu
     let object = object.read();
     match &*object {
         ObjectEntry::Pipe(pipe) => pipe.write(data),
-        ObjectEntry::Event(_) | ObjectEntry::Socket(_) => Err(BrokerError::InvalidRights),
+        ObjectEntry::Event(_) | ObjectEntry::Socket(_) | ObjectEntry::Timer(_) => {
+            Err(BrokerError::InvalidRights)
+        }
         ObjectEntry::Reserved => Err(BrokerError::Internal),
     }
 }