#!/usr/bin/env python3 """One block, reliable exactly-once unordered messages, atomic nonblocking steps. No eviction, failure, timeout, cancellation, finite buffers, or cross-block ordering. A deliberately conservative teaching protocol: home waits for Installed. The latest field is a test oracle, never read by protocol decisions. """ import sys sys.dont_write_bytecode = True from collections import deque from dataclasses import dataclass from copy import deepcopy from itertools import product import json import random def require(ok, why): if not ok: raise AssertionError(why) @dataclass(frozen=True) class Message: kind: str src: int dst: int q: tuple value: object = None class Directory: HOME = -1 def __init__(self, n, value=0): if type(n) is not int or n < 1 or type(value) is not int: raise ValueError('positive cache count and exact integer value') self.n = n self.cache = [['I', None] for _ in range(n)] self.pending = [None] * n self.seq = [0] * n self.memory = value self.owner = None self.sharers = set() self.requests = deque() self.active = None self.net = [] self.sent = 0 self.completed = [] self.latest = value # oracle only def send(self, kind, src, dst, q, value=None): self.net.append(Message(kind, src, dst, q, value)) self.sent += 1 def finish_op(self, c, op, value): if op == 'W': self.cache[c][1] = value self.latest = value else: require(value == self.latest, 'read must return last completed write') self.completed.append((c, op, value)) def issue(self, c, op, value=None): if type(c) is not int or not 0 <= c < self.n: raise ValueError('cache identity') if op not in ('R', 'W') or (op == 'W' and type(value) is not int): raise ValueError('read or exact-integer write') if self.pending[c] is not None: raise ValueError('one pending application operation per cache') state, old = self.cache[c] if (op == 'R' and state != 'I') or (op == 'W' and state == 'M'): self.finish_op(c, op, old if op == 'R' else value) else: self.seq[c] += 1 q = (c, self.seq[c]) self.pending[c] = (q, op, value) self.send('GetS' if op == 'R' else 'GetM', c, self.HOME, q) self.check() def start(self): if self.active is not None or not self.requests: return req = self.requests.popleft() c = req.src self.active = dict(req=req, phase='', wait=set(), old_owner=self.owner) if self.owner is not None: require(self.owner != c, 'a local M hit needs no remote request') self.active['phase'] = 'DATA' self.send('FwdS' if req.kind == 'GetS' else 'FwdM', self.HOME, self.owner, req.q) elif req.kind == 'GetM': wait = self.sharers - {c} self.active['wait'] = wait self.active['phase'] = 'ACK' for other in wait: self.send('Inv', self.HOME, other, req.q) if not wait: self.grant() else: self.grant() def grant(self): a = self.active req = a['req'] c = req.src require(not a['wait'], 'all invalidation acknowledgments required') if req.kind == 'GetM': self.owner = c self.sharers.clear() kind = 'GrantM' else: self.owner = None self.sharers.add(c) kind = 'GrantS' a['phase'] = 'INSTALLED' self.send(kind, self.HOME, c, req.q, self.memory) def deliver(self, index=0): msg = self.net.pop(index) # deliberate selectable list: O(number in flight) c, q = msg.dst, msg.q if c == self.HOME: if msg.kind in ('GetS', 'GetM'): self.requests.append(msg) self.start() else: a = self.active require(a is not None and q == a['req'].q, 'active transaction identity') if msg.kind == 'InvAck': require(a['phase'] == 'ACK' and msg.src in a['wait'], 'expected sender') a['wait'].remove(msg.src) if not a['wait']: self.grant() elif msg.kind == 'Data': require(a['phase'] == 'DATA' and msg.src == a['old_owner'], 'dirty owner') self.memory = msg.value self.owner = None self.sharers = {msg.src} if a['req'].kind == 'GetS' else set() self.grant() elif msg.kind == 'Installed': require(a['phase'] == 'INSTALLED' and msg.src == a['req'].src, 'installation acknowledgment') self.active = None self.start() else: raise AssertionError('unknown home response') elif msg.kind == 'Inv': require(self.cache[c][0] == 'S', 'exact directory sharer') self.cache[c] = ['I', None] # Crucially, pending[c] survives: its older S copy has been revoked. self.send('InvAck', c, self.HOME, q) elif msg.kind in ('FwdS', 'FwdM'): require(self.cache[c][0] == 'M', 'installed dirty owner') value = self.cache[c][1] self.cache[c] = ['S', value] if msg.kind == 'FwdS' else ['I', None] self.send('Data', c, self.HOME, q, value) elif msg.kind in ('GrantS', 'GrantM'): pending = self.pending[c] require(pending is not None and pending[0] == q, 'matching local request') op, value = pending[1:] require((msg.kind == 'GrantS') == (op == 'R'), 'requested permission') self.cache[c] = ['S' if op == 'R' else 'M', msg.value] # Verify the transferred value BEFORE a store could hide a stale-data bug. require(msg.value == self.latest, 'grant carries current block') self.finish_op(c, op, msg.value if op == 'R' else value) self.pending[c] = None self.send('Installed', c, self.HOME, q) else: raise AssertionError('unknown cache message') self.check() return msg def check(self): readers = {i for i, x in enumerate(self.cache) if x[0] == 'S'} writers = {i for i, x in enumerate(self.cache) if x[0] == 'M'} require(len(writers) <= 1 and not (writers and readers), 'SWMR') require(all(x[1] == self.latest for x in self.cache if x[0] != 'I'), 'valid data') if not writers and self.memory != self.latest: # Old owner may already have relinquished M; its data is still in transit. require(self.active is not None and self.active['phase'] == 'DATA', 'handoff phase') require(any(m.kind == 'Data' and m.q == self.active['req'].q and m.value == self.latest for m in self.net), 'latest in-flight carrier') if self.active is None: require(not self.requests, 'idle home starts queued work immediately') if self.owner is None: require(not writers and readers == self.sharers and self.memory == self.latest, 'stable shared directory') else: require(writers == {self.owner} and not self.sharers, 'stable exclusive directory') def drain(self, rng=None): while self.net: self.deliver(0 if rng is None else rng.randrange(len(self.net))) require(self.active is None and not self.requests and not any(self.pending), 'drained') def state(self): a = self.active return dict(cache=deepcopy(self.cache), memory=self.memory, owner=self.owner, sharers=sorted(self.sharers), pending=deepcopy(self.pending), phase=None if a is None else a['phase'], wait=[] if a is None else sorted(a['wait']), queued=len(self.requests), sent=self.sent, completed=list(self.completed)) def pick(d, kind, src=None, dst=None): for i, m in enumerate(d.net): if m.kind == kind and (src is None or m.src == src) and (dst is None or m.dst == dst): return d.deliver(i) raise AssertionError(('missing message', kind, src, dst)) def endpoint(): d = Directory(3) d.issue(0, 'R'); d.drain() d.issue(1, 'R'); d.drain() stages = [('warm', d.state())] d.issue(2, 'W', 10); pick(d, 'GetM', 2) d.issue(0, 'W', 20); pick(d, 'GetM', 0) pick(d, 'Inv', dst=0); pick(d, 'InvAck', 0) require(d.pending[0] is not None and d.cache[0][0] == 'I', 'upgrade survives invalidation') d.issue(1, 'R') stages.append(('one_ack_missing', d.state())) pick(d, 'Inv', dst=1) d.issue(1, 'R'); pick(d, 'GetS', 1) pick(d, 'InvAck', 1); pick(d, 'GrantM', dst=2) d.issue(2, 'W', 11) stages.append(('installed_not_acknowledged', d.state())) pick(d, 'Installed', 2); pick(d, 'FwdM', dst=2) stages.append(('latest_in_data_message', d.state())) pick(d, 'Data', 2); pick(d, 'GrantM', dst=0); pick(d, 'Installed', 0) pick(d, 'FwdS', dst=0); pick(d, 'Data', 0) pick(d, 'GrantS', dst=1); pick(d, 'Installed', 1) require(d.sent == 23 and d.cache == [['S', 20], ['S', 20], ['I', None]], 'endpoint') require(d.memory == 20 and [x[2] for x in d.completed if x[1] == 'W'] == [10, 11, 20], 'write order') stages.append(('final', d.state())) return stages def randomized(): rng = random.Random(6202026) events = operations = 0 for run in range(1200): d = Directory(1 + run % 5, -7) left = [12] * d.n while any(left) or d.net: choices = [c for c in range(d.n) if left[c] and d.pending[c] is None] if choices and (not d.net or rng.randrange(3)): c = rng.choice(choices) op = rng.choice(('R', 'W')) d.issue(c, op, run * 1000 + operations if op == 'W' else None) left[c] -= 1 operations += 1 else: d.deliver(rng.randrange(len(d.net))) events += 1 d.drain() return dict(histories=1200, operations=operations, events=events) def exhaustive(): # Every next operation and every in-flight delivery for fixed, short client scripts. states = terminal = transitions = 0 for scripts in [((('R', None), ('W', 1)), (('W', 2),)), ((('R', None), ('W', 3)), (('R', None), ('W', 4)))]: start = Directory(2) todo = [(start, (0, 0))] seen = set() while todo: d, pos = todo.pop() # Logs/counters do not influence future protocol behavior. a = d.active key = repr((d.cache, d.pending, d.seq, d.memory, d.owner, sorted(d.sharers), list(d.requests), None if a is None else (a['req'], a['phase'], sorted(a['wait']), a['old_owner']), sorted(d.net, key=repr), d.latest, pos)) if key in seen: continue seen.add(key); states += 1 moves = [('msg', i) for i in range(len(d.net))] moves += [('op', c) for c in range(2) if pos[c] < len(scripts[c]) and d.pending[c] is None] if not moves: require(pos == tuple(map(len, scripts)) and d.active is None, 'no blocked terminal') terminal += 1 for kind, i in moves: x = deepcopy(d) pp = list(pos) if kind == 'msg': x.deliver(i) else: op, value = scripts[i][pp[i]] pp[i] += 1 x.issue(i, op, value) todo.append((x, tuple(pp))); transitions += 1 return dict(states=states, transitions=transitions, terminals=terminal) def counterexamples(): found = {} def rejected(label, d, kind, **where): try: pick(d, kind, **where) except AssertionError as error: found[label] = str(error) else: raise AssertionError('mutant escaped: ' + label) d = Directory(3) for c in (0, 1): d.issue(c, 'R'); d.drain() d.issue(2, 'W', 10); pick(d, 'GetM', 2) pick(d, 'Inv', dst=0); pick(d, 'InvAck', 0) d.active['wait'].clear() # MUTANT: fabricate B's missing acknowledgment. d.grant() rejected('grant_before_all_revocations', d, 'GrantM', dst=2) d = Directory(2) d.issue(0, 'W', 11); d.drain() d.issue(1, 'R'); pick(d, 'GetS', 1); pick(d, 'FwdS', dst=0) i = next(i for i, m in enumerate(d.net) if m.kind == 'Data') m = d.net[i] d.net[i] = Message(m.kind, m.src, m.dst, m.q, d.memory) # MUTANT: stale 0, not 11. rejected('discard_dirty_data', d, 'Data', src=0) d = Directory(2) d.issue(0, 'W', 11); pick(d, 'GetM', 0) d.issue(1, 'R'); pick(d, 'GetS', 1) d.active = None # MUTANT: bypass Installed although GrantM is still in flight. d.start() rejected('forward_before_installation', d, 'FwdS', dst=0) # Reverse the two competing writers; the queued read must now observe C's 10. d = Directory(3) for c in (0, 1): d.issue(c, 'R'); d.drain() d.issue(0, 'W', 20); d.issue(2, 'W', 10) pick(d, 'GetM', 0); pick(d, 'GetM', 2) d.drain() d.issue(1, 'R'); d.drain() require(d.completed[-1] == (1, 'R', 10), 'reordered competition changes final read') found['reversed_request_order_final_read'] = 10 return found def main(): result = dict(endpoint=endpoint(), random=randomized(), exhaustive=exhaustive(), boundaries=counterexamples()) print(json.dumps(result, ensure_ascii=False, indent=2)) if __name__ == '__main__': main()