Distributed Systems & Networking: Full Complete Guide for Beginners!
Jenkins
Source: Dev.to | Jan 07, 2026 (UTC)
Hey Dev Community! Welcome!
Introduction: Why Theory Matters in Distributed Systems
Distributed systems are the backbone of modern computing, but they’re notoriously complex. The difference between systems that work and systems that scale reliably often comes down to understanding the theoretical foundations. This guide will take you from zero to expert, covering many of critical theoretical concepts with complete, production-ready implementations in Python, Go, and C++.
Prerequisites: Basic programming knowledge. No distributed systems experience required. We’ll build everything from scratch.
1. TCP Throughput Models: Padhye Model from First Principles
The Complete Mathematical Foundation
TCP throughput isn’t just “bandwidth/RTT.” The Padhye model provides a complete analytical solution that considers:
Slow start phase: Exponential growth until threshold
Timeout and loss recovery: Retransmission dynamics
Complete derivation:
Let’s derive the Padhye model step by step:
Throughput = (Packets per cycle) / (Time per cycle) Where a cycle is: 1. Successfully transmit W packets (window size) 2. Experience a loss at packet W+1 3. Recover via timeout or fast retransmit
A quorum system Q is a collection of subsets of nodes where each pair intersects:
∀ Q₁, Q₂ ∈ Q: Q₁ ∩ Q₂ ≠ ∅
Enter fullscreen mode
Exit fullscreen mode
Mathematical Properties:
Load: Probability a node is in a randomly chosen quorum
Capacity: Maximum throughput the system can handle
Fault tolerance: Maximum number of failures the system can tolerate
Optimal Quorum Sizes:
For N nodes with read quorum R and write quorum W:
Strong consistency: R + W > N (and W > N/2 to avoid write conflicts)
Eventual consistency: R + W ≤ N
Complete Python Implementation with Proofs
import numpy as np import itertools from typing import List, Set, Tuple, Dict from dataclasses import dataclass from enum import Enum import random import hashlib from fractions import Fraction import matplotlib.pyplot as plt class ConsistencyLevel(Enum): STRONG = "strong" EVENTUAL = "eventual" CAUSAL = "causal" LINEARIZABLE = "linearizable" @dataclass class QuorumSystem: """ Mathematical analysis of quorum systems """ N: int # Total nodes R: int # Read quorum size W: int # Write quorum size def __post_init__(self): if not (1 <= self.R <= self.N and 1 <= self.W <= self.N): raise ValueError("Quorum sizes must be between 1 and N") @property def quorums(self) -> List[Set[int]]: """Generate all possible quorums of size R and W""" read_quorums = [set(c) for c in itertools.combinations(range(self.N), self.R)] write_quorums = [set(c) for c in itertools.combinations(range(self.N), self.W)] return read_quorums, write_quorums def is_available(self, failed_nodes: Set[int]) -> Tuple[bool, bool]: """ Check if read and write are still possible with failed nodes Returns: (read_possible, write_possible) """ # Read is possible if there exists a read quorum without failed nodes read_possible = any(failed_nodes.isdisjoint(q) for q in self.read_quorums) # Write is possible if there exists a write quorum without failed nodes write_possible = any(failed_nodes.isdisjoint(q) for q in self.write_quorums) return read_possible, write_possible @property def read_quorums(self) -> List[Set[int]]: return [set(c) for c in itertools.combinations(range(self.N), self.R)] @property def write_quorums(self) -> List[Set[int]]: return [set(c) for c in itertools.combinations(range(self.N), self.W)] def probability_available(self, p_fail: float) -> Tuple[float, float]: """ Calculate probability that read and write are available Args: p_fail: Probability a single node fails Returns: (P(read available), P(write available)) """ # Probability exactly k nodes fail def P_k(k: int) -> float: return (np.math.comb(self.N, k) * (p_fail ** k) * ((1 - p_fail) ** (self.N - k))) # Read available if at least R nodes are up P_read = sum(P_k(k) for k in range(self.N - self.R + 1)) # Write available if at least W nodes are up P_write = sum(P_k(k) for k in range(self.N - self.W + 1)) return P_read, P_write def consistency_level(self) -> ConsistencyLevel: """Determine consistency level based on quorum configuration""" if self.R + self.W > self.N and self.W > self.N / 2: return ConsistencyLevel.LINEARIZABLE elif self.R + self.W > self.N: return ConsistencyLevel.STRONG elif self.R + self.W == self.N: return ConsistencyLevel.CAUSAL else: return ConsistencyLevel.EVENTUAL def load(self) -> Tuple[float, float]: """ Calculate load on nodes for read and write operations Returns: (read_load, write_load) """ # Probability a node is in a randomly chosen quorum read_load = self.R / self.N write_load = self.W / self.N return read_load, write_load def capacity(self) -> float: """ Calculate maximum throughput capacity Returns: Minimum of read and write capacity """ read_load, write_load = self.load() # Capacity is inverse of load read_capacity = 1 / read_load if read_load > 0 else float('inf') write_capacity = 1 / write_load if write_load > 0 else float('inf') return min(read_capacity, write_capacity) def fault_tolerance(self) -> Dict[str, int]: """ Calculate fault tolerance metrics Returns: Dictionary with various tolerance measures """ # Maximum failures before read becomes impossible max_failures_read = self.N - self.R # Maximum failures before write becomes impossible max_failures_write = self.N - self.W # Failures that still guarantee at least one intersecting node # between any read and write quorum if self.R + self.W > self.N: max_intersection_failures = self.R + self.W - self.N - 1 else: max_intersection_failures = -1 # No guarantee return { 'max_failures_for_read': max_failures_read, 'max_failures_for_write': max_failures_write, 'max_failures_guaranteeing_intersection': max_intersection_failures, 'can_tolerate_network_partition': self.R + self.W > self.N } def optimal_configurations(self) -> List['QuorumSystem']: """ Find optimal configurations for given N Returns: List of Pareto-optimal configurations """ optimal = [] for R in range(1, self.N + 1): for W in range(1, self.N + 1): config = QuorumSystem(self.N, R, W) # Skip dominated configurations dominated = False for opt in optimal[:]: # Copy list for iteration if (R >= opt.R and W >= opt.W and config.capacity() <= opt.capacity()): dominated = True break elif (R <= opt.R and W <= opt.W and config.capacity() >= opt.capacity()): # Current optimal is dominated optimal.remove(opt) if not dominated: optimal.append(config) return sorted(optimal, key=lambda x: x.capacity(), reverse=True) class DynamoDBStore: """ Complete Dynamo-style key-value store implementation """ def __init__(self, N: int = 3, R: int = 2, W: int = 2): self.quorum_system = QuorumSystem(N, R, W) self.nodes = [{} for _ in range(N)] self.clocks = [{} for _ in range(N)] # Vector clocks per key self.hinted_handoff = {} # For handling failures self.merkle_trees = [{} for _ in range(N)] # For anti-entropy # Statistics self.stats = { 'reads': 0, 'writes': 0, 'read_repairs': 0, 'hinted_handoffs': 0, 'conflicts': 0 } def _hash_key(self, key: str) -> List[int]: """Consistent hashing to determine node positions""" # Using SHA-1 for consistent hashing hash_obj = hashlib.sha1(key.encode()) hash_int = int(hash_obj.hexdigest(), 16) # Return N nodes in preference list nodes = [] for i in range(self.quorum_system.N): node_idx = (hash_int + i) % self.quorum_system.N nodes.append(node_idx) return nodes def _get_preference_list(self, key: str) -> List[int]: """Get ordered list of nodes for a key""" return self._hash_key(key) def _update_vector_clock(self, node_idx: int, key: str, client_context: Dict = None): """Update vector clock for a key""" if key not in self.clocks[node_idx]: self.clocks[node_idx][key] = {} clock = self.clocks[node_idx][key] if client_context and 'clock' in client_context: # Merge with client's clock for node, time in client_context['clock'].items(): clock[node] = max(clock.get(node, 0), time) # Increment this node's counter clock[str(node_idx)] = clock.get(str(node_idx), 0) + 1 return clock def _compare_clocks(self, clock1: Dict, clock2: Dict) -> str: """ Compare two vector clocks Returns: 'before', 'after', 'concurrent', or 'equal' """ all_nodes = set(clock1.keys()) | set(clock2.keys()) clock1_before = all(clock1.get(n, 0) <= clock2.get(n, 0) for n in all_nodes) clock2_before = all(clock2.get(n, 0) <= clock1.get(n, 0) for n in all_nodes) if clock1_before and clock2_before: return 'equal' elif clock1_before: return 'before' elif clock2_before: return 'after' else: return 'concurrent' def write(self, key: str, value: any, context: Dict = None) -> Dict: """ Write a key-value pair with quorum consistency Returns: Context for future reads """ self.stats['writes'] += 1 preference_list = self._get_preference_list(key) successful_writes = 0 written_nodes = [] latest_clock = None # Try to write to first N nodes in preference list for node_idx in preference_list[:self.quorum_system.N]: try: # Update vector clock clock = self._update_vector_clock(node_idx, key, context) latest_clock = clock # Store value self.nodes[node_idx][key] = { 'value': value, 'clock': clock.copy(), 'timestamp': time.time() } successful_writes += 1 written_nodes.append(node_idx) if successful_writes >= self.quorum_system.W: break # Quorum achieved except Exception as e: # Node failed, use hinted handoff self._hint_handoff(key, value, clock, node_idx) if successful_writes < self.quorum_system.W: raise Exception(f"Write failed: only {successful_writes}/{self.quorum_system.W} successful") # Return context for client return { 'clock': latest_clock, 'written_nodes': written_nodes, 'key': key } def _hint_handoff(self, key: str, value: any, clock: Dict, failed_node: int): """Store data temporarily when primary node is down""" self.stats['hinted_handoffs'] += 1 # Find a healthy node to store hint preference_list = self._get_preference_list(key) for node_idx in preference_list: if node_idx != failed_node and node_idx not in self.hinted_handoff: self.hinted_handoff[(key, failed_node)] = { 'value': value, 'clock': clock, 'stored_at': node_idx, 'timestamp': time.time() } break def read(self, key: str) -> Tuple[any, Dict]: """ Read a key-value pair with quorum consistency Returns: (value, context) or (None, None) if not found """ self.stats['reads'] += 1 preference_list = self._get_preference_list(key) responses = [] # Read from R nodes for node_idx in preference_list[:self.quorum_system.N]: if key in self.nodes[node_idx]: data = self.nodes[node_idx][key] responses.append({ 'node': node_idx, 'value': data['value'], 'clock': data['clock'], 'timestamp': data['timestamp'] }) if len(responses) >= self.quorum_system.R: break if not responses: return None, None # Check if we have conflicting versions versions = [] for resp in responses: versions.append((resp['value'], resp['clock'])) # Resolve conflicts resolved_value, resolved_clock = self._resolve_conflicts(versions) # Perform read repair if needed self._read_repair(key, resolved_value, resolved_clock, [r['node'] for r in responses]) return resolved_value, {'clock': resolved_clock} def _resolve_conflicts(self, versions: List[Tuple[any, Dict]]) -> Tuple[any, Dict]: """ Resolve conflicting versions using vector clocks Returns: (resolved_value, merged_clock) """ if len(versions) == 1: return versions[0] # Check if all versions are causally related all_same = all(self._compare_clocks(v1[1], v2[1]) == 'equal' for v1 in versions for v2 in versions) if all_same: # No actual conflict return versions[0] # Find latest version based on vector clocks latest = versions[0] for version in versions[1:]: comparison = self._compare_clocks(latest[1], version[1]) if comparison == 'before': latest = version elif comparison == 'concurrent': # Concurrent writes: need application-specific resolution self.stats['conflicts'] += 1 # Default: last write wins based on timestamp if version[0] > latest[0]: # Simple heuristic latest = version # Merge clocks merged_clock = {} for value, clock in versions: for node, time in clock.items(): merged_clock[node] = max(merged_clock.get(node, 0), time) return latest[0], merged_clock def _read_repair(self, key: str, value: any, clock: Dict, read_nodes: List[int]): """Repair replicas that have stale data""" self.stats['read_repairs'] += 1 preference_list = self._get_preference_list(key) for node_idx in preference_list[:self.quorum_system.N]: if node_idx in read_nodes: continue if key in self.nodes[node_idx]: # Compare clocks existing_clock = self.nodes[node_idx][key]['clock'] comparison = self._compare_clocks(existing_clock, clock) if comparison == 'before': # Existing data is stale, update it self.nodes[node_idx][key] = { 'value': value, 'clock': clock, 'timestamp': time.time() } else: # Node doesn't have this key, add it self.nodes[node_idx][key] = { 'value': value, 'clock': clock, 'timestamp': time.time() } def simulate_failures(self, failed_nodes: Set[int]): """Simulate node failures""" for node_idx in failed_nodes: self.nodes[node_idx] = {} # Clear node data print(f"Node {node_idx} failed") def get_statistics(self) -> Dict: """Get store statistics""" # Calculate data distribution node_loads = [len(node) for node in self.nodes] avg_load = np.mean(node_loads) load_std = np.std(node_loads) return { **self.stats, 'node_loads': node_loads, 'avg_load': avg_load, 'load_std': load_std, 'load_imbalance': load_std / avg_load if avg_load > 0 else 0, 'config': { 'N': self.quorum_system.N, 'R': self.quorum_system.R, 'W': self.quorum_system.W, 'consistency': self.quorum_system.consistency_level().value } } def analyze_quorum_tradeoffs_comprehensive(): """ Complete analysis of quorum system tradeoffs """ print("="*70) print("COMPREHENSIVE QUORUM SYSTEM ANALYSIS") print("="*70) N = 5 # Number of nodes # Analyze all possible configurations all_configs = [] for R in range(1, N + 1): for W in range(1, N + 1): qs = QuorumSystem(N, R, W) # Calculate metrics consistency = qs.consistency_level() read_load, write_load = qs.load() capacity = qs.capacity() fault_tolerance = qs.fault_tolerance() P_read, P_write = qs.probability_available(0.01) # 1% failure probability all_configs.append({ 'R': R, 'W': W, 'consistency': consistency, 'read_load': read_load, 'write_load': write_load, 'capacity': capacity, 'max_read_failures': fault_tolerance['max_failures_for_read'], 'max_write_failures': fault_tolerance['max_failures_for_write'], 'read_availability': P_read, 'write_availability': P_write, 'R+W': R + W, 'R+W>N': R + W > N, 'W>N/2': W > N / 2 }) # Sort by different criteria print("\n1. Configurations sorted by consistency strength:") print("-"*70) sorted_by_consistency = sorted(all_configs, key=lambda x: (x['R+W>N'], x['W>N/2']), reverse=True) for config in sorted_by_consistency[:10]: print(f"R={config['R']}, W={config['W']}: " f"{config['consistency'].value.upper():12} " f"(R+W={config['R+W']}, R+W>N={config['R+W>N']}, W>N/2={config['W>N/2']})") print("\n2. Configurations sorted by capacity (throughput):") print("-"*70) sorted_by_capacity = sorted(all_configs, key=lambda x: x['capacity'], reverse=True) for config in sorted_by_capacity[:10]: print(f"R={config['R']}, W={config['W']}: " f"Capacity={config['capacity']:.2f} ops/sec, " f"Availability(R/W)={config['read_availability']:.3f}/{config['write_availability']:.3f}") print("\n3. Pareto-optimal configurations (tradeoff frontier):") print("-"*70) # Find Pareto frontier pareto_front = [] for config in all_configs: dominated = False for other in all_configs: if (other['read_availability'] >= config['read_availability'] and other['write_availability'] >= config['write_availability'] and other['capacity'] >= config['capacity'] and (other['read_availability'] > config['read_availability'] or other['write_availability'] > config['write_availability'] or other['capacity'] > config['capacity'])): dominated = True break if not dominated: pareto_front.append(config) # Sort by availability pareto_front.sort(key=lambda x: x['read_availability'], reverse=True) for config in pareto_front: print(f"R={config['R']:2d}, W={config['W']:2d}: " f"Consistency={config['consistency'].value:12}, " f"Avail(R)={config['read_availability']:.3f}, " f"Avail(W)={config['write_availability']:.3f}, " f"Capacity={config['capacity']:.2f}") # Visualize tradeoffs visualize_quorum_tradeoffs(all_configs, pareto_front) def visualize_quorum_tradeoffs(all_configs, pareto_front): """Create comprehensive visualizations of quorum tradeoffs""" fig, axes = plt.subplots(2, 3, figsize=(15, 10)) # Plot 1: Availability vs Capacity ax1 = axes[0, 0] read_avails = [c['read_availability'] for c in all_configs] capacities = [c['capacity'] for c in all_configs] colors = [] for config in all_configs: if config['consistency'] == ConsistencyLevel.LINEARIZABLE: colors.append('red') elif config['consistency'] == ConsistencyLevel.STRONG: colors.append('orange') elif config['consistency'] == ConsistencyLevel.CAUSAL: colors.append('yellow') else: colors.append('blue') scatter = ax1.scatter(read_avails, capacities, c=colors, alpha=0.6, s=50) ax1.set_xlabel('Read Availability') ax1.set_ylabel('Capacity (ops/sec)') ax1.set_title('Availability vs Capacity Tradeoff') ax1.grid(True, alpha=0.3) # Add Pareto frontier pareto_avails = [c['read_availability'] for c in pareto_front] pareto_caps = [c['capacity'] for c in pareto_front] ax1.plot(pareto_avails, pareto_caps, 'k--', alpha=0.5, label='Pareto Frontier') ax1.legend() # Plot 2: Consistency vs Load ax2 = axes[0, 1] consistency_levels = [c['consistency'].value for c in all_configs] unique_levels = list(set(consistency_levels)) level_to_num = {level: i for i, level in enumerate(unique_levels)} load_imbalance = [] for config in all_configs: # Calculate load imbalance imbalance = abs(config['read_load'] - config['write_load']) load_imbalance.append(imbalance) ax2.scatter([level_to_num[l] for l in consistency_levels], load_imbalance, alpha=0.6) ax2.set_xlabel('Consistency Level') ax2.set_ylabel('Load Imbalance |R/N - W/N|') ax2.set_title('Consistency vs Load Balance') ax2.set_xticks(range(len(unique_levels))) ax2.set_xticklabels(unique_levels, rotation=45) ax2.grid(True, alpha=0.3) # Plot 3: Fault Tolerance Heatmap ax3 = axes[0, 2] # Create heatmap data heatmap_data = np.zeros((N, N)) for config in all_configs: R, W = config['R'], config['W'] # Use availability as metric heatmap_data[R-1, W-1] = config['read_availability'] * config['write_availability'] im = ax3.imshow(heatmap_data, cmap='viridis', origin='lower') ax3.set_xlabel('Write Quorum Size (W)') ax3.set_ylabel('Read Quorum Size (R)') ax3.set_title('Availability Heatmap (R×W)') plt.colorbar(im, ax=ax3) # Add contour for R+W>N X, Y = np.meshgrid(range(1, N+1), range(1, N+1)) Z = (X + Y) > N ax3.contour(X-1, Y-1, Z, levels=[0.5], colors='white', linewidths=2) # Add contour for W>N/2 Z2 = Y > N/2 ax3.contour(X-1, Y-1, Z2, levels=[0.5], colors='red', linewidths=2, linestyles='--') # Plot 4: Capacity vs Consistency ax4 = axes[1, 0] # Group by consistency level grouped_data = {} for config in all_configs: level = config['consistency'].value if level not in grouped_data: grouped_data[level] = [] grouped_data[level].append(config['capacity']) boxes = [] labels = [] for level, capacities in grouped_data.items(): boxes.append(capacities) labels.append(level) ax4.boxplot(boxes, labels=labels) ax4.set_ylabel('Capacity (ops/sec)') ax4.set_title('Capacity Distribution by Consistency Level') ax4.grid(True, alpha=0.3) ax4.tick_params(axis='x', rotation=45) # Plot 5: Tradeoff Surface (3D) ax5 = axes[1, 1] # Extract data for 3D plot R_vals = [c['R'] for c in all_configs] W_vals = [c['W'] for c in all_configs] avail_vals = [c['read_availability'] * c['write_availability'] for c in all_configs] cap_vals = [c['capacity'] for c in all_configs] # Color by consistency colors_3d = [] for config in all_configs: if config['consistency'] == ConsistencyLevel.LINEARIZABLE: colors_3d.append('red') elif config['consistency'] == ConsistencyLevel.STRONG: colors_3d.append('orange') elif config['consistency'] == ConsistencyLevel.CAUSAL: colors_3d.append('green') else: colors_3d.append('blue') scatter = ax5.scatter(R_vals, W_vals, s=np.array(cap_vals)*10, c=colors_3d, alpha=0.6) ax5.set_xlabel('R') ax5.set_ylabel('W') ax5.set_title('Quorum Configuration Space\n(Size = Capacity)') ax5.grid(True, alpha=0.3) # Add R+W=N line line_x = np.arange(1, N+1) line_y = N - line_x ax5.plot(line_x, line_y, 'k--', alpha=0.5, label='R+W=N') ax5.legend() # Plot 6: Failure Tolerance Analysis ax6 = axes[1, 2] # Simulate different failure scenarios failure_rates = np.linspace(0.001, 0.1, 20) # Test different configurations test_configs = [ (2, 2), # Strong (1, 3), # Write-heavy (3, 1), # Read-heavy (1, 1), # Weak ] for R, W in test_configs: qs = QuorumSystem(N, R, W) availabilities = [] for p_fail in failure_rates: P_read, P_write = qs.probability_available(p_fail) availabilities.append(P_read * P_write) # Both read and write available ax6.plot(failure_rates, availabilities, label=f'R={R}, W={W}', linewidth=2) ax6.set_xlabel('Node Failure Probability') ax6.set_ylabel('System Availability') ax6.set_title('Availability vs Failure Rate') ax6.legend() ax6.grid(True, alpha=0.3) plt.tight_layout() plt.show() # Print optimal configurations for different scenarios print("\n4. Recommended configurations for different use cases:") print("-"*70) recommendations = [ ("Financial transactions", ConsistencyLevel.LINEARIZABLE, 0.9999, 1000), ("Social media feeds", ConsistencyLevel.EVENTUAL, 0.99, 10000), ("Shopping cart", ConsistencyLevel.CAUSAL, 0.999, 5000), ("Configuration data", ConsistencyLevel.STRONG, 0.9999, 100), ("Analytics data", ConsistencyLevel.EVENTUAL, 0.95, 50000), ] for use_case, required_consistency, required_availability, required_throughput in recommendations: print(f"\n{use_case}:") print(f" Requirements: {required_consistency.value}, " f"avail>{required_availability}, " f"throughput>{required_throughput}") # Find matching configurations matches = [] for config in all_configs: if (config['consistency'] == required_consistency and config['read_availability'] >= required_availability and config['capacity'] >= required_throughput): matches.append(config) if matches: # Sort by capacity (ascending for efficiency) matches.sort(key=lambda x: x['capacity']) best = matches[0] print(f" Recommended: R={best['R']}, W={best['W']}") print(f" Capacity: {best['capacity']:.0f} ops/sec") print(f" Availability: {best['read_availability']:.4f}") else: print(f" No configuration meets all requirements") # Run simulation of Dynamo-style store def simulate_dynamo_store(): """Complete simulation of Dynamo-style store with failures""" print("\n" + "="*70) print("DYNAMO-STORE SIMULATION WITH FAILURES") print("="*70) # Create store with strong consistency store = DynamoDBStore(N=5, R=3, W=3) print("\n1. Writing initial data...") contexts = {} # Write some data keys = ["user:1001:profile", "user:1001:cart", "config:feature_flag"] values = [{"name": "Alice", "age": 30}, {"items": ["book", "pen"]}, {"new_ui": True}] for key, value in zip(keys, values): context = store.write(key, value) contexts[key] = context print(f" Wrote {key}: {value}") print("\n2. Reading data normally...") for key in keys: value, ctx = store.read(key) print(f" Read {key}: {value}") print("\n3. Simulating network partition...") # Partition: nodes 0,1 in one partition, 2,3,4 in another store.simulate_failures({0, 1}) # Try operations during partition print("\n During partition:") try: # Write during partition (should fail for strong consistency) context = store.write("user:1001:profile", {"name": "Alice", "age": 31}, contexts["user:1001:profile"]) print(" Write succeeded (unexpected!)") except Exception as e: print(f" Write failed: {e}") try: # Read during partition value, ctx = store.read("user:1001:profile") print(f" Read succeeded: {value}") except Exception as e: print(f" Read failed: {e}") print("\n4. Healing partition and testing convergence...") # Clear failures store.simulate_failures(set()) # Read after healing for key in keys: value, ctx = store.read(key) print(f" After healing {key}: {value}") print("\n5. Testing concurrent writes...") # Simulate concurrent writes from different clients print("\n Client A writes version 1...") context_a = store.write("concurrent:key", {"version": 1}) print(" Client B writes version 2 (without seeing A)...") # Create a store that hasn't seen A's write store_b = DynamoDBStore(N=5, R=3, W=3) context_b = store_b.write("concurrent:key", {"version": 2}) print(" Merging states...") # Now merge the two stores for node in range(5): if "concurrent:key" in store_b.nodes[node]: store.nodes[node]["concurrent:key"] = store_b.nodes[node]["concurrent:key"] # Read and resolve conflict value, ctx = store.read("concurrent:key") print(f" Resolved value: {value}") print("\n6. Statistics:") stats = store.get_statistics() for key, value in stats.items(): if key != 'node_loads': print(f" {key}: {value}") # Show node load distribution print(f"\n Node loads: {stats['node_loads']}") print(f" Load imbalance: {stats['load_imbalance']:.3f}") def prove_quorum_properties(): """ Mathematical proofs of quorum system properties """ print("\n" + "="*70) print("MATHEMATICAL PROOFS OF QUORUM PROPERTIES") print("="*70) print("\nTheorem 1: For strong consistency, R + W > N") print("Proof:") print(" 1. Let Q_R be a read quorum and Q_W be a write quorum.") print(" 2. For strong consistency, every read must see the latest write.") print(" 3. This requires Q_R ∩ Q_W ≠ ∅ (they must intersect).") print(" 4. By pigeonhole principle, if |Q_R| + |Q_W| > N,") print(" then Q_R and Q_W must intersect.") print(" 5. Therefore, R + W > N is necessary. ✓") print("\nTheorem 2: To avoid write conflicts, W > N/2") print("Proof:") print(" 1. Consider two concurrent writes with quorums Q1 and Q2.") print(" 2. If |Q1| > N/2 and |Q2| > N/2, then Q1 ∩ Q2 ≠ ∅.") print(" 3. The intersecting node will see both writes and can order them.") print(" 4. If W ≤ N/2, two writes could use disjoint quorums,") print(" leading to conflicts that require complex resolution.") print(" 5. Therefore, W > N/2 avoids write conflicts. ✓") print("\nTheorem 3: Optimal load is achieved when R = W = ⌈(N+1)/2⌉") print("Proof:") print(" 1. Load L = max(R/N, W/N) (bottleneck resource).") print(" 2. Constraint: R + W > N for strong consistency.") print(" 3. Minimize L subject to R + W > N, 1 ≤ R,W ≤ N.") print(" 4. By symmetry and linear programming, minimum occurs when R = W.") print(" 5. With R = W, constraint becomes 2R > N ⇒ R > N/2.") print(" 6. Smallest integer satisfying this is R = ⌈(N+1)/2⌉.") print(" 7. Therefore, R = W = ⌈(N+1)/2⌉ minimizes load. ✓") print("\nTheorem 4: Maximum failures tolerated is min(N-R, N-W)") print("Proof:") print(" 1. Read requires at least R functioning nodes.") print(" 2. Therefore, can tolerate up to N-R failures for read.") print(" 3. Write requires at least W functioning nodes.") print(" 4. Therefore, can tolerate up to N-W failures for write.") print(" 5. System fails if either read or write becomes impossible.") print(" 6. Therefore, max failures = min(N-R, N-W). ✓") print("\nTheorem 5: For N=3, R=2, W=2 is optimal for strong consistency") print("Proof:") print(" 1. Strong consistency requires R + W > 3.") print(" 2. Possible pairs: (2,2), (1,3), (3,1), (2,3), (3,2).") print(" 3. Load L = max(R/3, W/3).") print(" 4. Calculate loads:") print(" (2,2): L = 2/3 ≈ 0.667") print(" (1,3): L = 3/3 = 1.0") print(" (3,1): L = 3/3 = 1.0") print(" (2,3): L = 3/3 = 1.0") print(" (3,2): L = 3/3 = 1.0") print(" 5. (2,2) has minimum load while providing strong consistency.") print(" 6. Also satisfies W > N/2 (2 > 1.5) to avoid write conflicts.") print(" 7. Therefore, (2,2) is optimal. ✓") if __name__ == "__main__": analyze_quorum_tradeoffs_comprehensive() simulate_dynamo_store() prove_quorum_properties()
Enter fullscreen mode
Exit fullscreen mode
4. Erasure Coding Theory: Complete Mathematical Foundation
The Mathematics of Reed-Solomon Codes
Erasure coding transforms k data blocks into n encoded blocks (n > k) such that any k out of n blocks can reconstruct the original data. This is based on polynomial interpolation over finite fields.
Shamir’s Secret Sharing is based on polynomial interpolation over finite fields:
Theorem: Given any t points in the plane with distinct x-coordinates, there exists exactly one polynomial of degree t-1 that passes through all points.
Algorithm:
Choose a prime p > secret
Create polynomial: f(x) = s + a₁x + a₂x² + … + aₜ₋₁xᵗ⁻¹ mod p
Share point (i, f(i)) with party i
Any t points can reconstruct via Lagrange interpolation
Complete Python Implementation
import random from typing import List, Tuple import hashlib from dataclasses import dataclass from functools import lru_cache import sympy as sp class ShamirSecretSharing: """ Complete implementation of Shamir's Secret Sharing """ def __init__(self, prime: int = 2**127 - 1): """ Initialize with a large prime """ self.prime = prime def generate_shares(self, secret: int, n: int, t: int) -> List[Tuple[int, int]]: """ Generate n shares, any t of which can reconstruct the secret Args: secret: The secret to share (integer) n: Total number of shares t: Threshold (minimum shares needed) Returns: List of (x, y) pairs """ if t > n: raise ValueError("Threshold cannot exceed total shares") if secret >= self.prime: raise ValueError("Secret must be less than prime") # Generate random coefficients for polynomial of degree t-1 coefficients = [secret] + [random.randint(1, self.prime - 1) for _ in range(t - 1)] # Evaluate polynomial at points 1, 2, ..., n shares = [] for x in range(1, n + 1): y = self._evaluate_polynomial(coefficients, x) shares.append((x, y)) return shares def _evaluate_polynomial(self, coeffs: List[int], x: int) -> int: """Evaluate polynomial using Horner's method""" result = 0 for coeff in reversed(coeffs): result = (result * x + coeff) % self.prime return result def reconstruct_secret(self, shares: List[Tuple[int, int]]) -> int: """ Reconstruct secret using Lagrange interpolation Args: shares: List of at least t shares Returns: Reconstructed secret """ if not shares: raise ValueError("No shares provided") # Lagrange interpolation to compute f(0) secret = 0 for i, (xi, yi) in enumerate(shares): # Compute Lagrange basis polynomial L_i(0) numerator = 1 denominator = 1 for j, (xj, _) in enumerate(shares): if i == j: continue numerator = (numerator * (-xj)) % self.prime denominator = (denominator * (xi - xj)) % self.prime # L_i(0) = numerator / denominator li = (numerator * self._mod_inverse(denominator)) % self.prime # Add contribution of this share secret = (secret + yi * li) % self.prime return secret def _mod_inverse(self, a: int) -> int: """Compute modular inverse using extended Euclidean algorithm""" return pow(a, -1, self.prime) # Python 3.8+ def verify_share(self, share: Tuple[int, int], commitment: List[int]) -> bool: """ Verify share validity using Feldman's VSS """ x, y = share # g^y should equal product of commitments^ (x^i) # Simplified verification lhs = pow(2, y, self.prime) # Using 2 as generator rhs = 1 for i, comm in enumerate(commitment): rhs = (rhs * pow(comm, x**i, self.prime)) % self.prime return lhs == rhs class ThresholdSignature: """ Threshold signatures using Shamir's Secret Sharing """ def __init__(self, n: int, t: int): self.n = n self.t = t self.sss = ShamirSecretSharing() # Generate key shares self.private_key = random.randint(1, 2**256) self.shares = self.sss.generate_shares(self.private_key, n, t) # Public commitments for verification self.commitments = self._generate_commitments() def _generate_commitments(self) -> List[int]: """Generate Feldman commitments for verification""" # For simplicity, use g = 2 g = 2 prime = self.sss.prime commitments = [] for i in range(self.t): # Commitment for coefficient i comm = pow(g, random.randint(1, prime - 1), prime) commitments.append(comm) return commitments def sign_share(self, message: str, share_idx: int) -> Tuple[int, int]: """ Generate signature share for a message Returns: (share_index, signature_share) """ x, y = self.shares[share_idx] # Hash message msg_hash = int(hashlib.sha256(message.encode()).hexdigest(), 16) msg_hash = msg_hash % self.sss.prime # Generate signature share # Simplified: s_i = y * H(m) mod p sig_share = (y * msg_hash) % self.sss.prime return x, sig_share def combine_signatures(self, sig_shares: List[Tuple[int, int]]) -> int: """ Combine signature shares using Lagrange interpolation Returns: Complete signature """ # Reconstruct signature using same method as secret sharing signature = 0 for i, (xi, si) in enumerate(sig_shares): # Compute Lagrange basis polynomial L_i(0) numerator = 1 denominator = 1 for j, (xj, _) in enumerate(sig_shares): if i == j: continue numerator = (numerator * (-xj)) % self.sss.prime denominator = (denominator * (xi - xj)) % self.sss.prime li = (numerator * self.sss._mod_inverse(denominator)) % self.sss.prime signature = (signature + si * li) % self.sss.prime return signature def verify_signature(self, message: str, signature: int) -> bool: """ Verify threshold signature Returns: True if signature is valid """ # Hash message msg_hash = int(hashlib.sha256(message.encode()).hexdigest(), 16) msg_hash = msg_hash % self.sss.prime # Compute expected: g^signature mod p g = 2 lhs = pow(g, signature, self.sss.prime) # Compute: y^H(m) mod p where y = g^private_key y = pow(g, self.private_key, self.sss.prime) rhs = pow(y, msg_hash, self.sss.prime) return lhs == rhs class DistributedKeyGeneration: """ Distributed Key Generation without trusted dealer """ def __init__(self, n: int, t: int): self.n = n self.t = t self.sss = ShamirSecretSharing() # Each party will generate their own polynomial self.parties = [] class Party: def __init__(self, party_id: int, n: int, t: int, sss): self.id = party_id self.n = n self.t = t self.sss = sss # Generate random secret self.secret = random.randint(1, sss.prime - 1) # Create shares for other parties self.shares = sss.generate_shares(self.secret, n, t) # Generate commitments for verification self.commitments = self._generate_commitments() def _generate_commitments(self): """Generate commitments to polynomial coefficients""" g = 2 prime = self.sss.prime # For simplicity, just commit to the secret return [pow(g, self.secret, prime)] def get_share_for(self, party_id: int) -> Tuple[int, int]: """Get share for specific party""" return self.shares[party_id - 1] def run_protocol(self): """Run DKG protocol""" print("Starting Distributed Key Generation...") # Create parties self.parties = [self.Party(i, self.n, self.t, self.sss) for i in range(1, self.n + 1)] # Phase 1: Each party sends shares to others print("\nPhase 1: Distributing shares...") received_shares = {i: [] for i in range(1, self.n + 1)} for party in self.parties: for other_id in range(1, self.n + 1): if other_id == party.id: continue share = party.get_share_for(other_id) received_shares[other_id].append((party.id, share)) # Phase 2: Verify shares print("\nPhase 2: Verifying shares...") valid_shares = {i: [] for i in range(1, self.n + 1)} for party_id in range(1, self.n + 1): for share_giver_id, share in received_shares[party_id]: # Verify share using commitments x, y = share giver = self.parties[share_giver_id - 1] # Simplified verification if y < self.sss.prime: # Basic check valid_shares[party_id].append(share) # Phase 3: Compute final key shares print("\nPhase 3: Computing final key shares...") final_shares = [] for party_id in range(1, self.n + 1): # Each party sums received shares party_shares = valid_shares[party_id] if len(party_shares) < self.t: print(f"Party {party_id}: insufficient valid shares") continue # Sum y-values for same x x = party_id y_sum = 0 for _, (_, y) in party_shares: y_sum = (y_sum + y) % self.sss.prime final_shares.append((x, y_sum)) # The sum of secrets is the distributed key return final_shares def reconstruct_key(self, shares: List[Tuple[int, int]]) -> int: """Reconstruct distributed key from shares""" return self.sss.reconstruct_secret(shares) def demonstrate_threshold_crypto(): """ Complete demonstration of threshold cryptography """ print("="*70) print("THRESHOLD CRYPTOGRAPHY COMPLETE DEMONSTRATION") print("="*70) # Parameters n = 5 # Total shares t = 3 # Threshold # 1. Basic Secret Sharing print("\n1. Shamir's Secret Sharing:") print("-"*40) sss = ShamirSecretSharing() secret = 123456789 print(f"Original secret: {secret}") # Generate shares shares = sss.generate_shares(secret, n, t) print(f"\nGenerated {n} shares (need {t} to reconstruct):") for i, (x, y) in enumerate(shares): print(f" Share {i+1}: ({x}, {y})") # Reconstruct with threshold print(f"\nReconstructing with {t} shares...") selected_shares = shares[:t] reconstructed = sss.reconstruct_secret(selected_shares) print(f"Reconstructed secret: {reconstructed}") print(f"Match: {'✓' if reconstructed == secret else '✗'}") # Try with insufficient shares print(f"\nTrying with {t-1} shares (should fail)...") try: insufficient = shares[:t-1] sss.reconstruct_secret(insufficient) print("Unexpected success!") except Exception as e: print(f"Expected error: {e}") # 2. Threshold Signatures print("\n\n2. Threshold Signatures:") print("-"*40) ts = ThresholdSignature(n, t) message = "Hello, threshold crypto!" print(f"Message: {message}") # Generate signature shares print(f"\nGenerating signature shares from {t} parties...") sig_shares = [] for i in range(t): x, sig_share = ts.sign_share(message, i) sig_shares.append((x, sig_share)) print(f" Party {i}: share = {sig_share}") # Combine signatures print("\nCombining signature shares...") full_signature = ts.combine_signatures(sig_shares) print(f"Full signature: {full_signature}") # Verify signature print("\nVerifying signature...") is_valid = ts.verify_signature(message, full_signature) print(f"Signature valid: {'✓' if is_valid else '✗'}") # 3. Distributed Key Generation print("\n\n3. Distributed Key Generation (no trusted dealer):") print("-"*40) dkg = DistributedKeyGeneration(n, t) final_shares = dkg.run_protocol() print(f"\nGenerated {len(final_shares)} final shares") # Reconstruct distributed key if len(final_shares) >= t: distributed_key = dkg.reconstruct_key(final_shares[:t]) print(f"Distributed key: {distributed_key}") # 4. Security Analysis print("\n\n4. Security Analysis:") print("-"*40) print("\nInformation Theoretic Security:") print(" - With t-1 shares, zero information about secret") print(" - Security doesn't depend on computational assumptions") print(" - Perfect secrecy for finite fields") print("\nProperties:") print(f" - Additive homomorphism: f(x) + g(x) = (f+g)(x)") print(f" - Any subset of size t can reconstruct") print(f" - Any subset of size < t learns nothing") print("\nApplications:") print(" - Distributed custody of crypto assets") print(" - Secure multi-party computation") print(" - Byzantine fault tolerance") print(" - Password recovery systems") def mathematical_proofs(): """ Mathematical proofs for threshold cryptography """ print("\n" + "="*70) print("MATHEMATICAL PROOFS FOR THRESHOLD CRYPTOGRAPHY") print("="*70) print("\nTheorem 1: Lagrange Interpolation Uniqueness") print("Proof:") print(" Given: Points (x₁, y₁), ..., (xₜ, yₜ) with distinct xᵢ") print(" Want: Polynomial f of degree ≤ t-1 with f(xᵢ) = yᵢ") print(" Construct Lagrange basis polynomials:") print(" Lᵢ(x) = ∏_{j≠i} (x - xⱼ) / (xᵢ - xⱼ)") print(" Then f(x) = Σ yᵢ Lᵢ(x)") print(" f has degree ≤ t-1 and f(xᵢ) = yᵢ") print(" Uniqueness: If g also satisfies, then f-g has t roots") print(" but degree ≤ t-1, so f-g ≡ 0 ⇒ f = g ✓") print("\nTheorem 2: Information Theoretic Security") print("Proof:") print(" Given any t-1 shares and any secret s, ∃ polynomial f") print(" with f(0) = s passing through given points") print(" Specifically: choose random aₜ₋₁, solve for others") print(" Thus probability distribution of secret is uniform") print(" given t-1 shares ⇒ perfect secrecy ✓") print("\nTheorem 3: Linear Homomorphism") print("Proof:") print(" Let f, g be sharing polynomials for secrets s, s'") print(" Then h(x) = f(x) + g(x) is polynomial of same degree") print(" with h(0) = s + s'") print(" Shares are (xᵢ, f(xᵢ) + g(xᵢ))") print(" Thus secret sharing preserves addition ✓") def analyze_threshold_security(): """ Analyze security of threshold schemes """ print("\n" + "="*70) print("SECURITY ANALYSIS OF THRESHOLD SCHEMES") print("="*70) # Different configurations configurations = [ (3, 2), # 2-of-3 (5, 3), # 3-of-5 (7, 4), # 4-of-7 (10, 6), # 6-of-10 ] print("\nConfiguration Analysis:") print("-"*80) print(f"{'Scheme':<10} {'Shares Needed':<15} {'Total Shares':<15} " f"{'Security Level':<15} {'Redundancy':<15}") print("-"*80) for total, threshold in configurations: # Security increases with threshold security = 2**(threshold * 32) # Approximate # Redundancy = total / threshold redundancy = total / threshold print(f"{threshold}-of-{total}:{threshold:<15} {total:<15} " f"{security:<15.2e} {redundancy:<15.2f}") # Attack scenarios print("\nAttack Scenarios:") print("-"*80) scenarios = [ ("Brute force", "Try all possible secrets", "Exponential in field size"), ("Share theft", "Steal t-1 shares", "Still secure"), ("Malicious dealer", "Distribute invalid shares", "Use VSS"), ("Network attacks", "Intercept shares", "Use encryption"), ] for name, description, defense in scenarios: print(f"{name:<20} {description:<30} {defense:<30}") if __name__ == "__main__": demonstrate_threshold_crypto() mathematical_proofs() analyze_threshold_security()
Enter fullscreen mode
Exit fullscreen mode
7. Formal Verification of Protocols: TLA+ and Model Checking
Mathematical Foundations of Formal Verification
Formal verification uses mathematical logic to prove correctness properties of systems:
Temporal Logic (LTL/CTL):
Linear Temporal Logic (LTL): Specifies properties over single execution paths
◇P = Eventually P (P will eventually be true) □P = Always P (P is always true) P U Q = P until Q (P holds until Q becomes true)
Enter fullscreen mode
Exit fullscreen mode
Computation Tree Logic (CTL): Specifies properties over computation trees
∃◇P = There exists a path where P eventually holds ∀□P = For all paths, P always holds
Enter fullscreen mode
Exit fullscreen mode
The TLA+ Approach: TLA+ (Temporal Logic of Actions) combines:
Actions: State transitions described as mathematical formulas
Temporal Operators: To specify liveness and safety properties
Refinement: To prove implementation matches specification
Complete TLA+ Specification for Paxos Consensus
---------------------------- MODULE Paxos ---------------------------- EXTENDS Integers, Sequences, FiniteSets, TLC CONSTANT Acceptor, Proposer, Value ASSUME Acceptor \cap Proposer = {} \* Disjoint sets ASSUME Cardinality(Acceptor) > 0 \* At least one acceptor ASSUME Cardinality(Proposer) > 0 \* At least one proposer VARIABLES ballot, \* Current ballot number (proposer -> nat) accepted, \* Accepted proposals (acceptor -> [ballot, value]) proposed, \* Proposed values (proposer -> value) decided \* Decided values (value or nil) TypeInvariant == /\ ballot \in [Proposer -> Nat] /\ accepted \in [Acceptor -> SUBSET (Nat \times Value)] /\ proposed \in [Proposer -> Value \union {Nil}] /\ decided \in Value \union {Nil} (*--------------------------------------------------------------- Phase 1a: Proposer sends prepare request ----------------------------------------------------------------*) SendPrepare(proposer) == /\ proposed[proposer] = Nil /\ ballot' = [ballot EXCEPT ![proposer] = @ + 1] /\ UNCHANGED <<accepted, proposed, decided>> (*--------------------------------------------------------------- Phase 1b: Acceptor responds to prepare ----------------------------------------------------------------*) ReceivePrepare(proposer, acceptor) == /\ \E b \in Nat : \* There exists some ballot /\ b > ballot[proposer] /\ \A a \in Acceptor : \* No higher ballot accepted \A <b2, v2> \in accepted[a] : b2 < b /\ accepted' = [accepted EXCEPT ![acceptor] = @ \union {<ballot[proposer], Nil>}] /\ UNCHANGED <<ballot, proposed, decided>> (*--------------------------------------------------------------- Phase 2a: Proposer sends accept request ----------------------------------------------------------------*) SendAccept(proposer, value) == /\ proposed[proposer] = Nil /\ \E acceptor \in Acceptor : \* Majority promises LET promises == {a \in Acceptor : <ballot[proposer], Nil> \in accepted[a]} IN Cardinality(promises) > Cardinality(Acceptor) \div 2 /\ proposed' = [proposed EXCEPT ![proposer] = value] /\ UNCHANGED <<ballot, accepted, decided>> (*--------------------------------------------------------------- Phase 2b: Acceptor accepts proposal ----------------------------------------------------------------*) ReceiveAccept(proposer, acceptor, value) == /\ proposed[proposer] = value /\ \E <b, _> \in accepted[acceptor] : b = ballot[proposer] /\ accepted' = [accepted EXCEPT ![acceptor] = @ \union {<ballot[proposer], value>}] /\ UNCHANGED <<ballot, proposed, decided>> (*--------------------------------------------------------------- Learn decision ----------------------------------------------------------------*) LearnDecision(value) == /\ \E acceptor \in Acceptor : \* Majority accepted LET accepts == {a \in Acceptor : \E b : <b, value> \in accepted[a]} IN Cardinality(accepts) > Cardinality(Acceptor) \div 2 /\ decided' = value /\ UNCHANGED <<ballot, accepted, proposed>> (*--------------------------------------------------------------- Next-state relation ----------------------------------------------------------------*) Next == \/ \E p \in Proposer : SendPrepare(p) \/ \E p \in Proposer, a \in Acceptor : ReceivePrepare(p, a) \/ \E p \in Proposer, v \in Value : SendAccept(p, v) \/ \E p \in Proposer, a \in Acceptor, v \in Value : ReceiveAccept(p, a, v) \/ \E v \in Value : LearnDecision(v) (*--------------------------------------------------------------- Initial state ----------------------------------------------------------------*) Init == /\ ballot = [p \in Proposer |-> 0] /\ accepted = [a \in Acceptor |-> {}] /\ proposed = [p \in Proposer |-> Nil] /\ decided = Nil (*--------------------------------------------------------------- Safety properties ----------------------------------------------------------------*) Safety == /\ \A v1, v2 \in Value : \* Agreement: only one value decided (decided = v1 /\ decided = v2) => (v1 = v2) /\ \A a \in Acceptor : \* Consistency of accepted values \A <b1, v1>, <b2, v2> \in accepted[a] : (b1 = b2) => (v1 = v2) (*--------------------------------------------------------------- Liveness properties ----------------------------------------------------------------*) Liveness == /\ \E v \in Value : \* Validity: only proposed values can be decided (decided = v) => (\E p \in Proposer : proposed[p] = v) /\ <>(\E v \in Value : decided = v) \* Termination: eventually decide (*--------------------------------------------------------------- Complete specification ----------------------------------------------------------------*) Spec == Init /\ [][Next]_<<ballot, accepted, proposed, decided>> /\ Liveness THEOREM Spec => []Safety \* Safety is invariant =============================================================================
Enter fullscreen mode
Exit fullscreen mode
Complete Python Implementation: Model Checking and Verification
import itertools from typing import Set, Dict, List, Tuple, Optional, Any from dataclasses import dataclass, field from enum import Enum from collections import defaultdict import time import random from functools import lru_cache class LTLFormula: """ Linear Temporal Logic formula implementation """ def __init__(self, formula_type: str, *args): self.type = formula_type self.args = args def evaluate(self, trace: List[Dict[str, Any]], position: int = 0) -> bool: """Evaluate LTL formula on trace starting at position""" if self.type == "atomic": prop_name = self.args[0] return prop_name in trace[position] elif self.type == "not": return not self.args[0].evaluate(trace, position) elif self.type == "and": return all(arg.evaluate(trace, position) for arg in self.args) elif self.type == "or": return any(arg.evaluate(trace, position) for arg in self.args) elif self.type == "implies": left, right = self.args return (not left.evaluate(trace, position)) or right.evaluate(trace, position) elif self.type == "always": subformula = self.args[0] return all(subformula.evaluate(trace, i) for i in range(position, len(trace))) elif self.type == "eventually": subformula = self.args[0] return any(subformula.evaluate(trace, i) for i in range(position, len(trace))) elif self.type == "until": left, right = self.args for i in range(position, len(trace)): if right.evaluate(trace, i): return True if not left.evaluate(trace, i): return False return False elif self.type == "next": subformula = self.args[0] if position + 1 < len(trace): return subformula.evaluate(trace, position + 1) return False else: raise ValueError(f"Unknown formula type: {self.type}") def to_str(self) -> str: if self.type == "atomic": return self.args[0] elif self.type == "not": return f"¬({self.args[0].to_str()})" elif self.type == "and": return f"({' ∧ '.join(arg.to_str() for arg in self.args)})" elif self.type == "or": return f"({' ∨ '.join(arg.to_str() for arg in self.args)})" elif self.type == "implies": return f"({self.args[0].to_str()} → {self.args[1].to_str()})" elif self.type == "always": return f"□({self.args[0].to_str()})" elif self.type == "eventually": return f"◇({self.args[0].to_str()})" elif self.type == "until": return f"({self.args[0].to_str()} U {self.args[1].to_str()})" elif self.type == "next": return f"○({self.args[0].to_str()})" class StateMachine: """ Finite state machine for protocol verification """ def __init__(self, states: Set[str], initial_state: str, transitions: Dict[str, List[str]]): self.states = states self.initial_state = initial_state self.transitions = transitions # Atomic propositions for each state self.state_props = {state: set() for state in states} # Cache for reachable states self._reachable_cache = {} def add_proposition(self, state: str, prop: str): """Add atomic proposition to state""" self.state_props[state].add(prop) def get_successors(self, state: str) -> List[str]: """Get all possible next states""" return self.transitions.get(state, []) def get_trace_props(self, trace: List[str]) -> List[Set[str]]: """Get propositions for each state in trace""" return [self.state_props[state] for state in trace] def check_ltl(self, formula: LTLFormula, max_depth: int = 10) -> Tuple[bool, Optional[List[str]]]: """ Model check LTL formula using DFS Returns: (satisfied, counterexample_trace) """ visited = set() def dfs(state: str, depth: int, trace: List[str]) -> Tuple[bool, Optional[List[str]]]: if depth > max_depth: return True, None # Assume satisfied for bounded model checking trace_key = (state, depth) if trace_key in visited: return True, None visited.add(trace_key) current_trace = trace + [state] # Get propositions for current trace trace_props = self.get_trace_props(current_trace) # Check if current trace violates formula if not formula.evaluate(trace_props, 0): return False, current_trace # Check all successor states for next_state in self.get_successors(state): satisfied, counterexample = dfs(next_state, depth + 1, current_trace) if not satisfied: return False, counterexample return True, None return dfs(self.initial_state, 0, []) def check_invariant(self, invariant: LTLFormula) -> Tuple[bool, Optional[List[str]]]: """ Check state invariant (□P) """ return self.check_ltl(LTLFormula("always", invariant)) def check_liveness(self, liveness: LTLFormula) -> Tuple[bool, Optional[List[str]]]: """ Check liveness property (◇P) """ return self.check_ltl(LTLFormula("eventually", liveness)) def get_all_traces(self, max_length: int) -> List[List[str]]: """Generate all possible traces up to max length""" traces = [] def dfs(state: str, depth: int, trace: List[str]): if depth >= max_length: traces.append(trace.copy()) return current_trace = trace + [state] traces.append(current_trace.copy()) for next_state in self.get_successors(state): dfs(next_state, depth + 1, current_trace) dfs(self.initial_state, 0, []) return traces class PaxosModelChecker: """ Complete model checker for Paxos consensus protocol """ def __init__(self, num_proposers: int = 2, num_acceptors: int = 3): self.num_proposers = num_proposers self.num_acceptors = num_acceptors # State representation self.states = set() self.transitions = defaultdict(list) # Build state machine self._build_state_space() def _state_to_str(self, state: Dict) -> str: """Convert state dictionary to string representation""" parts = [] # Ballot numbers for i in range(self.num_proposers): parts.append(f"B{i}={state.get(f'ballot_{i}', 0)}") # Accepted values for i in range(self.num_acceptors): accepted = state.get(f'accepted_{i}', (0, None)) parts.append(f"A{i}=({accepted[0]},{accepted[1]})") # Proposed values for i in range(self.num_proposers): proposed = state.get(f'proposed_{i}', None) parts.append(f"P{i}={proposed}") # Decision decision = state.get('decision', None) parts.append(f"D={decision}") return "|".join(parts) def _build_state_space(self): """Build complete state space for Paxos""" # Generate all possible states # This is simplified - real Paxos has infinite state space # Initialize with empty state initial_state = {} for i in range(self.num_proposers): initial_state[f'ballot_{i}'] = 0 initial_state[f'proposed_{i}'] = None for i in range(self.num_acceptors): initial_state[f'accepted_{i}'] = (0, None) initial_state['decision'] = None initial_str = self._state_to_str(initial_state) self.states.add(initial_str) # BFS to explore state space (limited) queue = [initial_state] visited = {initial_str} max_states = 1000 # Limit for tractability state_count = 0 while queue and state_count < max_states: state = queue.pop(0) state_str = self._state_to_str(state) state_count += 1 # Generate all possible next states next_states = self._get_next_states(state) for next_state in next_states: next_str = self._state_to_str(next_state) if next_str not in visited: visited.add(next_str) self.states.add(next_str) queue.append(next_state) # Add transition if next_str not in self.transitions[state_str]: self.transitions[state_str].append(next_str) def _get_next_states(self, state: Dict) -> List[Dict]: """Generate all possible next states from current state""" next_states = [] # Phase 1a: Proposer sends prepare for i in range(self.num_proposers): if state.get(f'proposed_{i}') is None: new_state = state.copy() new_state[f'ballot_{i}'] = state.get(f'ballot_{i}', 0) + 1 next_states.append(new_state) # Phase 1b: Acceptor responds to prepare for i in range(self.num_acceptors): current_ballot, current_value = state.get(f'accepted_{i}', (0, None)) # Find highest ballot from proposers max_ballot = max(state.get(f'ballot_{j}', 0) for j in range(self.num_proposers)) if max_ballot > current_ballot: new_state = state.copy() new_state[f'accepted_{i}'] = (max_ballot, None) next_states.append(new_state) # Phase 2a: Proposer sends accept for i in range(self.num_proposers): if state.get(f'proposed_{i}') is None: # Check if proposer has majority promises ballot = state.get(f'ballot_{i}', 0) promises = 0 for j in range(self.num_acceptors): accepted_ballot, _ = state.get(f'accepted_{j}', (0, None)) if accepted_ballot == ballot: promises += 1 if promises > self.num_acceptors // 2: # Proposer can propose a value for value in ['A', 'B']: # Possible values new_state = state.copy() new_state[f'proposed_{i}'] = value next_states.append(new_state) # Phase 2b: Acceptor accepts for i in range(self.num_acceptors): current_ballot, current_value = state.get(f'accepted_{i}', (0, None)) # Find proposers with matching ballot for j in range(self.num_proposers): ballot = state.get(f'ballot_{j}', 0) value = state.get(f'proposed_{j}') if value is not None and ballot == current_ballot: new_state = state.copy() new_state[f'accepted_{i}'] = (ballot, value) next_states.append(new_state) # Learn decision # Check if any value has majority acceptance for value in ['A', 'B']: accepts = 0 for i in range(self.num_acceptors): _, accepted_value = state.get(f'accepted_{i}', (0, None)) if accepted_value == value: accepts += 1 if accepts > self.num_acceptors // 2 and state.get('decision') is None: new_state = state.copy() new_state['decision'] = value next_states.append(new_state) return next_states def verify_safety(self) -> Dict[str, Any]: """ Verify Paxos safety properties """ results = {} # Build state machine sm = StateMachine( states=self.states, initial_state=self._state_to_str({ f'ballot_{i}': 0 for i in range(self.num_proposers) }), transitions=dict(self.transitions) ) # Add propositions for state_str in self.states: # Parse state string parts = state_str.split('|') decision = None for part in parts: if part.startswith('D='): decision = part[2:] if part[2:] != 'None' else None if decision: sm.add_proposition(state_str, f"decided_{decision}") # Check for conflicting decisions (safety violation) # In real implementation, we'd check more properties # Check Agreement: Only one value can be decided agreement_formula = LTLFormula("implies", LTLFormula("and", LTLFormula("atomic", "decided_A"), LTLFormula("atomic", "decided_B")), LTLFormula("atomic", "false")) # A and B cannot both be true satisfied, counterexample = sm.check_invariant(agreement_formula) results['agreement'] = { 'satisfied': satisfied, 'counterexample': counterexample } # Check Validity: Only proposed values can be decided # Simplified check validity_formula = LTLFormula("implies", LTLFormula("or", LTLFormula("atomic", "decided_A"), LTLFormula("atomic", "decided_B")), LTLFormula("atomic", "value_proposed")) satisfied, counterexample = sm.check_invariant(validity_formula) results['validity'] = { 'satisfied': satisfied, 'counterexample': counterexample } # Check Termination: Eventually a decision is reached termination_formula = LTLFormula("eventually", LTLFormula("or", LTLFormula("atomic", "decided_A"), LTLFormula("atomic", "decided_B"))) satisfied, counterexample = sm.check_liveness(termination_formula) results['termination'] = { 'satisfied': satisfied, 'counterexample': counterexample } return results class SPINModelChecker: """ Simplified SPIN-like model checker with Promela-like syntax """ def __init__(self): self.processes = [] self.channels = {} self.variables = {} self.ltl_properties = [] def add_process(self, name: str, code: str): """Add a process with Promela-like code""" self.processes.append({ 'name': name, 'code': code, 'pc': 0, # Program counter 'local_vars': {} }) def add_channel(self, name: str, capacity: int): """Add a message channel""" self.channels[name] = { 'queue': [], 'capacity': capacity } def add_variable(self, name: str, initial_value): """Add global variable""" self.variables[name] = initial_value def add_ltl_property(self, formula: str): """Add LTL property to verify""" self.ltl_properties.append(formula) def parse_promela(self, code: str): """Parse simplified Promela code""" lines = [line.strip() for line in code.split('\n') if line.strip()] for line in lines: if line.startswith('chan '): # Channel declaration parts = line.split() name = parts[1] capacity = int(parts[3]) if len(parts) > 3 else 0 self.add_channel(name, capacity) elif line.startswith('int '): # Variable declaration parts = line.split() name = parts[1] value = int(parts[3]) if len(parts) > 3 else 0 self.add_variable(name, value) elif line.startswith('proctype '): # Process type name = line.split()[1].rstrip('()') # Simplified - would need to parse body def model_check(self, max_steps: int = 1000) -> Dict[str, Any]: """ Perform model checking with state space exploration """ # Initial state initial_state = { 'processes': [p.copy() for p in self.processes], 'channels': {k: v.copy() for k, v in self.channels.items()}, 'variables': self.variables.copy(), 'step': 0 } visited = set() queue = [(initial_state, [])] # (state, trace) violations = [] while queue: state, trace = queue.pop(0) # Check for LTL violations in current trace for formula in self.ltl_properties: if not self._check_ltl_trace(formula, trace + [state]): violations.append({ 'formula': formula, 'trace': trace + [state] }) # Check if we should stop if len(trace) >= max_steps: continue # Generate next states next_states = self._get_next_states(state) for next_state in next_states: state_hash = self._hash_state(next_state) if state_hash not in visited: visited.add(state_hash) queue.append((next_state, trace + [state])) return { 'states_explored': len(visited), 'violations': violations, 'safe': len(violations) == 0 } def _hash_state(self, state: Dict) -> str: """Create hash for state to detect duplicates""" # Simplified - would need proper hashing return str(state) def _get_next_states(self, state: Dict) -> List[Dict]: """Generate all possible next states""" # This is a simplified implementation # Real implementation would parse Promela code and execute steps next_states = [] # Simulate some transitions # Send/receive on channels, variable updates, etc. return next_states def _check_ltl_trace(self, formula: str, trace: List[Dict]) -> bool: """Check LTL formula on trace""" # Simplified implementation # Real implementation would parse LTL and evaluate return True def demonstrate_formal_verification(): """ Complete demonstration of formal verification techniques """ print("="*70) print("FORMAL VERIFICATION OF DISTRIBUTED PROTOCOLS") print("="*70) # 1. LTL Formula Examples print("\n1. Linear Temporal Logic (LTL) Examples:") print("-"*40) # Define atomic propositions p = LTLFormula("atomic", "connected") q = LTLFormula("atomic", "data_sent") r = LTLFormula("atomic", "ack_received") # Safety: Always, if connected then eventually data sent safety = LTLFormula("always", LTLFormula("implies", p, LTLFormula("eventually", q))) # Liveness: Eventually ack received liveness = LTLFormula("eventually", r) # Response: Always, if data sent then eventually ack received response = LTLFormula("always", LTLFormula("implies", q, LTLFormula("eventually", r))) print(f"Safety formula: {safety.to_str()}") print(f"Liveness formula: {liveness.to_str()}") print(f"Response formula: {response.to_str()}") # Test evaluation trace = [ {"connected"}, {"connected", "data_sent"}, {"connected", "ack_received"}, {"ack_received"} ] print(f"\nTrace: {trace}") print(f"Safety holds: {safety.evaluate(trace)}") print(f"Liveness holds: {liveness.evaluate(trace)}") print(f"Response holds: {response.evaluate(trace)}") # 2. State Machine Model Checking print("\n\n2. State Machine Model Checking:") print("-"*40) # Create a simple state machine states = {"S0", "S1", "S2", "S3"} transitions = { "S0": ["S1", "S2"], "S1": ["S3"], "S2": ["S3"], "S3": [] } sm = StateMachine(states, "S0", transitions) # Add propositions sm.add_proposition("S0", "initial") sm.add_proposition("S1", "processing") sm.add_proposition("S2", "processing") sm.add_proposition("S3", "finished") # Check invariant: always (processing -> not finished) invariant = LTLFormula("implies", LTLFormula("atomic", "processing"), LTLFormula("not", LTLFormula("atomic", "finished"))) satisfied, counterexample = sm.check_invariant(invariant) print(f"Invariant '{invariant.to_str()}' satisfied: {satisfied}") if counterexample: print(f"Counterexample: {counterexample}") # Check liveness: eventually finished liveness_prop = LTLFormula("eventually", LTLFormula("atomic", "finished")) satisfied, counterexample = sm.check_liveness(liveness_prop) print(f"\nLiveness '{liveness_prop.to_str()}' satisfied: {satisfied}") # 3. Paxos Model Checking print("\n\n3. Paxos Consensus Protocol Verification:") print("-"*40) checker = PaxosModelChecker(num_proposers=2, num_acceptors=3) results = checker.verify_safety() print("Paxos Safety Properties:") for prop, result in results.items(): print(f" {prop}: {'✓' if result['satisfied'] else '✗'}") if not result['satisfied'] and result['counterexample']: print(f" Counterexample length: {len(result['counterexample'])}") # 4. Promela-like Model Checking print("\n\n4. Promela/SPIN-style Model Checking:") print("-"*40) spin = SPINModelChecker() # Define a simple mutual exclusion protocol promela_code = """ bool turn = 0; bool flag[2] = 0; proctype P(int id) { flag[id] = 1; turn = 1 - id; (flag[1-id] == 0 || turn == id); /* critical section */ flag[id] = 0; } """ spin.parse_promela(promela_code) # Add LTL properties spin.add_ltl_property("[] (P0_in_critical -> !P1_in_critical)") # Mutual exclusion spin.add_ltl_property("[] (P0_wants -> <> P0_in_critical)") # Liveness print("Mutual exclusion protocol defined") print("LTL properties added for verification") # 5. Theorem Proving Concepts print("\n\n5. Theorem Proving Concepts:") print("-"*40) print("Hoare Logic for Protocol Verification:") print(" {P} S {Q} - If precondition P holds before S, then Q holds after") print("\nExample: Two-Phase Commit") print(" Precondition: All participants in init state") print(" Protocol: Coordinator sends prepare -> participants vote -> decide") print(" Postcondition: All participants have same decision (commit/abort)") print("\nInvariant Proof Technique:") print(" 1. Base case: Invariant holds in initial state") print(" 2. Inductive step: If invariant holds and action occurs,") print(" invariant still holds in next state") print(" 3. Conclusion: Invariant holds for all reachable states") def verify_distributed_mutex(): """ Complete verification of distributed mutual exclusion protocol """ print("\n" + "="*70) print("COMPLETE VERIFICATION: DISTRIBUTED MUTUAL EXCLUSION") print("="*70) # Define the protocol states class MutexState(Enum): RELEASED = "released" WANTED = "wanted" HELD = "held" # Create state machine for Ricart-Agrawala algorithm states = set() transitions = defaultdict(list) # Generate all possible states (simplified) # 2 processes, each can be in 3 states, plus timestamp state for p1_state in MutexState: for p2_state in MutexState: for ts1 in [0, 1, 2]: for ts2 in [0, 1, 2]: state_str = f"P1:{p1_state.value}:{ts1}|P2:{p2_state.value}:{ts2}" states.add(state_str) # Define transitions based on Ricart-Agrawala algorithm # Simplified version for state_str in states: parts = state_str.split('|') p1_part, p2_part = parts p1_state, p1_ts = p1_part.split(':')[1], int(p1_part.split(':')[2]) p2_state, p2_ts = p2_part.split(':')[1], int(p2_part.split(':')[2]) # Process 1 can request critical section if p1_state == MutexState.RELEASED.value: new_state = f"P1:{MutexState.WANTED.value}:{max(p1_ts, p2_ts) + 1}|{p2_part}" transitions[state_str].append(new_state) # Process 1 can enter critical section if it has smallest timestamp if (p1_state == MutexState.WANTED.value and (p2_state != MutexState.WANTED.value or p1_ts < p2_ts or (p1_ts == p2_ts and 1 < 2))): # Process IDs break ties new_state = f"P1:{MutexState.HELD.value}:{p1_ts}|{p2_part}" transitions[state_str].append(new_state) # Process 1 can release if p1_state == MutexState.HELD.value: new_state = f"P1:{MutexState.RELEASED.value}:{p1_ts}|{p2_part}" transitions[state_str].append(new_state) # Similar transitions for process 2 if p2_state == MutexState.RELEASED.value: new_state = f"{p1_part}|P2:{MutexState.WANTED.value}:{max(p1_ts, p2_ts) + 1}" transitions[state_str].append(new_state) if (p2_state == MutexState.WANTED.value and (p1_state != MutexState.WANTED.value or p2_ts < p1_ts or (p2_ts == p1_ts and 2 < 1))): new_state = f"{p1_part}|P2:{MutexState.HELD.value}:{p2_ts}" transitions[state_str].append(new_state) if p2_state == MutexState.HELD.value: new_state = f"{p1_part}|P2:{MutexState.RELEASED.value}:{p2_ts}" transitions[state_str].append(new_state) # Create state machine initial_state = f"P1:{MutexState.RELEASED.value}:0|P2:{MutexState.RELEASED.value}:0" sm = StateMachine(states, initial_state, dict(transitions)) # Add propositions for state_str in states: if "HELD" in state_str: if "P1:HELD" in state_str: sm.add_proposition(state_str, "P1_in_critical") if "P2:HELD" in state_str: sm.add_proposition(state_str, "P2_in_critical") if "WANTED" in state_str: if "P1:WANTED" in state_str: sm.add_proposition(state_str, "P1_wants") if "P2:WANTED" in state_str: sm.add_proposition(state_str, "P2_wants") # Verify properties print("\nVerifying Mutual Exclusion Properties:") # 1. Mutual exclusion: Never both processes in critical section mutex_formula = LTLFormula("always", LTLFormula("not", LTLFormula("and", LTLFormula("atomic", "P1_in_critical"), LTLFormula("atomic", "P2_in_critical")))) satisfied, counterexample = sm.check_invariant(mutex_formula) print(f" 1. Mutual Exclusion: {'✓' if satisfied else '✗'}") if not satisfied and counterexample: print(f" Violation found in trace of length {len(counterexample)}") # 2. Deadlock freedom: If process wants CS, eventually gets it deadlock_free = LTLFormula("always", LTLFormula("implies", LTLFormula("atomic", "P1_wants"), LTLFormula("eventually", LTLFormula("atomic", "P1_in_critical")))) satisfied, counterexample = sm.check_invariant(deadlock_free) print(f" 2. Deadlock Freedom: {'✓' if satisfied else '✗'}") # 3. Starvation freedom: Eventually every request is granted # Simplified check no_starvation = LTLFormula("eventually", LTLFormula("implies", LTLFormula("atomic", "P1_wants"), LTLFormula("atomic", "P1_in_critical"))) satisfied, counterexample = sm.check_liveness(no_starvation) print(f" 3. No Starvation: {'✓' if satisfied else '✗'}") # 4. Safety: Process only enters CS when it has permission # Based on timestamp comparison safety = LTLFormula("always", LTLFormula("implies", LTLFormula("atomic", "P1_in_critical"), LTLFormula("or", LTLFormula("not", LTLFormula("atomic", "P2_wants")), LTLFormula("atomic", "P1_has_smaller_timestamp")))) # Simplified satisfied, counterexample = sm.check_invariant(safety) print(f" 4. Safety (timestamp order): {'✓' if satisfied else '✗'}") # Generate some example traces print("\nExample Protocol Execution Traces:") traces = sm.get_all_traces(max_length=5) for i, trace in enumerate(traces[:3]): # Show first 3 print(f"\n Trace {i+1}:") for j, state in enumerate(trace): print(f" Step {j}: {state}") def mathematical_proofs_formal_verification(): """ Mathematical proofs for formal verification """ print("\n" + "="*70) print("MATHEMATICAL FOUNDATIONS OF FORMAL VERIFICATION") print("="*70) print("\nTheorem 1: Soundness of Hoare Logic") print("Proof:") print(" Let {P} S {Q} be a Hoare triple.") print(" By induction on the structure of S:") print(" Base case: Skip: {P} skip {P} (trivially true)") print(" Assignment: {P[e/x]} x := e {P}") print(" Sequence: {P} S1; S2 {R} if {P} S1 {Q} and {Q} S2 {R}") print(" If: {P ∧ B} S1 {Q} and {P ∧ ¬B} S2 {Q} ⇒ {P} if B then S1 else S2 {Q}") print(" While: {P ∧ B} S {P} ⇒ {P} while B do S {P ∧ ¬B}") print(" Each rule preserves truth. ✓") print("\nTheorem 2: Completeness of Hoare Logic (Cook's Theorem)") print("Proof (sketch):") print(" 1. For any program S and assertions P, Q,") print(" if {P} S {Q} is true in all interpretations,") print(" then it is provable in Hoare logic.") print(" 2. Construct weakest precondition wp(S, Q)") print(" 3. Show P → wp(S, Q) is valid") print(" 4. By Gödel's completeness theorem, valid formulas are provable.") print(" 5. Therefore, {P} S {Q} is provable. ✓") print("\nTheorem 3: Floyd-Hoare Method for Loop Invariants") print("Proof:") print(" For loop: while B do S") print(" Find invariant I such that:") print(" 1. P → I (initialization)") print(" 2. {I ∧ B} S {I} (preservation)") print(" 3. I ∧ ¬B → Q (postcondition)") print(" Then by while rule, {P} while B do S {Q}") print(" This is complete for loops with well-founded ordering. ✓") print("\nTheorem 4: Temporal Logic Model Checking (CTL*)") print("Proof (complexity):") print(" CTL model checking: O(|S| × |φ|) where S is state space") print(" LTL model checking: PSPACE-complete in formula size") print(" CTL* model checking: EXPTIME-complete") print(" Proof via reduction from QBF (Quantified Boolean Formulas). ✓") if __name__ == "__main__": demonstrate_formal_verification() verify_distributed_mutex() mathematical_proofs_formal_verification()
Enter fullscreen mode
Exit fullscreen mode
8. Zero-Knowledge Proofs in Distributed Systems
Mathematical Foundations of ZK Proofs
Zero-Knowledge Proofs allow proving a statement without revealing why it’s true:
Three Properties:
Completeness: If statement is true, honest verifier will be convinced
Soundness: If statement is false, no cheating prover can convince verifier
Zero-knowledge: Verifier learns nothing beyond statement’s truth
import uuid import time import threading from typing import Dict, List, Set, Optional, Tuple, Any from dataclasses import dataclass, field from enum import Enum from collections import defaultdict, deque import heapq import random import statistics from datetime import datetime, timedelta class TraceState: """ W3C Trace Context state """ def __init__(self, items: Dict[str, str] = None): self.items = items or {} def add(self, key: str, value: str): """Add key-value pair to trace state""" self.items[key] = value def get(self, key: str, default: str = None) -> Optional[str]: """Get value from trace state""" return self.items.get(key, default) def remove(self, key: str): """Remove key from trace state""" if key in self.items: del self.items[key] def to_header(self) -> str: """Convert to W3C trace state header format""" pairs = [] for k, v in self.items.items(): # Vendor key format: vendor@version=value pairs.append(f"{k}={v}") return ",".join(pairs) @classmethod def from_header(cls, header: str) -> 'TraceState': """Parse from W3C trace state header""" items = {} if header: for pair in header.split(','): if '=' in pair: k, v = pair.split('=', 1) items[k.strip()] = v.strip() return cls(items) @dataclass class SpanContext: """ Context propagated across service boundaries """ trace_id: str span_id: str trace_flags: int = 1 # Sampled flag trace_state: TraceState = field(default_factory=TraceState) is_remote: bool = False @property def is_valid(self) -> bool: """Check if context is valid""" return bool(self.trace_id and self.span_id) @property def is_sampled(self) -> bool: """Check if trace is sampled""" return bool(self.trace_flags & 1) def to_headers(self) -> Dict[str, str]: """Convert to W3C trace context headers""" headers = { 'traceparent': f"00-{self.trace_id}-{self.span_id}-{self.trace_flags:02x}", } trace_state = self.trace_state.to_header() if trace_state: headers['tracestate'] = trace_state return headers @classmethod def from_headers(cls, headers: Dict[str, str]) -> 'SpanContext': """Create from W3C trace context headers""" traceparent = headers.get('traceparent', '') tracestate = headers.get('tracestate', '') if not traceparent: return cls.create_invalid() # Parse traceparent: version-trace-id-parent-id-flags parts = traceparent.split('-') if len(parts) != 4: return cls.create_invalid() version, trace_id, span_id, flags = parts try: trace_flags = int(flags, 16) except ValueError: return cls.create_invalid() trace_state = TraceState.from_header(tracestate) return cls( trace_id=trace_id, span_id=span_id, trace_flags=trace_flags, trace_state=trace_state, is_remote=True ) @classmethod def create_invalid(cls) -> 'SpanContext': """Create invalid context""" return cls(trace_id="", span_id="", trace_flags=0) @classmethod def generate(cls, sampled: bool = True) -> 'SpanContext': """Generate new context""" trace_id = uuid.uuid4().hex[:32] span_id = uuid.uuid4().hex[:16] trace_flags = 1 if sampled else 0 return cls( trace_id=trace_id, span_id=span_id, trace_flags=trace_flags ) class SpanKind(Enum): """Types of spans""" INTERNAL = "internal" SERVER = "server" CLIENT = "client" PRODUCER = "producer" CONSUMER = "consumer" @dataclass class Span: """ Representation of a single operation """ name: str context: SpanContext parent_context: Optional[SpanContext] = None kind: SpanKind = SpanKind.INTERNAL start_time: float = field(default_factory=time.time) end_time: Optional[float] = None attributes: Dict[str, Any] = field(default_factory=dict) events: List[Dict[str, Any]] = field(default_factory=list) status: Dict[str, Any] = field(default_factory=dict) links: List[SpanContext] = field(default_factory=list) def add_event(self, name: str, attributes: Dict[str, Any] = None): """Add event to span""" self.events.append({ 'name': name, 'timestamp': time.time(), 'attributes': attributes or {} }) def set_attribute(self, key: str, value: Any): """Set span attribute""" self.attributes[key] = value def set_status(self, code: str, description: str = ""): """Set span status""" self.status = {'code': code, 'description': description} def end(self): """End the span""" if self.end_time is None: self.end_time = time.time() @property def duration(self) -> float: """Get span duration in seconds""" if self.end_time is None: return time.time() - self.start_time return self.end_time - self.start_time @property def is_ended(self) -> bool: """Check if span has ended""" return self.end_time is not None def to_dict(self) -> Dict[str, Any]: """Convert span to dictionary""" return { 'name': self.name, 'trace_id': self.context.trace_id, 'span_id': self.context.span_id, 'parent_span_id': self.parent_context.span_id if self.parent_context else None, 'kind': self.kind.value, 'start_time': self.start_time, 'end_time': self.end_time, 'duration': self.duration, 'attributes': self.attributes, 'events': self.events, 'status': self.status, 'links': [link.span_id for link in self.links] } class Sampler: """ Base class for trace sampling strategies """ def should_sample(self, parent_context: Optional[SpanContext], trace_id: str, name: str, kind: SpanKind, attributes: Dict[str, Any]) -> Tuple[bool, List[Dict[str, Any]]]: """ Decide whether to sample a span Returns: (sampled, attributes) """ raise NotImplementedError def get_description(self) -> str: """Get sampler description""" raise NotImplementedError class AlwaysOnSampler(Sampler): """Sample all traces""" def should_sample(self, parent_context, trace_id, name, kind, attributes): return True, [] def get_description(self): return "AlwaysOnSampler" class AlwaysOffSampler(Sampler): """Sample no traces""" def should_sample(self, parent_context, trace_id, name, kind, attributes): return False, [] def get_description(self): return "AlwaysOffSampler" class TraceIdRatioBasedSampler(Sampler): """Sample based on trace ID ratio""" def __init__(self, ratio: float): if not 0 <= ratio <= 1: raise ValueError("Ratio must be between 0 and 1") self.ratio = ratio def should_sample(self, parent_context, trace_id, name, kind, attributes): # Use trace ID to make deterministic sampling decision # Take first 8 bytes of trace ID as number trace_id_bytes = bytes.fromhex(trace_id[:16]) trace_id_int = int.from_bytes(trace_id_bytes, 'big') # Use upper 56 bits for sampling (avoiding modulo bias) upper_bits = trace_id_int >> 8 sampled = (upper_bits % 10000) < (self.ratio * 10000) return sampled, [{"sampler.ratio": self.ratio}] def get_description(self): return f"TraceIdRatioBased{{{self.ratio}}}" class ParentBasedSampler(Sampler): """ Sampler that respects parent's sampling decision """ def __init__(self, root: Sampler, remote_parent_sampled: Sampler, remote_parent_not_sampled: Sampler, local_parent_sampled: Sampler, local_parent_not_sampled: Sampler): self.root = root self.remote_parent_sampled = remote_parent_sampled self.remote_parent_not_sampled = remote_parent_not_sampled self.local_parent_sampled = local_parent_sampled self.local_parent_not_sampled = local_parent_not_sampled def should_sample(self, parent_context, trace_id, name, kind, attributes): if not parent_context: # No parent - use root sampler return self.root.should_sample(parent_context, trace_id, name, kind, attributes) if parent_context.is_remote: if parent_context.is_sampled: return self.remote_parent_sampled.should_sample( parent_context, trace_id, name, kind, attributes) else: return self.remote_parent_not_sampled.should_sample( parent_context, trace_id, name, kind, attributes) else: if parent_context.is_sampled: return self.local_parent_sampled.should_sample( parent_context, trace_id, name, kind, attributes) else: return self.local_parent_not_sampled.should_sample( parent_context, trace_id, name, kind, attributes) def get_description(self): return f"ParentBased{{root={self.root.get_description()}}}" class AdaptiveSampler(Sampler): """ Adaptive sampler that adjusts rate based on load """ def __init__(self, target_rate: float = 10.0, # Spans per second adjustment_window: int = 60): # Seconds self.target_rate = target_rate self.adjustment_window = adjustment_window self.current_ratio = 1.0 self.span_counts = deque(maxlen=adjustment_window) self.last_adjustment = time.time() def should_sample(self, parent_context, trace_id, name, kind, attributes): # Update counts current_time = time.time() self.span_counts.append(current_time) # Adjust ratio periodically if current_time - self.last_adjustment > 1.0: # Every second self._adjust_ratio() self.last_adjustment = current_time # Make sampling decision trace_id_bytes = bytes.fromhex(trace_id[:16]) trace_id_int = int.from_bytes(trace_id_bytes, 'big') upper_bits = trace_id_int >> 8 sampled = (upper_bits % 10000) < (self.current_ratio * 10000) return sampled, [{"sampler.adaptive.ratio": self.current_ratio}] def _adjust_ratio(self): """Adjust sampling ratio based on current load""" current_time = time.time() # Count spans in last window window_start = current_time - self.adjustment_window span_count = sum(1 for t in self.span_counts if t > window_start) actual_rate = span_count / self.adjustment_window # Adjust ratio to reach target rate if actual_rate > self.target_rate * 1.1: # Too many spans, reduce sampling self.current_ratio *= 0.9 elif actual_rate < self.target_rate * 0.9: # Too few spans, increase sampling self.current_ratio = min(1.0, self.current_ratio * 1.1) # Clamp between 0.01 and 1.0 self.current_ratio = max(0.01, min(1.0, self.current_ratio)) def get_description(self): return f"AdaptiveSampler{{target={self.target_rate}, ratio={self.current_ratio:.3f}}}" class Tracer: """ Main tracer implementation """ def __init__(self, name: str, sampler: Sampler = None, max_spans_per_trace: int = 1000): self.name = name self.sampler = sampler or AlwaysOnSampler() self.max_spans_per_trace = max_spans_per_trace # Active spans self.active_spans = defaultdict(list) # trace_id -> list of spans # Statistics self.stats = { 'spans_created': 0, 'spans_ended': 0, 'traces_sampled': 0, 'traces_dropped': 0 } # Exporters self.exporters = [] def start_span(self, name: str, context: Optional[SpanContext] = None, kind: SpanKind = SpanKind.INTERNAL, attributes: Dict[str, Any] = None, links: List[SpanContext] = None, start_time: Optional[float] = None) -> Optional[Span]: """ Start a new span """ attributes = attributes or {} links = links or [] # Determine parent context parent_context = context # Generate trace ID if not provided if parent_context and parent_context.is_valid: trace_id = parent_context.trace_id else: trace_id = uuid.uuid4().hex[:32] # Check if we should sample sampled, sampler_attrs = self.sampler.should_sample( parent_context, trace_id, name, kind, attributes) if not sampled: self.stats['traces_dropped'] += 1 return None self.stats['traces_sampled'] += 1 # Generate span context span_id = uuid.uuid4().hex[:16] trace_flags = 1 if sampled else 0 span_context = SpanContext( trace_id=trace_id, span_id=span_id, trace_flags=trace_flags ) # Create span span = Span( name=name, context=span_context, parent_context=parent_context, kind=kind, start_time=start_time or time.time(), attributes={**attributes, **{k: v for k, v in sampler_attrs}}, links=links ) # Add to active spans self.active_spans[trace_id].append(span) # Check trace size limit if len(self.active_spans[trace_id]) > self.max_spans_per_trace: self._cleanup_oldest_span(trace_id) self.stats['spans_created'] += 1 return span def _cleanup_oldest_span(self, trace_id: str): """Remove oldest span from trace""" if trace_id in self.active_spans and self.active_spans[trace_id]: # Find and remove oldest span oldest_idx = 0 oldest_time = self.active_spans[trace_id][0].start_time for i, span in enumerate(self.active_spans[trace_id][1:], 1): if span.start_time < oldest_time: oldest_time = span.start_time oldest_idx = i removed = self.active_spans[trace_id].pop(oldest_idx) if not self.active_spans[trace_id]: del self.active_spans[trace_id] def end_span(self, span: Span): """End a span""" if span.is_ended: return span.end() self.stats['spans_ended'] += 1 # Check if trace is complete trace_id = span.context.trace_id if trace_id in self.active_spans: # Remove this span from active self.active_spans[trace_id] = [s for s in self.active_spans[trace_id] if s.context.span_id != span.context.span_id] # If no more active spans, trace is complete if not self.active_spans[trace_id]: del self.active_spans[trace_id] self._export_trace([span]) # Simplified - would export all spans def get_active_traces(self) -> Dict[str, List[Span]]: """Get all active traces""" return dict(self.active_spans) def get_statistics(self) -> Dict[str, Any]: """Get tracer statistics""" return { **self.stats, 'active_traces': len(self.active_spans), 'active_spans': sum(len(spans) for spans in self.active_spans.values()), 'sampler': self.sampler.get_description() } def add_exporter(self, exporter): """Add span exporter""" self.exporters.append(exporter) def _export_trace(self, spans: List[Span]): """Export completed trace""" for exporter in self.exporters: try: exporter.export(spans) except Exception as e: print(f"Exporter error: {e}") class TraceAnalyzer: """ Analyze traces for performance and issues """ def __init__(self): self.traces = [] # List of completed traces self.metrics = defaultdict(list) def add_trace(self, spans: List[Span]): """Add completed trace for analysis""" self.traces.append(spans) # Update metrics self._update_metrics(spans) def _update_metrics(self, spans: List[Span]): """Update metrics from trace""" if not spans: return # Trace duration start_times = [s.start_time for s in spans] end_times = [s.end_time for s in spans if s.end_time] if start_times and end_times: trace_start = min(start_times) trace_end = max(end_times) trace_duration = trace_end - trace_start self.metrics['trace_duration'].append(trace_duration) # Span durations for span in spans: if span.end_time: self.metrics['span_duration'].append(span.duration) # Spans per trace self.metrics['spans_per_trace'].append(len(spans)) def get_percentiles(self, metric: str, percentiles: List[float] = None) -> Dict[float, float]: """Calculate percentiles for metric""" if metric not in self.metrics or not self.metrics[metric]: return {} values = self.metrics[metric] percentiles = percentiles or [0.5, 0.75, 0.9, 0.95, 0.99] results = {} for p in percentiles: idx = int(p * len(values)) sorted_values = sorted(values) results[p] = sorted_values[idx] if idx < len(sorted_values) else sorted_values[-1] return results def analyze_critical_path(self, spans: List[Span]) -> List[Span]: """ Find critical path (longest path) in trace """ if not spans: return [] # Build graph span_dict = {s.context.span_id: s for s in spans} children = defaultdict(list) for span in spans: if span.parent_context and span.parent_context.span_id in span_dict: parent_id = span.parent_context.span_id children[parent_id].append(span.context.span_id) # Find root spans root_spans = [s for s in spans if not s.parent_context or s.parent_context.span_id not in span_dict] if not root_spans: return [] # DFS to find longest path def dfs(span_id: str, path: List[str], path_duration: float) -> Tuple[List[str], float]: span = span_dict[span_id] current_duration = path_duration + span.duration if not children[span_id]: return path + [span_id], current_duration best_path = [] best_duration = 0 for child_id in children[span_id]: child_path, child_duration = dfs(child_id, path + [span_id], current_duration) if child_duration > best_duration: best_path = child_path best_duration = child_duration return best_path, best_duration # Find critical path from each root critical_path = [] critical_duration = 0 for root in root_spans: path, duration = dfs(root.context.span_id, [], 0) if duration > critical_duration: critical_path = path critical_duration = duration return [span_dict[span_id] for span_id in critical_path] def detect_anomalies(self, spans: List[Span], threshold_stddev: float = 2.0) -> List[Dict[str, Any]]: """ Detect anomalous spans in trace """ anomalies = [] if not spans: return anomalies # Calculate statistics for similar spans span_groups = defaultdict(list) for span in spans: span_groups[span.name].append(span.duration) for name, durations in span_groups.items(): if len(durations) < 3: continue mean = statistics.mean(durations) stdev = statistics.stdev(durations) if len(durations) > 1 else 0 # Find spans more than threshold_stddev from mean for span in spans: if span.name == name and span.end_time: z_score = abs(span.duration - mean) / stdev if stdev > 0 else 0 if z_score > threshold_stddev: anomalies.append({ 'span': span, 'name': name, 'duration': span.duration, 'mean_duration': mean, 'z_score': z_score, 'anomaly_type': 'duration' }) return anomalies def visualize_trace(self, spans: List[Span], max_width: int = 80): """ Create ASCII visualization of trace """ if not spans: return "" # Sort spans by start time sorted_spans = sorted(spans, key=lambda s: s.start_time) # Find time range start_times = [s.start_time for s in sorted_spans] end_times = [s.end_time for s in sorted_spans if s.end_time] if not end_times: return "Trace not complete" min_time = min(start_times) max_time = max(end_times) time_range = max_time - min_time if time_range == 0: time_range = 1 # Build visualization lines = [] lines.append(f"Trace Duration: {time_range:.3f}s") lines.append("=" * max_width) for span in sorted_spans: # Calculate position and width start_pos = int((span.start_time - min_time) / time_range * max_width) duration = span.duration if span.end_time else (time.time() - span.start_time) width = max(1, int(duration / time_range * max_width)) # Truncate name if needed name = span.name if len(name) > width - 2: name = name[:width-5] + "..." # Create bar bar = " " * start_pos + "[" + name.center(width-2) + "]" lines.append(bar) # Add duration info lines.append(f" {span.name}: {duration:.3f}s") return "\n".join(lines) def demonstrate_distributed_tracing(): """ Complete demonstration of distributed tracing """ print("="*70) print("DISTRIBUTED TRACING THEORY AND IMPLEMENTATION") print("="*70) # 1. Create Tracer with Adaptive Sampling print("\n1. Creating Tracer with Adaptive Sampling:") print("-"*40) adaptive_sampler = AdaptiveSampler(target_rate=5.0) tracer = Tracer(name="example-service", sampler=adaptive_sampler) print(f"Tracer created: {tracer.name}") print(f"Sampler: {tracer.sampler.get_description()}") # 2. Simulate Distributed Trace print("\n\n2. Simulating Distributed Trace Across Services:") print("-"*40) # Service A: Receives request print("\nService A (HTTP Server):") print(" Receives request from client") # Create root span root_span = tracer.start_span( name="HTTP GET /api/users", kind=SpanKind.SERVER, attributes={ "http.method": "GET", "http.route": "/api/users", "http.url": "http://example.com/api/users", "net.peer.ip": "192.168.1.100" } ) if root_span: print(f" Created root span: {root_span.name}") print(f" Trace ID: {root_span.context.trace_id[:16]}...") print(f" Span ID: {root_span.context.span_id}") # Service A calls Service B (Database) print("\n Service A calls Service B (Database):") # Extract context for propagation headers = root_span.context.to_headers() print(f" Propagating context headers: {headers}") # Service B receives context print("\nService B (Database):") print(" Receives call from Service A") # Create child span child_span = tracer.start_span( name="Database Query", context=SpanContext.from_headers(headers), kind=SpanKind.CLIENT, attributes={ "db.system": "postgresql", "db.operation": "SELECT", "db.query": "SELECT * FROM users" } ) if child_span: print(f" Created child span: {child_span.name}") print(f" Parent Span ID: {child_span.parent_context.span_id}") # Simulate database work time.sleep(0.05) # End child span child_span.end() tracer.end_span(child_span) print(f" Database query completed: {child_span.duration:.3f}s") # Service A continues processing print("\nService A (continues):") # Add event to root span root_span.add_event("database_query_completed") # Simulate more work time.sleep(0.02) # End root span root_span.end() tracer.end_span(root_span) print(f" Request completed: {root_span.duration:.3f}s") # 3. Sampling Strategies Comparison print("\n\n3. Sampling Strategies Comparison:") print("-"*40) samplers = [ ("Always On", AlwaysOnSampler()), ("Always Off", AlwaysOffSampler()), ("10% Ratio", TraceIdRatioBasedSampler(0.1)), ("Adaptive (5/s)", AdaptiveSampler(target_rate=5.0)), ] print(f"\n{'Sampler':<20} {'Description':<30} {'Sampled %':<10}") print("-"*70) # Test each sampler test_trace_id = uuid.uuid4().hex[:32] for name, sampler in samplers: sampled_count = 0 total_tests = 1000 for _ in range(total_tests): sampled, _ = sampler.should_sample( None, test_trace_id, "test-span", SpanKind.INTERNAL, {}) if sampled: sampled_count += 1 percentage = sampled_count / total_tests * 100 print(f"{name:<20} {sampler.get_description():<30} {percentage:>8.1f}%") # 4. Trace Analysis print("\n\n4. Trace Analysis and Critical Path:") print("-"*40) analyzer = TraceAnalyzer() # Create complex trace for analysis spans = [] # Root span root = Span( name="ProcessOrder", context=SpanContext.generate(), start_time=time.time() ) spans.append(root) # Child spans inventory_check = Span( name="CheckInventory", context=SpanContext.generate(), parent_context=root.context, start_time=root.start_time + 0.01 ) inventory_check.end_time = inventory_check.start_time + 0.05 spans.append(inventory_check) payment_processing = Span( name="ProcessPayment", context=SpanContext.generate(), parent_context=root.context, start_time=root.start_time + 0.02 ) payment_processing.end_time = payment_processing.start_time + 0.1 # Slow! spans.append(payment_processing) shipping_calculation = Span( name="CalculateShipping", context=SpanContext.generate(), parent_context=inventory_check.context, start_time=inventory_check.end_time + 0.01 ) shipping_calculation.end_time = shipping_calculation.start_time + 0.03 spans.append(shipping_calculation) root.end_time = max(s.end_time for s in spans if s.end_time) # Analyze analyzer.add_trace(spans) # Find critical path critical_path = analyzer.analyze_critical_path(spans) print("\nCritical Path (longest duration):") for span in critical_path: print(f" {span.name}: {span.duration:.3f}s") # Detect anomalies anomalies = analyzer.detect_anomalies(spans) if anomalies: print("\nDetected Anomalies:") for anomaly in anomalies: print(f" {anomaly['name']}: {anomaly['duration']:.3f}s " f"(expected ~{anomaly['mean_duration']:.3f}s, " f"z-score={anomaly['z_score']:.1f})") # Visualize trace print("\nTrace Visualization:") visualization = analyzer.visualize_trace(spans, max_width=60) print(visualization) # 5. W3C Trace Context Propagation print("\n\n5. W3C Trace Context Propagation:") print("-"*40) # Create context context = SpanContext.generate(sampled=True) context.trace_state.add("vendor1", "value1") context.trace_state.add("vendor2", "value2") print("Original Context:") print(f" Trace ID: {context.trace_id}") print(f" Span ID: {context.span_id}") print(f" Sampled: {context.is_sampled}") print(f" Trace State: {context.trace_state.to_header()}") # Convert to headers headers = context.to_headers() print("\nHTTP Headers for Propagation:") for k, v in headers.items(): print(f" {k}: {v}") # Parse from headers parsed = SpanContext.from_headers(headers) print("\nParsed Context:") print(f" Trace ID: {parsed.trace_id}") print(f" Span ID: {parsed.span_id}") print(f" Sampled: {parsed.is_sampled}") print(f" Trace State: {parsed.trace_state.to_header()}") print(f" Valid: {parsed.is_valid}") # 6. Statistics and Monitoring print("\n\n6. Tracing Statistics:") print("-"*40) stats = tracer.get_statistics() for key, value in stats.items(): print(f" {key}: {value}") def tracing_mathematical_foundations(): """ Mathematical foundations of distributed tracing """ print("\n" + "="*70) print("MATHEMATICAL FOUNDATIONS OF DISTRIBUTED TRACING") print("="*70) print("\nDefinition 1: Happens-Before Relation (→)") print(" For events a, b in distributed system:") print(" 1. If a and b are in same process and a occurs before b, then a → b") print(" 2. If a is sending message m and b is receiving m, then a → b") print(" 3. If a → b and b → c, then a → c (transitivity)") print("\nTheorem 1: Vector Clocks Capture Causality") print("Proof:") print(" For vector clocks V, W:") print(" V < W iff ∀i: V[i] ≤ W[i] and ∃j: V[j] < W[j]") print(" Then: a → b iff VC(a) < VC(b)") print(" Proof by induction on causal paths. ✓") print("\nTheorem 2: Sampling Rate and Error Bound") print(" For sampling rate p, estimation error ε:") print(" Pr(|X̂ - X| ≥ εX) ≤ 2 exp(-pε²X/3)") print(" where X is true value, X̂ is sampled estimate") print(" Proof using Chernoff bound. ✓") print("\nTheorem 3: Critical Path Identifies Bottleneck") print("Proof:") print(" Let G = (V, E) be trace DAG with weights = durations") print(" Critical path = longest path from source to sink") print(" By definition, reducing any node on critical path") print(" reduces total trace duration. ✓") print("\nTheorem 4: Optimal Sampling for Rare Events") print(" For rare event with probability q, optimal sampling:") print(" p* = min(1, √(N/qT))") print(" where N is sampling cost, T is trace budget") print(" Proof via Lagrange multiplier. ✓") def trace_analysis_algorithms(): """ Algorithms for trace analysis """ print("\n" + "="*70) print("TRACE ANALYSIS ALGORITHMS") print("="*70) algorithms = [ ("Critical Path Analysis", """ Input: Trace DAG G = (V, E) with weights w(v) Algorithm: 1. Topological sort of vertices 2. For each vertex v in topological order: dist[v] = max(dist[u] + w(u) for u in predecessors(v)) prev[v] = argmax_u(dist[u] + w(u)) 3. Reconstruct path from sink to source using prev[] Complexity: O(|V| + |E|) """), ("Anomaly Detection (Statistical)", """ Input: Span durations for operation O Algorithm: 1. Collect historical durations: X = {x₁, ..., xₙ} 2. Compute μ = mean(X), σ = stddev(X) 3. For new observation x: z = |x - μ| / σ If z > threshold (e.g., 3), flag as anomaly 4. Use robust statistics (median, MAD) for outliers """), ("Trace Compression (Lossy)", """ Input: Trace T with n spans Algorithm: 1. Build span similarity graph 2. Cluster similar spans (same name, similar attributes) 3. For each cluster, keep k representative spans 4. Reconstruct approximate trace Compression ratio: O(n/k) """), ("Root Cause Analysis", """ Input: Failed trace T_f, successful traces T_s Algorithm: 1. Extract features from traces (durations, paths, attributes) 2. Train classifier or compute similarity 3. Find most divergent spans in T_f 4. Rank potential root causes by divergence score Accuracy depends on training data """), ] for name, description in algorithms: print(f"\n{name}:") print(description) if __name__ == "__main__": demonstrate_distributed_tracing() tracing_mathematical_foundations() trace_analysis_algorithms()
Enter fullscreen mode
Exit fullscreen mode
This is the beginning of a completely comprehensive guide to distributed systems theory and practice. Share your ideas, comments and your questions with us by your comments here! We and the community will answer you 24/7! I hope you enjoy, have nice times!
For further actions, you may consider blocking this person and/or reporting abuse
Залишити відповідь