data-dataflow-coordination

# Dataflow Coordination **Scope**: Coordination primitives, barrier synchronization, epoch markers, distributed snapshots, consistency **Lines**: 370 **Last Updated**: 2025-10-27 **Format Version**: 1.0 (Atomic) ## When to Use This Skill Use this skill when: - Coordinating multiple parallel dataflow streams - Implementing barrier synchronization across workers - Managing epochs and checkpointing in distributed systems - Building consistent snapshots of distributed state - Coordinating multi-stage pipeline computations - Implementing exactly-once semantics with coordination - Synchronizing external systems with dataflow progress - Handling backpressure across distributed workers ## Core Concepts ### Coordination Mechanisms ``` Barriers → Synchronization points across parallel streams → All workers must reach barrier before proceeding → Use for: Global aggregations, checkpointing → Cost: Latency spike at barrier Epochs → Logical time units for batching → Watermarks: "No data before T will arrive" → Use for: Windowing, progress tracking → Cost: Buffering until epoch closes Snapshots → Consistent global state capture → Chandy-Lamport algorithm → Use for: Fault recovery, migration → Cost: Storage and I/O overhead ``` ### Progress Tracking Models ``` Global Frontier → Minimum timestamp across all workers → Conservative: Waits for slowest worker → Guarantees: Completeness at timestamp Per-Worker Frontiers → Independent progress per worker → Optimistic: Faster workers proceed → Requires: Careful synchronization Hierarchical Frontiers → Per-operator, per-worker tracking → Fine-grained progress visibility → Complex but enables optimization ``` ### Consistency Models ``` Strong Consistency → All workers see same state → Requires: Global coordination → Use for: Financial transactions Eventual Consistency → Workers converge over time → Minimal coordination → Use for: Analytics, monitoring Causal Consistency → Respects causal relationships → Vector clocks, happens-before → Use for: Distributed collaboration ``` ## Patterns ### Pattern 1: Barrier Synchronization (Rust/Timely) ```rust use timely::dataflow::{Scope, Stream}; use timely::dataflow::channels::pact::Pipeline; use timely::dataflow::operators::generic::operator::Operator; use std::collections::HashMap; // Barrier that emits when all inputs reach same timestamp fn barrier<G: Scope>( streams: Vec<&Stream<G, i32>>, ) -> Stream<G, Vec<i32>> { assert!(!streams.is_empty()); let mut builder = timely::dataflow::operators::generic::builder_rc::OperatorBuilder::new( "Barrier".to_string(), streams[0].scope(), ); // Create inputs for all streams let mut inputs: Vec<_> = streams.iter() .map(|stream| builder.new_input(stream, Pipeline)) .collect(); let (mut output, stream) = builder.new_output(); builder.build(move |_capability| { let num_inputs = inputs.len(); let mut buffers: HashMap<G::Timestamp, Vec<Vec<i32>>> = HashMap::new(); move |_frontiers| { // Collect data from all inputs for (idx, input) in inputs.iter_mut().enumerate() { input.for_each(|time, data| { let entry = buffers.entry(time.time().clone()) .or_insert_with(|| vec![Vec::new(); num_inputs]); entry[idx].extend(data.iter().cloned()); }); } // Emit when all inputs have data at timestamp let ready: Vec<G::Timestamp> = buffers.iter() .filter(|(_, vecs)| vecs.iter().all(|v| !v.is_empty())) .map(|(t, _)| t.clone()) .collect(); for time in ready { if let Some(mut data_vecs) = buffers.remove(&time) { let mut session = output.session(&time); // Combine data from all inputs let combined: Vec<i32> = data_vecs.into_iter() .flat_map(|v| v) .collect(); session.give(combined); } } } }); stream } // Usage fn main() { use timely::dataflow::operators::{ToStream, Inspect}; timely::execute_from_args(std::env::args(), |worker| { worker.dataflow::<u64, _, _>(|scope| { let stream1 = (0..5).to_stream(scope); let stream2 = (10..15).to_stream(scope); barrier(vec![&stream1, &stream2]) .inspect(|data| println!("Barrier output: {:?}", data)); }); }).expect("Execution failed"); } ``` ### Pattern 2: Epoch Markers (Go) ```go package main import ( "fmt" "sync" "time" ) // EpochCoordinator manages epoch transitions across workers type EpochCoordinator struct { currentEpoch int64 numWorkers int barriers map[int64]*sync.WaitGroup mu sync.Mutex } func NewEpochCoordinator(numWorkers int) *EpochCoordinator { return &EpochCoordinator{ currentEpoch: 0, numWorkers: numWorkers, barriers: make(map[int64]*sync.WaitGroup), } } // ArriveAtEpoch signals that worker has reached epoch func (ec *EpochCoordinator) ArriveAtEpoch(workerID int, epoch int64) { ec.mu.Lock() defer ec.mu.Unlock() // Create barrier for this epoch if needed if _, exists := ec.barriers[epoch]; !exists { ec.barriers[epoch] = &sync.WaitGroup{} ec.barriers[epoch].Add(ec.numWorkers) } fmt.Printf("Worker %d arrived at epoch %d\n", workerID, epoch) ec.barriers[epoch].Done() } // WaitForEpoch blocks until all workers reach epoch func (ec *EpochCoordinator) WaitForEpoch(epoch int64) { ec.mu.Lock() barrier := ec.barriers[epoch] ec.mu.Unlock() if barrier != nil { barrier.Wait() fmt.Printf("Epoch %d complete\n", epoch) // Clean up old barriers ec.mu.Lock() delete(ec.barriers, epoch) ec.mu.Unlock() } } // AdvanceEpoch moves to next epoch func (ec *EpochCoordinator) AdvanceEpoch() int64 { ec.mu.Lock() defer ec.mu.Unlock() ec.currentEpoch++ return ec.currentEpoch } // Worker simulates dataflow worker with epoch coordination func worker(id int, coordinator *EpochCoordinator, wg *sync.WaitGroup) { defer wg.Done() for epoch := int64(0); epoch < 5; epoch++ { // Simulate work time.Sleep(time.Duration(id*100) * time.Millisecond) fmt.Printf("Worker %d processing epoch %d\n", id, epoch) // Signal arrival at epoch barrier coordinator.ArriveAtEpoch(id, epoch) // Wait for all workers to reach epoch coordinator.WaitForEpoch(epoch) } } func main() { numWorkers := 3 coordinator := NewEpochCoordinator(numWorkers) var wg sync.WaitGroup wg.Add(numWorkers) for i := 0; i < numWorkers; i++ { go worker(i, coordinator, &wg) } wg.Wait() fmt.Println("All workers complete") } ``` ### Pattern 3: Distributed Snapshot (Chandy-Lamport) ```python import threading import queue from dataclasses import dataclass from typing import Dict, List, Set from enum import Enum class MessageType(Enum): DATA = 1 MARKER = 2 @dataclass class Message: msg_type: MessageType data: any snapshot_id: int = 0 class SnapshotWorker: """Implements Chandy-Lamport snapshot algorithm""" def __init__(self, worker_id: int, neighbors: List[int]): self.worker_id = worker_id self.neighbors = neighbors self.state = {} # Local state self.channels: Dict[int, queue.Queue] = {} # Input channels # Snapshot state self.recording: Dict[int, bool] = {} # Per snapshot self.recorded_state: Dict[int, dict] = {} # Snapshot ID -> state self.recorded_messages: Dict[int, Dict[int, List]] = {} # Snapshot -> channel -> messages self.markers_received: Dict[int, Set[int]] = {} # Snapshot -> set of channels # Initialize channels for neighbor in neighbors: self.channels[neighbor] = queue.Queue() def start_snapshot(self, snapshot_id: int): """Initiate snapshot (only by coordinator)""" print(f"Worker {self.worker_id}: Starting snapshot {snapshot_id}") # Record local state self.recorded_state[snapshot_id] = self.state.copy() self.recording[snapshot_id] = True self.recorded_messages[snapshot_id] = {n: [] for n in self.neighbors} self.markers_received[snapshot_id] = set() # Send markers to all neighbors for neighbor in self.neighbors: self.send_message(neighbor, Message(MessageType.MARKER, None, snapshot_id)) def receive_marker(self, snapshot_id: int, from_channel: int): """Handle marker message""" if snapshot_id not in self.recording: # First marker for this snapshot print(f"Worker {self.worker_id}: Received first marker for snapshot {snapshot_id}") # Record state self.recorded_state[snapshot_id] = self.state.copy() self.recording[snapshot_id] = True self.recorded_messages[snapshot_id] = {n: [] for n in self.neighbors} self.markers_received[snapshot_id] = {from_channel} # Channel from which marker arrived is empty self.recorded_messages[snapshot_id][from_channel] = [] # Send markers to all neighbors for neighbor in self.neighbors: self.send_message(neighbor, Message(MessageType.MARKER, None, snapshot_id)) else: # Subsequent marker print(f"Worker {self.worker_id}: Received marker from channel {from_channel}") self.markers_received[snapshot_id].add(from_channel) # Check if snapshot complete if len(self.markers_received[snapshot_id]) == len(self.neighbors): self.finalize_snapshot(snapshot_id) def receive_data(self, data: any, from_channel: int): """Handle data message""" # Process data (update state) self.state[f'key_{len(self.state)}'] = data # Record message if snapshot in progress for snapshot_id, recording in self.recording.items(): if recording and from_channel not in self.markers_received.get(snapshot_id, set()): # Recording messages on this channel self.recorded_messages[snapshot_id][from_channel].append(data) def finalize_snapshot(self, snapshot_id: int): """Snapshot complete for this worker""" print(f"Worker {self.worker_id}: Snapshot {snapshot_id} complete") print(f" State: {self.recorded_state[snapshot_id]}") print(f" Messages: {self.recorded_messages[snapshot_id]}") self.recording[snapshot_id] = False def send_message(self, to_worker: int, message: Message): """Send message to another worker (simulated)""" # In real system, this would send over network pass # Example usage def main(): # Create 3 workers in ring topology: 0 -> 1 -> 2 -> 0 workers = [ SnapshotWorker(0, [1, 2]), SnapshotWorker(1, [0, 2]), SnapshotWorker(2, [0, 1]), ] # Simulate some processing workers[0].state = {'count': 10} workers[1].state = {'count': 20} workers[2].state = {'count': 30} # Initiate snapshot from worker 0 workers[0].start_snapshot(snapshot_id=1) # Simulate marker propagation workers[1].receive_marker(snapshot_id=1, from_channel=0) workers[2].receive_marker(snapshot_id=1, from_channel=1) workers[0].receive_marker(snapshot_id=1, from_channel=2) workers[1].receive_marker(snapshot_id=1, from_channel=2) workers[2].receive_marker(snapshot_id=1, from_channel=0) if __name__ == '__main__': main() ``` ### Pattern 4: Backpressure Coordination (Python) ```python import asyncio from dataclasses import dataclass from typing import Optional import time @dataclass class Watermark: """Progress indicator for backpressure""" timestamp: int worker_id: int class BackpressureCoordinator: """Coordinates flow control across pipeline stages""" def __init__(self, num_workers: int, buffer_size: int = 100): self.num_workers = num_workers self.buffer_size = buffer_size self.worker_watermarks = [0] * num_workers self.global_watermark = 0 self.lock = asyncio.Lock() async def update_watermark(self, worker_id: int, timestamp: int): """Update watermark for worker""" async with self.lock: self.worker_watermarks[worker_id] = timestamp old_global = self.global_watermark self.global_watermark = min(self.worker_watermarks) if self.global_watermark > old_global: print(f"Global watermark advanced to {self.global_watermark}") async def can_proceed(self, worker_id: int, timestamp: int) -> bool: """Check if worker can proceed without overwhelming downstream""" async with self.lock: # Allow if within buffer_size of slowest worker