distributed.camp
A Python port of the Invisible Internet Project (I2P)
1"""Clock — time with offset management.
2
3Ported from net.i2p.util.Clock (lives in util/ in Java, but logically
4belongs with the time module).
5"""
6
7import threading
8import time as _time
9from typing import Protocol, Set
10
11from i2p_time.build_time import BuildTime
12
13
14class ClockUpdateListener(Protocol):
15 """Listener for clock offset changes."""
16
17 def offset_changed(self, delta: int) -> None: ...
18
19
20def _system_millis() -> int:
21 """Current system time in milliseconds since epoch."""
22 return int(_time.time() * 1000)
23
24
25class Clock:
26 """Alternate time source that maintains an offset from system clock.
27
28 The offset compensates for system clock drift relative to a reference
29 (e.g., NTP). Positive offset means system clock is slow; negative
30 means system clock is fast.
31 """
32
33 MAX_OFFSET = 3 * 24 * 60 * 60 * 1000 # 3 days
34 MAX_LIVE_OFFSET = 10 * 60 * 1000 # 10 minutes
35 MIN_OFFSET_CHANGE = 5 * 1000 # 5 seconds
36
37 def __init__(self) -> None:
38 self._lock = threading.Lock()
39 self._listeners: Set[ClockUpdateListener] = set()
40 self._offset: int = 0
41 self._already_changed: bool = False
42 self._stat_created: bool = False
43
44 now = _system_millis()
45 min_time = BuildTime.get_earliest_time()
46 max_time = BuildTime.get_latest_time()
47
48 if now < min_time:
49 self._offset = min_time - now
50 self._is_system_clock_bad = True
51 now = min_time
52 elif now > max_time:
53 self._offset = max_time - now
54 self._is_system_clock_bad = True
55 now = max_time
56 else:
57 self._is_system_clock_bad = False
58
59 self._started_on = now
60
61 def now(self) -> int:
62 """Current time adjusted by offset, in milliseconds since epoch."""
63 return self._offset + _system_millis()
64
65 def get_offset(self) -> int:
66 """Current delta from system clock in milliseconds."""
67 with self._lock:
68 return self._offset
69
70 def get_updated_successfully(self) -> bool:
71 """Whether the clock offset has been set at least once."""
72 return self._already_changed
73
74 def set_offset(self, offset_ms: int, force: bool = False) -> None:
75 """Set the clock offset.
76
77 Args:
78 offset_ms: Delta from system time in ms. Positive = system slow.
79 force: If True, bypass sanity checks.
80 """
81 with self._lock:
82 delta = offset_ms - self._offset
83
84 if not force:
85 if not self._is_system_clock_bad and abs(offset_ms) > self.MAX_OFFSET:
86 return
87
88 if self._already_changed and (
89 _system_millis() - self._started_on > 10 * 60 * 1000
90 ):
91 if abs(delta) > self.MAX_LIVE_OFFSET:
92 return
93
94 if abs(delta) < self.MIN_OFFSET_CHANGE:
95 self._already_changed = True
96 return
97
98 self._already_changed = True
99 self._offset = offset_ms
100
101 self._fire_offset_changed(delta)
102
103 def set_now(self, real_time: int, stratum: int = 0) -> None:
104 """Set the clock to a known-correct time.
105
106 Args:
107 real_time: The actual current time in ms since epoch.
108 stratum: NTP stratum (1-15, lower is better). Ignored in base Clock.
109 """
110 if (
111 real_time < BuildTime.get_earliest_time()
112 or real_time > BuildTime.get_latest_time()
113 ):
114 return
115 diff = real_time - _system_millis()
116 self.set_offset(diff)
117
118 def add_update_listener(self, listener: ClockUpdateListener) -> None:
119 self._listeners.add(listener)
120
121 def remove_update_listener(self, listener: ClockUpdateListener) -> None:
122 self._listeners.discard(listener)
123
124 def _fire_offset_changed(self, delta: int) -> None:
125 for listener in self._listeners:
126 listener.offset_changed(delta)