data-dataflow-coordination

name	data-dataflow-coordination
description	Coordination patterns for distributed dataflow systems including barriers, epochs, and distributed snapshots

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)

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<>>> = HashMap::();

         |_frontiers| {
            
             (idx, input)  inputs.().() {
                input.for_each(|time, data| {
                      = buffers.(time.().())
                        .(|| [::(); num_inputs]);
                    entry[idx].(data.().());
                });
            }

            
             : <G::Timestamp> = buffers.()
                .(|(_, vecs)| vecs.().(|v| !v.()))
                .(|(t, _)| t.())
                .();

               ready {
                  ( data_vecs) = buffers.(&time) {
                      = output.(&time);

                    
                     : <> = data_vecs.()
                        .(|v| v)
                        .();

                    session.(combined);
                }
            }
        }
    });

    stream
}


 () {
     timely::dataflow::operators::{ToStream, Inspect};

    timely::(std::env::(), |worker| {
        worker.dataflow::<, _, _>(|scope| {
              = (..).(scope);
              = (..).(scope);

            ([&stream1, &stream2])
                .(|data| (, data));
        });
    }).();
}

Pattern 2: Epoch Markers (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()
        (ec.barriers, epoch)
        ec.mu.Unlock()
    }
}


 AdvanceEpoch()  {
    ec.mu.Lock()
     ec.mu.Unlock()

    ec.currentEpoch++
     ec.currentEpoch
}


 {
     wg.Done()

     epoch := (); epoch < ; epoch++ {
        
        time.Sleep(time.Duration(id*) * time.Millisecond)
        fmt.Printf(, id, epoch)

        
        coordinator.ArriveAtEpoch(id, epoch)

        
        coordinator.WaitForEpoch(epoch)
    }
}

 {
    numWorkers := 
    coordinator := NewEpochCoordinator(numWorkers)

     wg sync.WaitGroup
    wg.Add(numWorkers)

     i := ; i < numWorkers; i++ {
         worker(i, coordinator, &wg)
    }

    wg.Wait()
    fmt.Println()
}

Pattern 3: Distributed Snapshot (Chandy-Lamport)

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] = {}  
        .recorded_messages: [, [, ]] = {}  
        .markers_received: [, []] = {}  

        
         neighbor  neighbors:
            .channels[neighbor] = queue.Queue()

     ():
        
        ()

        
        .recorded_state[snapshot_id] = .state.copy()
        .recording[snapshot_id] = 
        .recorded_messages[snapshot_id] = {n: []  n  .neighbors}
        .markers_received[snapshot_id] = ()

        
         neighbor  .neighbors:
            .send_message(neighbor, Message(MessageType.MARKER, , snapshot_id))

     ():
        
         snapshot_id   .recording:
            
            ()

            
            .recorded_state[snapshot_id] = .state.copy()
            .recording[snapshot_id] = 
            .recorded_messages[snapshot_id] = {n: []  n  .neighbors}
            .markers_received[snapshot_id] = {from_channel}

            
            .recorded_messages[snapshot_id][from_channel] = []

            
             neighbor  .neighbors:
                .send_message(neighbor, Message(MessageType.MARKER, , snapshot_id))
        :
            
            ()
            .markers_received[snapshot_id].add(from_channel)

        
         (.markers_received[snapshot_id]) == (.neighbors):
            .finalize_snapshot(snapshot_id)

     ():
        
        
        .state[] = data

        
         snapshot_id, recording  .recording.items():
             recording  from_channel   .markers_received.get(snapshot_id, ()):
                
                .recorded_messages[snapshot_id][from_channel].append(data)

     ():
        
        ()
        ()
        ()

        .recording[snapshot_id] = 

     ():
        
        
        


 ():
    
    workers = [
        SnapshotWorker(, [, ]),
        SnapshotWorker(, [, ]),
        SnapshotWorker(, [, ]),
    ]

    
    workers[].state = {: }
    workers[].state = {: }
    workers[].state = {: }

    
    workers[].start_snapshot(snapshot_id=)

    
    workers[].receive_marker(snapshot_id=, from_channel=)
    workers[].receive_marker(snapshot_id=, from_channel=)
    workers[].receive_marker(snapshot_id=, from_channel=)
    workers[].receive_marker(snapshot_id=, from_channel=)
    workers[].receive_marker(snapshot_id=, from_channel=)

 __name__ == :
    main()

Pattern 4: Backpressure Coordination (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 .global_watermark > old_global:
                ()

      () -> :
        
          .lock:
            
             timestamp <= .global_watermark + .buffer_size

      () -> :
          .lock:
             .global_watermark

 :
    

     ():
        .stage_id = stage_id
        .coordinator = coordinator
        .input_queue = input_queue
        .output_queue = output_queue
        .current_timestamp = 

      ():
        
         :
            :
                data =  asyncio.wait_for(
                    .input_queue.get(),
                    timeout=
                )

                 data  :  
                     .output_queue:
                         .output_queue.put()
                    

                timestamp, value = data

                
                   .coordinator.can_proceed(.stage_id, timestamp):
                    ()
                     asyncio.sleep()

                
                 asyncio.sleep()
                processed = value * 

                
                .current_timestamp = timestamp
                 .coordinator.update_watermark(.stage_id, timestamp)

                
                 .output_queue:
                     .output_queue.put((timestamp, processed))

             asyncio.TimeoutError:
                

  ():
    num_stages = 
    coordinator = BackpressureCoordinator(num_stages, buffer_size=)

    
    queues = [asyncio.Queue()  _  (num_stages)]

    stages = [
        PipelineStage(, coordinator, queues[], queues[]),
        PipelineStage(, coordinator, queues[], queues[]),
        PipelineStage(, coordinator, queues[], ),
    ]

    
    tasks = [asyncio.create_task(stage.process())  stage  stages]

    
     i  ():
         queues[].put((i, i))
         asyncio.sleep()  

    
     queues[].put()

    
     asyncio.gather(*tasks)

 __name__ == :
    asyncio.run(main())

Pattern 5: Causal Consistency with Vector Clocks (Go)

package main

import (
    "fmt"
    "sync"
)

// VectorClock tracks causal relationships
type VectorClock map[int]int

func (vc VectorClock) Copy() VectorClock {
    copy := make(VectorClock)
    for k, v := range vc {
        copy[k] = v
    }
    return copy
}

func (vc VectorClock) Increment(nodeID int) {
    vc[nodeID]++
}

func (vc VectorClock) Merge(other VectorClock) {
    for nodeID, timestamp := range other {
        if current, exists := vc[nodeID]; !exists || timestamp > current {
            vc[nodeID] = timestamp
        }
    }
}

func (vc VectorClock) HappensBefore(other VectorClock) bool {
    lessOrEqual := true
    strictlyLess := false

    for nodeID := range vc {
        if vc[nodeID] > other[nodeID] {
            return false // Not happens-before
        }
        if vc[nodeID] < other[nodeID] {
            strictlyLess = true
        }
    }

    return lessOrEqual && strictlyLess
}

// Event with causal timestamp
type Event struct {
    NodeID 
    Data   
    Clock  VectorClock
}


 CausalBroadcast  {
    nodeID   
    clock    VectorClock
    pending  []Event
    delivered []
    mu       sync.Mutex
}

 *CausalBroadcast {
    clock := (VectorClock)
     i := ; i < numNodes; i++ {
        clock[i] = 
    }

     &CausalBroadcast{
        nodeID:    nodeID,
        clock:     clock,
        pending:   []Event{},
        delivered: ([]),
    }
}

 Send(data ) Event {
    cb.mu.Lock()
     cb.mu.Unlock()

    cb.clock.Increment(cb.nodeID)

    event := Event{
        NodeID: cb.nodeID,
        Data:   data,
        Clock:  cb.clock.Copy(),
    }

    fmt.Printf(,
        cb.nodeID, data, event.Clock)

     event
}

 Receive(event Event) {
    cb.mu.Lock()
     cb.mu.Unlock()

    
     cb.canDeliver(event) {
        cb.deliver(event)
        cb.checkPending()
    }  {
        
        cb.pending = (cb.pending, event)
        fmt.Printf(,
            cb.nodeID, event.Data)
    }
}

 canDeliver(event Event)  {
    
    
    

    expected := cb.clock[event.NodeID] + 
     event.Clock[event.NodeID] != expected {
         
    }

     nodeID, timestamp :=  event.Clock {
         nodeID != event.NodeID && timestamp > cb.clock[nodeID] {
              
        }
    }

     
}

 deliver(event Event) {
    cb.clock.Merge(event.Clock)
    cb.delivered[event.Data] = 

    fmt.Printf(,
        cb.nodeID, event.Data, event.Clock)
}

 checkPending() {
    
     stillPending []Event

     _, event :=  cb.pending {
         cb.canDeliver(event) {
            cb.deliver(event)
        }  {
            stillPending = (stillPending, event)
        }
    }

    cb.pending = stillPending
}

 {
    
    nodes := []*CausalBroadcast{
        NewCausalBroadcast(, ),
        NewCausalBroadcast(, ),
        NewCausalBroadcast(, ),
    }

    
    eventA := nodes[].Send()

    
    eventB := nodes[].Send()

    
    nodes[].Receive(eventB) 
    nodes[].Receive(eventA) 
}

Quick Reference

Coordination Patterns

Barrier: Wait for all workers at sync point
Epoch: Logical time units for batching
Watermark: "No data before T will arrive"
Snapshot: Consistent global state capture
Vector Clock: Track causal dependencies

Trade-offs

Strong Coordination
  Pros: Consistency, simplicity
  Cons: Latency, throughput impact

Weak Coordination
  Pros: Low latency, high throughput
  Cons: Complex, eventual consistency

Anti-Patterns

❌ NEVER: Use global locks in hot path
   → Use lock-free coordination or partitioning

❌ NEVER: Block all workers for slow worker
   → Use timeout or skip slow worker with compensation

❌ NEVER: Ignore stragglers in barrier
   → Implement timeout and speculative execution

❌ NEVER: Take snapshots synchronously in critical path
   → Use background checkpointing

❌ NEVER: Use barriers for every record
   → Batch into epochs for efficiency

❌ NEVER: Assume synchronized clocks
   → Use logical clocks (Lamport, vector)

❌ NEVER: Coordinate without backpressure
   → Fast producers overwhelm slow consumers

❌ NEVER: Hardcode barrier counts
   → Use dynamic registration for elasticity

Related Skills

• timely-dataflow.md - Progress tracking in timely dataflow
• differential-dataflow.md - Incremental computation
• streaming-aggregations.md - Windowing with watermarks
• stream-processing.md - High-level stream processing

Last Updated: 2025-10-27 Format Version: 1.0 (Atomic)