mixle.utils.parallel.sdc_audit module

Silent-data-corruption (SDC) audit for the resilient mp EM backend (K5).

Status audit this module leans on (see K4’s docstring in resilient_em.py, which this module builds directly on top of): fits are deterministic given seed (#115) and sufficient statistics are ADDITIVE. K4 already turns those two facts into exact, cheap failure recovery (a crashed worker’s shard is just redone). K5’s contribution is different: it turns the SAME two facts into a cheap corruption detector.

The idea. Silent data corruption (SDC) is a bit-flip in memory or compute that produces a WRONG result with no exception and no crash – the fit just silently converges to a corrupted answer. For gradient-based training this is expensive to catch (you’d have to redo an entire gradient step, or use redundant hardware). mixle’s additive-stat EM makes it cheap: recompute the SAME shard’s accumulator TWICE, from the SAME raw shard bytes / same estimator / same model / same sub_chunks split, once on the shard’s normal (“primary”) rank and once on a DIFFERENT (“audit”) rank.

The verdict is deliberately not based on pickle-byte equality. Healthy heterogeneous workers can differ in serialization order or in the last numerical bits because of CPU/GPU kernels, BLAS implementations, fused operations, or library versions. Payloads are decoded into a closed typed statistic, compared under the explicitly configured audit_rtol/audit_atol numeric envelope, and separately canonicalized for exact evidence digests. Every mismatch receipt records that envelope, maximum observed errors, the typed canonical digests, and the raw payload digests. Exact raw agreement remains useful evidence but is never assumed to be a portable proof of worker health.

NaN/Inf watchdog. mixle.models._neural_serial.check_finite already guards individual density evaluations. K5 extends that same “fail loud, immediately, with the offending location named” philosophy to the accumulator-fold boundary: finite_guarded_fold() is a drop-in replacement for checkpointed_fold() that checks finiteness of the running accumulator’s value() immediately after EVERY combine() call, not just once at the end – so a NaN/Inf introduced while folding payload i is caught at payload i’s combine() boundary, before it silently propagates into payload i+1..n. The fully tied value is checked again after key_merge/key_replace so shared-key transformations cannot introduce an unchecked invalid state. Scope note: this wraps operations made by THIS module’s own fold loop only – it deliberately does NOT touch SequenceEncodableStatisticAccumulator.combine()’s contract itself, which every other caller in the codebase still uses unguarded, exactly as before.

class AuditedMPEncodedData(data, estimator=None, encoder=None, num_workers=None, sub_chunks=1, max_retries=2, audit_rate=0.1, audit_rtol=1.0e-12, audit_atol=1.0e-12, rng=None, worker_timeout_s=30.0)[source]

Bases: ResilientMPEncodedData

ResilientMPEncodedData (K4) plus a continuous SDC audit (K5).

Every round (pysp_seq_estimate / pysp_stream_accumulate), a random audit_rate fraction of shards are ALSO recomputed on a second, different rank via the same ad hoc "update_shard" wire command K4 already uses for shard recovery (see ResilientMPEncodedData._recover_shard) – no new worker-side machinery. The primary and audit (count, value()) payloads are compared as typed statistics under an explicit numerical envelope. A mismatch:

  1. is recorded as a structured SDCAuditReceipt (self.audit_receipts, self.last_round_audit_mismatches);

  2. obtains a third independent witness before identifying a suspect. A two-way disagreement alone is recorded but never used to quarantine either worker;

  3. stages every affected shard on validated survivors before atomically committing the suspect’s retirement. If no safe witness or placement exists, quarantine is deferred and that decision is recorded in the receipt.

The main EM round itself (retry, blacklisting on repeated failure, checkpointed fold) is entirely K4’s, untouched, reused via inheritance – K5 only adds the audit phase (run BEFORE the main round each call, so a quarantine decided by the audit is already reflected in that same round’s live-worker set) and swaps K4’s plain checkpointed_fold for finite_guarded_fold() via the _fold_fn hook K4 exposes for exactly this purpose.

Parameters:

  • audit_rate (float) – fraction of shards (0..1) redundantly recomputed each round.

  • rng (np.random.RandomState | None) – drives which shards are audited each round and which live rank is picked as the second (“audit”) rank.

  • audit_atol (float) – finite nonnegative tolerances used for floating and complex statistic leaves. Integer and categorical leaves remain exact.

  • data (Sequence[Any])

  • estimator (Any | None)

  • encoder (Any | None)

  • num_workers (int | None)

  • sub_chunks (int)

  • max_retries (int)

  • audit_rtol (float)

  • audit_atol

  • worker_timeout_s (float)

Testing hook:

arm_corruption() registers a one-shot-per-round hook that can mutate a payload right after it comes off the wire from an ad hoc "update_shard" recompute, letting a test inject a deterministic, reproducible corruption (see inject_bit_flip()) into a chosen (rank, shard, role) combination without touching worker internals – the same “observe the wire, mutate deterministically” pattern K4’s arm_kill uses.

arm_corruption(hook)[source]

Register hook(worker_id, shard_id, role, payload_bytes) -> payload_bytes (role is "primary" or "audit"), applied to every ad hoc "update_shard" payload this instance receives until cleared. Unlike arm_kill this is NOT one-shot by default (an SDC fault is typically persistent, e.g. a stuck bit in one DIMM) – clear it explicitly with arm_corruption(None) to simulate a transient fault.

Parameters:

hook (Any)

Return type:

None

pysp_seq_estimate(estimator, prev_estimate)[source]

One distributed EM step, tolerant of a worker dying mid-accumulation.

Parameters:

  • estimator (Any)

  • prev_estimate (Any)

Return type:

Any

pysp_stream_accumulate(estimator, model)[source]

Return globally folded batch sufficient statistics for streaming EM, chaos-tolerant.

Parameters:
Return type:

tuple[float, Any]

class SDCAuditReceipt(round, shard_id, primary_worker, audit_worker, primary_nbytes, audit_nbytes, first_diff_byte_offset, primary_sha256, audit_sha256, comparison_mode='typed_numeric_envelope/v1', audit_rtol=0.0, audit_atol=0.0, max_abs_error=0.0, max_rel_error=0.0, mismatch_reason='', primary_canonical_sha256='', audit_canonical_sha256='', witness_worker=None, suspected_workers=(), quarantine_status='not_requested', primary_value_repr='', audit_value_repr='')[source]

Bases: object

A structured record of one detected primary-vs-audit mismatch: which shard, which two ranks, and a summary of what actually differed (never just a bare “mismatch” boolean).

Parameters:

  • round (int)

  • shard_id (int)

  • primary_worker (int)

  • audit_worker (int)

  • primary_nbytes (int)

  • audit_nbytes (int)

  • first_diff_byte_offset (int | None)

  • primary_sha256 (str)

  • audit_sha256 (str)

  • comparison_mode (str)

  • audit_rtol (float)

  • audit_atol (float)

  • max_abs_error (float)

  • max_rel_error (float)

  • mismatch_reason (str)

  • primary_canonical_sha256 (str)

  • audit_canonical_sha256 (str)

  • witness_worker (int | None)

  • suspected_workers (tuple[int, ...])

  • quarantine_status (str)

  • primary_value_repr (str)

  • audit_value_repr (str)

finite_guarded_fold(estimator, payloads, where='AuditedMPEncodedData.combine')[source]

checkpointed_fold, but check finiteness of the running accumulator immediately after EVERY combine() call (see module docstring for why this is a wrapper around this module’s own fold loop rather than a change to combine()’s contract).

Parameters:
Return type:

tuple[float, Any]

inject_bit_flip(payload, bit_offset=None)[source]

Flip exactly one bit of payload, deterministically, and return the corrupted bytes.

A real, reproducible stand-in for a hardware/software bit-flip: this is the corruption primitive the acceptance tests inject via AuditedMPEncodedData.arm_corruption(). bit_offset defaults to the middle bit of the payload (an arbitrary but fixed choice – determinism of the test does not depend on which bit, only that the two payloads it is applied to differ afterward).

Parameters:

  • payload (bytes)

  • bit_offset (int | None)

Return type:

bytes