Distributed Data Processing Pipeline (DDP)

C++20 telemetry ETL with mmap zero-copy ingest, a lock-free SPSC ring buffer, AVX2 deserialization, per-metric aggregation, and reproducible latency/throughput benches.

Build

cmake -B build -DCMAKE_BUILD_TYPE=Release -DCMAKE_CXX_COMPILER=g++
cmake --build build -j

AVX2 is on by default (-mavx2). Disable with -DDDP_ENABLE_AVX2=OFF. Prefer g++ if clang++ cannot find libstdc++.

Quick start

mkdir -p data
./build/generate_telemetry --output data/telemetry.bin --count 5000000
./build/pipeline --input data/telemetry.bin --backend lockfree --threads 4 \
  --output data/metrics.csv --sink csv

Pipeline stages: mmap ingest → queue (lock-free or mutex) → SIMD parse → per-metric aggregate → optional CSV/binary sink.

Generate synthetic telemetry

./build/generate_telemetry --output data/telemetry.bin --count 1000000

Each record is a packed 64-byte TelemetryRecord (timestamp, device, metric, four floats, checksum).

Verify mmap ingest

./build/generate_telemetry --output data/telemetry.bin --count 100000
./build/verify_mmap data/telemetry.bin

MmapReader maps the file read-only and exposes a std::span<const TelemetryRecord> over the mapped bytes (no per-record copy).

Full pipeline CLI

./build/pipeline --input data/telemetry.bin \
  --backend lockfree \
  --threads 4 \
  --capacity 65536 \
  --chunk 256 \
  --output data/metrics.csv \
  --sink csv

Flag	Meaning
--input	Binary telemetry file
--threads	Worker threads (lock-free: one SPSC ring each; mutex: shared queue)
--capacity	Ring/queue capacity
--backend	lockfree (default) or mutex
--output / --sink	Optional aggregate sink (csv, binary, or auto from extension)
--verify	Count checksum failures while scanning

CSV sink columns: metric_id,count,sum,min,max.

Latency: mutex vs lock-free ring

./build/ring_buffer_smoke
./build/latency_bench --backend compare --ops 200000 --capacity 1024 --warmup 20000

Reports enqueue→dequeue handoff latency and latency_*_reduction_pct of lock-free vs mutex. Use --backend mutex or --backend lockfree alone as needed.

Throughput: AVX2 zero-copy parse

./build/generate_telemetry --output data/telemetry.bin --count 5000000
./build/throughput_bench --input data/telemetry.bin --mode parse --repeats 4
./build/throughput_bench --input data/telemetry.bin --mode pipeline --repeats 2

Target: ≥10M records/sec in Release with AVX2.

Reproducing resume-style numbers

Numbers vary by CPU; measure on your machine and cite the methodology.

Lock-free ingest latency reduction (vs mutex queue):

cmake -B build -DCMAKE_BUILD_TYPE=Release -DCMAKE_CXX_COMPILER=g++
cmake --build build -j --target latency_bench
./build/latency_bench --backend compare --ops 200000 --capacity 1024 --warmup 20000

Read latency_p50_reduction_pct / latency_p99_reduction_pct from the output (aspirational resume figure ~85%; report what you actually measure).

≥10M telemetry records/sec (SIMD + zero-copy):

./build/generate_telemetry --output data/telemetry.bin --count 5000000
./build/throughput_bench --input data/telemetry.bin --mode parse --repeats 4
./build/pipeline --input data/telemetry.bin --backend lockfree --threads 4

Read throughput_rec_per_sec. Direct parse and pipeline modes both exercise mmap + AVX2 paths without per-record memcpy.

Layout

Path	Role
include/ddp/telemetry_record.hpp	64-byte record schema
include/ddp/mmap_reader.hpp	mmap reader API
include/ddp/mutex_queue.hpp	mutex-backed bounded queue
include/ddp/lockfree_ring_buffer.hpp	SPSC lock-free ring buffer
include/ddp/simd_parser.hpp	AVX2/scalar parse API
include/ddp/aggregator.hpp	per-metric aggregates + sinks
src/mmap_reader.cpp	POSIX mmap implementation
src/simd_parser.cpp	AVX2 zero-copy deserializer
apps/pipeline_main.cpp	full ETL CLI
tools/generate_telemetry.cpp	synthetic data writer
tools/verify_mmap.cpp	smoke check for mapped records
tools/ring_buffer_smoke.cpp	SPSC push/pop integrity check
bench/latency_bench.cpp	mutex vs lock-free latency harness
bench/throughput_bench.cpp	mmap → ring → SIMD throughput harness