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

mountos_native/
events.rs
