crabshell_services/
clock.rs

1//! The clock: the time, truncated to a whole second or minute.
2
3use std::cell::RefCell;
4use std::collections::HashMap;
5use std::future::poll_fn;
6use std::path::Path;
7use std::pin::Pin;
8use std::rc::Rc;
9use std::sync::Once;
10use std::task::Poll;
11use std::time::{Duration, Instant};
12
13use calloop::RegistrationToken;
14use calloop::timer::TimeoutAction;
15use crabshell_reactive::{ReadSignal, Signal};
16use crabshell_ui::Cx;
17use jiff::tz::TimeZone;
18use jiff::{SignedDuration, Timestamp, Zoned};
19use zbus::export::futures_core::Stream;
20use zbus::{MatchRule, Message, MessageStream};
21
22use crate::Shared;
23
24#[cfg(test)]
25mod tests;
26
27/// How often a clock changes.
28#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
29pub enum Precision {
30  /// On every whole second.
31  Second,
32  /// On every whole minute.
33  Minute,
34}
35
36impl Precision {
37  fn seconds(self) -> i64 {
38    match self {
39      Precision::Second => 1,
40      Precision::Minute => 60,
41    }
42  }
43}
44
45/// How early a timer may fire and still count as on its boundary.
46const EARLY: SignedDuration = SignedDuration::from_millis(50);
47
48/// The clocks of one shell: one per precision, each started on first use.
49pub(crate) struct Clocks {
50  shared: Rc<Shared>,
51  started: RefCell<Vec<Clock>>,
52}
53
54struct Clock {
55  precision: Precision,
56  signal: Signal<Zoned>,
57  /// Its timer in the event loop, set for the next boundary.
58  timer: Option<RegistrationToken>,
59}
60
61/// `cx.clock(precision)`.
62pub(crate) fn clock(cx: &Cx, precision: Precision) -> ReadSignal<Zoned> {
63  if let Some(clocks) = cx.service::<Clocks>() {
64    return clocks.get(precision);
65  }
66  static LOGGED: Once = Once::new();
67  LOGGED.call_once(|| {
68    tracing::warn!("cx.clock() needs the services the shell provides; this clock stands still");
69  });
70  let (shown, _) = boundaries(Timestamp::now(), precision);
71  cx.signal(shown.to_zoned(TimeZone::system())).read_only()
72}
73
74impl Clocks {
75  pub(crate) fn new(shared: Rc<Shared>) -> Clocks {
76    Clocks {
77      shared,
78      started: RefCell::new(Vec::new()),
79    }
80  }
81
82  /// The clock for `precision`, started if it is the first ask.
83  fn get(self: &Rc<Self>, precision: Precision) -> ReadSignal<Zoned> {
84    let started = self.started.borrow();
85    if let Some(clock) = started.iter().find(|clock| clock.precision == precision) {
86      return clock.signal.read_only();
87    }
88    let first = started.is_empty();
89    drop(started);
90    if first && let Some(bus) = self.shared.config.system_bus.clone() {
91      self.shared.spawn(watch(Rc::clone(self), bus));
92    }
93    let (shown, deadline) = current(precision, &self.zone());
94    let signal = self.shared.root.signal(shown);
95    let timer = self.timer(precision, deadline);
96    self.started.borrow_mut().push(Clock {
97      precision,
98      signal,
99      timer,
100    });
101    signal.read_only()
102  }
103
104  /// Shows the right time at once and sets every timer afresh from the
105  /// wall clock. Timers count monotonic time, which stops in a suspend.
106  fn resync(self: &Rc<Self>) {
107    let zone = self.zone();
108    let Some(event_loop) = self.shared.event_loop() else {
109      return;
110    };
111    for clock in self.started.borrow_mut().iter_mut() {
112      let (shown, deadline) = current(clock.precision, &zone);
113      clock.signal.try_set(shown);
114      if let Some(old) = clock.timer.take() {
115        event_loop.remove(old);
116      }
117      clock.timer = self.timer(clock.precision, deadline);
118    }
119  }
120
121  /// Acts on a signal from logind or timedated. A bad one is logged.
122  fn on_message(self: &Rc<Self>, message: &Message) {
123    let body = message.body();
124    match message.header().member().map(|member| member.as_str()) {
125      Some("PrepareForSleep") => match body.deserialize::<bool>() {
126        Ok(true) => {}
127        Ok(false) => self.resync(),
128        Err(error) => tracing::warn!(%error, "a bad PrepareForSleep from logind"),
129      },
130      Some("PropertiesChanged") => {
131        match body.deserialize::<(&str, HashMap<&str, zbus::Variant<'_>>, Vec<&str>)>() {
132          Ok((_, changed, invalidated))
133            if changed.contains_key("Timezone") || invalidated.contains(&"Timezone") =>
134          {
135            tracing::info!(zone = self.zone().iana_name(), "the time zone changed");
136            self.resync();
137          }
138          Ok(_) => {}
139          Err(error) => tracing::warn!(%error, "a bad PropertiesChanged from timedated"),
140        }
141      }
142      _ => {}
143    }
144  }
145
146  /// A timer that ticks the `precision` clock at `deadline`, and then at
147  /// every boundary after it.
148  fn timer(self: &Rc<Self>, precision: Precision, deadline: Instant) -> Option<RegistrationToken> {
149    let clocks = Rc::clone(self);
150    self
151      .shared
152      .event_loop()?
153      .insert_timer(deadline, Box::new(move || clocks.tick(precision)))
154  }
155
156  /// Shows the boundary just passed on the `precision` clock and says
157  /// when to tick next. Stops once the clock's signal is disposed.
158  fn tick(&self, precision: Precision) -> TimeoutAction {
159    let (shown, deadline) = current(precision, &self.zone());
160    let started = self.started.borrow();
161    let clock = started.iter().find(|clock| clock.precision == precision);
162    match clock {
163      Some(clock) if clock.signal.try_set(shown) => TimeoutAction::ToInstant(deadline),
164      _ => TimeoutAction::Drop,
165    }
166  }
167
168  /// Read on every tick, since `/etc/localtime` can change with no
169  /// timedated to say so, as under OpenRC.
170  fn zone(&self) -> TimeZone {
171    system_zone(self.shared.config.localtime.as_deref())
172  }
173}
174
175/// Re-syncs `clocks` when the machine resumes and when timedated changes
176/// the time zone, for as long as the bus is there.
177async fn watch(clocks: Rc<Clocks>, bus: zbus::Address) {
178  let (mut sleep, mut zone) = match subscribe(bus).await {
179    Ok(streams) => streams,
180    Err(error) => {
181      tracing::info!(
182          %error,
183          "no system bus: the clock won't re-sync after a suspend or a time zone change"
184      );
185      return;
186    }
187  };
188  loop {
189    let next = poll_fn(|cx| match Pin::new(&mut sleep).poll_next(cx) {
190      Poll::Pending => Pin::new(&mut zone).poll_next(cx),
191      ready => ready,
192    });
193    match next.await {
194      Some(Ok(message)) => clocks.on_message(&message),
195      Some(Err(error)) => tracing::warn!(%error, "a bad message on the system bus"),
196      None => {
197        tracing::info!("the system bus closed: the clock no longer re-syncs");
198        return;
199      }
200    }
201  }
202}
203
204/// Connects to the system bus and subscribes to logind's `PrepareForSleep`
205/// and timedated's `PropertiesChanged`. Match rules only: a proxy would
206/// read properties first, which starts logind and timedated if they are
207/// not running.
208async fn subscribe(bus: zbus::Address) -> zbus::Result<(MessageStream, MessageStream)> {
209  let connection = zbus::connection::Builder::address(bus).build().await?;
210  let signal = || MatchRule::builder().msg_type(zbus::message::Type::Signal);
211  let sleep = signal()
212    .sender("org.freedesktop.login1")
213    .path("/org/freedesktop/login1")
214    .interface("org.freedesktop.login1.Manager")
215    .member("PrepareForSleep")
216    .build()?;
217  let zone = signal()
218    .sender("org.freedesktop.timedate1")
219    .path("/org/freedesktop/timedate1")
220    .interface("org.freedesktop.DBus.Properties")
221    .member("PropertiesChanged")
222    .arg(0, "org.freedesktop.timedate1")
223    .build()?;
224  Ok((
225    MessageStream::for_match_rule(sleep, &connection, None).await?,
226    MessageStream::for_match_rule(zone, &connection, None).await?,
227  ))
228}
229
230/// What a `precision` clock shows now in `zone`, and the instant its timer
231/// should fire next.
232fn current(precision: Precision, zone: &TimeZone) -> (Zoned, Instant) {
233  let now = Timestamp::now();
234  let (shown, next) = boundaries(now, precision);
235  let wait = Duration::try_from(next.duration_since(now)).unwrap_or_default();
236  // Read after the wall clock, so the deadline errs late, never early.
237  (shown.to_zoned(zone.clone()), Instant::now() + wait)
238}
239
240/// The boundary `now` has just passed, and the one after it. A `now` up to
241/// [`EARLY`] before a boundary counts as on it, so a timer that fires a
242/// hair early shows the new value instead of waking again ms later.
243fn boundaries(now: Timestamp, precision: Precision) -> (Timestamp, Timestamp) {
244  let step = precision.seconds();
245  let shown = (now + EARLY).as_second().div_euclid(step) * step;
246  let at = |second| {
247    Timestamp::from_second(second).expect("the system clock is within jiff's years -9999 to 9999")
248  };
249  (at(shown), at(shown + step))
250}
251
252/// The system time zone, read afresh each call. jiff's
253/// `TimeZone::system()` keeps its answer for 5 minutes, so the
254/// `/etc/localtime` link that timedated rewrites is read here. A copied
255/// file, or `TZ` (`localtime` is `None`), is left to jiff.
256fn system_zone(localtime: Option<&Path>) -> TimeZone {
257  let target = localtime.and_then(|path| std::fs::read_link(path).ok());
258  let name = target
259    .as_deref()
260    .and_then(Path::to_str)
261    .and_then(|target| target.rsplit_once("zoneinfo/"))
262    .map(|(_, name)| name);
263  match name.map(|name| jiff::tz::db().get(name)) {
264    Some(Ok(zone)) => zone,
265    _ => TimeZone::system(),
266  }
267}
crabshell_services/clock.rs

crabshell_services/
clock.rs
