Chapter 9: Build Your Own
HLD, LLD, Class Diagrams & Implementation Roadmap
High-Level Design (HLD)
💡 System Overview
PCCL consists of 3 main components communicating over TCP/IP:
- Master Server - Coordination & consensus
- Peer Client - Training & collective operations
- P2P Mesh - Direct data transfer between peers
┌─────────────────────────────────────────────────────────────────────────┐
│ HIGH-LEVEL ARCHITECTURE │
└─────────────────────────────────────────────────────────────────────────┘
┌─────────────────┐
│ MASTER SERVER │
│ │
│ • Membership │
│ • Consensus │
│ • Topology │
└────────┬────────┘
│
Control Plane │ (lightweight messages)
────────────────┼────────────────
│
┌──────────────────────────┼──────────────────────────┐
│ │ │
▼ ▼ ▼
┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
│ PEER CLIENT │ │ PEER CLIENT │ │ PEER CLIENT │
│ │ │ │ │ │
│ • Model/Grads │◄────►│ • Model/Grads │◄────►│ • Model/Grads │
│ • Optimizer │ │ • Optimizer │ │ • Optimizer │
│ • Collectives │ │ • Collectives │ │ • Collectives │
└─────────────────┘ └─────────────────┘ └─────────────────┘
▲ ▲ ▲
│ │ │
└──────────────────────────┴──────────────────────────┘
Data Plane (P2P Mesh)
(heavy tensor data)
Component Responsibilities
| Component | Responsibilities | Communicates With |
|---|---|---|
| Master Server | Track peers, coordinate state transitions, optimize topology | All Peers (control) |
| Peer Client | Hold model weights, run training, execute collectives | Master + Other Peers |
| P2P Connection | Transfer tensor data directly between peers | Peer ↔ Peer only |
Entity Relationship Diagram
┌─────────────────────────────────────────────────────────────────────────┐
│ ENTITY RELATIONSHIP DIAGRAM │
└─────────────────────────────────────────────────────────────────────────┘
┌──────────────┐ 1:N ┌──────────────┐
│ Master │─────────────────►│ Peer │
│──────────────│ │──────────────│
│ id │ │ id (rank) │
│ address │ │ address │
│ port │ │ port │
│ topology │ │ phase │
└──────────────┘ │ state │
│ hash │
└──────┬───────┘
│
│ N:N
▼
┌──────────────┐
│ P2PConnection│
│──────────────│
│ peer_a_id │
│ peer_b_id │
│ bandwidth │
│ latency │
│ socket │
└──────────────┘
│
│ 1:N
▼
┌──────────────┐ N:1 ┌──────────────┐
│ Operation │◄─────────────────│ Tensor │
│──────────────│ │──────────────│
│ id (tag) │ │ id │
│ type │ │ data_ptr │
│ state │ │ size │
│ participants │ │ dtype │
└──────────────┘ │ device │
└──────────────┘
Low-Level Design (LLD)
Phase 1: Core Infrastructure
┌─────────────────────────────────────────────────────────────────────────┐
│ PHASE 1: CLASS DIAGRAM - CORE │
└─────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────┐
│ TcpServer │
├─────────────────────────────────┤
│ - address: String │
│ - port: u16 │
│ - listener: TcpListener │
│ - connections: Vec │
├─────────────────────────────────┤
│ + bind(addr, port) │
│ + accept() -> Connection │
│ + broadcast(msg) │
│ + shutdown() │
└─────────────────────────────────┘
▲
│ uses
│
┌─────────────────────────────────┐ ┌─────────────────────────────────┐
│ Master │ │ Peer │
├─────────────────────────────────┤ ├─────────────────────────────────┤
│ - peers: HashMap │ │ - id: PeerId │
│ - topology: RingTopology │ │ - phase: ConnectionPhase │
│ - state: MasterState │ │ - state: ConnectionState │
├─────────────────────────────────┤ │ - master_conn: TcpStream │
│ + register_peer(peer) │ │ - p2p_conns: Vec │
│ + accept_peer(peer_id) │ ├─────────────────────────────────┤
│ + remove_peer(peer_id) │ │ + connect_master(addr) │
│ + broadcast_abort(op_id) │ │ + connect_peer(peer_id) │
│ + optimize_topology() │ │ + send_to_master(msg) │
│ + get_ring_order() -> Vec │ │ + recv_from_master() -> Msg │
└─────────────────────────────────┘ └─────────────────────────────────┘
│
│ has
▼
┌─────────────────────────────────┐
│ P2PConnection │
├─────────────────────────────────┤
│ - remote_peer_id: PeerId │
│ - socket: TcpStream │
│ - send_buffer: Vec │
│ - recv_buffer: Vec │
│ - bandwidth: f64 │
├─────────────────────────────────┤
│ + send(data: &[u8]) │
│ + recv(buf: &mut [u8]) │
│ + measure_bandwidth() -> f64 │
│ + close() │
└─────────────────────────────────┘
State Machine Classes
┌─────────────────────────────────────────────────────────────────────────┐
│ STATE ENUMS & TRANSITIONS │
└─────────────────────────────────────────────────────────────────────────┘
┌─────────────────────┐ ┌─────────────────────┐ ┌─────────────────────┐
│ ConnectionPhase │ │ ConnectionState │ │ CollectiveState │
├─────────────────────┤ ├─────────────────────┤ ├─────────────────────┤
│ • Registered │ │ • Idle │ │ • VoteInitiate │
│ • Accepted │ │ • VotingAcceptPeers │ │ • Performing │
└─────────────────────┘ │ • SyncingState │ │ • VoteComplete │
│ • CollectiveRunning │ └─────────────────────┘
└─────────────────────┘
State Transition Rules:
─────────────────────────────────────────────────────────────────────────────
Registered ──[vote passes]──► Accepted
Idle ──[peers pending]──► VotingAcceptPeers ──[vote done]──► Idle
Idle ──[hash mismatch]──► SyncingState ──[sync done]──► Idle
Idle ──[start collective]──► CollectiveRunning ──[complete]──► Idle
VoteInitiate ──[all agree]──► Performing ──[done]──► VoteComplete ──[all ack]──► (end)
Phase 2: Collective Operations
┌─────────────────────────────────────────────────────────────────────────┐
│ PHASE 2: CLASS DIAGRAM - COLLECTIVES │
└─────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────┐
│ CollectiveOperation │ (abstract)
├─────────────────────────────────┤
│ - tag: u32 │
│ - state: CollectiveState │
│ - participants: Vec │
│ - buffer_backup: Option │
├─────────────────────────────────┤
│ + execute() -> Result │
│ + abort() │
│ + restore_backup() │
└───────────────┬─────────────────┘
│
┌────────────────────┼────────────────────┐
│ │ │
▼ ▼ ▼
┌───────────────────────┐ ┌───────────────────────┐ ┌───────────────────────┐
│ AllReduce │ │ Broadcast │ │ AllGather │
├───────────────────────┤ ├───────────────────────┤ ├───────────────────────┤
│ - reduce_op: ReduceOp │ │ - root: PeerId │ │ - recv_counts: Vec │
├───────────────────────┤ ├───────────────────────┤ ├───────────────────────┤
│ + reduce_scatter() │ │ + execute() │ │ + execute() │
│ + all_gather() │ └───────────────────────┘ └───────────────────────┘
│ + execute() │
└───────────────────────┘
┌─────────────────────────────────┐ ┌─────────────────────────────────┐
│ ReduceOp │ │ RingTopology │
├─────────────────────────────────┤ ├─────────────────────────────────┤
│ • Sum │ │ - order: Vec │
│ • Avg │ │ - bandwidth_matrix: Matrix │
│ • Min │ ├─────────────────────────────────┤
│ • Max │ │ + get_next(peer) -> PeerId │
└─────────────────────────────────┘ │ + get_prev(peer) -> PeerId │
│ + optimize() -> Vec │
│ + compute_chunk_bounds(n) -> Vec│
└─────────────────────────────────┘
Phase 3: Fault Tolerance
┌─────────────────────────────────────────────────────────────────────────┐
│ PHASE 3: CLASS DIAGRAM - FAULT TOLERANCE │
└─────────────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────┐
│ FailureDetector │
├─────────────────────────────────┤
│ - heartbeat_interval: Duration │
│ - timeout: Duration │
│ - last_seen: HashMap │
├─────────────────────────────────┤
│ + start_monitoring() │
│ + record_heartbeat(peer_id) │
│ + get_failed_peers() -> Vec │
│ + is_alive(peer_id) -> bool │
└─────────────────────────────────┘
┌─────────────────────────────────┐ ┌─────────────────────────────────┐
│ AbortController │ │ StateRecovery │
├─────────────────────────────────┤ ├─────────────────────────────────┤
│ - abort_flag: AtomicBool │ │ - backup_store: HashMap │
│ - abort_reason: Option │ ├─────────────────────────────────┤
├─────────────────────────────────┤ │ + save_backup(op_id, data) │
│ + signal_abort(reason) │ │ + restore_backup(op_id) -> Data │
│ + check_abort() -> bool │ │ + clear_backup(op_id) │
│ + reset() │ │ + has_backup(op_id) -> bool │
└─────────────────────────────────┘ └─────────────────────────────────┘
┌─────────────────────────────────┐
│ SharedStateSync │
├─────────────────────────────────┤
│ - hasher: SimpleHash │
│ - local_hash: u64 │
├─────────────────────────────────┤
│ + compute_hash(tensors) -> u64 │
│ + compare_hashes(peers) -> Vec │ (returns outliers)
│ + sync_from_peer(peer_id) │
│ + sync_to_peer(peer_id) │
└─────────────────────────────────┘
Sequence Diagrams
Peer Join Flow
┌─────────────────────────────────────────────────────────────────────────┐
│ SEQUENCE: PEER JOIN │
└─────────────────────────────────────────────────────────────────────────┘
NewPeer Master ExistingPeers
│ │ │
│──── connect() ────►│ │
│ │ │
│◄─── REGISTERED ────│ │
│ │ │
│ │◄── vote_accept? ─────│
│ │ │
│ │──── vote_accept ────►│
│ │ │
│ │◄─── all_agree ───────│
│ │ │
│◄─── ACCEPTED ──────│ │
│ │ │
│◄─────────────── establish P2P ───────────►│
│ │ │
│ │◄── sync_state ───────│
│◄──────────── receive state ──────────────│
│ │ │
│ READY TO PARTICIPATE │
▼ ▼ ▼
All-Reduce Flow
┌─────────────────────────────────────────────────────────────────────────┐
│ SEQUENCE: ALL-REDUCE │
└─────────────────────────────────────────────────────────────────────────┘
Peer0 Peer1 Peer2 Master
│ │ │ │
│───────────── vote_initiate ──────────────────►│
│ │ │ │
│◄──────────── all_agreed ─────────────────────│
│ │ │ │
│ backup_buffer │ backup_buffer │ backup_buffer │
│ │ │ │
│ ══════════ REDUCE-SCATTER (N-1 steps) ══════ │
│ │ │ │
│──── chunk ───►│ │ │
│ │──── chunk ───►│ │
│◄───────────── chunk ─────────│ │
│ │ │ │
│ (accumulate received chunks locally) │
│ │ │ │
│ ══════════ ALL-GATHER (N-1 steps) ═══════════│
│ │ │ │
│──── chunk ───►│ │ │
│ │──── chunk ───►│ │
│◄───────────── chunk ─────────│ │
│ │ │ │
│───────────── vote_complete ──────────────────►│
│ │ │ │
│◄──────────── all_acked ──────────────────────│
│ │ │ │
│ clear_backup │ clear_backup │ clear_backup │
▼ ▼ ▼ ▼
Failure Recovery Flow
┌─────────────────────────────────────────────────────────────────────────┐
│ SEQUENCE: FAILURE DURING COLLECTIVE │
└─────────────────────────────────────────────────────────────────────────┘
Peer0 Peer1(dies) Peer2 Master
│ │ │ │
│ ══════════ ALL-REDUCE IN PROGRESS ══════════ │
│ │ │ │
│──── chunk ───►│ CRASH! 💀 │ │
│ X │ │
│ X │ │
│ (timeout waiting for Peer1) │
│ │ │ │
│────────────── report_failure ────────────────►│
│ │ │ │
│◄───────────── ABORT signal ──────────────────│
│ │ │◄───── ABORT ──│
│ │ │ │
│ restore_backup│ │ restore_backup│
│ │ │ │
│ │ │◄─ remove_peer─│
│◄──────────── topology_update ────────────────│
│ │ │ │
│ ══════════ RETRY ALL-REDUCE (2 peers) ══════ │
│ │ │ │
│◄─────────────────────────────►│ │
│ │ │ │
│ SUCCESS! │ │
▼ ▼ ▼
Language Showdown
Recommended Stack
| Component | Language | Why |
|---|---|---|
| Core Library | Rust or C++ | Performance, memory control |
| Master Node | Go or Rust | Concurrency, simplicity |
| Python Bindings | PyO3 / pybind11 | PyTorch integration |
Implementation Checklist
Phase 1: Core Infrastructure
- ☐ TcpServer / TcpClient classes
- ☐ Master with peer registry
- ☐ Peer with master connection
- ☐ P2PConnection class
- ☐ ConnectionPhase enum
- ☐ ConnectionState enum
- ☐ Message serialization (protobuf/msgpack)
Phase 2: Collective Operations
- ☐ RingTopology class
- ☐ CollectiveOperation base class
- ☐ AllReduce (reduce-scatter + all-gather)
- ☐ ReduceOp enum (Sum, Avg, Min, Max)
- ☐ CollectiveState enum
- ☐ Async operation support
- ☐ Multiple concurrent operations
Phase 3: Fault Tolerance
- ☐ FailureDetector (heartbeat-based)
- ☐ AbortController (atomic flag)
- ☐ StateRecovery (backup/restore)
- ☐ SharedStateSync (hash comparison)
- ☐ SimpleHash implementation
- ☐ Graceful peer removal
- ☐ Operation retry logic
Phase 4: Optimization
- ☐ Bandwidth measurement
- ☐ ATSP solver for ring order
- ☐ Connection pooling
- ☐ Quantization support
Phase 5: Integration
- ☐ Python bindings (PyO3/pybind11)
- ☐ PyTorch tensor support
- ☐ CUDA memory handling
- ☐ Stress testing