mountos_native/
events.rs

1//! Server-pushed notices. Readers decode notify frames into `Event`s and hand them to the hub.
2//! The hub settles warm waits and watch queues at once, and queues every event for the dispatcher
3//! thread, which calls the listeners in order.
4
5use crate::client::Client;
6use crate::error::{Error, Result};
7use crate::wire::{self, WatchEventKind};
8use std::collections::{HashMap, VecDeque};
9use std::io;
10use std::panic::{catch_unwind, AssertUnwindSafe};
11use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
12use std::sync::{Arc, Condvar, Mutex, MutexGuard};
13use std::time::{Duration, Instant, SystemTime};
14
15/// Capacity of the listener queue.
16const LISTENER_QUEUE: usize = 1024;
17/// Finished warms kept for `wait_warm`.
18const WARM_CACHE: usize = 1024;
19
20/// Receives log lines. The line has no prefix and no newline.
21pub type Logger = Arc<dyn Fn(&str) + Send + Sync>;
22
23pub(crate) fn lock<T>(m: &Mutex<T>) -> MutexGuard<'_, T> {
24  m.lock().unwrap_or_else(|e| e.into_inner())
25}
26
27/// Kind of an event. New kinds can be added, so a `match` needs a wildcard arm.
28#[derive(Clone, Copy, Debug, PartialEq, Eq)]
29#[non_exhaustive]
30pub enum EventKind {
31  /// A warm request finished. `ref_id` is the warm id.
32  WarmDone,
33  /// A warm request failed. `error` says why and `path` names the file.
34  WarmFailed,
35  /// A watched path changed. `ref_id` is the watch id and `change` says how.
36  Change,
37  /// Events were dropped. `ref_id` is the watch id, or 0 when the loss is not tied to one watch.
38  /// A local queue that overflowed also produces this event.
39  Overflow,
40  /// The server is about to close the connection, for example on unmount.
41  Closing,
42  /// A connection ended without a Closing notice or a local close. `error` says why.
43  /// This event is local, the server does not send it.
44  ConnectionLost,
45}
46
47/// How a watched path changed. New kinds can be added, so a `match` needs a wildcard arm.
48#[derive(Clone, Copy, Debug, PartialEq, Eq)]
49#[non_exhaustive]
50pub enum ChangeKind {
51  Create,
52  Remove,
53  Write,
54  /// `path` is the new name and `old_path` the old one.
55  Rename,
56  Attrib,
57}
58
59/// One event from the server, or a local one (`Overflow`, `ConnectionLost`). The client builds
60/// it. Callers only read the fields.
61#[derive(Debug)]
62#[non_exhaustive]
63pub struct Event {
64  pub kind: EventKind,
65  /// Server sequence number. 0 for local events.
66  pub sequence: u64,
67  pub ref_id: u64,
68  pub path: String,
69  pub old_path: String,
70  /// Set for `Change`.
71  pub change: Option<ChangeKind>,
72  /// `Error::Server` for `WarmFailed`, an `Error::Io` for `ConnectionLost`.
73  pub error: Option<Error>,
74  /// When the client received the event.
75  pub time: SystemTime,
76}
77
78impl Clone for Event {
79  fn clone(&self) -> Event {
80    Event {
81      kind: self.kind,
82      sequence: self.sequence,
83      ref_id: self.ref_id,
84      path: self.path.clone(),
85      old_path: self.old_path.clone(),
86      change: self.change,
87      error: self.error.as_ref().map(Error::dup),
88      time: self.time,
89    }
90  }
91}
92
93impl Event {
94  pub(crate) fn local(kind: EventKind, ref_id: u64, error: Option<Error>) -> Event {
95    Event {
96      kind,
97      sequence: 0,
98      ref_id,
99      path: String::new(),
100      old_path: String::new(),
101      change: None,
102      error,
103      time: SystemTime::now(),
104    }
105  }
106
107  /// Decodes a notice. Returns `None` for a kind this client does not know.
108  pub(crate) fn from_wire(m: wire::Notice) -> Option<Event> {
109    let mut change = None;
110    let kind = match wire::NoticeKind::try_from(m.kind).ok()? {
111      wire::NoticeKind::WarmDone => EventKind::WarmDone,
112      wire::NoticeKind::WarmFailed => EventKind::WarmFailed,
113      wire::NoticeKind::Closing => EventKind::Closing,
114      wire::NoticeKind::Overflow => EventKind::Overflow,
115      wire::NoticeKind::Change => {
116        change = match WatchEventKind::try_from(m.change_kind).ok()? {
117          WatchEventKind::Create => Some(ChangeKind::Create),
118          WatchEventKind::Remove => Some(ChangeKind::Remove),
119          WatchEventKind::Write => Some(ChangeKind::Write),
120          WatchEventKind::Rename => Some(ChangeKind::Rename),
121          WatchEventKind::Attrib => Some(ChangeKind::Attrib),
122          WatchEventKind::Overflow => None,
123        };
124        if change.is_some() {
125          EventKind::Change
126        } else {
127          EventKind::Overflow
128        }
129      }
130    };
131    let error = (kind == EventKind::WarmFailed).then_some(Error::Server {
132      code: m.code,
133      message: m.message,
134    });
135    Some(Event {
136      kind,
137      sequence: m.sequence,
138      ref_id: m.ref_id,
139      path: m.path,
140      old_path: m.old_path,
141      change,
142      error,
143      time: SystemTime::now(),
144    })
145  }
146
147  fn log_line(&self) -> String {
148    let err = self
149      .error
150      .as_ref()
151      .map_or_else(|| "none".to_string(), |e| e.to_string());
152    format!(
153      "event kind={:?} ref_id={} path={:?} error={}",
154      self.kind, self.ref_id, self.path, err
155    )
156  }
157}
158
159/// Handle for `Client::remove_listener`.
160#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
161pub struct ListenerId(u64);
162
163struct ChanState {
164  items: VecDeque<Event>,
165  /// Set when the oldest event was dropped. The next pop returns a local Overflow first.
166  overflowed: bool,
167  closed: bool,
168}
169
170/// Bounded event queue. A full queue drops its oldest event and reports one local Overflow.
171pub(crate) struct Chan {
172  cap: usize,
173  /// Stamped on the local Overflow event.
174  ref_id: AtomicU64,
175  st: Mutex<ChanState>,
176  cv: Condvar,
177}
178
179impl Chan {
180  pub(crate) fn new(cap: usize, ref_id: u64) -> Chan {
181    Chan {
182      cap,
183      ref_id: AtomicU64::new(ref_id),
184      st: Mutex::new(ChanState {
185        items: VecDeque::new(),
186        overflowed: false,
187        closed: false,
188      }),
189      cv: Condvar::new(),
190    }
191  }
192
193  pub(crate) fn set_ref_id(&self, id: u64) {
194    self.ref_id.store(id, Ordering::SeqCst);
195  }
196
197  /// Queues `ev`. Returns true when this push started an overflow episode (it dropped the
198  /// oldest event and no Overflow was pending yet).
199  pub(crate) fn push(&self, ev: Event) -> bool {
200    let mut st = lock(&self.st);
201    if st.closed {
202      return false;
203    }
204    let mut started = false;
205    if st.items.len() >= self.cap {
206      st.items.pop_front();
207      started = !st.overflowed;
208      st.overflowed = true;
209    }
210    st.items.push_back(ev);
211    drop(st);
212    self.cv.notify_one();
213    started
214  }
215
216  /// Ends the queue. With `discard`, queued events are dropped, otherwise they are still delivered.
217  pub(crate) fn close(&self, discard: bool) {
218    let mut st = lock(&self.st);
219    st.closed = true;
220    if discard {
221      st.items.clear();
222      st.overflowed = false;
223    }
224    drop(st);
225    self.cv.notify_all();
226  }
227
228  /// Waits for the next event. `None` timeout blocks. Returns `None` on timeout, or when the
229  /// queue is closed and empty.
230  pub(crate) fn pop(&self, timeout: Option<Duration>) -> Option<Event> {
231    let deadline = timeout.and_then(|t| Instant::now().checked_add(t));
232    let mut st = lock(&self.st);
233    loop {
234      if st.overflowed {
235        st.overflowed = false;
236        return Some(Event::local(
237          EventKind::Overflow,
238          self.ref_id.load(Ordering::SeqCst),
239          None,
240        ));
241      }
242      if let Some(ev) = st.items.pop_front() {
243        return Some(ev);
244      }
245      if st.closed {
246        return None;
247      }
248      st = match deadline {
249        None => self.cv.wait(st).unwrap_or_else(|e| e.into_inner()),
250        Some(d) => {
251          let left = d.saturating_duration_since(Instant::now());
252          if left.is_zero() {
253            return None;
254          }
255          self.cv.wait_timeout(st, left).unwrap_or_else(|e| e.into_inner()).0
256        }
257      };
258    }
259  }
260}
261
262type Listener = Arc<dyn Fn(&Event) + Send + Sync>;
263
264#[derive(Default)]
265struct Warms {
266  /// Finished warms by id: Ok, or the failure (code, message).
267  done: HashMap<u64, std::result::Result<(), (String, String)>>,
268  order: VecDeque<u64>,
269  closed: bool,
270  /// Server instance of the latest Hello. 0 when that server reports none.
271  instance: u64,
272  /// True once a Hello has been seen, so a first connect differs from a reconnect to a server
273  /// that reports no id.
274  instance_known: bool,
275  /// Counts server instance changes. A waiter that sees it move gives up.
276  restarts: u64,
277}
278
279struct WatchEntry {
280  conn: u64,
281  chan: Arc<Chan>,
282}
283
284/// Shared by the client, its connection readers and the dispatcher thread.
285pub(crate) struct Hub {
286  logger: Logger,
287  listeners: Mutex<Vec<(ListenerId, Listener)>>,
288  next_listener: AtomicU64,
289  queue: Chan,
290  warms: Mutex<Warms>,
291  warm_cv: Condvar,
292  watches: Mutex<HashMap<u64, WatchEntry>>,
293  /// Counts the background threads of the client that are still running.
294  threads: Arc<AtomicUsize>,
295}
296
297/// Counts a running background thread until dropped.
298pub(crate) struct ThreadGuard(Arc<AtomicUsize>);
299
300impl Drop for ThreadGuard {
301  fn drop(&mut self) {
302    self.0.fetch_sub(1, Ordering::SeqCst);
303  }
304}
305
306impl Hub {
307  /// Call before spawning a thread and move the guard into it.
308  pub(crate) fn thread_guard(&self) -> ThreadGuard {
309    self.threads.fetch_add(1, Ordering::SeqCst);
310    ThreadGuard(self.threads.clone())
311  }
312
313  pub(crate) fn thread_probe(&self) -> Arc<AtomicUsize> {
314    self.threads.clone()
315  }
316
317  pub(crate) fn new(logger: Option<Logger>) -> Hub {
318    Hub {
319      logger: logger.unwrap_or_else(|| Arc::new(|line: &str| eprintln!("mountos: {line}"))),
320      listeners: Mutex::new(Vec::new()),
321      next_listener: AtomicU64::new(1),
322      queue: Chan::new(LISTENER_QUEUE, 0),
323      warms: Mutex::new(Warms::default()),
324      warm_cv: Condvar::new(),
325      watches: Mutex::new(HashMap::new()),
326      threads: Arc::new(AtomicUsize::new(0)),
327    }
328  }
329
330  pub(crate) fn log(&self, line: &str) {
331    (self.logger)(line);
332  }
333
334  pub(crate) fn add_watch(&self, id: u64, conn: u64, chan: Arc<Chan>) {
335    chan.set_ref_id(id);
336    lock(&self.watches).insert(id, WatchEntry { conn, chan });
337  }
338
339  pub(crate) fn remove_watch(&self, id: u64) {
340    lock(&self.watches).remove(&id);
341  }
342
343  /// Ends the watches of a connection that is gone.
344  pub(crate) fn conn_lost(&self, conn: u64) {
345    lock(&self.watches).retain(|_, w| {
346      if w.conn == conn {
347        w.chan.close(false);
348        false
349      } else {
350        true
351      }
352    });
353  }
354
355  /// Handles one event: settles warm waits, feeds the watch queue, queues it for listeners.
356  /// `conn` is the connection the event came from, or `None` for a local event. A server
357  /// Overflow without a watch id applies to every watch of that connection.
358  pub(crate) fn publish(&self, conn: Option<u64>, ev: Event) {
359    match ev.kind {
360      EventKind::WarmDone | EventKind::WarmFailed => self.record_warm(&ev),
361      EventKind::Overflow if ev.ref_id == 0 => {
362        if let Some(conn) = conn {
363          let chans: Vec<Arc<Chan>> = lock(&self.watches)
364            .values()
365            .filter(|w| w.conn == conn)
366            .map(|w| w.chan.clone())
367            .collect();
368          for c in chans {
369            c.push(ev.clone());
370          }
371        }
372      }
373      EventKind::Change | EventKind::Overflow if ev.ref_id != 0 => {
374        let chan = lock(&self.watches).get(&ev.ref_id).map(|w| w.chan.clone());
375        if let Some(c) = chan {
376          if c.push(ev.clone()) {
377            self.log(&format!("watch {} queue full, dropped oldest events", ev.ref_id));
378          }
379        }
380      }
381      _ => {}
382    }
383    if self.queue.push(ev) {
384      self.log("listener queue full, dropped oldest events");
385    }
386  }
387
388  fn record_warm(&self, ev: &Event) {
389    let r = match &ev.error {
390      Some(Error::Server { code, message }) => Err((code.clone(), message.clone())),
391      _ => Ok(()),
392    };
393    let mut w = lock(&self.warms);
394    if w.done.insert(ev.ref_id, r).is_none() {
395      w.order.push_back(ev.ref_id);
396    }
397    while w.order.len() > WARM_CACHE {
398      if let Some(old) = w.order.pop_front() {
399        w.done.remove(&old);
400      }
401    }
402    drop(w);
403    self.warm_cv.notify_all();
404  }
405
406  /// Records the server instance of a Hello. Warm ids restart per server instance, so a new
407  /// instance drops the finished warms and fails the waiters. This includes a change from or to
408  /// 0 (a server that reports no id). A reconnect between two servers that both report 0 cannot
409  /// be told apart and keeps the results.
410  pub(crate) fn note_instance(&self, id: u64) {
411    let mut w = lock(&self.warms);
412    let known = std::mem::replace(&mut w.instance_known, true);
413    if known && w.instance == id {
414      return;
415    }
416    w.instance = id;
417    if !known {
418      return;
419    }
420    w.done.clear();
421    w.order.clear();
422    w.restarts += 1;
423    drop(w);
424    self.warm_cv.notify_all();
425  }
426
427  pub(crate) fn wait_warm(&self, id: u64, timeout: Duration) -> Result<()> {
428    let deadline = Instant::now().checked_add(timeout);
429    let mut w = lock(&self.warms);
430    let restarts = w.restarts;
431    loop {
432      if w.restarts != restarts {
433        return Err(Error::Io(io::Error::new(
434          io::ErrorKind::ConnectionAborted,
435          "server restarted, warm result lost",
436        )));
437      }
438      if w.closed {
439        return Err(Error::Io(io::Error::new(
440          io::ErrorKind::ConnectionAborted,
441          "client closed",
442        )));
443      }
444      match w.done.get(&id) {
445        Some(Ok(())) => return Ok(()),
446        Some(Err((code, message))) => {
447          return Err(Error::Server {
448            code: code.clone(),
449            message: message.clone(),
450          })
451        }
452        None => {}
453      }
454      let left = match deadline {
455        Some(d) => d.saturating_duration_since(Instant::now()),
456        None => Duration::from_secs(3600),
457      };
458      if left.is_zero() {
459        return Err(Error::Io(io::Error::new(
460          io::ErrorKind::TimedOut,
461          "wait_warm timed out",
462        )));
463      }
464      w = self.warm_cv.wait_timeout(w, left).unwrap_or_else(|e| e.into_inner()).0;
465    }
466  }
467
468  /// Stops the dispatcher queue, ends all watches and wakes warm waiters.
469  pub(crate) fn shutdown(&self) {
470    self.queue.close(true);
471    for (_, w) in lock(&self.watches).drain() {
472      w.chan.close(false);
473    }
474    lock(&self.warms).closed = true;
475    self.warm_cv.notify_all();
476  }
477
478  /// Drops all listeners, which releases anything they captured.
479  pub(crate) fn clear_listeners(&self) {
480    let old = std::mem::take(&mut *lock(&self.listeners));
481    drop(old);
482  }
483
484  /// Dispatcher thread body. Runs until `shutdown`.
485  pub(crate) fn dispatch_loop(&self) {
486    while let Some(ev) = self.queue.pop(None) {
487      let ls: Vec<Listener> = lock(&self.listeners).iter().map(|(_, l)| l.clone()).collect();
488      if ls.is_empty() {
489        self.log(&ev.log_line());
490        continue;
491      }
492      for l in ls {
493        if catch_unwind(AssertUnwindSafe(|| l(&ev))).is_err() {
494          self.log(&format!("listener panicked on {}", ev.log_line()));
495        }
496      }
497    }
498  }
499}
500
501impl Client {
502  /// Adds a listener. Every listener gets every event, in order, on the dispatcher thread of
503  /// this client. The call must return quickly. A slow listener makes the queue (1024 events)
504  /// drop its oldest events, and every listener then gets one local `Overflow` event.
505  /// A panic in a listener is caught and logged. With no listener, events go to the logger.
506  /// A listener that holds a `Client` clone keeps the client open until `close` is called.
507  pub fn add_listener(&self, f: impl Fn(&Event) + Send + Sync + 'static) -> ListenerId {
508    let hub = &self.inner.hub;
509    let id = ListenerId(hub.next_listener.fetch_add(1, Ordering::SeqCst));
510    lock(&hub.listeners).push((id, Arc::new(f)));
511    id
512  }
513
514  /// Counter of running background threads (readers and the dispatcher), for tests.
515  #[doc(hidden)]
516  pub fn thread_probe(&self) -> Arc<AtomicUsize> {
517    self.inner.hub.thread_probe()
518  }
519
520  /// Removes a listener. An unknown id does nothing.
521  pub fn remove_listener(&self, id: ListenerId) {
522    lock(&self.inner.hub.listeners).retain(|(i, _)| *i != id);
523  }
524
525  /// Waits until warm request `id` is done. A failed warm returns its failure as `Error::Server`.
526  /// Waiting longer than `timeout` returns an I/O error of kind `TimedOut`. The client keeps the
527  /// last 1024 finished warms, so a warm that ended before this call is not missed. The server
528  /// is not polled.
529  pub fn wait_warm(&self, id: u64, timeout: Duration) -> Result<()> {
530    self.inner.hub.wait_warm(id, timeout)
531  }
532}
mountos_native/events.rs

mountos_native/
events.rs
