外观
distributed/raft_model.py
配套源码，运行方法见同目录 README。返回实验总览。
python
"""Deterministic fixed-membership Raft/KV teaching model; no sockets or disk.
Persistent fields survive restart() inside this process. Calls are atomic and
ordered by the test harness. This is not a production Raft implementation.
"""
from dataclasses import dataclass, field
from copy import deepcopy
@dataclass(eq=True)
class Entry:
term: int
command: tuple | None
@dataclass
class Node:
ident: int
term: int = 0
voted_for: int | None = None
log: list = field(default_factory=lambda: [Entry(0, None)])
commit: int = 0
applied: int = 0
role: str = "follower"
state: dict = field(default_factory=dict)
results: dict = field(default_factory=dict)
match: dict = field(default_factory=dict)
def observe(self, term):
if term > self.term:
self.term, self.voted_for, self.role = term, None, "follower"
def vote(self, term, candidate, last_index, last_term):
self.observe(term)
fresh = (last_term, last_index) >= (self.log[-1].term, len(self.log)-1)
granted = term == self.term and fresh and self.voted_for in (None, candidate)
if granted:
self.voted_for = candidate
return self.term, granted
def receive(self, term, prev_index, prev_term, entries, leader_commit):
self.observe(term)
if term < self.term:
return self.term, False
self.role = "follower"
if prev_index >= len(self.log) or self.log[prev_index].term != prev_term:
return self.term, False
for offset, entry in enumerate(entries, prev_index + 1):
if offset < len(self.log) and self.log[offset].term != entry.term:
assert offset > self.commit, "a committed prefix must never be replaced"
del self.log[offset:]
if offset == len(self.log):
self.log.append(deepcopy(entry))
# Limit to the last index proven consistent by THIS AppendEntries.
self.commit = max(self.commit, min(leader_commit, prev_index + len(entries)))
self.apply()
return self.term, True
def apply(self):
while self.applied < self.commit:
self.applied += 1
command = self.log[self.applied].command
if command is None:
continue
client, seq, op, key, value = command
request = (client, seq)
if request in self.results:
continue
if op == "put":
# A log entry is a value snapshot, not mutable state storage.
self.state[key] = deepcopy(value)
elif op == "add":
self.state[key] = self.state.get(key, 0) + value
elif op != "get":
raise ValueError(op)
self.results[request] = deepcopy(self.state.get(key))
class Cluster:
def __init__(self, count=5):
self.nodes = [Node(i) for i in range(count)]
self.links = [[True]*count for _ in range(count)]
self.alive = [True]*count
self.majority = count//2+1
def connected(self, a, b):
return self.alive[a] and self.alive[b] and self.links[a][b] and self.links[b][a]
def partition(self, *groups):
members = [i for group in groups for i in group]
if sorted(members) != list(range(len(self.nodes))):
raise ValueError("partition must list every member exactly once")
self.links = [[any(a in g and b in g for g in groups)
for b in range(len(self.nodes))] for a in range(len(self.nodes))]
def heal(self):
n = len(self.nodes)
self.links = [[True]*n for _ in range(n)]
def elect(self, candidate):
if not self.alive[candidate]:
return False
node = self.nodes[candidate]
node.term += 1
node.role, node.voted_for = "candidate", candidate
term, votes = node.term, 1
for other in self.nodes:
if other.ident == candidate or not self.connected(candidate, other.ident):
continue
observed, granted = other.vote(term, candidate, len(node.log)-1, node.log[-1].term)
node.observe(observed)
votes += granted
if node.term == term and node.role == "candidate" and votes >= self.majority:
node.role = "leader"
node.match = {i: 0 for i in range(len(self.nodes))}
node.match[candidate] = len(node.log)-1
return True
return False
def replicate(self, leader_id):
leader = self.nodes[leader_id]
if not self.alive[leader_id] or leader.role != "leader":
return False
# Full-prefix transfer is deliberately inefficient; it exposes conflict
# deletion without burying the model in nextIndex optimizations.
for other in self.nodes:
if other.ident == leader_id or not self.connected(leader_id, other.ident):
continue
term, ok = other.receive(leader.term, 0, 0, leader.log[1:], leader.commit)
leader.observe(term)
if leader.role != "leader":
return False
if ok:
leader.match[other.ident] = len(leader.log)-1
leader.match[leader_id] = len(leader.log)-1
for index in range(len(leader.log)-1, leader.commit, -1):
votes = sum(v >= index for v in leader.match.values())
if leader.log[index].term == leader.term and votes >= self.majority:
leader.commit = index
break
leader.apply()
for other in self.nodes:
if other.ident != leader_id and self.connected(leader_id, other.ident):
term, _ = other.receive(leader.term, 0, 0, leader.log[1:], leader.commit)
leader.observe(term)
return leader.role == "leader"
def submit(self, leader_id, command):
node = self.nodes[leader_id]
if not self.alive[leader_id] or node.role != "leader":
return False, None
# A read is also a log entry here: no stale local-only reads.
# Model the RPC serialization boundary: callers cannot mutate a log
# record after submitting it through a shared Python object reference.
node.log.append(Entry(node.term, deepcopy(command)))
index = len(node.log)-1
if not self.replicate(leader_id) or node.commit < index:
return False, None # outcome UNKNOWN to caller; not proof of abort
result = node.results.get(command[:2]) if command is not None else None
return True, deepcopy(result)
def restart(self, ident):
n = self.nodes[ident]
self.nodes[ident] = Node(ident, n.term, n.voted_for, deepcopy(n.log))
self.alive[ident] = True
if __name__ == "__main__":
c = Cluster()
print("election A:", c.elect(0))
print("put x=1:", c.submit(0, ("client", 1, "put", "x", 1)))
c.partition([0, 1], [2, 3, 4])
print("minority write (unknown):", c.submit(0, ("client", 2, "put", "x", 9)))
print("election C:", c.elect(2))
print("majority write:", c.submit(2, ("other", 1, "put", "x", 2)))
c.heal(); c.replicate(2)
print("after heal:", [n.state for n in c.nodes])