Rescue: reactive @wl.signal engine + per-sample signal perf (was orphaned branch signal-refinements)#260
Merged
Merged
Conversation
A single self-contained PyTorch example showing the three ways to declare
per-sample signals in WeightsLab, plus an honest overhead breakdown:
- base signals (from logits) pushed with wl.save_signals
- a live derived @wl.signal (loss normalized by a running mean)
- end-of-run derived signals from each sample's full loss trajectory:
a six-class shape as a text categorical tag (100% coverage) + loss_cv /
loss_drop, written back by sample_id
The run reports a per-step cost breakdown (baseline / loss-logging / signal
compute / persist) vs plain PyTorch, CUDA-synced, showing the signal math is
cheap and persistence dominates.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
The ledger default (flush_max_rows=100) flushes every ~1.5 steps at batch 64, which dominates the per-step cost. A sweep (batch 64, MNIST, single GPU): flush=4 +2086% flush=512 +440% flush=64 +1577% flush=4096 +210% flush=100 +631% flush=8192 +200% (H5 on/off no longer matters) Setting flush_max_rows well above the batch size (here 8192) flushes only a few times per epoch and cuts the measured overhead ~6x (185 -> 32 ms/step) with the report unchanged (100% coverage). A value <= batch size is a landmine (it flushes mid-batch). Flush frequency governs both the persist and the watched-loss logging cost; H5-persistence on/off is irrelevant once flush is high. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
The dynamic signal executor calls func(ctx) once per sample in a Python loop,
allocating a SignalContext per sample. For signals that read the ledger this is
catastrophic: each of the B calls does its own query_sample_history (lock +
flush + scan).
Add an opt-in batched contract:
- BatchSignalContext carries the whole batch as arrays (sample_ids,
subscribed_values) and exposes .history(name) as ONE query_per_sample for
the batch (not one per sample).
- executor branches on meta['batched']: build one context, call func once,
expect a length-B array back.
- fully backward compatible; the per-sample path is unchanged.
Measured (batch 64, MNIST, single GPU, flush 8192):
L1 compute-only (subscribed_value): 31 -> 18 ms/step (-43%)
L2 history read (per-sample query): 254 -> 63 ms/step (-75%, 4x)
subscriber invocations: per-sample 1728 -> batched 27 (one per step)
values identical to the per-sample path (max abs diff 0).
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
A dynamic signal subscribes to ONE trigger metric, but can now ingest any number
of OTHER signals by reading their current value from the ledger:
@wl.signal(name="sig/hardness", subscribe_to="train/loss_sample", batched=True)
def hardness(bctx):
return bctx.subscribed_values * bctx.latest("sig/entropy") \
* (1.0 - bctx.latest("sig/confidence"))
- BatchSignalContext.latest(name): most recent value of another signal for the
whole batch in ONE query -> (B,) array.
- SignalContext.latest(name): same for a single sample.
Requirements: the ingested signals must be logged (a watched metric, another
@wl.signal, or save_signals(..., log=True)) and written earlier in the step than
the trigger, so latest() sees this step's values.
Verified: hardness == loss*entropy*(1-confidence) to 2e-7 over 640 rows.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…ing errors A signal that ingests other signals (subscribes to one trigger, reads others via ctx.latest) could silently read STALE or missing inputs if they weren't logged, or were written after the trigger. Make that a loud error instead: - ctx.step / bctx.step carry the trigger's step. - ctx.latest(name, require_fresh=True) / bctx.latest(...) raise StaleSignalError unless the ingested signal has a value at the current step. - @wl.signal(..., ingests=["sig/a","sig/b"]) declares inputs; the executor validates their freshness for the whole batch BEFORE calling the signal. - StaleSignalError now propagates out of the subscriber dispatch (previously ALL subscriber exceptions were debug-logged and swallowed — silent failures). Verified: correct order (inputs logged before trigger) runs; wrong order raises StaleSignalError naming the signal, step, and affected samples. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…+ chaining Unify subscribe_to and ingests into one dependency declaration. A signal with inputs=[a, b, c] fires when ALL its inputs are present at a step, regardless of the order they were logged — eliminating the ordering footgun instead of just detecting it. Fired signals are themselves inputs, so signals chain. - _react_dependents(): iterative, dedup'd (fires each signal at most once/step), reads inputs fresh from the logger, persists results (log=True), and cascades via a work-queue (no re-entrant dispatch). - wired into wrappered_fwd (watched metric) and save_signals(log=True) so ANY logged signal can satisfy a dependent — that's what makes it order-independent. - subscribe_to keeps the legacy dispatch (handles non-per-sample metrics); the two paths coexist without double-firing (inputs= vs subscribe_to). Verified (batch 64): order-independent (entropy logged AFTER loss still fires the loss*entropy signal), exactly-once/step, chaining (B<-A<-entropy), values correct, and the existing subscribe_to signal still fires. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Move reactive signal computation off the training thread. When hyperparam ledger_signal_worker=True, the reactive dispatch is handed to a background worker (WL-Signal-Worker) instead of running inline; the train thread just detaches + enqueues (seed_names, ids, step). The seed signals are logged synchronously, so the worker only defers the derived compute + persistence. - _dispatch_or_enqueue() at both injection sites; drain_signals() (exported) joins the queue; write_dataframe() auto-drains so the report is complete. - _gather_inputs_fresh selects the value AT the job's step (not "latest"), so a lagging worker still gathers that step's inputs. - save_signals now honors an explicit step= (as its docstring says) instead of letting the live model age override it — required so reactive backfill logs derived values at the job's step; otherwise chained signals can't find them. Verified worker on vs off: exactly-once/step, order-independent, chaining (B<-A<-entropy) all identical; drain makes the report complete. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…warning Reactive-signal overhead was dominated by redundant, growing ledger reads. Logger query cache (backend/logger.py): - Memoize query_per_sample via functools.lru_cache keyed (signal, ids, hash, version[signal]); _stage_sample_row bumps the per-signal version (per-signal, not global, so persisting a derived signal doesn't invalidate the loss its siblings are still reading). - Step-scoped: clear both caches when the training step advances. The loader reshuffles ids and the version bumps every step, so a key never recurs across steps — cross-step entries are pure dead weight. Bounds the cache to one step (currsize ~6 vs ~1600 accumulating), killing the churn while keeping the intra-step reuse (10 signals sharing the loss -> 1 query). - query_per_sample_at_step(name, ids, step): O(batch) value-at-step read (WHERE step=? in-engine) behind its own step-cache. Core (src.py): - _gather_inputs_fresh uses the value-at-step read -> flat as history grows (single query 4.7->13ms/2.8x before; 2.1->2.4ms/1.1x after). - Circular-dependency detection: _detect_signal_cycles (grey/black DFS over the inputs graph) + _warn_on_signal_cycles, lazy+idempotent, fired from start_training AND first reactive dispatch so it works headless (no wl.serve). Warns, does not raise; cyclic signals just stay dark. Measured (100-epoch MNIST headless, no wl.serve): per-step time flat (~157ms) vs rising 170->248ms; total wall 430s->286s (-33%). Reactive verify, multi-level/multi-parent chain, and cycle tests all pass.
…lar dict Two profile-driven per-step wins (−30% wall at N=60k, full reactive config; 124.8 → 87.2 ms median), correctness unchanged (signal stress verify all pass). #2 (logger.py): serve query_per_sample_at_step from the in-memory staging buffer. The reactive gather reads the CURRENT step's value, which was just staged and isn't in DuckDB yet — so a reverse scan of the staging list (early break once all ids found) skips the flush -> register(pandas) -> INSERT -> unregister -> SELECT round-trip that profiling showed was ~a third of per-step cost. Falls through to DuckDB only when an id isn't in the buffer (mid-step flush moved it / older step). Also flattens the epoch-over-epoch drift: current -step reads no longer scan the growing per_sample table (measured flat ~75 ms over 98 epochs at 70k, vs the old code rising 110→137 in 7). #3 (src.py): vectorize the id->scalar dict build — one .tolist() (a single device sync) instead of a per-element .item() comprehension (B device syncs). Note: a row-wise executemany flush was tried and REVERTED — it was ~6x slower; register(pandas)+INSERT SELECT is DuckDB's fast bulk path (comment left in _flush_stage to prevent re-attempts).
…ds_raw gate Makes @wl.signal the efficient path (no user save_signals needed), verified by the signal stress test (all checks pass). - ctx.logits/preds/targets on SignalContext + BatchSignalContext: a subscribe_to signal can compute from the model output (entropy, confidence, ...) — logit-derived signals become @wl.signal, no save_signals push. - Subscriber results logged id-keyed (dict), so reactive dependents can read them; reactive dispatch seeded with subscriber results so chains like conf_r<-entropy<-logits fire. - Batched reactive persist: _react_dependents accumulates all fired signals in-memory (chains read from there) and persists once, not per signal. DuckDB execute 1404->936, register 702->468 (-33%); the @wl.signal path is now faster than batched save_signals. - ledger_store_preds_raw hyperparam (default on): gate off to skip persisting the (B,C) logits every step (~7% saving); ctx.logits still works. - Trim verbose comments ~70% throughout.
… reactive) 460->180 lines. Per-step user code is just the watched loss; entropy is a @wl.signal from ctx.logits, loss_norm/hardness are reactive, shapes classified at the end. No save_signals.
…-pass A reactive-derived input only lives in the in-memory fresh_cache during the pass; querying the ledger for it (when a dependent is attempted before it fires) was a guaranteed-miss flush every step. Skip cheaply instead. Per-step 107->55ms (-48%), 1 flush/step -> 0; verify all pass.
…ve loss_shape signal, overhead readout)
… cycle detection, ctx.logits, freshness, reactive-derived gather skip)
…SIZE, default 2048) Addresses PR #244 review (Guillaume): let users size the dedicated query cache.
…review #3) Two-layer, reusable over any per-sample signal: - classify_loss_shape(values, ...): default shape classifier, all thresholds named + configurable (natural defaults; tune without rewriting). - write_signal_shapes(signal, tag, classifier): engine — classify every sample's trajectory of ANY signal (loss, accuracy, a second loss, any metric) into a categorical tag; returns the {label: count} distribution. - write_loss_shapes(): convenience wrapper for the loss signal. - write_dataframe(loss_shape_signal=...): runs the summary on report write (tags + logs the distribution), so a report written every N steps stays current. Exported: classify_loss_shape, write_loss_shapes, write_signal_shapes, LOSS_SHAPES. Tests added; existing signal/logger suites pass.
- trajectory_stats(values): generic reusable feature layer (renamed from loss_trajectory_stats — works on any trajectory). - classify_loss_shape: built on trajectory_stats; every threshold a named param. - write_signal_shapes: engine over any per-sample signal; write_loss_shapes: wrapper. - enable_loss_shape_signal(loss_signal, every=N): live per-step @wl.signal toggle (the heavier live counterpart to the report-time write_loss_shapes). - example ws-signals-mnist: drop the local classifier, use the core hook wl.write_dataframe(loss_shape_signal=...). Exports + tests updated.
# Conflicts: # weightslab/src.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Rescuing the reactive per-sample signal engine work. This branch (
signal-refinements) was deleted from origin but still held 21 unmerged commits in a local checkout — re-pushing so it's preserved and reviewable.Scope (mostly
weightslab/src.py+backend/logger.py):@wl.signalengine — vectorizedbatched=True, multi-input viactx.latest()/inputs=[...], logit-derived signals, in-coreledger_signal_workerthread,StaleSignalErrorfresh-ingest enforcement.ledger_flush_max_rowstuning.devhas moved substantially since this was cut — expect a rebase / conflict pass before merge. Opening primarily to preserve the work and surface how mergeable it is.🤖 Generated with Claude Code