Interactivity on 100GB+ datasets: O(change) view sync and ledger writes

docs/perf/o_change_register.md

@@ -0,0 +1,170 @@
+# Interactivity for 100GB+ datasets — register of O(data) operations
+
+Goal: every per-step and per-request operation should cost **O(change)** or
+**O(page)**, never **O(dataset)**. Today several do, so cost grows with the
+dataset while the actual work stays constant.
+
+> **Baseline caveat.** The measurements below were taken against a stale
+> in-place copy of weightslab (`~/weightslab_src`, 1.3.3+multiview), which
+> differs from `dev` in 51 files. Two serving costs listed there are **already
+> fixed on dev** (see §B). Serving numbers must be re-measured on this branch
+> before any serving change is attributed an improvement. The storage findings
+> (§A) were re-verified against dev and still hold.
+
+Reference measurements (UltraEdit, 3,959,093 rows, ~19 cols, A10G box):
+
+| | measured |
+|---|---|
+| bare-torch step (no WL) | 1,162 ms → 20.66 samples/s |
+| with WL, no UI client | ~5.5 samples/s (**3.8× slower**) |
+| with WL + image requests | ~1.5 samples/s (**13.8× slower**) |
+| per-step signal write itself | **4 ms (0.34%)** — already fine |
+| grid page latency (64 imgs) | p50 5.2 s, p95 9.6 s |
+
+The signal path is not the problem. Storage write-amplification and the view
+rebuild are.
+
+---
+
+## A. STORAGE — `data/h5_dataframe_store.py`
+
+`upsert()` **receives** only dirty rows but **implements** a full table
+replacement.
+
+| line | operation | cost |
+|---|---|---|
+| 693 | `_create_backup()` — full file copy **before every upsert** | O(file) |
+| 708 | `existing = store.select(key)` — read entire table | O(N) |
+| ~768 | `pd.concat([existing, delta])` | O(N) |
+| ~772 | `existing[~existing.index.duplicated()]` — dedupe all rows | O(N) |
+| ~785 | `_decategorize_for_storage(existing)` | O(N) |
+| 801 | `store.remove(key)` — drop table | O(N) |
+| 804 | `store.append(..., data_columns=True)` — rewrite + index **every** column | O(N·cols) |
+| 846–883 | same read/remove/rewrite in the column-delete path | O(N) |
+
+**Amplification:** ~5 KB of changed signals per flush → ~200 MB written,
+roughly **40,000×**. At `ledger_flush_interval=3.0s` vs ~1.5 s steps, that is a
+full-table rewrite about every 2 steps.
+
+Fix direction: append new rows; modify existing rows in place
+(`select_as_coordinates` + `table.modify_rows`). Backup incrementally, not per
+upsert. No `data_columns=True` — no `store.select()` in this file uses `where=`,
+so those per-column indexes are built and never read.
+
+*(A previous attempt to narrow `data_columns` broke the write path entirely —
+678 upsert failures, zero persisted data. Any change here needs a
+write→read→assert-contents check, not just a timing check.)*
+
+## B. SERVING — `trainer/services/data_service.py`
+
+`_pull_into_all_data_view_df()` (line 938) runs several full-frame passes.
+
+**Already fixed on dev — do not re-report as wins:**
+- the collapse no longer re-enters `get_combined_df()`; the pulled frame is
+  passed in, so the frame is copied once, not twice
+- `array_proxy` no longer does a per-cell `.apply(convert_to_proxy)` (was
+  1,660 ms at 4M rows)
+
+Remaining, to be **re-measured on this branch**:
+
+| line | operation | cost (measured @4M) |
+|---|---|---|
+| 946 | `get_combined_df()` → `dataframe_manager:2140 self._df.copy()` | 101 ms–1.2 s |
+| — | `get_collapse_annotations_to_samples_df(df)` — groupby collapse | 6,326 ms* |
+| — | `safe_reset_index(df)` | 1,912 ms |
+| — | `set_index([origin, sample_id])` | O(N) |
+| 3636 | `updated_df.reindex(target_order)` | 290 ms |
+
+Callers — each one is a full O(N) rebuild: lines **440, 851, 3593, 4605, 4626**,
+reached from `GetDataSamples`, `GetMetaData`, `EditDataSample`, `GetDataSplits`.
+
+Held under `_update_lock`, which the trainer also needs → measured lock holds of
+39–126 s and the 3.7× training penalty while browsing.
+
+**The collapse is provably a no-op when `annotation_id.max() == 0`** (UltraEdit
+is exactly 1:1) yet still costs 6.3 s per rebuild.
+
+Fix direction: serve a page from the source frame by index (O(page)); rebuild
+the full view only for genuinely global operations (histogram, global sort);
+apply deltas rather than rebuilding; never hold the writer lock across a
+rebuild — build off-lock and swap the reference.
+
+## C. OTHER FULL SCANS
+
+| location | note |
+|---|---|
+| `dataframe_manager:1875` `data_snapshot.iterrows()` | input is O(change), but row-wise Python per flush |
+| `dataframe_manager:2400` `.apply(lambda …)` | per cell |
+| `data_service:1200` `_compute_natural_sort_stats` | builds a list of one Series per row (4M objects). Gated off (`compute_natural_sort=False`) — latent |
+| `data_service:538` PreviewCache | bounded by `WL_MAX_PREVIEW_CACHE_SIZE` — OK |
+
+## D. ALREADY O(change) — keep
+
+- `self._pending` dirty-row set (`dataframe_manager:95, 751, 761`)
+- flush work set: `work = list(self._pending)` (`:1827`)
+- `_origin_revisions` per-origin version counters (`:94, 1235`)
+
+The bookkeeping needed for differential updates already exists; the storage and
+view layers just don't use it.
+
+\* measured on the stale copy; re-measure on dev.
+
+## Measurement protocol
+
+Fixed workload: **1,000 train samples** = 41 steps at batch 24. Every change is
+reported as:
+
+1. wall-clock for the 41 steps, vs the bare-torch floor
+2. bytes written to H5 for those steps
+3. grid-page latency (64 images) and training throughput **while** serving
+4. **ledger contents verified** — signal columns present, measured-row count
+
+(4) is not optional: a previous "10× win" was writes silently failing.
+
+---
+
+# E. Triage — which call sites need a full reconstruction
+
+`_slowUpdateInternals()` rebuilds the whole view: `copy → collapse → reset_index
+→ set_index → reindex`. It has 18 call sites, and almost none of them need
+that. Most just want **fresh values for rows the trainer touched**, which is
+`O(change)`.
+
+`_fastUpdateInternals()` applies only dirty rows, via a maintained
+`sample_id → position` map (`_rebuild_view_pos_map`), and returns `False` —
+falling back to the full rebuild — whenever it cannot safely apply:
+
+- no view yet, or no position map (first build)
+- a dirty `sample_id` absent from the map (new rows ⇒ structural change)
+- backlog > `max_dirty` (a rebuild is genuinely cheaper)
+
+So the worst case is today's behaviour, never wrong data.
+
+| site | routing | why |
+|---|---|---|
+| `_bg_view_refresh` | **fast** | exists purely to refresh values after a stale read — the textbook differential case, and the one that holds `_lock` against the trainer |
+| `_process_get_data_samples` | **fast** | grid fetch needs current values, not a new frame |
+| `_compute_custom_signals` | **fast** | writes new signal *values*; schema unchanged |
+| `GetDataSplits` | **fast** | read-only summary |
+| `EditDataSample` ×5 | **fast** | per-sample value edits |
+| `EditDataSample` ×3 (`df.modify`, `df.drop_column`) | **full** | changes the schema — differential cannot add/remove columns |
+| `ApplyDataQuery` `@reset`/`@clear` | **full** | clears `_is_filtered` to restore the full universe; a differential updates values but cannot restore *dropped rows* |
+| `ApplyDataQuery` filter + agent paths | **full** (deferred) | a forced rebuild preserves `_is_filtered` (`:3691`), so swapping in a differential changes which rows the user sees. Rare, user-initiated, low perf value, high blast radius — not worth the risk until the filter semantics are pinned down |
+| `_compute_natural_sort_stats` | **full** | gated off (`compute_natural_sort=False`); latent |
+| `_manual_save_data_state` | **full** | explicit user save; correctness over speed |
+
+**Kill-switch:** `WL_FAST_VIEW=0` disables the differential and the position-map
+build, reproducing prior behaviour exactly. This is what makes a like-for-like
+A/B possible from a single tree.
+
+## Why the no-client benchmark cannot show this
+
+A 41-step run with no UI client attached records **0 rebuild events** — nothing
+calls `_slowUpdateInternals` at all, so the fast path has nothing to improve and
+correctly measures as no change. The rebuild cost only materialises when a
+client is attached, which is the case that measured **3.7× slower** with p50
+grid latency of 5.2 s.
+
+The A/B is therefore run under load: baseline → under-load → recovery phases
+within one training process (`t_imgload.py`), so each arm is normalised against
+its own idle throughput.

weightslab/backend/dataloader_interface.py

@@ -43,6 +43,44 @@
 _DENY_LIST_REFRESH_INTERVAL = 32
 
 
+def _close_inherited_h5_fds(worker_id: int = 0) -> None:
+    """DataLoader worker_init_fn: drop HDF5 handles inherited from the parent.
+
+    torch forks workers, so every fd the parent had open at fork time is
+    duplicated into the child -- including the ledger store. The child never
+    uses them, but their mere existence makes HDF5 refuse the parent's
+    read-write open, which silently kills ledger persistence.
+
+    Closes rather than just dropping the Python object: the fd is what holds
+    the file, and the child has no Python-level reference to it at all.
+    """
+    import os
+    try:
+        fd_dir = "/proc/self/fd"
+        for entry in os.listdir(fd_dir):
+            try:
+                target = os.readlink(os.path.join(fd_dir, entry))
+            except OSError:
+                continue
+            if target.endswith(".h5") or target.endswith(".h5.lock"):
+                try:
+                    os.close(int(entry))
+                except OSError:
+                    pass
+    except Exception:
+        # Never let cleanup break a worker: a leaked handle degrades
+        # persistence, a raising worker_init_fn kills the run.
+        pass
+
+
+def _with_worker_init(kwargs: dict, num_workers: int) -> dict:
+    """Attach the fd cleanup unless the caller supplied its own init."""
+    if num_workers and not kwargs.get("worker_init_fn"):
+        kwargs = dict(kwargs)
+        kwargs["worker_init_fn"] = _close_inherited_h5_fds
+    return kwargs
+
+
 def _resolve_safe_num_workers(dataset: Any, num_workers: int, loader_name: Optional[str] = None) -> int:
     """Clamp worker count for datasets that cannot be pickled by Windows spawn."""
     try:
@@ -177,6 +215,12 @@ def _get_deny_list_revision(self) -> Optional[tuple[str, int]]:
         try:
             origin = self._get_current_origin()
             df_manager = get_dataframe()
+            if origin and df_manager is not None and hasattr(df_manager, "get_discard_revision"):
+                # Deliberately NOT get_origin_revision: that moves on every
+                # per-sample signal write, i.e. every training step, so the
+                # cache below could never hit and __len__ rescanned 3.96M rows
+                # per batch. Discard state is what the deny-list depends on.
+                return ("discard", int(df_manager.get_discard_revision(origin)))
             if origin and df_manager is not None and hasattr(df_manager, "get_origin_revision"):
                 return ("origin", int(df_manager.get_origin_revision(origin)))
         except Exception:
@@ -514,6 +558,7 @@ def __init__(
                 self.tracked_dataset,
                 batch_sampler=batch_sampler,
                 num_workers=num_workers,
+                worker_init_fn=_close_inherited_h5_fds,
                 pin_memory=pin_memory,
                 collate_fn=collate_fn,
                 persistent_workers=self._should_persist_workers(num_workers),
@@ -1134,6 +1179,7 @@ def restore_iteration_state(self, state: dict) -> None:
                     self.tracked_dataset,
                     batch_sampler=sampler,
                     num_workers=num_workers,
+                    worker_init_fn=_close_inherited_h5_fds,
                     pin_memory=pin_memory,
                     collate_fn=collate_fn,
                     # Ensure no conflicting args are passed alongside batch_sampler
@@ -1220,6 +1266,7 @@ def set_batch_size(self, new_batch_size: int) -> None:
                         self.tracked_dataset,
                         batch_sampler=sampler,
                         num_workers=num_workers,
+                        worker_init_fn=_close_inherited_h5_fds,
                         pin_memory=pin_memory,
                         collate_fn=collate_fn,
                         persistent_workers=self._should_persist_workers(num_workers),
@@ -1232,6 +1279,7 @@ def set_batch_size(self, new_batch_size: int) -> None:
                         batch_size=batch_size,
                         shuffle=shuffle,
                         num_workers=num_workers,
+                        worker_init_fn=_close_inherited_h5_fds,
                         drop_last=drop_last,
                         pin_memory=pin_memory,
                         collate_fn=collate_fn,

weightslab/backend/logger.py

@@ -36,7 +36,7 @@
 import os
 import threading
 import time
-from collections import defaultdict
+from collections import defaultdict, deque
 
 import duckdb
 import pandas as pd
@@ -64,6 +64,18 @@
 _STAGE_FLUSH_THRESHOLD = 50_000
 
 # How often the background flush thread wakes up (see LoggerQueue._flush_loop).
+def _default_history_tail() -> int:
+    """Recent points kept per sample for signal-DAG history reads.
+
+    Bounded so history() costs O(batch) instead of scanning the whole
+    per_sample table (140ms at 20M rows, once per step, and growing).
+    """
+    try:
+        return max(0, int(os.environ.get("WL_HISTORY_TAIL", "16")))
+    except (TypeError, ValueError):
+        return 16
+
+
 def _default_flush_interval_seconds() -> float:
     try:
         return float(os.environ.get("WL_LOGGER_FLUSH_INTERVAL_SECONDS", "2.0"))
@@ -276,6 +288,10 @@ def __init__(self, register: bool = True, db_path: str = ":memory:") -> None:
         _qps_maxsize = int(os.environ.get("WL_QUERY_CACHE_MAXSIZE", "2048"))
         self._qps_version: dict = defaultdict(int)
         self._qps_cache_step: int = -1
+        # {signal: {sample_id: deque}} -- the recent tail of each sample's
+        # per-sample values, maintained on write so history() never scans.
+        self._tail_len = _default_history_tail()
+        self._recent_tail: dict = defaultdict(dict)
         self._qps_cache = functools.lru_cache(maxsize=_qps_maxsize)(self._query_per_sample_uncached)
         self._qps_step_cache = functools.lru_cache(maxsize=_qps_maxsize)(self._query_per_sample_at_step_uncached)
 
@@ -757,6 +773,20 @@ def _next_seq(self) -> int:
         self._seq += 1
         return s
 
+    def recent_per_sample(self, graph_name: str, sample_ids):
+        """Recent in-memory values per sample: ``{sample_id: [values]}``.
+
+        O(batch). Samples not written in this process are absent rather than
+        empty-listed; callers treat both as "not enough history yet".
+        """
+        tail = self._recent_tail.get(graph_name) or {}
+        out = {}
+        for s in sample_ids:
+            q = tail.get(str(s))
+            if q:
+                out[s] = list(q)
+        return out
+
     def _maybe_autoflush(self) -> None:
         if (len(self._stage_signals) + len(self._stage_sample)
                 + len(self._stage_instance)) >= _STAGE_FLUSH_THRESHOLD:
@@ -823,6 +853,13 @@ def _stage_sample_row(self, graph_name, exp_hash, sample_id, step, value):
         )
         self._qps_version[graph_name] += 1   # invalidate this signal's cached reads
         self._loss_shape_dirty_samples[(graph_name, exp_hash)].add(str(sample_id))
+        if self._tail_len:
+            _tail = self._recent_tail[graph_name]
+            _sid = str(sample_id)
+            _q = _tail.get(_sid)
+            if _q is None:
+                _q = _tail[_sid] = deque(maxlen=self._tail_len)
+            _q.append(float(value))
         self._maybe_autoflush()
 
     def _invalidate_qps_cache(self) -> None: