En este tutorial, implementamos un simulador de tolerancia práctica a fallas bizantinas (PBFT) de un extremo a otro utilizando asyncio. Modelamos una red distribuida realista con paso de mensajes asíncrono, retrasos configurables y nodos bizantinos que se desvían intencionalmente del protocolo. Al implementar explícitamente las fases de preparación previa, preparación y compromiso, exploramos cómo PBFT logra consenso en condiciones adversas respetando al mismo tiempo el límite teórico 3f+1. También instrumentamos el sistema para medir la latencia del consenso y las tasas de éxito a medida que aumenta la cantidad de nodos maliciosos, lo que nos permite observar empíricamente los límites de la tolerancia a fallas bizantinas.
importar asyncio importar tiempo de importación aleatorio importar hashlib desde clases de datos importar clase de datos, campo al escribir importar Dict, Set, Tuple, Opcional, Lista importar matplotlib.pyplot como plt PREPREPARE = "PREPREPARE" PREPARE = "PREPARE" COMMIT = "COMMIT" @dataclass(frozen=True) class Msg: typ: str view: int seq: int digest: str sender: int @dataclass class NetConfig: min_delay_ms: int = 5 max_delay_ms: int = 40 drop_prob: float = 0,0 reorder_prob: float = 0,0
Establecemos las bases del simulador importando las bibliotecas necesarias y definiendo los tipos de mensajes PBFT principales. Formalizamos mensajes y parámetros de red utilizando clases de datos para garantizar una comunicación estructurada y consistente. También definimos constantes que representan las tres fases PBFT utilizadas en todo el sistema.
clase Red: def __init__(self, cfg: NetConfig): self.cfg = cfg self.nodes: Dict[int, "Nodo"] = {} def registro(self, nodo: "Nodo"): self.nodes[nodo.nid] = nodo async def enviar(self, dst: int, msg: Msg): if random.random() < self.cfg.drop_prob: return d = random.uniform(self.cfg.min_delay_ms, self.cfg.max_delay_ms) / 1000.0 espera asyncio.sleep(d) si random.random() < self.cfg.reorder_prob: espera asyncio.sleep(random.uniform(0.0, 0.02)) espera self.nodes[dst].inbox.put(msg) async def transmisión(self, src: int, msg: Msg): tareas =[]para nid en self.nodes.keys(): tareas.append(asyncio.create_task(self.send(nid, msg))) await asyncio.gather(*tasks)
Implementamos una capa de red asíncrona que simula la entrega de mensajes del mundo real con retrasos, reordenamiento y posibles caídas. Registramos nodos dinámicamente y utilizamos tareas asincio para transmitir mensajes a través de la red simulada. Modelamos un comportamiento de comunicación no determinista que impacta directamente en la latencia y solidez del consenso.
@dataclass clase NodeConfig: n: int f: int Primary_id: int = 0 vista: int = 0 timeout_s: float = 2.0 clase Nodo: def __init__(self, nid: int, net: Network, cfg: NodeConfig, byzantine: bool = False): self.nid = nid self.net = net self.cfg = cfg self.byzantine = byzantine self.inbox: asyncio.Queue[Msg] = asyncio.Queue() self.preprepare_seen: Dict[int, str] = {} self.prepare_votes: Dict[Tuple[int, str], Set[int]] = {} self.commit_votes: Dict[Tuple[int, str], Set[int]] = {} self.committed: Dict[int, str] = {} self.running = Verdadero @property def f(self) -> int: regresa self.cfg.f def _q_prepare(self) -> int: regresa 2 * self.f + 1 def _q_commit(self) -> int: regresa 2 * self.f + 1 @staticmethod def digest_of(carga útil: str) -> str: regresa hashlib.sha256(payload.encode("utf-8")).hexdigest()
Definimos la configuración y el estado interno de cada nodo PBFT que participa en el protocolo. Inicializamos estructuras de datos para realizar un seguimiento de la preparación, preparación y confirmación de los votos, al mismo tiempo que apoyamos el comportamiento honesto y bizantino. También implementamos lógica de umbral de quórum y generación de resumen determinista para la validación de solicitudes.
async def proponer(self, payload: str, seq: int): if self.nid != self.cfg.primary_id: rise ValueError("Solo el primario puede proponer en este simulador simplificado.") if not self.byzantine: dig = self.digest_of(payload) msg = Msg(PREPREPARE, self.cfg.view, seq, dig, self.nid) await self.net.broadcast(self.nid, msg) regresa para dst en self.net.nodes.keys(): variante = f"{payload}::to={dst}::salt={random.randint(0,10**9)}" dig = self.digest_of(variant) msg = Msg(PREPREPARE, self.cfg.view, seq, dig, self.nid) espera self.net.send(dst, msg) async def handle_preprepare(self, msg: Msg): seq = msg.seq dig = msg.digest if self.byzantine: if random.random() < 0.5: return fake_dig = dig if random.random() < 0.5 else self.digest_of(dig + "::fake") out = Msg(PREPARE, msg.view, seq, fake_dig, self.nid) await self.net.broadcast(self.nid, out) regresa si la secuencia no está en self.preprepare_seen: self.preprepare_seen[seq] = excavar = Msg(PREPARE, msg.view, seq, dig, self.nid) await self.net.broadcast(self.nid, out) async def handle_prepare(self, msg: Msg): seq, dig = msg.seq, msg.digest clave = (seq, dig) votantes = self.prepare_votes.setdefault(key, set()) votantes.add(msg.sender) if self.byzantine: return if self.preprepare_seen.get(seq) != dig: return if len(voters) >= self._q_prepare(): out = Msg(COMMIT, msg.view, seq, dig, self.nid) await self.net.broadcast(self.nid, out) async def handle_commit(self, msg: Msg): seq, dig = msg.seq, msg.digest key = (seq, dig) votantes = self.commit_votes.setdefault(key, set()) votantes.add(msg.sender) if self.byzantine: return if self.preprepare_seen.get(seq) != dig: devolver si seq en self.committed: devolver si len(votantes) >= self._q_commit(): self.committed[seq] = dig
Implementamos la lógica central del protocolo PBFT, incluido el manejo de propuestas y las fases de preparación y preparación previa. Modelamos explícitamente la ambigüedad bizantina al permitir que nodos maliciosos envíen resúmenes conflictivos a diferentes pares. Avanzamos el protocolo a la fase de compromiso una vez que se alcanza el quórum de preparación requerido.
async def run(self): mientras self.running: msg = await self.inbox.get() if msg.typ == PREPREPARE: await self.handle_preprepare(msg) elif msg.typ == PREPARE: await self.handle_prepare(msg) elif msg.typ == COMMIT: await self.handle_commit(msg) def stop(self): self.running = False def pbft_params(n: int) -> int: return (n – 1) // 3 async def run_single_consensus( n: int, malicioso: int, net_cfg: NetConfig, carga útil: str = "tx: pay Alice->Bob 5", seq: int = 1, timeout_s: float = 2.0, semilla: Opcional[int] = Ninguno) -> Dict[str, object]: si la semilla no es Ninguna: random.seed(seed) f_max = pbft_params(n) f = f_max net = Network(net_cfg) cfg = NodeConfig(n=n, f=f, Primary_id=0, view=0, timeout_s=timeout_s) mal_set = set(random.sample(range(n), k=min(malicious, n))) nodos: Lista[Nodo] =[]para i en rango(n): nodo = Nodo(i, net, cfg, byzantine=(i en mal_set)) net.register(nodo) nodos.append(nodo) tareas = [asyncio.create_task(node.run()) para nodo en nodos] t0 = time.perf_counter() espera nodos[cfg.primary_id].propose(payload, seq) honesto = [nodo para nodo en nodos si no nodo.byzantine] objetivo = max(1, len(honest)) comprometido_honesto = 0 latencia = Ninguno async def poll_commits(): no local comprometido_honesto, latencia mientras Verdadero: comprometido_honesto = suma(1 para nodo en honesto si secuencia en nodo.committed) si comprometido_honesto >= objetivo: latencia = time.perf_counter() – t0 retorno en espera asyncio.sleep(0.005) intente: await asyncio.wait_for(poll_commits(), timeout=timeout_s) éxito = Verdadero excepto asyncio.TimeoutError: éxito = Falso latencia = Ninguno para nodo en nodos: node.stop() para tarea en tareas: task.cancel() await asyncio.gather(*tasks, return_exceptions=True) digest_set = set(node.committed.get(seq) for node in honest if seq in node.committed) acordado = (len(digest_set) == 1) si tiene éxito else False return { "n": n, "f": f, "malicious": malicioso, "mal_set": mal_set, "success": éxito, "latency_s": latencia, "honest_committed": comprometido_honest, "honest_total": len(honesto), "agreed_digest": de acuerdo, }
Completamos la máquina de estado PBFT procesando mensajes de confirmación y finalizando decisiones una vez que se satisfacen los quórumes de confirmación. Ejecutamos el bucle de eventos del nodo para procesar continuamente los mensajes entrantes de forma asincrónica. También incluimos controles del ciclo de vida para detener los nodos de forma segura después de cada ejecución del experimento.
async def latency_sweep( n: int = 10, max_malicious: Opcional[int] = Ninguno, testing_per_point: int = 5, timeout_s: float = 2.0, net_cfg: Opcional[NetConfig] = Ninguno, semilla: int = 7): si net_cfg es Ninguno: net_cfg = NetConfig(min_delay_ms=5, max_delay_ms=35, drop_prob=0.0, reorder_prob=0.05) si max_malicious es Ninguno: max_malicious = n resultados =[]random.seed(seed) para m en el rango(0, max_malicious + 1): latencias =[]éxitos = 0 acuerdos = 0 para t en rango (pruebas_por_punto): salida = espera run_single_consensus( n=n, malicioso=m, net_cfg=net_cfg, timeout_s=timeout_s, semilla=semilla + 1000*m + t ) resultados.append(fuera) si fuera["éxito"]: éxitos += 1 latencias.append(out["latency_s"]) if out["agreed_digest"]: acuerdos += 1 avg_lat = suma(latencias)/len(latencias) si latencias else Ninguna print( f"malicious={m:2d} | Success={successes}/{trials_per_point} " f"| avg_latency={avg_lat si avg_lat no es Ninguno más 'NA'} " f"| digest_agreement={agreements}/{successes if Successes else 1}" ) devuelve resultados def plot_latency(resultados: List[Dict[str, object]], juicios_por_punto: int): by_m = {} for r en resultados: m = r["malicious"] by_m.setdefault(m,[]).append(r) xs, ys =[],[]tasa_éxito =[]para m en ordenado(by_m.keys()): grupo = by_m[m] lats = [g["latency_s"] para g en el grupo si g["latency_s"] no es Ninguno] succ = suma(1 para g en el grupo si g["success"]) xs.append(m) ys.append(sum(lats)/len(lats) if lats else float("nan")) Success_rate.append(succ / len(grupo)) plt.figure() plt.plot(xs, ys, marcador="o") plt.xlabel("Número de nodos maliciosos (bizantinos)") plt.ylabel("Latencia de consenso (segundos) – promedio de éxitos") plt.title("Simulador PBFT: latencia frente a nodos maliciosos") plt.grid(True) plt.show() plt.figure() plt.plot(xs, Success_rate, Marker="o") plt.xlabel("Número de nodos maliciosos (bizantinos)") plt.ylabel("Tasa de éxito") plt.title("Simulador PBFT: tasa de éxito frente a nodos maliciosos") plt.ylim(-0.05, 1.05) plt.grid(True) plt.show() async def main(): n = 10 pruebas = 6 f = pbft_params(n) print(f"n={n} => PBFT máximo teórico f = piso((n-1)/3) = {f}") print("Teoría: seguridad/vivacidad típicamente asumida cuando malicioso <= f y suposiciones de tiempo son válidas.n") resultados = await latency_sweep( n=n, max_malicious=min(n, f + 6), juicios_por_punto=ensayos, timeout_s=2.0, net_cfg=NetConfig(min_delay_ms=5, max_delay_ms=35, drop_prob=0.0, reorder_prob=0.05), seed=11) plot_latency(resultados, pruebas) await main()
Orquestamos experimentos a gran escala barriendo diferentes números de nodos maliciosos y recopilando estadísticas de latencia. Agregamos resultados para analizar las tasas de éxito del consenso y visualizar el comportamiento del sistema mediante gráficos. Ejecutamos todo el proceso de experimentos y observamos cómo PBFT se degrada a medida que el número de fallas bizantinas se acerca y excede los límites teóricos.
En conclusión, obtuvimos información práctica sobre cómo se comporta PBFT más allá de las garantías de los libros de texto y cómo la presión adversaria afecta tanto la latencia como la vitalidad en la práctica. Vimos cómo los umbrales de quórum refuerzan la seguridad, por qué el consenso se rompe una vez que los nodos bizantinos exceden el límite tolerado y cómo las redes asincrónicas amplifican estos efectos. Esta implementación proporciona una base práctica para experimentar con conceptos de sistemas distribuidos más avanzados, como cambios de vista, rotación de líderes o mensajes autenticados. Nos ayuda a desarrollar la intuición para las compensaciones de diseño que sustentan la cadena de bloques moderna y los sistemas de confianza distribuidos.
Consulte los códigos completos aquí. Además, no dude en seguirnos en Twitter y no olvide unirse a nuestro SubReddit de más de 120.000 ML y suscribirse a nuestro boletín. ¡Esperar! estas en telegrama? Ahora también puedes unirte a nosotros en Telegram.
