"""A Snowflake-style 64-bit ID generator, and the one integer that makes it unique. Layout (64 bits, sign bit always zero so an ID is a positive int64): [ 1 unused ][ 41 timestamp ms ][ 10 machine id ][ 12 sequence ] The timestamp and the machine id are the parts everyone talks about: the timestamp makes IDs roughly sortable, and the machine id is what lets 1024 generators run with no coordination between them. Neither of them makes an ID unique. Two IDs from one machine in one millisecond differ only in the sequence counter, and the sequence counter is only correct because `last_ts` remembers which millisecond the generator was last in. That memory is one integer, it lives in RAM, and it is the whole uniqueness argument. So the behaviour worth studying is what happens when the wall clock stops being monotonic. Three policies for that, switchable, so one trace can price each of them: blind ignore it and hand out an old millisecond again refuse raise Exhausted until the clock catches up (Twitter's original) clamp ts = max(now, last_ts): keep issuing under the last ms used and two for a sequence that runs out inside one millisecond: spin block until the wall clock passes the millisecond (tilNextMillis) borrow issue a millisecond that has not happened yet Determinism: nothing here imports `time`. `next_id(now)` takes the wall clock as an argument, `Wall` derives it from the simulation's own counter, and a tilNextMillis block is a `Spin` exception the caller resolves by advancing that counter -- so a 50 ms stall costs no real seconds and the demo prints identical output on every machine. Single-threaded throughout: the lock the real implementation wraps `next_id` in is out of scope. """ BLIND, REFUSE, CLAMP = "blind", "refuse", "clamp" # rewind policies SPIN, BORROW = "spin", "borrow" # overflow policies RAM, RESUME, SKIP = "ram", "resume", "skip" # startup rules OK, REFUSED = "ok", "refused" class Exhausted(Exception): """The generator refused to issue an ID at all.""" class Spin(Exception): """The generator wants the caller to block until the wall clock reads `until`.""" def __init__(self, until): super().__init__("spin until wall >= %d" % until) self.until = until class Snowflake: def __init__(self, machine_id, epoch=0, ts_bits=41, mid_bits=10, seq_bits=12, rewind=BLIND, overflow=SPIN): assert 1 + ts_bits + mid_bits + seq_bits == 64 self.machine_id = machine_id self.epoch = epoch self.ts_bits, self.mid_bits, self.seq_bits = ts_bits, mid_bits, seq_bits self.seq_mask = (1 << seq_bits) - 1 # 4095 slots per ms at 12 bits self.mid_max = (1 << mid_bits) - 1 # 1023 at 10 bits self.rewind = rewind self.overflow = overflow # The amnesia. Everything above came from config and survives a # restart; these two are the state, and they do not. self.last_ts = -1 self.seq = 0 self.refusals = self.spins = self.borrowed = 0 def pack(self, ts, seq): return (((ts - self.epoch) << (self.mid_bits + self.seq_bits)) | ((self.machine_id & self.mid_max) << self.seq_bits) | seq) def unpack(self, i): seq = i & self.seq_mask mid = (i >> self.seq_bits) & self.mid_max ts = (i >> (self.seq_bits + self.mid_bits)) + self.epoch return ts, mid, seq def next_id(self, now): """`now` is a wall-clock reading in ms, supplied by the caller. Returns (id, ts); raises Exhausted or Spin when a policy says so. """ if now < self.last_ts: # The clock went backwards. Somebody has to decide what that means. if self.rewind == REFUSE: self.refusals += 1 raise Exhausted("clock moved backwards by %d ms" % (self.last_ts - now)) if self.rewind == CLAMP: now = self.last_ts # pin to the highest ts ever issued # BLIND: fall through and cheerfully reuse an old millisecond if now == self.last_ts: self.seq = (self.seq + 1) & self.seq_mask if self.seq == 0: # wrapped: 4096 IDs spent in this ms return self._overflow(now) else: self.seq = 0 self.last_ts = now return self.pack(self.last_ts, self.seq), self.last_ts def _overflow(self, _now): if self.overflow == BORROW: self.borrowed += 1 self.last_ts += 1 # a millisecond that has not happened self.seq = 0 return self.pack(self.last_ts, self.seq), self.last_ts self.spins += 1 # tilNextMillis, as an exception self.seq = self.seq_mask raise Spin(self.last_ts + 1) class Wall: """wall(t) = t, except from `step_at` onwards, where it jumps back `step_ms`. `t` is the simulation's own monotonic millisecond counter -- true time, which no process can read. The generator only ever sees `read(t)`, so an NTP step backwards is one subtraction rather than a mocked syscall. """ def __init__(self, step_at=None, step_ms=0): self.step_at, self.step_ms = step_at, step_ms def read(self, t): if self.step_at is not None and t >= self.step_at: return int(t - self.step_ms) return int(t) def reaches(self, target, t_from): """Smallest true time >= t_from whose reading is >= target.""" for c in sorted([float(target), float(target + self.step_ms)]): if c >= t_from and self.read(c) >= target: return c t = t_from # piecewise inverse missed: scan while self.read(t) < target: t += 0.001 return t class Result: """What one simulated run produced. Records are (true_t, wall, id, outcome).""" def __init__(self, records, stalls, restarts, restart_log, borrowed, gen): self.records, self.stalls, self.gen = records, stalls, gen self.restarts, self.restart_log = restarts, restart_log self.borrowed = borrowed self.ids = [r[2] for r in records if r[3] == OK] self.issued = len(self.ids) self.refused = sum(1 for r in records if r[3] == REFUSED) self.dupes = self.issued - len(set(self.ids)) self.stall = sum(stalls) self.max_stall = max(stalls, default=0.0) def collisions(self): """(t_first, t_again, id) for every ID handed out more than once.""" seen, out = {}, [] for t, _w, i, outcome in self.records: if outcome != OK: continue if i in seen: out.append((seen[i], t, i)) else: seen[i] = t return out def _spawn(cfg, store, start_rule, cls): """A fresh process. Config survives; `last_ts` is whatever startup says. RAM is the real generator: a new process starts at -1 and believes the clock. RESUME and SKIP are the durable variants -- see the commentary for why the obvious one of the two is still wrong. `cls` exists so a demo can run a one-line variant of the generator down the identical trace. """ gen = cls(**cfg) if start_rule == RESUME and store["hwm"] >= 0: gen.last_ts = store["hwm"] elif start_rule == SKIP and store["hwm"] >= 0: gen.last_ts = store["hwm"] + 1 return gen def run(rewind, overflow=SPIN, rate=8, duration=200, machine_id=7, step_at=100.0, step_ms=50, fail_after=0, restart_at=None, start_rule=RAM, mid_bits=10, seq_bits=12, epoch=0, cls=Snowflake): """Fire `rate` requests per ms of true time for `duration` ms. Two ways for the process to be replaced, because they are not the same experiment. `restart_at` is a restart at a chosen instant -- a deploy. `fail_after` is a liveness probe: K consecutive refusals and the supervisor replaces the process, which means a policy that refuses is a policy that arranges its own restart. """ cfg = dict(machine_id=machine_id, epoch=epoch, mid_bits=mid_bits, seq_bits=seq_bits, rewind=rewind, overflow=overflow) wall = Wall(step_at=step_at, step_ms=step_ms) store = {"hwm": -1} gen = _spawn(cfg, store, start_rule, cls) records, stalls, restart_log = [], [], [] restarts = borrowed = consecutive = 0 restarted, t_free = False, 0.0 for i in range(int(rate * duration)): t = max(i / rate, t_free) if restart_at is not None and not restarted and t >= restart_at: borrowed += gen.borrowed gen = _spawn(cfg, store, start_rule, cls) restarted = True restarts += 1 restart_log.append((t, wall.read(t))) stalled = 0.0 while True: w = wall.read(t) try: ident, _ts = gen.next_id(w) except Exhausted: records.append((t, w, None, REFUSED)) consecutive += 1 if fail_after and consecutive >= fail_after: borrowed += gen.borrowed gen = _spawn(cfg, store, start_rule, cls) restarts += 1 restart_log.append((t, w)) consecutive = 0 t_free = t break except Spin as spin: nxt = wall.reaches(spin.until, t) stalled += nxt - t t = nxt continue store["hwm"] = max(store["hwm"], gen.last_ts) records.append((t, w, ident, OK)) if stalled: stalls.append(stalled) consecutive = 0 t_free = t break return Result(records, stalls, restarts, restart_log, borrowed + gen.borrowed, gen)