Distributed training framework

A from-scratch data-parallel training framework with fault injection and continuous correctness verification. Independent workers, their own data shards, gradients synchronised every step by ring all-reduce or a parameter server, and a chaos harness that kills workers mid-step while checking six invariants on every step.

The model is deliberately trivial — softmax regression on synthetic blobs — for the same reason raft-chaos-testing put a toy KV store under a real consensus implementation. The engineering is the distribution.

Report →

---

Headline result

Seven hand-built scenarios across both strategies, plus a 600-run randomized sweep over 3–8 workers, covering 1,960 checked training steps with 22 workers killed mid-training: no correctness violation.

Worst gradient-exactness error across every run: 8.9e-16, against a float noise floor near 5e-16 and a smallest detectable real bug near 2e-3. The margin is what makes that a result rather than a tolerance that happens to pass.

That is a negative result and its limits are worth stating: no violation was found by these faults. The sweep models worker crashes, stragglers and dropped or delayed links. It does not model a parameter-server failure, disk corruption of a checkpoint, or Byzantine workers returning plausible but wrong gradients.

Four of the six findings on the report are defects in this repository's own harness and tests, including two that would have produced a confidently wrong result. They are published rather than quietly fixed.

---

The property everything rests on

Data-parallel SGD is not an approximation of single-process SGD. For a mean-reduction loss over a partitioned batch they are the same computation:

∇L  =  Σ_k (n_k / N) · ∇L_k        N = Σ_k n_k

Two consequences drive the design:

1. The reference is exact, so the checker is a comparison rather than a heuristic. Every step, the applied update is compared against what one process would have computed over the union of the contributing batches, rebuilt by a route that shares no arithmetic with the production path. Lost gradients, double-counted gradients and wrong denominators all change that number.

2. The combination is weighted by shard size. An unweighted mean of per-worker means is correct only while every batch is the same size — which stops being true the moment a worker dies or an epoch ends on a short batch. It is invisible on a loss curve, and it is the bug this project was built to catch.

What is verified

Six properties, checked on every step so the report can say when:

1. Gradient exactness — the applied update equals the single-process update.
2. Replica agreement — every live worker holds bitwise identical parameters after a barrier. Bitwise is achievable because the reduction sums in canonical worker-id order; float addition is not associative, so without a fixed order this fails intermittently and looks like a race.
3. Contribution accounting — worker k's gradient for step t is counted exactly once, at step t.
4. Sample accounting — shards stay disjoint and cover the dataset, including after a death re-shards it.
5. Checkpoint fidelity — resuming reproduces an uninterrupted run bitwise.
6. Step monotonicity — the step counter advances by exactly one per update.

Deliberately not violations, reported separately: a dead worker's in-flight gradient vanishing (correct, and the analogue of an uncommitted Raft entry); loss rising on a given step (SGD is stochastic, convergence is a trend); a shrinking effective batch after deaths (the consequence of losing workers).

Two engines, cross-validated

	engines/sim.py	engines/procs.py
Workers	objects, one process	one OS process each
Transport	queued in-memory messages	multiprocessing.Pipe
Death	a flag	real SIGKILL
Reproducible	byte for byte from a seed	no, OS scheduling
Speed	fast	slow

The simulator is primary because a violation on seed 4193 replays exactly. But everything it verifies, it verifies against a model of failure. tests/ test_cross_engine.py runs the same training through both and asserts the weights match bitwise, including through a real kill — which is what makes a simulator result attributable to the framework.

Layout

dtf/
  model.py        softmax regression, float64, mean-reduction loss
  data.py         synthetic blobs; deterministic disjoint sharding
  protocol.py     message types; gradients carry (worker, step, n_samples)
  worker.py       local step only; message-passing, no shared state
  reduce.py       the weighted reduction, shared by both strategies
  allreduce.py    ring all-reduce, canonical reduction order
  paramserver.py  central aggregator, duplicate/stale rejection
  coordinator.py  barrier, straggler timeout, membership, fencing
  checkpoint.py   weights + RNG state + iterator position, atomic writes
  checker.py      the six invariants, per step
  events.py       structured event stream behind the report
engines/          sim.py (deterministic) and procs.py (real processes)
chaos/            faults.py, runner.py, scenarios.py, fuzz.py
bench/            cost_model.py (modelled), run_bench.py (modelled + measured)
tests/            equivalence, checkpoint, checker mutation, faults, cross-engine
run_experiments.py  -> site/results.json
build_site.py       -> site/index.html (self-contained)

Running it

python -m venv .venv && .venv\Scripts\activate   # source .venv/bin/activate elsewhere
pip install numpy pytest

pytest                              # 68 tests
pytest -m "not slow"                # skip the ones that spawn real processes
python -m chaos.fuzz 300            # randomized sweep
python run_experiments.py           # -> site/results.json
python build_site.py                # -> site/index.html

site/index.html inlines its data, so it opens from the filesystem as well as over HTTP.

Anything driving ProcEngine needs a real module and an if __name__ == "__main__" guard. The engine forces spawn on every platform — fork would let children inherit the parent's address space, which is exactly the shared state this project claims not to use — and spawn re-imports the main module in each child. A script piped to python - has no importable main module and its children will deadlock.

Benchmarks

Every figure is tagged modelled or measured and the two are never mixed.

• Modelled applies bench/cost_model.py to the real byte and hop counts the framework actually scheduled. Its assumptions are published beside its numbers.
• Measured is wall-clock from real processes. It is ground truth and it is narrow: workers are processes on one machine, so the "network" is a local pipe, which understates communication cost against any real cluster.

Two results worth reading with the caveats attached:

The parameter server beats all-reduce here, and that is not a general result. At 288 bytes of parameters the run is latency-bound, and a ring pays 2(N−1) sequential hops. All-reduce starts winning at roughly 64 KB, consistently from 4 to 32 workers. This model sits two orders of magnitude below that.

Measured speedup below 1.0 is the workload, not the framework. Four real processes are slower than one at 288 bytes. Speedup at four workers goes 0.62× → 1.28× → 1.59× as the model grows 288 B → 33 KB → 512 KB.

Deployment

.github/workflows/pages.yml runs the test suite, executes the scenarios and sweep, builds the report and publishes site/ to GitHub Pages. Results are baked at build time; there is no backend.