1mod system_clock;
2
3use smallvec::SmallVec;
4use std::{
5 cmp::{self, Ordering},
6 fmt, iter,
7};
8
9pub use system_clock::*;
10
11/// A unique identifier for each distributed node.
12pub type ReplicaId = u16;
13
14/// A [Lamport sequence number](https://en.wikipedia.org/wiki/Lamport_timestamp).
15pub type Seq = u32;
16
17/// A [Lamport timestamp](https://en.wikipedia.org/wiki/Lamport_timestamp),
18/// used to determine the ordering of events in the editor.
19#[derive(Clone, Copy, Default, Eq, Hash, PartialEq)]
20pub struct Lamport {
21 pub replica_id: ReplicaId,
22 pub value: Seq,
23}
24
25/// A [vector clock](https://en.wikipedia.org/wiki/Vector_clock).
26#[derive(Clone, Default, Hash, Eq, PartialEq)]
27pub struct Global(SmallVec<[u32; 8]>);
28
29impl Global {
30 pub fn new() -> Self {
31 Self::default()
32 }
33
34 pub fn get(&self, replica_id: ReplicaId) -> Seq {
35 self.0.get(replica_id as usize).copied().unwrap_or(0) as Seq
36 }
37
38 pub fn observe(&mut self, timestamp: Lamport) {
39 if timestamp.value > 0 {
40 let new_len = timestamp.replica_id as usize + 1;
41 if new_len > self.0.len() {
42 self.0.resize(new_len, 0);
43 }
44
45 let entry = &mut self.0[timestamp.replica_id as usize];
46 *entry = cmp::max(*entry, timestamp.value);
47 }
48 }
49
50 pub fn join(&mut self, other: &Self) {
51 if other.0.len() > self.0.len() {
52 self.0.resize(other.0.len(), 0);
53 }
54
55 for (left, right) in self.0.iter_mut().zip(&other.0) {
56 *left = cmp::max(*left, *right);
57 }
58 }
59
60 pub fn meet(&mut self, other: &Self) {
61 if other.0.len() > self.0.len() {
62 self.0.resize(other.0.len(), 0);
63 }
64
65 let mut new_len = 0;
66 for (ix, (left, right)) in self
67 .0
68 .iter_mut()
69 .zip(other.0.iter().chain(iter::repeat(&0)))
70 .enumerate()
71 {
72 if *left == 0 {
73 *left = *right;
74 } else if *right > 0 {
75 *left = cmp::min(*left, *right);
76 }
77
78 if *left != 0 {
79 new_len = ix + 1;
80 }
81 }
82 self.0.resize(new_len, 0);
83 }
84
85 pub fn observed(&self, timestamp: Lamport) -> bool {
86 self.get(timestamp.replica_id) >= timestamp.value
87 }
88
89 pub fn observed_any(&self, other: &Self) -> bool {
90 let mut lhs = self.0.iter();
91 let mut rhs = other.0.iter();
92 loop {
93 if let Some(left) = lhs.next() {
94 if let Some(right) = rhs.next() {
95 if *right > 0 && left >= right {
96 return true;
97 }
98 } else {
99 return false;
100 }
101 } else {
102 return false;
103 }
104 }
105 }
106
107 pub fn observed_all(&self, other: &Self) -> bool {
108 let mut lhs = self.0.iter();
109 let mut rhs = other.0.iter();
110 loop {
111 if let Some(left) = lhs.next() {
112 if let Some(right) = rhs.next() {
113 if left < right {
114 return false;
115 }
116 } else {
117 return true;
118 }
119 } else {
120 return rhs.next().is_none();
121 }
122 }
123 }
124
125 pub fn changed_since(&self, other: &Self) -> bool {
126 if self.0.len() > other.0.len() {
127 return true;
128 }
129 for (left, right) in self.0.iter().zip(other.0.iter()) {
130 if left > right {
131 return true;
132 }
133 }
134 false
135 }
136
137 pub fn iter(&self) -> impl Iterator<Item = Lamport> + '_ {
138 self.0.iter().enumerate().map(|(replica_id, seq)| Lamport {
139 replica_id: replica_id as ReplicaId,
140 value: *seq,
141 })
142 }
143}
144
145impl FromIterator<Lamport> for Global {
146 fn from_iter<T: IntoIterator<Item = Lamport>>(locals: T) -> Self {
147 let mut result = Self::new();
148 for local in locals {
149 result.observe(local);
150 }
151 result
152 }
153}
154
155impl Ord for Lamport {
156 fn cmp(&self, other: &Self) -> Ordering {
157 // Use the replica id to break ties between concurrent events.
158 self.value
159 .cmp(&other.value)
160 .then_with(|| self.replica_id.cmp(&other.replica_id))
161 }
162}
163
164impl PartialOrd for Lamport {
165 fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
166 Some(self.cmp(other))
167 }
168}
169
170impl Lamport {
171 pub const MIN: Self = Self {
172 replica_id: ReplicaId::MIN,
173 value: Seq::MIN,
174 };
175
176 pub const MAX: Self = Self {
177 replica_id: ReplicaId::MAX,
178 value: Seq::MAX,
179 };
180
181 pub fn new(replica_id: ReplicaId) -> Self {
182 Self {
183 value: 1,
184 replica_id,
185 }
186 }
187
188 pub fn tick(&mut self) -> Self {
189 let timestamp = *self;
190 self.value += 1;
191 timestamp
192 }
193
194 pub fn observe(&mut self, timestamp: Self) {
195 self.value = cmp::max(self.value, timestamp.value) + 1;
196 }
197}
198
199impl fmt::Debug for Lamport {
200 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
201 write!(f, "Lamport {{{}: {}}}", self.replica_id, self.value)
202 }
203}
204
205impl fmt::Debug for Global {
206 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
207 write!(f, "Global {{")?;
208 for timestamp in self.iter() {
209 if timestamp.replica_id > 0 {
210 write!(f, ", ")?;
211 }
212 write!(f, "{}: {}", timestamp.replica_id, timestamp.value)?;
213 }
214 write!(f, "}}")
215 }
216}