In this tutorial, we implement an end-to-end Practical Byzantine Fault Tolerance (PBFT) simulator using asyncio. We model a realistic distributed network with asynchronous message passing, configurable delays, and Byzantine nodes that intentionally deviate from the protocol. By explicitly implementing the pre-prepare, prepare, and commit phases, we explore how PBFT achieves consensus under adversarial conditions while respecting the theoretical 3f+1 bound. We also instrument the system to measure consensus latency and success rates as the number of malicious nodes increases, allowing us to empirically observe the limits of Byzantine fault tolerance.<\/p>\n
import asyncio\nimport random\nimport time\nimport hashlib\nfrom dataclasses import dataclass, field\nfrom typing import Dict, Set, Tuple, Optional, List\nimport matplotlib.pyplot as plt\n\n\nPREPREPARE = \"PREPREPARE\"\nPREPARE = \"PREPARE\"\nCOMMIT = \"COMMIT\"\n\n\n@dataclass(frozen=True)\nclass Msg:\n typ: str\n view: int\n seq: int\n digest: str\n sender: int\n\n\n@dataclass\nclass NetConfig:\n min_delay_ms: int = 5\n max_delay_ms: int = 40\n drop_prob: float = 0.0\n reorder_prob: float = 0.0<\/code><\/pre>\n<\/div>\n<\/div>\nWe establish the simulator\u2019s foundation by importing the required libraries and defining the core PBFT message types. We formalize network messages and parameters using dataclasses to ensure structured, consistent communication. We also define constants representing the three PBFT phases used throughout the system.<\/p>\n
\n\n\n\n<\/div>\n<\/div>\n<\/div>\n<\/div>\nCopy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\nclass Network:\n def __init__(self, cfg: NetConfig):\n self.cfg = cfg\n self.nodes: Dict[int, \"Node\"] = {}\n\n\n def register(self, node: \"Node\"):\n self.nodes[node.nid] = node\n\n\n async def send(self, dst: int, msg: Msg):\n if random.random() < self.cfg.drop_prob:\n return\n\n\n d = random.uniform(self.cfg.min_delay_ms, self.cfg.max_delay_ms) \/ 1000.0\n await asyncio.sleep(d)\n\n\n if random.random() < self.cfg.reorder_prob:\n await asyncio.sleep(random.uniform(0.0, 0.02))\n\n\n await self.nodes[dst].inbox.put(msg)\n\n\n async def broadcast(self, src: int, msg: Msg):\n tasks = []\n for nid in self.nodes.keys():\n tasks.append(asyncio.create_task(self.send(nid, msg)))\n await asyncio.gather(*tasks)<\/code><\/pre>\n<\/div>\n<\/div>\nWe implement an asynchronous network layer that simulates real-world message delivery with delays, reordering, and potential drops. We register nodes dynamically and use asyncio tasks to broadcast messages across the simulated network. We model non-deterministic communication behavior that directly impacts consensus latency and robustness.<\/p>\n
\n\n\n\n<\/div>\n<\/div>\n<\/div>\n<\/div>\nCopy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n@dataclass\nclass NodeConfig:\n n: int\n f: int\n primary_id: int = 0\n view: int = 0\n timeout_s: float = 2.0\n\n\nclass Node:\n def __init__(self, nid: int, net: Network, cfg: NodeConfig, byzantine: bool = False):\n self.nid = nid\n self.net = net\n self.cfg = cfg\n self.byzantine = byzantine\n\n\n self.inbox: asyncio.Queue[Msg] = asyncio.Queue()\n\n\n self.preprepare_seen: Dict[int, str] = {}\n self.prepare_votes: Dict[Tuple[int, str], Set[int]] = {}\n self.commit_votes: Dict[Tuple[int, str], Set[int]] = {}\n\n\n self.committed: Dict[int, str] = {}\n self.running = True\n\n\n @property\n def f(self) -> int:\n return self.cfg.f\n\n\n def _q_prepare(self) -> int:\n return 2 * self.f + 1\n\n\n def _q_commit(self) -> int:\n return 2 * self.f + 1\n\n\n @staticmethod\n def digest_of(payload: str) -> str:\n return hashlib.sha256(payload.encode(\"utf-8\")).hexdigest()<\/code><\/pre>\n<\/div>\n<\/div>\nWe define the configuration and internal state of each PBFT node participating in the protocol. We initialize data structures for tracking pre-prepare, prepare, and commit votes while supporting both honest and Byzantine behavior. We also implement quorum threshold logic and deterministic digest generation for request validation.<\/p>\n
\n\n\n\n<\/div>\n<\/div>\n<\/div>\n<\/div>\nCopy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n async def propose(self, payload: str, seq: int):\n if self.nid != self.cfg.primary_id:\n raise ValueError(\"Only the primary can propose in this simplified simulator.\")\n\n\n if not self.byzantine:\n dig = self.digest_of(payload)\n msg = Msg(PREPREPARE, self.cfg.view, seq, dig, self.nid)\n await self.net.broadcast(self.nid, msg)\n return\n\n\n for dst in self.net.nodes.keys():\n variant = f\"{payload}::to={dst}::salt={random.randint(0,10**9)}\"\n dig = self.digest_of(variant)\n msg = Msg(PREPREPARE, self.cfg.view, seq, dig, self.nid)\n await self.net.send(dst, msg)\n\n\n async def handle_preprepare(self, msg: Msg):\n seq = msg.seq\n dig = msg.digest\n\n\n if self.byzantine:\n if random.random() < 0.5:\n return\n fake_dig = dig if random.random() < 0.5 else self.digest_of(dig + \"::fake\")\n out = Msg(PREPARE, msg.view, seq, fake_dig, self.nid)\n await self.net.broadcast(self.nid, out)\n return\n\n\n if seq not in self.preprepare_seen:\n self.preprepare_seen[seq] = dig\n out = Msg(PREPARE, msg.view, seq, dig, self.nid)\n await self.net.broadcast(self.nid, out)\n\n\n async def handle_prepare(self, msg: Msg):\n seq, dig = msg.seq, msg.digest\n key = (seq, dig)\n voters = self.prepare_votes.setdefault(key, set())\n voters.add(msg.sender)\n\n\n if self.byzantine:\n return\n\n\n if self.preprepare_seen.get(seq) != dig:\n return\n\n\n if len(voters) >= self._q_prepare():\n out = Msg(COMMIT, msg.view, seq, dig, self.nid)\n await self.net.broadcast(self.nid, out)\n\n\n async def handle_commit(self, msg: Msg):\n seq, dig = msg.seq, msg.digest\n key = (seq, dig)\n voters = self.commit_votes.setdefault(key, set())\n voters.add(msg.sender)\n\n\n if self.byzantine:\n return\n\n\n if self.preprepare_seen.get(seq) != dig:\n return\n\n\n if seq in self.committed:\n return\n\n\n if len(voters) >= self._q_commit():\n self.committed[seq] = dig<\/code><\/pre>\n<\/div>\n<\/div>\nWe implement the core PBFT protocol logic, including proposal handling and the pre-prepare and prepare phases. We explicitly model Byzantine equivocation by allowing malicious nodes to send conflicting digests to different peers. We advance the protocol to the commit phase once the required prepare quorum is reached.<\/p>\n
\n\n\n\n<\/div>\n<\/div>\n<\/div>\n<\/div>\nCopy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n async def run(self):\n while self.running:\n msg = await self.inbox.get()\n if msg.typ == PREPREPARE:\n await self.handle_preprepare(msg)\n elif msg.typ == PREPARE:\n await self.handle_prepare(msg)\n elif msg.typ == COMMIT:\n await self.handle_commit(msg)\n\n\n def stop(self):\n self.running = False\n\n\ndef pbft_params(n: int) -> int:\n return (n - 1) \/\/ 3\n\n\nasync def run_single_consensus(\n n: int,\n malicious: int,\n net_cfg: NetConfig,\n payload: str = \"tx: pay Alice->Bob 5\",\n seq: int = 1,\n timeout_s: float = 2.0,\n seed: Optional[int] = None\n) -> Dict[str, object]:\n if seed is not None:\n random.seed(seed)\n\n\n f_max = pbft_params(n)\n f = f_max\n\n\n net = Network(net_cfg)\n cfg = NodeConfig(n=n, f=f, primary_id=0, view=0, timeout_s=timeout_s)\n\n\n mal_set = set(random.sample(range(n), k=min(malicious, n)))\n nodes: List[Node] = []\n for i in range(n):\n node = Node(i, net, cfg, byzantine=(i in mal_set))\n net.register(node)\n nodes.append(node)\n\n\n tasks = [asyncio.create_task(node.run()) for node in nodes]\n\n\n t0 = time.perf_counter()\n await nodes[cfg.primary_id].propose(payload, seq)\n\n\n honest = [node for node in nodes if not node.byzantine]\n target = max(1, len(honest))\n\n\n committed_honest = 0\n latency = None\n\n\n async def poll_commits():\n nonlocal committed_honest, latency\n while True:\n committed_honest = sum(1 for node in honest if seq in node.committed)\n if committed_honest >= target:\n latency = time.perf_counter() - t0\n return\n await asyncio.sleep(0.005)\n\n\n try:\n await asyncio.wait_for(poll_commits(), timeout=timeout_s)\n success = True\n except asyncio.TimeoutError:\n success = False\n latency = None\n\n\n for node in nodes:\n node.stop()\n for task in tasks:\n task.cancel()\n await asyncio.gather(*tasks, return_exceptions=True)\n\n\n digest_set = set(node.committed.get(seq) for node in honest if seq in node.committed)\n agreed = (len(digest_set) == 1) if success else False\n\n\n return {\n \"n\": n,\n \"f\": f,\n \"malicious\": malicious,\n \"mal_set\": mal_set,\n \"success\": success,\n \"latency_s\": latency,\n \"honest_committed\": committed_honest,\n \"honest_total\": len(honest),\n \"agreed_digest\": agreed,\n }<\/code><\/pre>\n<\/div>\n<\/div>\nWe complete the PBFT state machine by processing commit messages and finalizing decisions once commit quorums are satisfied. We run the node event loop to continuously process incoming messages asynchronously. We also include lifecycle controls to safely stop nodes after each experiment run.<\/p>\n
\n\n\n\n<\/div>\n<\/div>\n<\/div>\n<\/div>\nCopy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\nasync def latency_sweep(\n n: int = 10,\n max_malicious: Optional[int] = None,\n trials_per_point: int = 5,\n timeout_s: float = 2.0,\n net_cfg: Optional[NetConfig] = None,\n seed: int = 7\n):\n if net_cfg is None:\n net_cfg = NetConfig(min_delay_ms=5, max_delay_ms=35, drop_prob=0.0, reorder_prob=0.05)\n\n\n if max_malicious is None:\n max_malicious = n\n\n\n results = []\n random.seed(seed)\n\n\n for m in range(0, max_malicious + 1):\n latencies = []\n successes = 0\n agreements = 0\n\n\n for t in range(trials_per_point):\n out = await run_single_consensus(\n n=n,\n malicious=m,\n net_cfg=net_cfg,\n timeout_s=timeout_s,\n seed=seed + 1000*m + t\n )\n results.append(out)\n if out[\"success\"]:\n successes += 1\n latencies.append(out[\"latency_s\"])\n if out[\"agreed_digest\"]:\n agreements += 1\n\n\n avg_lat = sum(latencies)\/len(latencies) if latencies else None\n print(\n f\"malicious={m:2d} | success={successes}\/{trials_per_point} \"\n f\"| avg_latency={avg_lat if avg_lat is not None else 'NA'} \"\n f\"| digest_agreement={agreements}\/{successes if successes else 1}\"\n )\n\n\n return results\n\n\ndef plot_latency(results: List[Dict[str, object]], trials_per_point: int):\n by_m = {}\n for r in results:\n m = r[\"malicious\"]\n by_m.setdefault(m, []).append(r)\n\n\n xs, ys = [], []\n success_rate = []\n for m in sorted(by_m.keys()):\n group = by_m[m]\n lats = [g[\"latency_s\"] for g in group if g[\"latency_s\"] is not None]\n succ = sum(1 for g in group if g[\"success\"])\n xs.append(m)\n ys.append(sum(lats)\/len(lats) if lats else float(\"nan\"))\n success_rate.append(succ \/ len(group))\n\n\n plt.figure()\n plt.plot(xs, ys, marker=\"o\")\n plt.xlabel(\"Number of malicious (Byzantine) nodes\")\n plt.ylabel(\"Consensus latency (seconds) \u2014 avg over successes\")\n plt.title(\"PBFT Simulator: Latency vs Malicious Nodes\")\n plt.grid(True)\n plt.show()\n\n\n plt.figure()\n plt.plot(xs, success_rate, marker=\"o\")\n plt.xlabel(\"Number of malicious (Byzantine) nodes\")\n plt.ylabel(\"Success rate\")\n plt.title(\"PBFT Simulator: Success Rate vs Malicious Nodes\")\n plt.ylim(-0.05, 1.05)\n plt.grid(True)\n plt.show()\n\n\nasync def main():\n n = 10\n trials = 6\n f = pbft_params(n)\n print(f\"n={n} => PBFT theoretical max f = floor((n-1)\/3) = {f}\")\n print(\"Theory: safety\/liveness typically assumed when malicious <= f and timing assumptions hold.n\")\n\n\n results = await latency_sweep(\n n=n,\n max_malicious=min(n, f + 6),\n trials_per_point=trials,\n timeout_s=2.0,\n net_cfg=NetConfig(min_delay_ms=5, max_delay_ms=35, drop_prob=0.0, reorder_prob=0.05),\n seed=11\n )\n plot_latency(results, trials)\n\n\nawait main()<\/code><\/pre>\n<\/div>\n<\/div>\nWe orchestrate large-scale experiments by sweeping across different numbers of malicious nodes and collecting latency statistics. We aggregate results to analyze consensus success rates and visualize system behavior using plots. We run the full experiment pipeline and observe how PBFT degrades as the number of Byzantine faults approaches and exceeds theoretical limits.<\/p>\n
In conclusion, we gained hands-on insight into how PBFT behaves beyond textbook guarantees and how adversarial pressure impacts both latency and liveness in practice. We saw how quorum thresholds enforce safety, why consensus breaks down once Byzantine nodes exceed the tolerated bound, and how asynchronous networks amplify these effects. This implementation provides a practical foundation for experimenting with more advanced distributed-systems concepts, such as view changes, leader rotation, or authenticated messaging. It helps us build intuition for the design trade-offs that underpin modern blockchain and distributed trust systems.<\/p>\n
\nCheck out the Full Codes here<\/a>.\u00a0<\/strong>Also,\u00a0feel free to follow us on\u00a0Twitter<\/mark><\/a><\/strong>\u00a0and don\u2019t forget to join our\u00a0120k+ ML SubReddit<\/a><\/strong>\u00a0and Subscribe to\u00a0our Newsletter<\/a><\/strong>. Wait! are you on telegram?\u00a0now you can join us on telegram as well.<\/a><\/strong><\/p>\nThe post A Coding Implementation to Simulate Practical Byzantine Fault Tolerance with Asyncio, Malicious Nodes, and Latency Analysis<\/a> appeared first on MarkTechPost<\/a>.<\/p>","protected":false},"excerpt":{"rendered":"In this tutorial, we implement…<\/p>\n","protected":false},"author":1,"featured_media":29,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-472","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-uncategorized"],"_links":{"self":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/472","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=472"}],"version-history":[{"count":0,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/472\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/media\/29"}],"wp:attachment":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=472"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=472"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=472"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}