Start gossip protocol: vector clocks

Author: plotnick

Cargo.lock

@@ -1,5 +1,5 @@
 [workspace]
-members = ["api", "client", "common", "server", "xtask"]
+members = ["api", "client", "common", "gossip", "server", "xtask"]
 resolver = "2"
 
 [workspace.dependencies]

Cargo.toml

@@ -0,0 +1,8 @@
+[package]
+name = "sush-gossip"
+description = "Oxide Support Shell Gossip Protocol"
+version = "0.1.0"
+edition = "2024"
+publish = false
+
+[dependencies]

gossip/Cargo.toml

@@ -0,0 +1,303 @@
+//! Simple [vector clocks](https://en.wikipedia.org/wiki/Vector_clock)
+//! for the sush gossip protocol.
+
+use std::cmp::{Ordering, PartialOrd};
+use std::ops::{Index, IndexMut};
+
+/// A node index in a fixed-size message-passing (gossip) network.
+///
+/// Vector clock owners are responsible for mapping from meaningful
+/// identifiers to these indices.
+pub type NodeIndex = usize;
+
+/// A Lamport (logical) clock; a counter. Should be impractical to
+/// overflow on realistic networks.
+type Clock = u64;
+
+/// A list of Lamport (logical) clocks used to impose a partial order
+/// over events in a fixed-size message-passing (gossip) network, which
+/// in turn induces a notion of causality between messages.
+///
+/// Each node in the network maintains its own vector clock, and each
+/// message is stamped by the sender with a time (a copy of its clock).
+/// Whenever the `i`th node sends or receives a message, it increments
+/// the `i`th entry of its own clock; on receive, that clock is then
+/// joined (in the lattice-theoretic sense, i.e., ∨, _lub_, _sup_) via
+/// `max` with the message timestamp.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct VectorClock<const N: usize>([Clock; N]);
+
+impl<const N: usize> VectorClock<N> {
+    pub fn new() -> Self {
+        Self([0; N])
+    }
+
+    pub fn tick(&mut self, i: NodeIndex) {
+        self[i] += 1;
+    }
+
+    pub fn join(&mut self, other: &Self) {
+        for i in 0..N {
+            self[i] = self[i].max(other[i]);
+        }
+    }
+
+    pub fn is_comparable_to(&self, other: &Self) -> bool {
+        self.partial_cmp(other).is_some()
+    }
+}
+
+impl<const N: usize> Index<NodeIndex> for VectorClock<N> {
+    type Output = Clock;
+
+    fn index(&self, index: NodeIndex) -> &Self::Output {
+        self.0.index(index)
+    }
+}
+
+impl<const N: usize> IndexMut<NodeIndex> for VectorClock<N> {
+    fn index_mut(&mut self, index: NodeIndex) -> &mut Self::Output {
+        self.0.index_mut(index)
+    }
+}
+
+/// Latice ordering where `None` denotes incomparable clocks.
+impl<const N: usize> PartialOrd for VectorClock<N> {
+    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
+        use Ordering::*;
+
+        let mut ordering = Equal;
+        for i in 0..N {
+            ordering = match (ordering, self[i].cmp(&other[i])) {
+                (Equal, ordering) => ordering,
+                (Less | Greater, Equal) => ordering,
+                (Less, Less) | (Greater, Greater) => ordering,
+                (Less, Greater) | (Greater, Less) => return None,
+            };
+        }
+        Some(ordering)
+    }
+}
+
+/// An event shared amongst `N` nodes in a causal gossip network.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct Message<E, const N: usize> {
+    pub event: E,
+    pub time: VectorClock<N>,
+}
+
+impl<E, const N: usize> Message<E, N> {
+    pub fn new(event: E, time: VectorClock<N>) -> Self {
+        Self { event, time }
+    }
+
+    pub fn happened_before(&self, other: &Message<E, N>) -> bool {
+        self.time < other.time
+    }
+}
+
+pub trait Node<E: Clone, const N: usize> {
+    fn send(&mut self, event: E) -> Message<E, N>;
+    fn recv(&mut self, message: Message<E, N>) -> Message<E, N>;
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[test]
+    fn point() {
+        let t = VectorClock::<0>::new();
+        assert_eq!(t, t);
+        assert!(!(t < t));
+        assert!(!(t > t));
+        // can't tick a trivial clock
+    }
+
+    #[test]
+    fn scalar() {
+        let mut t = VectorClock::<1>::new();
+        assert_eq!(t, t);
+        assert!(!(t < t));
+        assert!(!(t > t));
+
+        let t0 = t;
+        assert_eq!(t, t0);
+
+        t.tick(0);
+        assert_ne!(t, t0);
+        assert!(t > t0);
+
+        let t1 = t;
+        assert!(t1 > t0);
+
+        t.tick(0);
+        assert_ne!(t, t0);
+        assert_ne!(t, t1);
+        assert!(t > t0);
+        assert!(t > t1);
+    }
+
+    #[test]
+    fn vector() {
+        let mut t = VectorClock::<2>::new();
+        assert_eq!(t, t);
+        assert!(!(t < t));
+        assert!(!(t > t));
+
+        let t0 = t;
+        t.tick(0);
+        assert_ne!(t, t0);
+        assert!(t > t0);
+
+        let t1 = t;
+        assert!(t1 > t0);
+        t.tick(1);
+        assert_ne!(t, t0);
+        assert_ne!(t, t1);
+        assert!(t > t0);
+        assert!(t > t1);
+    }
+
+    /// A labeled event.
+    type TestEvent = &'static str;
+
+    /// A node in a manual message-passing network.
+    #[derive(Debug, Eq, PartialEq)]
+    struct TestNode<const N: usize> {
+        me: NodeIndex,
+        now: VectorClock<N>,
+        log: Vec<Message<TestEvent, N>>,
+    }
+
+    impl<const N: usize> TestNode<N> {
+        fn new(myself: NodeIndex) -> Self {
+            Self {
+                me: myself,
+                now: VectorClock::new(),
+                log: vec![],
+            }
+        }
+    }
+
+    /// Maintain vector clocks and log each message.
+    impl<const N: usize> Node<TestEvent, N> for TestNode<{ N }> {
+        fn send(&mut self, event: TestEvent) -> Message<TestEvent, N> {
+            self.now.tick(self.me);
+
+            let message = Message::new(event, self.now);
+            self.log.push(message.clone());
+            message
+        }
+
+        fn recv(&mut self, message: Message<TestEvent, N>) -> Message<TestEvent, N> {
+            self.now.tick(self.me);
+
+            let Message { event, time } = message;
+            self.now.join(&time);
+            let message = Message::new(event, self.now);
+            self.log.push(message.clone());
+            message
+        }
+    }
+
+    /// Example from [Why Logical Clocks are Easy](https://queue.acm.org/detail.cfm?id=2917756)
+    /// by Carlos Baquero and Nuno Preguiça, figure 3:
+    /// ```
+    ///     ――― time ――→
+    ///        [1,0,0] [2,0,0] [3,0,0]
+    /// node A    a1 ――→  a2 ――→  a3
+    ///              ⭨
+    ///        [0,1,0] [2,2,0] [2,3,0]
+    /// node B    b1 ――→  b2 ――→  b3
+    ///                      ⭨
+    ///        [0,0,1] [0,0,2] [2,3,3]
+    /// node C    c1 ――→  c2 ――→  c3
+    /// ```
+    #[test]
+    fn baquero_preguiça_figure_3() {
+        let mut a = TestNode::<3>::new(0);
+        let mut b = TestNode::<3>::new(1);
+        let mut c = TestNode::<3>::new(2);
+        assert_eq!(a.now.0, [0, 0, 0], "clocks should start at 0");
+        assert_eq!(a.now, b.now, "clocks should be equal at start");
+        assert_eq!(b.now, c.now, "clocks should be equal at start");
+
+        // A
+        let a1 = a.send("a1");
+        assert!(!a1.happened_before(&a1));
+
+        let a2 = a.send("a2");
+        assert!(a1.happened_before(&a2));
+
+        let a3 = a.send("a3");
+        assert!(a1.happened_before(&a3));
+        assert!(a2.happened_before(&a3));
+
+        // B
+        let b1 = b.send("b1");
+        assert!(!b1.happened_before(&a1));
+        assert!(!b1.happened_before(&a2));
+        assert!(!b1.happened_before(&a3));
+
+        b.recv(a2.clone());
+        let b2 = b.send("b2");
+        assert!(b1.happened_before(&b2));
+        assert!(a2.happened_before(&b2));
+        assert!(!b2.happened_before(&a1));
+        assert!(!b2.happened_before(&a2));
+        assert!(!b2.happened_before(&a3));
+
+        // C
+        let c1 = c.send("c1");
+        assert!(!c1.happened_before(&a1));
+        assert!(!c1.happened_before(&a2));
+        assert!(!c1.happened_before(&a3));
+        assert!(!c1.happened_before(&b1));
+        assert!(!c1.happened_before(&b2));
+
+        let c2 = c.send("c2");
+        assert!(c1.happened_before(&c2));
+        assert!(!c2.happened_before(&a1));
+        assert!(!c2.happened_before(&a2));
+        assert!(!c2.happened_before(&a3));
+        assert!(!c2.happened_before(&b1));
+        assert!(!c2.happened_before(&b2));
+        assert!(!c2.happened_before(&b2));
+        c.recv(b2.clone());
+
+        // Logs
+        macro_rules! msg {
+            ($e:expr, $t:expr) => {
+                Message {
+                    event: $e,
+                    time: VectorClock($t),
+                }
+            };
+        }
+        assert_eq!(
+            a.log,
+            vec![
+                msg!("a1", [1, 0, 0]),
+                msg!("a2", [2, 0, 0]),
+                msg!("a3", [3, 0, 0]),
+            ]
+        );
+        assert_eq!(
+            b.log,
+            vec![
+                msg!("b1", [0, 1, 0]),
+                msg!("a2", [2, 2, 0]),
+                msg!("b2", [2, 3, 0]),
+            ]
+        );
+        assert_eq!(
+            c.log,
+            vec![
+                msg!("c1", [0, 0, 1]),
+                msg!("c2", [0, 0, 2]),
+                msg!("b2", [2, 3, 3]),
+            ]
+        );
+    }
+}

gossip/src/clocks.rs

@@ -0,0 +1,11 @@
+//! Oxide Support Shell Gossip Protocol.
+//!
+//! The gossip protocol is used only between `sush` servers,
+//! and is not exposed externally (to the customer). It allows
+//! distribution of jobs across the rack and enforces job order.
+//!
+//! The ordering constraint is complicated by the fact that the
+//! rack may not know what time it is; therefore, we must impose
+//! a _logical time_, using [vector clocks](https://en.wikipedia.org/wiki/Vector_clock).
+
+pub mod clocks;