Signal refinements: vectorized + reactive @wl.signal, overhead tuning, example#244
Open
AlexGrayBox wants to merge 9 commits into
Open
Signal refinements: vectorized + reactive @wl.signal, overhead tuning, example#244AlexGrayBox wants to merge 9 commits into
AlexGrayBox wants to merge 9 commits into
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).
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
1 participant
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.
Signal refinements: idiomatic example, overhead tuning, vectorized + reactive signals
A self-contained per-sample-signals example plus SDK improvements to the
@wl.signalsystem, driven by an overhead study. Each item is measured; nothing changes existing behavior unless opted into.Commits
examples/PyTorch/ws-signals-mnist/main.py) — base signals viasave_signals, a live derived@wl.signal, and end-of-run trajectory signals (6-shape categorical tag +loss_cv/loss_drop), with a per-step overhead breakdown.ledger_flush_max_rows— the default (100) flushes every ~1.5 steps at batch 64; setting it ≫ batch cuts measured overhead ~6× (185 → 32 ms/step), report unchanged.≤ batchis a landmine.@wl.signal(batched=True)—BatchSignalContext(arrays + a single batched.history()query). One call/step instead of one/sample. Measured: compute-only L1 −43%, history-read L2 −75% (4×); values bit-identical.ctx.latest(name)/bctx.latest(name): a signal subscribes to one trigger but ingests any number of other signals.ingests=[...]+StaleSignalError: reading a stale/missing ingested signal now raises instead of silently using bad data. Also: the executor no longer swallows all subscriber exceptions.inputs=[...]— unifiessubscribe_to+ingestsinto one dependency set. A signal fires when ALL its inputs are present at a step, order-independent, and fired signals chain. Verified: order-independence, exactly-once/step, chaining, correctness, back-compat.Notes
subscribe_tokeeps its legacy dispatch;inputs=[...]is the new reactive form; the two coexist without double-firing.overhead_study.py,thread_experiment.py,microbench_batchread.py) live outside the package; a consolidated "signal stress test" is being added toweightslab_kitchen.🤖 Generated with Claude Code