Dataflow Models: Deterministic Concurrent Computation

Purpose: Mathematical frameworks for modeling computation as data flowing through networks of interconnected processing nodes, enabling deterministic parallelism and compositional reasoning.

Core Insight: Computation is data transformation. Explicitly model data dependencies, achieving deterministic parallel execution and compositional system design.

Foundational Concepts
Kahn Process Networks
Synchronous Dataflow
Flow-Based Programming
Dataflow Languages and DSLs
Practical Applications
Stack-Specific Implementations
Integration Points

Foundational Concepts

What is Dataflow?

Dataflow: Computational model where execution is driven by data availability rather than control flow.

Key Properties:

Data-Driven Execution: Operations fire when inputs available
Explicit Dependencies: Data dependencies visible in graph structure
Implicit Parallelism: Independent operations execute concurrently
Determinism: Same inputs → same outputs (in most models)

Dataflow vs. Control Flow

Aspect	Control Flow	Dataflow
Execution Order	Sequential (program counter)	Data-driven (readiness)
Parallelism	Explicit (threads, async)	Implicit (data dependencies)
Dependencies	Hidden in code	Explicit in graph
Composition	Function calls	Graph wiring
Determinism	Not guaranteed	Often guaranteed

Dataflow Graph

Components:

Nodes: Processing elements (functions, operators)
Edges: Data channels (FIFO queues, streams)
Tokens: Data values flowing through edges

Execution:

Node fires when:
  - All input channels have required tokens
  - Output channels have space

Firing:
  1. Consume input tokens
  2. Execute computation
  3. Produce output tokens

Benefits of Dataflow

Advantages:

Automatic Parallelism: Dependencies explicit, scheduler exploits
Determinism: Reproducible results (critical for testing, debugging)
Modularity: Nodes are composable, reusable black boxes
Visual Programming: Graphs are intuitive representations
Optimization: Static analysis of graph enables compile-time optimization

Challenges:

State Management: Handling mutable state requires care
Deadlock: Cycles with insufficient buffering can deadlock
Memory Overhead: Buffering tokens between nodes
Debugging: Harder to trace execution than sequential code

Kahn Process Networks

Developed by: Gilles Kahn (1974)

Key Idea: Network of deterministic sequential processes communicating through unbounded FIFO channels. Produces deterministic results regardless of execution order.

KPN Formal Model

Definition:

Processes: Sequential computations
Channels: Unbounded FIFO queues
Operations: read(channel) (blocking), write(channel, value) (non-blocking)

Determinism Theorem (Kahn 1974):

For any KPN:

Determinism: Output sequences depend only on input sequences, not execution timing
Monotonicity: Partial results never retracted (outputs append-only)
Continuity: Finite outputs computed from finite prefixes of inputs

Proof Sketch: Process sees input channel as stream. Blocking read ensures process waits for data. Since channels unbounded, writes never block on full buffer. Thus, process execution determined solely by input arrival order, which is fixed.

KPN Properties

Determinism: Given same inputs, KPN produces same outputs regardless of:

Process scheduling order
Execution speed of individual processes
Parallel vs. sequential execution

Monotonicity: Once a value output, it’s never retracted. Allows incremental computation.

Deadlock: KPN can deadlock if:

Process blocks reading from empty channel, no process will write to it
Circular dependency with finite buffers (though KPN assumes unbounded)

Boundedness: Channel buffer size may grow unboundedly. Implementation challenge.

Example: Stream Processing Pipeline

[Generator] → channel_a → [Filter] → channel_b → [Mapper] → channel_c → [Sink]

Generator:
  for i in 1..∞:
    write(channel_a, i)

Filter:
  while true:
    val = read(channel_a)
    if val % 2 == 0:
      write(channel_b, val)

Mapper:
  while true:
    val = read(channel_b)
    write(channel_c, val * 2)

Sink:
  while true:
    result = read(channel_c)
    print(result)

Determinism: Given same generator sequence, output always same.

Parallelism: Filter, Mapper, Sink can run concurrently.

Parks’ Bounded Buffer Problem

Problem: KPN assumes unbounded channels. Real systems have finite memory.

Parks’ Algorithm (1995): Implements KPN with bounded buffers:

Maintain minimum buffer sizes to avoid artificial deadlock
Dynamically expand buffers when necessary
Detect “real” deadlock (cyclic wait) vs. “buffer deadlock”

Trade-off: Bounded buffers may introduce artificial deadlock if too small. Parks’ algorithm minimizes this.

KPN Scheduling

Goal: Determine process execution order to maximize throughput, minimize buffer usage.

Strategies:

Static Scheduling: Compile-time order (requires static rates)
Dynamic Scheduling: Runtime decisions based on data availability
Quasi-Static Scheduling: Hybrid - static with runtime adaptations

SDF (Synchronous Dataflow, next section) enables efficient static scheduling.

Synchronous Dataflow

Developed by: Lee & Messerschmitt (1987)

Key Idea: Restrict KPN by specifying fixed number of tokens consumed/produced per firing. Enables static scheduling and bounded memory.

SDF Formal Model

SDF Actor:

Ports: Named input/output channels
Consumption Rates: Fixed tokens consumed per port per firing
Production Rates: Fixed tokens produced per port per firing
Firing Rule: Fire when all inputs have enough tokens and outputs have space

Notation:

Actor A:
  inputs: in1 (consume 2), in2 (consume 1)
  outputs: out1 (produce 3)

Firing: Consumes 2 from in1, 1 from in2, produces 3 to out1

SDF Balance Equations

For SDF graph to execute in bounded memory with consistent firing:

Topology Matrix Γ:

Rows = channels
Columns = actors
Entry Γ[c,a] = (tokens produced by a on c) - (tokens consumed by a on c)

Balance Equation: Γq = 0

Where q = firing vector (number of times each actor fires per iteration).

Example:

[A] --2--> [B] --3--> [C]
     3 tokens/fire  6 tokens/fire

A produces 2, B consumes 2: Balance for A→B channel
B produces 3, C consumes 3: Balance for B→C channel

Firing vector: q_A, q_B, q_C
Balance equations:
  2*q_A - 2*q_B = 0  =>  q_A = q_B
  3*q_B - 3*q_C = 0  =>  q_B = q_C

Solution: q_A = q_B = q_C = k (any integer)
Minimal: q = [1, 1, 1]

Static Scheduling

Given: SDF graph with balance equations satisfied

Goal: Find execution order that:

Respects data dependencies
Minimizes buffer sizes
Executes in bounded memory

Algorithm:

Compute firing vector q from Γq = 0
Topologically sort actors respecting dependencies
Schedule q[a] firings of each actor a in order

Example Schedule:

Graph: [A(1,2)] → [B(2,3)] → [C(3,1)]

Firing vector: q = [6, 3, 2]  (GCD normalization)

Schedule: A A A A A A B B B C C

Bounded memory: Predictable buffer sizes

SDF Extensions

Cyclo-Static Dataflow (CSDF):

Rates vary in fixed cycle pattern
Extends SDF expressiveness while keeping static analysis

Boolean Dataflow (BDF):

Conditional execution (if-then-else)
Rates depend on token values (dynamic)
Loses static schedulability

Parameterized Dataflow:

Rates parameterized by constants
Static analysis with parameters
Enables reusable components

SDF Applications

Signal Processing:

Audio/video codecs
Wireless communication (modulation, filtering)
Radar signal processing

Embedded Systems:

Real-time constraints
Resource-constrained devices
Deterministic timing

Streaming Applications:

Video streaming pipelines
Data analytics streams
Network packet processing

Flow-Based Programming

Developed by: J. Paul Morrison (1970s)

Key Idea: Applications are networks of black-box processes exchanging data packets (Information Packets, IPs) over bounded connections.

FBP Concepts

Components:

Processes: Reusable, concurrent black boxes
Information Packets (IPs): Data with ownership semantics
Connections: Bounded FIFO queues between ports
Ports: Named input/output connection points

Process Lifecycle:

1. Wait for input IPs
2. Process IPs (read, transform, create new)
3. Send output IPs
4. Repeat or terminate

FBP vs. KPN/SDF

Feature	KPN	SDF	FBP
Channels	Unbounded	Bounded (static)	Bounded (configurable)
Rates	Dynamic	Static	Dynamic
Determinism	Guaranteed	Guaranteed	Not guaranteed (depends)
Scheduling	Dynamic	Static	Dynamic
State	Process-local	Process-local	Process-local + IP data

FBP Patterns

Generator:

Process: DataGenerator
Outputs: out

Loop:
  data = generateData()
  send(out, data)

Filter:

Process: Filter
Inputs: in
Outputs: out

Loop:
  ip = receive(in)
  if predicate(ip):
    send(out, ip)

Transformer:

Process: Mapper
Inputs: in
Outputs: out

Loop:
  ip = receive(in)
  result = transform(ip)
  send(out, result)

Splitter:

Process: Splitter
Inputs: in
Outputs: out1, out2

Loop:
  ip = receive(in)
  if condition(ip):
    send(out1, ip)
  else:
    send(out2, ip)

Merger:

Process: Merger
Inputs: in1, in2
Outputs: out

Loop:
  select:
    case ip1 = receive(in1):
      send(out, ip1)
    case ip2 = receive(in2):
      send(out, ip2)

FBP Example: Text Processing

[FileReader] → [Splitter] → [Counter] → [Sorter] → [Writer]
                    ↓
                [Filter] → [Formatter] → [Writer2]

FileReader: Read file line by line, emit IPs
Splitter: Split lines into words, emit word IPs
Counter: Count word frequencies, emit (word, count) IPs
Sorter: Sort by count descending
Writer: Write top 10 to file

Filter: Filter by word length > 3
Formatter: Format as JSON
Writer2: Write filtered results

Information Packets (IPs)

Ownership: IP owned by one process at a time. Ownership transfers on send/receive.

Types:

Normal IP: Carries data payload
Bracket IP: Mark start/end of substream (hierarchy)
Empty IP: Signal without data (e.g., end-of-stream)

Bracket IPs:

Process: GroupCounter
Inputs: in
Outputs: out

count = 0
Loop:
  ip = receive(in)
  if ip is OpenBracket:
    count = 0
  elif ip is CloseBracket:
    send(out, count)
  else:
    count += 1

Dataflow Languages and DSLs

Lustre: Synchronous Dataflow Language

Lustre: Declarative language for reactive systems (synchronous model).

Example:

node counter(reset: bool) returns (count: int);
let
  count = 0 -> if reset then 0 else pre(count) + 1;
tel

node edge_detector(x: bool) returns (edge: bool);
let
  edge = false -> x and not pre(x);
tel

Semantics: Clock-driven. All variables updated simultaneously each tick.

Applications: Safety-critical embedded systems (avionics, automotive).

LabVIEW: Visual Dataflow Programming

LabVIEW: Graphical programming environment using dataflow (called G language).

Features:

Visual block diagram (nodes and wires)
Parallel loops execute concurrently
Built-in data acquisition and instrument control

Use Cases: Test/measurement, embedded control, FPGA programming.

StreamIt: Stream Processing Language

StreamIt: High-level language for streaming applications.

Constructs:

filter Doubler() {
  work pop 1 push 1 {
    push(2 * pop());
  }
}

pipeline DoublerPipeline {
  add DataSource();
  add Doubler();
  add DataSink();
}

Compiler: Optimizes for multi-core, generates C code.

Practical Applications

1. Video Processing Pipeline

Problem: Decode video, apply filters, encode to multiple resolutions.

Solution: Dataflow pipeline with parallel branches.

[Decoder] → [ColorCorrect] → [Split] → [Resize_720p] → [Encode_720p]
                               ↓
                           [Resize_1080p] → [Encode_1080p]
                               ↓
                           [Resize_4K] → [Encode_4K]

Each node = process
Edges = bounded FIFO queues
Parallel execution of resize/encode stages

2. Machine Learning Inference Pipeline

Problem: Real-time inference on video stream with preprocessing.

Solution: SDF graph for deterministic latency.

[Camera] → [Preprocess] → [Model Inference] → [Postprocess] → [Display]

Static schedule:
  - Preprocess: 30 FPS (consume 1 frame, produce 1 tensor)
  - Inference: 10 FPS (consume 1 tensor, produce 1 result)
  - Postprocess: 10 FPS (consume 1 result, produce 1 frame)

Buffer between stages to handle rate mismatch

3. ETL Data Pipeline

Problem: Extract, transform, load large datasets.

Solution: FBP network with reusable components.

[CSVReader] → [Validator] → [Transformer] → [Enricher] → [DBWriter]
                  ↓                            ↑
              [ErrorLog]                  [APIFetch]

Components:
- CSVReader: Generator process
- Validator: Filter invalid records
- Transformer: Map fields
- Enricher: Join with external data (APIFetch)
- DBWriter: Sink process
- ErrorLog: Handle validation failures

4. Real-Time Audio Effects

Problem: Apply audio effects (reverb, EQ) in real-time with low latency.

Solution: SDF graph for predictable timing.

[AudioInput] → [FFT] → [EQ] → [Reverb] → [IFFT] → [AudioOutput]

SDF rates:
- AudioInput: 1 sample/fire
- FFT: 512 samples in, 512 frequency bins out
- EQ: 512 bins in, 512 bins out
- Reverb: 512 bins in, 512 bins out
- IFFT: 512 bins in, 512 samples out

Static schedule ensures deterministic latency

Stack-Specific Implementations

Rust: KPN with Async Channels

use tokio::sync::mpsc;
use tokio::task;

/// Kahn Process Network node trait
#[async_trait::async_trait]
trait KPNProcess: Send {
    async fn run(self: Box<Self>);
}

/// Example: Generator process
struct Generator {
    output: mpsc::UnboundedSender<i32>,
    count: usize,
}

#[async_trait::async_trait]
impl KPNProcess for Generator {
    async fn run(self: Box<Self>) {
        for i in 0..self.count {
            // Non-blocking write (unbounded channel)
            self.output.send(i as i32).unwrap();
        }
    }
}

/// Example: Filter process
struct Filter {
    input: mpsc::UnboundedReceiver<i32>,
    output: mpsc::UnboundedSender<i32>,
    predicate: fn(i32) -> bool,
}

#[async_trait::async_trait]
impl KPNProcess for Filter {
    async fn run(mut self: Box<Self>) {
        while let Some(value) = self.input.recv().await {
            // Blocking read
            if (self.predicate)(value) {
                self.output.send(value).unwrap();
            }
        }
    }
}

/// Example: Mapper process
struct Mapper {
    input: mpsc::UnboundedReceiver<i32>,
    output: mpsc::UnboundedSender<i32>,
    transform: fn(i32) -> i32,
}

#[async_trait::async_trait]
impl KPNProcess for Mapper {
    async fn run(mut self: Box<Self>) {
        while let Some(value) = self.input.recv().await {
            let result = (self.transform)(value);
            self.output.send(result).unwrap();
        }
    }
}

/// Build and run KPN
async fn kpn_example() {
    let (gen_tx, gen_rx) = mpsc::unbounded_channel();
    let (filter_tx, filter_rx) = mpsc::unbounded_channel();
    let (map_tx, map_rx) = mpsc::unbounded_channel();

    // Create processes
    let generator: Box<dyn KPNProcess> = Box::new(Generator {
        output: gen_tx,
        count: 100,
    });

    let filter: Box<dyn KPNProcess> = Box::new(Filter {
        input: gen_rx,
        output: filter_tx,
        predicate: |x| x % 2 == 0, // Even numbers only
    });

    let mapper: Box<dyn KPNProcess> = Box::new(Mapper {
        input: filter_rx,
        output: map_tx,
        transform: |x| x * 2,
    });

    // Spawn concurrent tasks
    let gen_handle = task::spawn(async move { generator.run().await });
    let filter_handle = task::spawn(async move { filter.run().await });
    let mapper_handle = task::spawn(async move { mapper.run().await });

    // Collect results
    task::spawn(async move {
        let mut receiver = map_rx;
        while let Some(value) = receiver.recv().await {
            println!("Result: {}", value);
        }
    });

    // Wait for completion
    let _ = tokio::join!(gen_handle, filter_handle, mapper_handle);
}

/// SDF Actor trait
trait SDFActor {
    /// Number of tokens consumed from each input port per firing
    fn consumption_rates(&self) -> Vec<usize>;

    /// Number of tokens produced to each output port per firing
    fn production_rates(&self) -> Vec<usize>;

    /// Fire actor: consume inputs, produce outputs
    fn fire(&mut self, inputs: Vec<Vec<i32>>) -> Vec<Vec<i32>>;
}

/// Example: SDF Upsampler (rate conversion)
struct Upsampler {
    factor: usize,
}

impl SDFActor for Upsampler {
    fn consumption_rates(&self) -> Vec<usize> {
        vec![1] // Consume 1 token per firing
    }

    fn production_rates(&self) -> Vec<usize> {
        vec![self.factor] // Produce 'factor' tokens per firing
    }

    fn fire(&mut self, inputs: Vec<Vec<i32>>) -> Vec<Vec<i32>> {
        let input = &inputs[0];
        let value = input[0];

        // Repeat value 'factor' times
        vec![vec![value; self.factor]]
    }
}

/// SDF Graph executor
struct SDFGraph {
    actors: Vec<Box<dyn SDFActor>>,
    topology: Vec<(usize, usize)>, // (src_actor, dst_actor) edges
    buffers: Vec<Vec<i32>>,         // Buffer for each edge
}

impl SDFGraph {
    fn new(actors: Vec<Box<dyn SDFActor>>, topology: Vec<(usize, usize)>) -> Self {
        let buffers = vec![Vec::new(); topology.len()];
        SDFGraph {
            actors,
            topology,
            buffers,
        }
    }

    /// Execute one iteration of static schedule
    fn execute_iteration(&mut self) {
        // Simplified: fire each actor in order
        // Real implementation would compute firing vector and schedule

        for (actor_idx, actor) in self.actors.iter_mut().enumerate() {
            // Gather inputs from incoming edges
            let mut inputs: Vec<Vec<i32>> = Vec::new();

            for (edge_idx, (src, dst)) in self.topology.iter().enumerate() {
                if *dst == actor_idx {
                    let required = actor.consumption_rates()[0];
                    if self.buffers[edge_idx].len() >= required {
                        let consumed: Vec<i32> = self.buffers[edge_idx]
                            .drain(0..required)
                            .collect();
                        inputs.push(consumed);
                    }
                }
            }

            if !inputs.is_empty() {
                // Fire actor
                let outputs = actor.fire(inputs);

                // Distribute outputs to outgoing edges
                for (edge_idx, (src, _dst)) in self.topology.iter().enumerate() {
                    if *src == actor_idx {
                        self.buffers[edge_idx].extend(&outputs[0]);
                    }
                }
            }
        }
    }
}

TypeScript: FBP with Streams

import { Readable, Writable, Transform } from 'stream';

/**
 * Information Packet
 */
interface IP<T> {
  type: 'data' | 'open-bracket' | 'close-bracket';
  data?: T;
}

/**
 * FBP Process base class
 */
abstract class FBPProcess<TIn, TOut> extends Transform {
  constructor() {
    super({ objectMode: true });
  }

  protected abstract process(ip: IP<TIn>): IP<TOut>[] | null;

  _transform(
    chunk: any,
    encoding: string,
    callback: Function
  ): void {
    const ip = chunk as IP<TIn>;
    const outputs = this.process(ip);

    if (outputs) {
      outputs.forEach(out => this.push(out));
    }

    callback();
  }
}

/**
 * Generator process
 */
class Generator extends Readable {
  private count: number = 0;
  private max: number;

  constructor(max: number) {
    super({ objectMode: true });
    this.max = max;
  }

  _read(): void {
    if (this.count < this.max) {
      const ip: IP<number> = {
        type: 'data',
        data: this.count++,
      };
      this.push(ip);
    } else {
      this.push(null); // End of stream
    }
  }
}

/**
 * Filter process
 */
class FilterProcess extends FBPProcess<number, number> {
  constructor(private predicate: (x: number) => boolean) {
    super();
  }

  protected process(ip: IP<number>): IP<number>[] | null {
    if (ip.type === 'data' && ip.data !== undefined) {
      if (this.predicate(ip.data)) {
        return [ip];
      }
    }
    return null;
  }
}

/**
 * Mapper process
 */
class MapperProcess extends FBPProcess<number, number> {
  constructor(private transform: (x: number) => number) {
    super();
  }

  protected process(ip: IP<number>): IP<number>[] | null {
    if (ip.type === 'data' && ip.data !== undefined) {
      return [{
        type: 'data',
        data: this.transform(ip.data),
      }];
    }
    return null;
  }
}

/**
 * Sink process
 */
class Sink extends Writable {
  private results: any[] = [];

  constructor() {
    super({ objectMode: true });
  }

  _write(
    chunk: any,
    encoding: string,
    callback: Function
  ): void {
    const ip = chunk as IP<any>;
    if (ip.type === 'data') {
      this.results.push(ip.data);
      console.log('Received:', ip.data);
    }
    callback();
  }

  getResults(): any[] {
    return this.results;
  }
}

/**
 * Build FBP network
 */
function buildFBPNetwork() {
  const generator = new Generator(10);
  const filter = new FilterProcess(x => x % 2 === 0);
  const mapper = new MapperProcess(x => x * 2);
  const sink = new Sink();

  // Wire network
  generator
    .pipe(filter)
    .pipe(mapper)
    .pipe(sink);

  return { generator, sink };
}

/**
 * SDF Actor interface
 */
interface SDFActor<TIn, TOut> {
  name: string;
  consumptionRates: number[];
  productionRates: number[];
  fire(inputs: TIn[][]): TOut[][];
}

/**
 * Example: SDF Downsampler
 */
class Downsampler implements SDFActor<number, number> {
  name = 'Downsampler';

  constructor(private factor: number) {}

  get consumptionRates(): number[] {
    return [this.factor]; // Consume 'factor' tokens
  }

  get productionRates(): number[] {
    return [1]; // Produce 1 token
  }

  fire(inputs: number[][]): number[][] {
    // Take first sample from each group
    const value = inputs[0][0];
    return [[value]];
  }
}

/**
 * SDF Graph
 */
class SDFGraph {
  private actors: SDFActor<any, any>[] = [];
  private topology: [number, number][] = []; // (src, dst) edges
  private buffers: any[][] = [];

  addActor(actor: SDFActor<any, any>): number {
    this.actors.push(actor);
    return this.actors.length - 1;
  }

  connect(src: number, dst: number): void {
    this.topology.push([src, dst]);
    this.buffers.push([]);
  }

  executeIteration(): void {
    // Simplified: fire each actor in order
    for (let actorIdx = 0; actorIdx < this.actors.length; actorIdx++) {
      const actor = this.actors[actorIdx];

      // Gather inputs
      const inputs: any[][] = [];
      for (let edgeIdx = 0; edgeIdx < this.topology.length; edgeIdx++) {
        const [src, dst] = this.topology[edgeIdx];
        if (dst === actorIdx) {
          const required = actor.consumptionRates[0];
          if (this.buffers[edgeIdx].length >= required) {
            inputs.push(this.buffers[edgeIdx].splice(0, required));
          }
        }
      }

      if (inputs.length > 0) {
        // Fire actor
        const outputs = actor.fire(inputs);

        // Distribute outputs
        for (let edgeIdx = 0; edgeIdx < this.topology.length; edgeIdx++) {
          const [src, dst] = this.topology[edgeIdx];
          if (src === actorIdx) {
            this.buffers[edgeIdx].push(...outputs[0]);
          }
        }
      }
    }
  }
}

PHP: Dataflow with Laravel Queues

<?php

namespace Dataflow;

use Illuminate\Support\Facades\Queue;
use Illuminate\Support\Facades\Redis;

/**
 * KPN Process abstraction using Laravel queues
 */
abstract class KPNProcess
{
    protected string $processId;
    protected array $inputQueues = [];
    protected array $outputQueues = [];

    public function __construct(string $id)
    {
        $this->processId = $id;
    }

    abstract protected function process(array $inputs): array;

    public function run(): void
    {
        while (true) {
            // Blocking read from input queues (KPN semantics)
            $inputs = $this->readInputs();

            if (empty($inputs)) {
                sleep(1); // Wait for data
                continue;
            }

            // Process data
            $outputs = $this->process($inputs);

            // Non-blocking write to output queues
            $this->writeOutputs($outputs);
        }
    }

    protected function readInputs(): array
    {
        $inputs = [];
        foreach ($this->inputQueues as $queueName) {
            $value = Queue::pop($queueName);
            if ($value !== null) {
                $inputs[$queueName] = $value;
            }
        }
        return $inputs;
    }

    protected function writeOutputs(array $outputs): void
    {
        foreach ($outputs as $queueName => $value) {
            if (in_array($queueName, $this->outputQueues)) {
                Queue::push($queueName, $value);
            }
        }
    }
}

/**
 * Generator process
 */
class Generator extends KPNProcess
{
    private int $count = 0;
    private int $max;

    public function __construct(string $id, int $max)
    {
        parent::__construct($id);
        $this->max = $max;
        $this->outputQueues = ['generator_out'];
    }

    public function run(): void
    {
        while ($this->count < $this->max) {
            $this->writeOutputs([
                'generator_out' => $this->count++
            ]);
            usleep(100000); // 100ms
        }
    }

    protected function process(array $inputs): array
    {
        return [];
    }
}

/**
 * Filter process
 */
class FilterProcess extends KPNProcess
{
    private $predicate;

    public function __construct(string $id, callable $predicate)
    {
        parent::__construct($id);
        $this->predicate = $predicate;
        $this->inputQueues = ['filter_in'];
        $this->outputQueues = ['filter_out'];
    }

    protected function process(array $inputs): array
    {
        $value = $inputs['filter_in'] ?? null;

        if ($value !== null && ($this->predicate)($value)) {
            return ['filter_out' => $value];
        }

        return [];
    }
}

/**
 * Mapper process
 */
class MapperProcess extends KPNProcess
{
    private $transform;

    public function __construct(string $id, callable $transform)
    {
        parent::__construct($id);
        $this->transform = $transform;
        $this->inputQueues = ['mapper_in'];
        $this->outputQueues = ['mapper_out'];
    }

    protected function process(array $inputs): array
    {
        $value = $inputs['mapper_in'] ?? null;

        if ($value !== null) {
            $result = ($this->transform)($value);
            return ['mapper_out' => $result];
        }

        return [];
    }
}

/**
 * SDF Actor interface
 */
interface SDFActor
{
    public function getName(): string;
    public function getConsumptionRates(): array;
    public function getProductionRates(): array;
    public function fire(array $inputs): array;
}

/**
 * Example: SDF FIR Filter
 */
class FIRFilter implements SDFActor
{
    private array $coefficients;
    private array $buffer = [];

    public function __construct(array $coefficients)
    {
        $this->coefficients = $coefficients;
    }

    public function getName(): string
    {
        return 'FIRFilter';
    }

    public function getConsumptionRates(): array
    {
        return [1]; // Consume 1 sample per firing
    }

    public function getProductionRates(): array
    {
        return [1]; // Produce 1 sample per firing
    }

    public function fire(array $inputs): array
    {
        $sample = $inputs[0][0];

        // Add to buffer
        array_unshift($this->buffer, $sample);

        // Keep only needed samples
        $this->buffer = array_slice(
            $this->buffer,
            0,
            count($this->coefficients)
        );

        // Compute filter output
        $output = 0;
        foreach ($this->coefficients as $i => $coef) {
            if (isset($this->buffer[$i])) {
                $output += $coef * $this->buffer[$i];
            }
        }

        return [[$output]];
    }
}

/**
 * SDF Graph executor
 */
class SDFGraph
{
    private array $actors = [];
    private array $topology = []; // [[src, dst]]
    private array $buffers = [];

    public function addActor(SDFActor $actor): int
    {
        $this->actors[] = $actor;
        return count($this->actors) - 1;
    }

    public function connect(int $src, int $dst): void
    {
        $this->topology[] = [$src, $dst];
        $this->buffers[] = [];
    }

    public function executeIteration(): void
    {
        foreach ($this->actors as $actorIdx => $actor) {
            // Gather inputs
            $inputs = [];

            foreach ($this->topology as $edgeIdx => [$src, $dst]) {
                if ($dst === $actorIdx) {
                    $required = $actor->getConsumptionRates()[0];

                    if (count($this->buffers[$edgeIdx]) >= $required) {
                        $consumed = array_splice(
                            $this->buffers[$edgeIdx],
                            0,
                            $required
                        );
                        $inputs[] = $consumed;
                    }
                }
            }

            if (!empty($inputs)) {
                // Fire actor
                $outputs = $actor->fire($inputs);

                // Distribute outputs
                foreach ($this->topology as $edgeIdx => [$src, $dst]) {
                    if ($src === $actorIdx) {
                        array_push(
                            $this->buffers[$edgeIdx],
                            ...$outputs[0]
                        );
                    }
                }
            }
        }
    }

    public function getBuffers(): array
    {
        return $this->buffers;
    }
}

/**
 * Example: Audio processing pipeline
 */
function createAudioPipeline(): SDFGraph
{
    $graph = new SDFGraph();

    // Create actors
    $firFilter = new FIRFilter([0.2, 0.5, 0.3]); // Lowpass
    $filterId = $graph->addActor($firFilter);

    // Add more actors and connections as needed

    return $graph;
}

Integration Points

With Actor Model

Dataflow nodes can be actors: Each actor has input/output ports
Message passing = data tokens: Actors communicate via dataflow edges
Supervision: Restart failed dataflow nodes

Example: Akka Streams implements dataflow on top of Akka actors.

With FRP

FRP networks are dataflow graphs: Signals flow through operators
Synchronous dataflow: FRP with discrete time steps
Behaviors and events: Continuous (behavior) vs. discrete (event) dataflow

Example: FRP switch operator is dataflow demultiplexer.

With Process Calculi

KPN is process calculus: Processes + FIFO channels = π-calculus variant
SDF has process algebra: Composition operators with rate annotations
Session types: Specify dataflow channel protocols

Example: Prove KPN determinism using process calculus semantics.

With Streaming Systems

Streaming frameworks use dataflow: Kafka Streams, Apache Flink are dataflow engines
Backpressure: Flow control in bounded dataflow channels
Windowing: Time-based grouping in dataflow

Example: Flink’s DataStream API is dataflow with windows.