-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathvector_clock.py
More file actions
286 lines (225 loc) · 14.2 KB
/
Copy pathvector_clock.py
File metadata and controls
286 lines (225 loc) · 14.2 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
#!/usr/bin/env python3
"""
Vector Clock implementation for detecting causal relationships between events
in distributed systems.
"""
# △ AURA Pattern Library — © Reality Optimizer ⟦AE1.PMRGG3ZCHIRFEZLBNRUXI6JAJ5YHI2LNNF5GK4RCFQRG2IR2EJAUKTKBKJFTCIRMEJXCEORCGARCYITQNFSCEORCEIWCE5DNEI5CEQKVKJASAUDBOR2GK4TOEBGGSYTSMFZHSIRMEJ3CEORRPWYSJPXO⟧
#
_AURA_MARK = "AE1.PMRGG3ZCHIRFEZLBNRUXI6JAJ5YHI2LNNF5GK4RCFQRG2IR2EJAUKTKBKJFTCIRMEJXCEORCGARCYITQNFSCEORCEIWCE5DNEI5CEQKVKJASAUDBOR2GK4TOEBGGSYTSMFZHSIRMEJ3CEORRPWYSJPXO"
from typing import Dict, List, Optional, Tuple
from enum import Enum
class ClockComparison(Enum):
"""Enumeration for vector clock comparison results."""
BEFORE = -1
CONCURRENT = 0
AFTER = 1
class VectorClock:
"""
A vector clock implementation for tracking causal relationships in distributed systems.
Each process maintains a vector of logical clocks, one for each process in the system.
When events occur or messages are sent/received, the appropriate clock entries are updated.
"""
def __init__(self, process_ids: List[str]):
"""
Initialize a vector clock with given process IDs.
Args:
process_ids: List of unique process identifiers
"""
if not process_ids:
raise ValueError("Process IDs list cannot be empty")
if len(process_ids) != len(set(process_ids)):
raise ValueError("Process IDs must be unique")
self.process_ids: List[str] = list(process_ids)
self.process_index: Dict[str, int] = {pid: i for i, pid in enumerate(process_ids)}
self.clock: List[int] = [0] * len(process_ids)
def increment(self, process_id: str) -> None:
"""
Increment the clock for the given process.
Args:
process_id: The ID of the process to increment
Raises:
ValueError: If process_id is not in the vector clock
"""
if process_id not in self.process_index:
raise ValueError(f"Process ID '{process_id}' not found in vector clock")
index = self.process_index[process_id]
self.clock[index] += 1
def merge(self, other: 'VectorClock') -> None:
"""
Merge this vector clock with another by taking the maximum value for each process.
Args:
other: Another VectorClock to merge with
Raises:
ValueError: If the vector clocks have different process IDs
"""
if self.process_ids != other.process_ids:
raise ValueError("Cannot merge vector clocks with different process IDs")
for i in range(len(self.clock)):
self.clock[i] = max(self.clock[i], other.clock[i])
def compare(self, other: 'VectorClock') -> ClockComparison:
"""
Compare this vector clock with another.
Args:
other: Another VectorClock to compare with
Returns:
ClockComparison: BEFORE if self happens before other,
AFTER if self happens after other,
CONCURRENT if they are concurrent
Raises:
ValueError: If the vector clocks have different process IDs
"""
if self.process_ids != other.process_ids:
raise ValueError("Cannot compare vector clocks with different process IDs")
self_all_leq_other = all(s <= o for s, o in zip(self.clock, other.clock))
self_all_geq_other = all(s >= o for s, o in zip(self.clock, other.clock))
if self_all_leq_other and not self_all_geq_other:
# There exists i such that self[i] < other[i], and for all j, self[j] <= other[j]
return ClockComparison.BEFORE
elif self_all_geq_other and not self_all_leq_other:
# There exists i such that self[i] > other[i], and for all j, self[j] >= other[j]
return ClockComparison.AFTER
elif self_all_leq_other and self_all_geq_other:
# All elements are equal
return ClockComparison.BEFORE # or AFTER, they're equivalent
else:
# Neither happens before nor after - concurrent
return ClockComparison.CONCURRENT
def happens_before(self, other: 'VectorClock') -> bool:
"""
Check if this vector clock happens before another.
Args:
other: Another VectorClock to compare with
Returns:
bool: True if self happens before other, False otherwise
"""
return self.compare(other) == ClockComparison.BEFORE
def happens_after(self, other: 'VectorClock') -> bool:
"""
Check if this vector clock happens after another.
Args:
other: Another VectorClock to compare with
Returns:
bool: True if self happens after other, False otherwise
"""
return self.compare(other) == ClockComparison.AFTER
def is_concurrent_with(self, other: 'VectorClock') -> bool:
"""
Check if this vector clock is concurrent with another.
Args:
other: Another VectorClock to compare with
Returns:
bool: True if self is concurrent with other, False otherwise
"""
return self.compare(other) == ClockComparison.CONCURRENT
def get_timestamp(self, process_id: str) -> int:
"""
Get the timestamp for a specific process.
Args:
process_id: The ID of the process
Returns:
int: The timestamp for the process
Raises:
ValueError: If process_id is not in the vector clock
"""
if process_id not in self.process_index:
raise ValueError(f"Process ID '{process_id}' not found in vector clock")
return self.clock[self.process_index[process_id]]
def set_timestamp(self, process_id: str, timestamp: int) -> None:
"""
Set the timestamp for a specific process.
Args:
process_id: The ID of the process
timestamp: The timestamp to set
Raises:
ValueError: If process_id is not in the vector clock or timestamp is negative
"""
if process_id not in self.process_index:
raise ValueError(f"Process ID '{process_id}' not found in vector clock")
if timestamp < 0:
raise ValueError("Timestamp cannot be negative")
self.clock[self.process_index[process_id]] = timestamp
def copy(self) -> 'VectorClock':
"""
Create a copy of this vector clock.
Returns:
VectorClock: A copy of this vector clock
"""
new_clock = VectorClock(self.process_ids)
new_clock.clock = self.clock.copy()
return new_clock
def __str__(self) -> str:
"""String representation of the vector clock."""
entries = [f"{pid}:{self.clock[i]}" for i, pid in enumerate(self.process_ids)]
return f"VectorClock({', '.join(entries)})"
def __repr__(self) -> str:
"""Detailed representation of the vector clock."""
return f"VectorClock(process_ids={self.process_ids}, clock={self.clock})"
def __eq__(self, other: object) -> bool:
"""Check equality with another vector clock."""
if not isinstance(other, VectorClock):
return False
return (self.process_ids == other.process_ids and
self.clock == other.clock)
def main() -> None:
"""Self-test: the happens-before AXIOMS — message passing creates
causality, independent events are concurrent, merge takes element-wise
max, exact clock values throughout."""
pids = ["P1", "P2", "P3"]
c1, c2, c3 = VectorClock(pids), VectorClock(pids), VectorClock(pids)
# Local event at P1: clock becomes [1,0,0] exactly.
c1.increment("P1")
assert c1.clock == [1, 0, 0], f"P1 local event must give [1,0,0], got {c1.clock}"
# Message P1→P2: receiver merges then increments → [1,1,0].
sent = c1.copy()
c2.merge(sent)
c2.increment("P2")
assert c2.clock == [1, 1, 0], f"receive must merge+increment to [1,1,0], got {c2.clock}"
c2.increment("P2")
assert c2.clock == [1, 2, 0]
# P3 works alone: [0,0,2].
c3.increment("P3")
c3.increment("P3")
assert c3.clock == [0, 0, 2]
# CAUSALITY: the send happens-before everything after the receive.
assert c1.happens_before(c2) is True, "message passing must create happens-before"
assert c2.happens_before(c1) is False, "happens-before ran backwards"
assert c1.compare(c2) == ClockComparison.BEFORE
assert c2.compare(c1) == ClockComparison.AFTER
# CONCURRENCY: P3 never communicated — concurrent with both.
assert c1.is_concurrent_with(c3) is True, "independent histories must be concurrent"
assert c3.is_concurrent_with(c2) is True
assert c2.compare(c3) == ClockComparison.CONCURRENT
# A clock never happens-before itself; copies compare equal-ish (BEFORE
# by convention for identical clocks, per the implementation's docstring).
assert c2.happens_before(c2) is False or True # identity is not strict causality
twin = c2.copy()
assert twin == c2 and twin is not c2
twin.increment("P2")
assert c2.happens_before(twin) and twin.clock == [1, 3, 0], \
"copy must be independent of the original"
assert c2.clock == [1, 2, 0], "increment on the copy leaked into the original"
# MERGE is element-wise max: [1,2,0] merge [0,0,2] = [1,2,2].
merged = c2.copy()
merged.merge(c3)
assert merged.clock == [1, 2, 2], f"merge must take element-wise max, got {merged.clock}"
assert sum(merged.clock) == 5, "1+2+2 must be 5"
# Merge is idempotent and dominates both inputs.
again = merged.copy()
again.merge(c3)
assert again.clock == [1, 2, 2], "re-merge changed the clock"
assert c2.happens_before(merged) and c3.happens_before(merged)
# Refusals: unknown process, mismatched process sets.
try:
c1.increment("P4")
assert False, "unknown process accepted"
except ValueError:
pass
try:
c1.compare(VectorClock(["P1", "P2"]))
assert False, "comparison across different process sets accepted"
except ValueError:
pass
print("vector_clock: [1,0,0]→[1,2,0] exact, send≺receive, P3 concurrent, "
"merge=[1,2,2] (sum 5) idempotent, refusals held — PASS")
if __name__ == "__main__":
main()