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

mountos_native/
events.rs
