M3: fundamental thumbnail-loader redesign — viewport is first at every queue level

The reported 'fast scroll → stop → long shimmer → small move → everything
pops' survived the first fix (scroll-handler re-submission) and the second
(central-queue prepend): re-submission only happened in on_scroll, and the
workers' FIFO mpsc still held stale transit work ahead of fresh batches.

Root causes and the full redesign:
- on_scroll now ONLY records the viewport (zero work). The 10 Hz tick is
  the sole authority AND sole submitter: CONSUME (paint viewport results,
  capped), EVICT (far pixels, MAX_PAINTED=120), FILL (viewport,
  center-first), PRUNE (queued jobs outside the retention band).
- Invariant restored unconditionally each tick and debug_asserted: every
  viewport cell is PAINTED, IN-FLIGHT, or FAILED. A visible cell is empty
  only while its decode is genuinely in flight — never until the next
  scroll event.
- Worker queues: Mutex<VecDeque>+Condvar (mpsc is FIFO-only). The
  dispatcher drains the central queue as priority-ordered bursts and
  PREPENDS them, so fresh viewport work always jumps ahead of stale
  transit work. Per-worker cap (96) drops the oldest tail with a
  'skipped' result (UI clears the in-flight marker; re-entry re-submits
  — disk-cache hit, never a re-decode). A burst (<= viewport ~45) always
  survives the cap.

Stress-verified on the emulator (debug build) before phone deployment:
- fast up/down x5 (the repro): fully covered 0-119 ms after stop
- cold landings: 1.0-1.2 s = pure decode throughput (p rises
  monotonically while stale jobs wait behind, harmlessly)
- chaos x6 + 25 s idle soak: covered before every stop; max oldest
  in-flight job 217 ms (pre-fix: 8,055 ms); max queue depth 45; 0
  invariant violations in ~1,450 ticks; 0 panics; idle -> 0 queued
Phone (release aarch64, 8,577 photos): 99.98% cache hits in stress
scroll, 0 panics/ANRs.

Tests: 15 pass incl. fresh_batch_preempts_stale_worker_queue (new) and
the existing scroll_back_cache_hit_not_starved; 5k engine test uses an
uncapped queue (usize::MAX cap param).
This commit is contained in:
Greg Pomerantz 2026-10-27 21:02:35 -04:00
parent ebb4023955
commit 2e9e5b78b6
3 changed files with 622 additions and 234 deletions

View File

@ -6,17 +6,24 @@ point for development: it captures every decision, gotcha, and measurement
from the framework evaluation that preceded it, so no learning has to be from the framework evaluation that preceded it, so no learning has to be
rediscovered. rediscovered.
**Status: M3 done (2026-10-27)** — M0 scaffold, M1 metadata scan + sqlite + **Status: M3 done (2026-10-27, twice overhauled)** — M0 scaffold, M1 metadata
instant list paint, M2 thumbnail engine (turbojpeg DCT-shrink decode, worker scan + sqlite + instant list paint, M2 thumbnail engine (turbojpeg DCT-shrink
pool, disk cache, cancellation), M3 grid UI (3-column ListView grid, shimmer decode, worker pool, disk cache, cancellation), M3 grid UI (3-column ListView
placeholders, scroll-driven prefetch, memory-bound cell retention) all grid, shimmer placeholders, memory-bound cell retention) all verified: 15
verified: 14 unit tests (incl. the scroll-back FIFO-starvation regression), unit tests (incl. two queue-priority regressions), 5k synthetic engine test,
5k synthetic engine test, on the emulator (real JPEGs: first paint 0.6 s and on a real Pixel 9 Pro (8,577 photos: 60 fps, 0 hitches). The thumbnail
cold, warm relaunch from disk cache, 0 ANRs through 16 fast swipes), and on loader was rewritten TWICE on 2026-10-27 because "stop → long shimmer → small
a real Pixel 9 Pro (8,577 photos: grid renders, scroll = 60 fps with 0 move → everything pops" kept surviving: first the scroll handler forgot cells
hitches). The reported scroll-back-to-top shimmer was root-caused (FIFO job when scrolling stopped (fix: **viewport reconciliation** — the 10 Hz tick is
queue starvation) and fixed (prepend-on-submit) + regression-tested. Next: the sole authority, the invariant "every viewport cell is painted / in flight
M4 viewer. / failed" is restored unconditionally each tick and `debug_assert`ed), then
the worker-level FIFO still let stale transit work delay a fresh viewport
(fix: **worker-queue prepend + cap** — bursts are prepended ahead of older
queued work; per-worker cap drops the oldest tail with a `skipped` result).
Stress-verified on the emulator (debug build) + phone before shipping:
fast up/down 0–120 ms to full coverage, cold landings bounded by decode
throughput, chaos scrolling max oldest in-flight job 217 ms (was 8 s),
0 invariant violations in ~1,450 ticks, 0 panics/ANRs. Next: M4 viewer.
--- ---
@ -414,17 +421,51 @@ pool threads), and `logcat` pool timing logs behind a `debug_timing` flag.
scroll; `on_scroll` only re-anchors the active range + submits prefetch scroll; `on_scroll` only re-anchors the active range + submits prefetch
(no model mutation), all mutation is in the 10 Hz drain timer. (no model mutation), all mutation is in the 10 Hz drain timer.
**Scroll-back-to-top shimmer (fixed 2026-10-27, reported from the phone):** **Thumbnail loading: viewport reconciliation + worker-level priority
scrolling down fills the worker queue with slow cold decodes of far cells; (rewritten 2026-10-27 after two field-reported bugs).** Design doc is the
scrolling back up re-submits the top cells as fast disk-cache HITS, but on header comment of `src/lib.rs`; engine in `src/thumb.rs`. The invariants:
the old FIFO job queue they waited behind the whole cold backlog — seconds
of shimmer (deterministic repro: 6.3 s behind a 180-cold backlog). Fix: the 1. *The viewport is the only source of truth.* `on_scroll` only records the
job queue is now a `Mutex<VecDeque>` + `Condvar` and `Engine::submit` viewport range (zero work). The 10 Hz tick is the sole authority AND the
**prepends** each new batch (most-recent intent first; stale far prefetch sole submitter: CONSUME finished results (paint viewport ones, capped),
can no longer starve the viewport). Regression test EVICT far pixels (hard cap `MAX_PAINTED=120`), FILL the viewport
`thumb::tests::scroll_back_cache_hit_not_starved` (6.3 s → 130 ms). Note (center-first), PRUNE queued jobs outside the retention band. After each
`mpsc` can't do this — it's FIFO-only; a priority queue needs the tick: every viewport cell is PAINTED, IN-FLIGHT, or FAILED — restored,
mutex+condvar (dispatcher exits via a `closed` flag set in `Engine::drop`). not maintained, so no scroll pattern can leave a visible cell forgotten
(the original bug: re-submission only happened in `on_scroll`, so when
scrolling STOPPED, forgotten cells stayed empty until the next small
move). `debug_assert`ed every tick.
2. *The viewport is first at EVERY queue level.* The central queue
(`Mutex<VecDeque>` + `Condvar`, prepend-on-submit) is drained as
priority-ordered bursts and PREPENDED to the per-worker queues (also
`Mutex<VecDeque>` + `Condvar`, NOT mpsc — mpsc is FIFO-only). Each
worker queue is capped (96 ≈ two screens): overflow is dropped from the
TAIL (oldest transit work) with a `skipped: true` result; the UI clears
the in-flight marker, the cell is NOT failed, re-entry re-submits it
(disk-cache hit). A burst ≤ viewport (~45) < cap, so fresh viewport work
is never dropped. `in_flight` accounting: +1 at submit, −1 at finish OR
at cap-drop; the UI's `inflight` map (path→time) is the dedupe key and
is cleared on every result (incl. `skipped`) — a dropped job whose
path has a newer queued job may be briefly re-submitted (one wasted
decode, self-cleans).
Regression tests: `thumb::tests::scroll_back_cache_hit_not_starved`
(central-level: 6.3 s FIFO starvation → 130 ms) and
`thumb::tests::fresh_batch_preempts_stale_worker_queue` (worker-level: a
fresh job submitted while a slow batch is queued must finish within the
first two results, never behind the whole old queue).
*Stress results (emulator x86_64 DEBUG build — slowest plausible hardware —
then release aarch64 on the Pixel 9 Pro):* fast up/down ×5 (the reported
repro): viewport fully covered 0–119 ms after stop (4/5 already covered at
stop); cold landings: 1.0–1.2 s = pure decode throughput (p rises
monotonically while stale jobs wait BEHIND, oldest growing 700→1900 ms
harmlessly); chaos ×6 + 25 s idle soak: covered before every stop, max
oldest in-flight job 217 ms (pre-fix: 8,055 ms), max queue depth 45 (one
viewport), 0 invariant violations in ~1,450 ticks, 0 panics, idle
converges to 0 queued. Phone smoke: 99.98% cache hits during 8k-cell
stress scroll, 0 panics/ANRs, one 541 ms transient stall alarm (cold
decode burst).
*Verified (emulator, 150 real JPEGs + 5k stubs): first paint 0.6 s cold; *Verified (emulator, 150 real JPEGs + 5k stubs): first paint 0.6 s cold;
warm relaunch from disk cache; 0 ANRs through 16 fast swipes; backlog warm relaunch from disk cache; 0 ANRs through 16 fast swipes; backlog

View File

@ -4,15 +4,61 @@
//! - scan thread: one per launch, walks storage, owns the scan DB connection. //! - scan thread: one per launch, walks storage, owns the scan DB connection.
//! - thumb engine: process-wide pool (dispatcher + N workers), created on the //! - thumb engine: process-wide pool (dispatcher + N workers), created on the
//! UI thread once; workers own one DB connection each (WAL). //! UI thread once; workers own one DB connection each (WAL).
//! - UI thread: owns the window + model; drains engine results 10x/s via a //! - UI thread: owns the window + model; runs the reconcile tick 10x/s via a
//! Slint timer; builds `slint::Image` here (Image is !Send, pixel bytes are). //! Slint timer; builds `slint::Image` here (Image is !Send, pixel bytes are).
//! //!
//! M3 memory bound: grid cells decode at 500 px (~0.7 MB RGB8 each), so the //! # Thumbnail loading: viewport reconciliation (rewritten 2026-10-27)
//! model may NOT keep pixels for the whole 8k+ entry list. Cells only hold //!
//! an image while inside the *active range* (visible rows ± margin). Results //! One source of truth for what matters: the **viewport** (visible rows ±
//! that finish out-of-range are held in a small deferred queue and painted //! `MARGIN_ROWS`), updated by scroll events — and only by them. Scroll
//! when the range catches up; cells that scroll out of range are dropped back //! handlers do no work; they move the target.
//! to shimmer (the disk cache repaints them in ~10 ms on re-entry). //!
//! The 10 Hz tick is the single authority. Every tick, looking only at
//! CURRENT state (never at scroll history):
//! 1. CONSUME: drain finished results; paint the viewport ones (capped);
//! drop the rest — every finished decode is on the disk cache, so a
//! dropped result costs a ~10 ms cache hit on re-entry, never a re-decode.
//! 2. EVICT: release pixels for painted cells far outside the viewport
//! (farthest first; row-budgeted, plus a hard `MAX_PAINTED` cap).
//! 3. FILL: submit every viewport cell that has no pixels and no in-flight
//! job, center-first.
//! 4. PRUNE: drop queued jobs outside the retention band (belt; the engine
//! also enforces this per-worker — see below).
//!
//! The invariant, RESTORED unconditionally each tick:
//!
//! every viewport cell is PAINTED, or has a job IN FLIGHT,
//! or is permanently FAILED (undecodable).
//!
//! Because it is restored (not maintained), no scroll pattern, cap hit,
//! dropped result, or list reset can leave a visible cell forgotten until
//! the next scroll event. It is `debug_assert`ed on every tick.
//!
//! "The viewport is first" holds at EVERY level of the pipeline (the engine,
//! thumb.rs):
//! - The central queue is drained as priority-ordered BURSTS (newest batch
//! first, center-first within a batch) and PREPENDED to the worker queues
//! — fresh viewport work always jumps ahead of older queued work.
//! - Each worker queue is capped (96 ≈ two screens): overflow is dropped
//! from the TAIL (oldest transit work) with a `skipped` result, so the UI
//! clears the in-flight marker and re-entry re-submits (disk-cache hit).
//! A burst is never bigger than the viewport (~45), so fresh viewport work
//! is never itself dropped.
//!
//! Consequences (stress-verified 2026-10-27, emulator debug build):
//! - After any stop — fast up/down, cold landing, or chaos flicking — the
//! viewport is fully covered within decode throughput time (0–120 ms warm,
//! ~1 s for a cold 45-cell landing on a slow emulator; ~0.25 s on a phone).
//! - No stale backlog exists: measured oldest in-flight job after chaos
//! scrolling was 217 ms (pre-fix design: 8 s), max queue depth one
//! viewport. A visible cell is empty only while its decode is genuinely
//! in flight — never "until the next small scroll move".
//!
//! # Memory bound
//! Grid cells decode at 500 px (~0.7 MB RGB8 each); the model may NOT keep
//! pixels for the 8k+ entry list. Pixels are held only near the viewport
//! (evicted at `MARGIN_ROWS` distance) and are hard-capped at `MAX_PAINTED`
//! cells. The disk cache (`thumbs/`) is the persistent store.
slint::include_modules!(); slint::include_modules!();
@ -21,11 +67,12 @@ pub mod images;
pub mod scan; pub mod scan;
pub mod thumb; pub mod thumb;
use std::collections::HashSet; use std::collections::{HashMap, HashSet};
use std::ops::Range; use std::ops::Range;
use std::path::Path; use std::path::Path;
use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Mutex, OnceLock}; use std::sync::{Mutex, OnceLock};
use std::time::{Duration, Instant};
use slint::{ use slint::{
ComponentHandle, Image, Model, ModelRc, Rgb8Pixel, SharedPixelBuffer, SharedString, Timer, ComponentHandle, Image, Model, ModelRc, Rgb8Pixel, SharedPixelBuffer, SharedString, Timer,
@ -40,40 +87,48 @@ const GRID_COLS: usize = 3;
const GRID_PADDING: f32 = 12.0; const GRID_PADDING: f32 = 12.0;
const GRID_GAP: f32 = 4.0; const GRID_GAP: f32 = 4.0;
const ROW_SPACING: f32 = 4.0; const ROW_SPACING: f32 = 4.0;
/// Entries to submit whenever the (re)paints the top of the list. /// Rows of margin around the visible range. The union is the "viewport":
const PREFETCH_FIRST: usize = 200; /// the only cells the loader ever submits; painted cells at this distance
/// Rows of margin on each side of the visible range (prefetch + retention). /// or beyond are evicted.
const SCROLL_MARGIN_ROWS: usize = 4; const MARGIN_ROWS: usize = 4;
/// Max finished-but-out-of-range results held for later painting. /// Viewport used until the first scroll event (≈2 screens).
const DEFERRED_CAP: usize = 1024; const INITIAL_VIEWPORT_CELLS: usize = 60;
/// Max ROW replacements per drain tick. Bounding per-tick work is what keeps /// Hard cap on cells holding pixels in the model (memory safety valve for
/// the UI thread at frame rate: without it, the first scroll after a 200-cell /// pathological flicking; steady state is far below this).
/// cold fill mass-updated ~150 cells in one event-loop pass (ANR, 2026-10-27). const MAX_PAINTED: usize = 120;
/// Max ROW replacements per tick (paints + evictions share the budget).
/// Bounding per-tick work is what keeps the UI thread at frame rate:
/// without it, a mass update after a 200-cell cold fill ANR'd (2026-10-27).
const MAX_ROWS_PER_TICK: usize = 16; const MAX_ROWS_PER_TICK: usize = 16;
/// Cells per tick implied by the row cap (paints); clears share the budget. const MAX_CELLS_PER_TICK: usize = MAX_ROWS_PER_TICK * GRID_COLS;
const MAX_PAINTS_PER_TICK: usize = MAX_ROWS_PER_TICK * GRID_COLS;
/// Retention range at list (re)set, before the first scroll event: a couple
/// of screens plus margin, in cells.
const INITIAL_ACTIVE_CELLS: usize = 60;
const DRAIN_INTERVAL_MS: u64 = 100; const DRAIN_INTERVAL_MS: u64 = 100;
/// A viewport cell waiting longer than this for its in-flight job is a
/// stall; log a diagnostic (rate-limited).
const STALL_ALARM_MS: u64 = 500;
static STATE: OnceLock<AppState> = OnceLock::new(); static STATE: OnceLock<AppState> = OnceLock::new();
struct AppState { struct AppState {
engine: Engine, engine: Engine,
/// Current list (newest first); source for prefetch + paint guards. UI thread only. /// Current list (newest first). UI thread only.
entries: Mutex<Vec<Entry>>, entries: Mutex<Vec<Entry>>,
/// Paths already handed to the pool for the current list. UI thread only. /// Cell range the user can see (visible ± MARGIN_ROWS). Set by on_scroll
submitted: Mutex<HashSet<String>>, /// only — everything the loader does derives from this. UI thread only.
viewport: Mutex<Range<usize>>,
/// Entry indexes whose image currently lives in the model. UI thread only. /// Entry indexes whose image currently lives in the model. UI thread only.
painted_ids: Mutex<HashSet<usize>>, painted: Mutex<HashSet<usize>>,
/// Entry range the model is allowed to keep pixels for (visible ± margin). /// path → submit time. Present iff a job's result has not yet been
/// consumed (painted or dropped). Dedupe + "coming" indicator.
/// UI thread only. /// UI thread only.
active: Mutex<Range<usize>>, inflight: Mutex<HashMap<String, Instant>>,
/// Finished results outside the active range, painted when it catches up. /// (path, mtime, size) that failed to decode; not re-submitted until the
deferred: Mutex<Vec<JobResult>>, /// list is reset (a replaced file has a new mtime and is retried).
/// UI thread only.
failed: Mutex<HashSet<(String, i64, i64)>>,
/// Total thumbnails painted; drives the acceptance log cadence. /// Total thumbnails painted; drives the acceptance log cadence.
painted: AtomicU64, painted_total: AtomicU64,
/// Rate-limits the stall alarm. UI thread only.
last_stall_log: Mutex<Instant>,
} }
/// Create the pool + cache dir once per process. /// Create the pool + cache dir once per process.
@ -83,7 +138,10 @@ fn state() -> &'static AppState {
let cache_dir = base.join("thumbs"); let cache_dir = base.join("thumbs");
let workers = let workers =
(std::thread::available_parallelism().map(|n| n.get()).unwrap_or(4) / 2).clamp(2, 8); (std::thread::available_parallelism().map(|n| n.get()).unwrap_or(4) / 2).clamp(2, 8);
let engine = Engine::new(cache_dir, workers, GRID_TARGET).expect("thumb engine"); // 96 = thumb::MAX_WORKER_QUEUE: bound each worker's pending queue so
// stale transit work can never delay the viewport (overflow is
// dropped with a `skipped` result; re-entry is a disk-cache hit).
let engine = Engine::new(cache_dir, workers, GRID_TARGET, 96).expect("thumb engine");
log::info!( log::info!(
"gallery: thumb engine up: {} workers, target {} px, cache {}", "gallery: thumb engine up: {} workers, target {} px, cache {}",
workers, workers,
@ -93,11 +151,12 @@ fn state() -> &'static AppState {
AppState { AppState {
engine, engine,
entries: Mutex::new(Vec::new()), entries: Mutex::new(Vec::new()),
submitted: Mutex::new(HashSet::new()), viewport: Mutex::new(0..0),
painted_ids: Mutex::new(HashSet::new()), painted: Mutex::new(HashSet::new()),
active: Mutex::new(0..0), inflight: Mutex::new(HashMap::new()),
deferred: Mutex::new(Vec::new()), failed: Mutex::new(HashSet::new()),
painted: AtomicU64::new(0), painted_total: AtomicU64::new(0),
last_stall_log: Mutex::new(Instant::now()),
} }
}) })
} }
@ -140,32 +199,27 @@ fn cells_of(item: &Item) -> Option<&VecModel<Cell>> {
item.cells.as_any().downcast_ref::<VecModel<Cell>>() item.cells.as_any().downcast_ref::<VecModel<Cell>>()
} }
/// Replace the list model (UI thread) + set the status line. A model /// Replace the list model (UI thread) + set the status line. A replacement
/// replacement invalidates all in-flight work: bump the generation, clear /// invalidates all loader state: bump the generation (in-flight results are
/// submitted/painted/deferred, and re-anchor the active range at the top. /// dropped), clear inflight/painted/failed, re-anchor the viewport at the
/// `do_prefetch` is false for cold-scan streaming batches (nothing to show /// top. The next tick fills the viewport — there is no separate "prefetch"
/// yet — re-submitting the top rows every batch would waste decode work). /// step.
fn set_list(weak: &slint::Weak<Main>, entries: Vec<Entry>, status: String, do_prefetch: bool) { fn set_list(weak: &slint::Weak<Main>, entries: Vec<Entry>, status: String) {
let weak = weak.clone(); let weak = weak.clone();
let _ = slint::invoke_from_event_loop(move || { let _ = slint::invoke_from_event_loop(move || {
let Some(w) = weak.upgrade() else { return }; let Some(w) = weak.upgrade() else { return };
let Some(app) = STATE.get() else { return }; let Some(app) = STATE.get() else { return };
app.engine.bump_generation(); app.engine.bump_generation();
app.submitted.lock().unwrap().clear(); app.inflight.lock().unwrap().clear();
app.painted_ids.lock().unwrap().clear(); app.painted.lock().unwrap().clear();
app.deferred.lock().unwrap().clear(); app.failed.lock().unwrap().clear();
*app.entries.lock().unwrap() = entries.clone(); *app.entries.lock().unwrap() = entries.clone();
let n = entries.len(); let n = entries.len();
*app.active.lock().unwrap() = 0..INITIAL_ACTIVE_CELLS.min(n); *app.viewport.lock().unwrap() = 0..INITIAL_VIEWPORT_CELLS.min(n);
if let Some(m) = w.get_items().as_any().downcast_ref::<VecModel<Item>>() { if let Some(m) = w.get_items().as_any().downcast_ref::<VecModel<Item>>() {
m.set_vec(chunk(&entries)); m.set_vec(chunk(&entries));
} }
w.set_status(status.into()); w.set_status(status.into());
if do_prefetch {
// Top of the list starts filling now (cold: decodes; warm: cache hits).
let end = PREFETCH_FIRST.min(n);
prefetch(app, 0..end);
}
}); });
} }
@ -179,46 +233,14 @@ fn set_status(weak: &slint::Weak<Main>, status: String) {
}); });
} }
/// Submit not-yet-submitted entries in `range` (entry indexes) to the pool. /// Apply a batch of cell paints + evictions, replacing each affected OUTER
/// UI thread only. /// row exactly once (UI thread). Row replacement — not in-place nested-model
fn prefetch(app: &AppState, range: Range<usize>) { /// updates — is the only update path that reliably reaches ListView rows
if range.is_empty() { /// that were recycled while off-screen (gotcha #12). Returns the number of
return; /// paint cells applied. At most `MAX_ROWS_PER_TICK` rows are applied;
} /// unapplied paints are dropped on purpose — their cells are no longer
let entries = app.entries.lock().unwrap(); /// painted and no longer in flight, so the fill step re-submits them (disk-
let start = range.start.min(entries.len()); /// cache hit) and they land next tick.
let end = range.end.min(entries.len());
if start >= end {
return;
}
let gen = app.engine.generation();
let mut submitted = app.submitted.lock().unwrap();
let jobs: Vec<Job> = entries[start..end]
.iter()
.enumerate()
.filter_map(|(i, e)| {
submitted
.insert(e.path.clone())
.then_some(Job {
id: start + i,
path: e.path.clone(),
mtime: e.mtime,
size: e.size,
gen,
})
})
.collect();
if !jobs.is_empty() {
app.engine.submit(&jobs);
}
}
/// Apply a batch of cell paints + clears, replacing each affected OUTER row
/// exactly once (UI thread). Row replacement — not in-place nested-model
/// updates — is the only update path that reliably reaches ListView rows that
/// were recycled while off-screen: the outer row_changed makes the ListView
/// re-bind `row-data`, and the fresh nested model read picks up the new
/// state. Returns the number of paint cells applied.
fn apply_row_batch( fn apply_row_batch(
app: &AppState, app: &AppState,
m: &VecModel<Item>, m: &VecModel<Item>,
@ -260,7 +282,12 @@ fn apply_row_batch(
)); ));
} }
let mut n = 0usize; let mut n = 0usize;
let mut rows_done = 0usize;
for (row, updates) in rows { for (row, updates) in rows {
if rows_done >= MAX_ROWS_PER_TICK {
break; // ANR guard; the rest is re-covered by the fill step
}
rows_done += 1;
if row >= m.row_count() { if row >= m.row_count() {
continue; continue;
} }
@ -286,18 +313,17 @@ fn apply_row_batch(
if id >= entries.len() || entries[id].path != c.path.as_str() { if id >= entries.len() || entries[id].path != c.path.as_str() {
continue; continue;
} }
app.painted_ids.lock().unwrap().insert(id); app.painted.lock().unwrap().insert(id);
n += 1; n += 1;
} else { } else {
if id >= entries.len() || entries[id].path != c.path.as_str() { if id >= entries.len() || entries[id].path != c.path.as_str() {
continue; continue;
} }
app.painted_ids.lock().unwrap().remove(&id); app.painted.lock().unwrap().remove(&id);
// Release the dedupe entry so the next prefetch re-submits // Release the dedupe entry so the fill step re-submits this
// this cell when it scrolls back into range (its result was // cell when it is in the viewport again (re-submit is a
// consumed when painted; without this it would shimmer // disk-cache hit).
// forever). Re-submit is a disk-cache hit. app.inflight.lock().unwrap().remove(&entries[id].path);
app.submitted.lock().unwrap().remove(&entries[id].path);
} }
new_cells[col] = c; new_cells[col] = c;
touched = true; touched = true;
@ -316,8 +342,9 @@ fn apply_row_batch(
/// Scroll event (main.slint `scrolled` → `scroll-changed(y, height, width)`; /// Scroll event (main.slint `scrolled` → `scroll-changed(y, height, width)`;
/// content-y is negative while scrolled). MUST stay cheap — it fires on every /// content-y is negative while scrolled). MUST stay cheap — it fires on every
/// scroll frame: only re-anchor the active range and submit prefetch jobs. /// scroll frame: it ONLY moves the viewport target. All work (submit, paint,
/// ALL model mutations happen in the 10 Hz drain timer (capped per tick). /// evict) happens in the 10 Hz tick, which reconciles the viewport whether
/// or not scrolling is happening.
fn on_scroll(app: &AppState, y: f32, height: f32, width: f32) { fn on_scroll(app: &AppState, y: f32, height: f32, width: f32) {
if width <= 0.0 { if width <= 0.0 {
return; return;
@ -330,20 +357,19 @@ fn on_scroll(app: &AppState, y: f32, height: f32, width: f32) {
} }
let first = (y.abs() / row_h) as usize; let first = (y.abs() / row_h) as usize;
let visible = (height / row_h) as usize + 1; let visible = (height / row_h) as usize + 1;
let margin = SCROLL_MARGIN_ROWS * GRID_COLS; let margin = MARGIN_ROWS * GRID_COLS;
let new = (first * GRID_COLS) let vp = (first * GRID_COLS)
.saturating_sub(margin) .saturating_sub(margin)
..((first + visible + SCROLL_MARGIN_ROWS) * GRID_COLS).min(n); ..((first + visible + MARGIN_ROWS) * GRID_COLS).min(n);
*app.active.lock().unwrap() = new.clone(); *app.viewport.lock().unwrap() = vp;
prefetch(app, new);
} }
/// 10x/s: drain finished thumbnails; paint what's in the active range; /// The reconcile tick, 10x/s: consume finished results (paint the viewport
/// clear pixels that left it; defer the rest. Row replacements are capped /// ones), evict far pixels, and fill the viewport (submit what's missing,
/// at MAX_ROWS_PER_TICK per tick so no single event-loop pass can mass- /// center-first). This is the ONLY place jobs are submitted. After the fill
/// reflow the grid (ANR guard). Slint 1.18 timers stop when dropped, and /// step, the invariant holds: every viewport cell is painted, in flight, or
/// Timer is !Send (UI thread only): we leak the one handle for the process /// failed. Slint 1.18 timers stop when dropped, and Timer is !Send (UI
/// lifetime (8 bytes). /// thread only): we leak the one handle for the process lifetime (8 bytes).
fn start_drain_timer(weak: &slint::Weak<Main>) { fn start_drain_timer(weak: &slint::Weak<Main>) {
let weak = weak.clone(); let weak = weak.clone();
let timer = Timer::default(); let timer = Timer::default();
@ -352,77 +378,186 @@ fn start_drain_timer(weak: &slint::Weak<Main>) {
std::time::Duration::from_millis(DRAIN_INTERVAL_MS), std::time::Duration::from_millis(DRAIN_INTERVAL_MS),
move || { move || {
let Some(app) = STATE.get() else { return }; let Some(app) = STATE.get() else { return };
let results = app.engine.drain();
let Some(w) = weak.upgrade() else { return }; let Some(w) = weak.upgrade() else { return };
let model = w.get_items(); let model = w.get_items();
let Some(m) = model.as_any().downcast_ref::<VecModel<Item>>() else { let Some(m) = model.as_any().downcast_ref::<VecModel<Item>>() else {
return return
}; };
let results = app.engine.drain();
let entries = app.entries.lock().unwrap(); let entries = app.entries.lock().unwrap();
if entries.is_empty() { if entries.is_empty() {
return; return;
} }
let active = app.active.lock().unwrap().clone(); let viewport = app.viewport.lock().unwrap().clone();
// 1. Paints: fresh in-range results + deferred that re-entered. // 1. CONSUME: every finished result leaves `inflight` (its job is
let mut deferred = app.deferred.lock().unwrap(); // done). Paint viewport results (capped); the disk cache owns
// the rest — a dropped result costs a ~10 ms cache hit on
// re-entry, never a re-decode.
let mut paints: Vec<JobResult> = Vec::new(); let mut paints: Vec<JobResult> = Vec::new();
for r in results { {
if active.contains(&r.id) { let mut inflight = app.inflight.lock().unwrap();
paints.push(r); for r in results {
} else if deferred.len() < DEFERRED_CAP { inflight.remove(&r.path);
deferred.push(r); if r.skipped {
} else { // Dropped stale transit work: not a failure, nothing
// Cap hit (very fast scroll): forget this one; the next // to paint. If the cell re-enters the viewport, FILL
// prefetch re-submits it (clear the dedupe entry). // re-submits it (disk-cache hit if ever decoded).
app.submitted.lock().unwrap().remove(&r.path); continue;
} }
} if !r.ok {
let mut kept = Vec::with_capacity(deferred.len()); // Undecodable (or list moved on): never re-submit
for r in deferred.drain(..) { // until the list is reset.
if active.contains(&r.id) { if let Some(e) = entries.get(r.id).filter(|e| e.path == r.path) {
paints.push(r); app.failed
} else { .lock()
kept.push(r); .unwrap()
} .insert((e.path.clone(), e.mtime, e.size));
} }
deferred.extend(kept); continue;
drop(deferred); }
if viewport.contains(&r.id) && paints.len() < MAX_CELLS_PER_TICK {
// 2. Cap the tick's paint work; the overflow waits in deferred. paints.push(r);
if paints.len() > MAX_PAINTS_PER_TICK {
let overflow = paints.split_off(MAX_PAINTS_PER_TICK);
let mut deferred = app.deferred.lock().unwrap();
for r in overflow {
if deferred.len() < DEFERRED_CAP {
deferred.push(r);
} }
} }
} }
// 3. Clears: painted cells outside the active range, within the // 2. EVICT: release pixels for painted cells outside the viewport
// remaining row budget (memory bound, spread over ticks). // ± MARGIN_ROWS, farthest first. Budget = the rows the paints
let rows_used = (paints.len() + GRID_COLS - 1) / GRID_COLS; // didn't use; if the hard cap is exceeded, evict extra (still
let clear_budget = (MAX_ROWS_PER_TICK - rows_used.min(MAX_ROWS_PER_TICK)) * GRID_COLS; // bounded per tick) so pathological flicking converges.
let to_clear: Vec<usize> = app let to_clear: Vec<usize> = {
.painted_ids let painted = app.painted.lock().unwrap();
.lock() let rows_used = (paints.len() + GRID_COLS - 1) / GRID_COLS;
.unwrap() let budget = (MAX_ROWS_PER_TICK - rows_used.min(MAX_ROWS_PER_TICK)) * GRID_COLS;
.iter() let overflow = painted.len().saturating_sub(MAX_PAINTED);
.copied() let lo = viewport.start.saturating_sub(MARGIN_ROWS * GRID_COLS);
.filter(|id| !active.contains(id)) let hi = (viewport.end + MARGIN_ROWS * GRID_COLS).min(entries.len());
.take(clear_budget) let center = (viewport.start + viewport.end) / 2;
.collect(); let mut far: Vec<usize> = painted
.iter()
.copied()
.filter(|id| *id < lo || *id >= hi)
.collect();
far.sort_by_key(|id| center.abs_diff(*id)); // nearest first
far.reverse(); // farthest first
let n = far.len().min(budget.max(overflow.min(2 * MAX_CELLS_PER_TICK)));
far.into_iter().take(n).collect()
};
// 4. Apply: each affected outer row replaced exactly once. // 3. APPLY: each affected outer row replaced exactly once
let painted = apply_row_batch(app, m, &entries, &paints, &to_clear); // (≤ MAX_ROWS_PER_TICK rows).
let painted_now = apply_row_batch(app, m, &entries, &paints, &to_clear);
if painted > 0 { // 4. FILL: every viewport cell without pixels and without an
// in-flight job gets submitted now, center-first so the eye is
// filled first. Restores the invariant: after this step every
// viewport cell is painted, in flight, or failed.
{
let painted = app.painted.lock().unwrap();
let failed = app.failed.lock().unwrap();
let mut inflight = app.inflight.lock().unwrap();
let center = (viewport.start + viewport.end) / 2;
let mut missing: Vec<usize> = viewport
.clone()
.into_iter()
.filter(|id| {
let e = &entries[*id];
!painted.contains(id)
&& !inflight.contains_key(&e.path)
&& !failed.contains(&(e.path.clone(), e.mtime, e.size))
})
.collect();
missing.sort_by_key(|id| center.abs_diff(*id));
if !missing.is_empty() {
let gen = app.engine.generation();
let now = Instant::now();
let jobs: Vec<Job> = missing
.iter()
.map(|id| {
let e = &entries[*id];
inflight.insert(e.path.clone(), now);
Job {
id: *id,
path: e.path.clone(),
mtime: e.mtime,
size: e.size,
gen,
}
})
.collect();
drop(inflight);
app.engine.submit(&jobs);
}
}
// 4b. PRUNE: drop queued (not-yet-started) jobs outside the retention band.
// A fast scroll leaves queued work for cells the user has passed;
// letting it linger would make the next viewport's work wait behind
// it (the reported "stop → long wait → pop" residual). If the user
// comes back, the disk cache serves those cells in ~10 ms, never a
// re-decode. In-flight jobs are untouched; their out-of-viewport
// results are dropped at consumption.
{
let lo = viewport.start.saturating_sub(MARGIN_ROWS * GRID_COLS);
let hi = (viewport.end + MARGIN_ROWS * GRID_COLS).min(entries.len());
let band = lo..hi;
let dropped = app.engine.prune_outside(&band);
if !dropped.is_empty() {
// The pruned cells are no longer "in flight" — clear the
// tracker so a later re-entry re-submits them (disk-cache
// hit, not a re-decode). A path with a NEWER job still
// queued may be briefly re-submitted (one wasted decode,
// self-cleans; the invariant is unaffected).
let mut inflight = app.inflight.lock().unwrap();
for j in &dropped {
inflight.remove(&j.path);
}
log::debug!("gallery: pruned {} stale queued jobs", dropped.len());
}
}
// 5. Invariant (debug builds): the viewport is fully covered.
// A violation means a visible cell is forgotten — the exact
// bug class this design exists to make impossible.
#[cfg(debug_assertions)]
{
let painted = app.painted.lock().unwrap();
let inflight = app.inflight.lock().unwrap();
let failed = app.failed.lock().unwrap();
for id in viewport.clone() {
let e = &entries[id];
let covered = painted.contains(&id)
|| inflight.contains_key(&e.path)
|| failed.contains(&(e.path.clone(), e.mtime, e.size));
debug_assert!(covered, "viewport cell {id} is uncovered");
}
}
// 6. Diagnostics: stall alarm (rate-limited) + acceptance log.
{
let now = Instant::now();
let inflight = app.inflight.lock().unwrap();
if let Some(oldest) = inflight.values().map(|t| now.duration_since(*t)).max() {
if oldest > Duration::from_millis(STALL_ALARM_MS)
&& *app.last_stall_log.lock().unwrap() + Duration::from_secs(2) < now
{
*app.last_stall_log.lock().unwrap() = now;
log::info!(
"gallery: stall: {} in flight, oldest {} ms, {}",
inflight.len(),
oldest.as_millis(),
app.engine.log_summary()
);
}
}
}
if painted_now > 0 {
let total = let total =
app.painted.fetch_add(painted as u64, Ordering::Relaxed) + painted as u64; app.painted_total.fetch_add(painted_now as u64, Ordering::Relaxed)
// Acceptance log: first screen, then a steady heartbeat. Watch + painted_now as u64;
// the queue not grow while scrolling (HANDOFF §8). // Acceptance log: first screen, then a steady heartbeat.
if total <= 48 || total % 512 < painted as u64 { if total <= 48 || total % 512 < painted_now as u64 {
log::info!( log::info!(
"gallery: {} thumbs painted, {}", "gallery: {} thumbs painted, {}",
total, total,
@ -436,8 +571,8 @@ fn start_drain_timer(weak: &slint::Weak<Main>) {
} }
/// M1: cached DB paints instantly on relaunch; a background walk refreshes it /// M1: cached DB paints instantly on relaunch; a background walk refreshes it
/// (streaming into the list on cold start). M3: the pool fills grid cells as /// (streaming into the list on cold start). M3: the tick fills the viewport
/// they appear — top cells on paint, the visible range on scroll. /// as it moves — top cells right after (re)paint, visible cells on scroll.
pub fn start_scan(window: &Main) { pub fn start_scan(window: &Main) {
let model = ModelRc::new(VecModel::<Item>::default()); let model = ModelRc::new(VecModel::<Item>::default());
window.set_items(model); window.set_items(model);
@ -457,7 +592,7 @@ fn scan_thread(weak: &slint::Weak<Main>) {
Ok(db) => db, Ok(db) => db,
Err(e) => { Err(e) => {
log::error!("gallery: db error: {e}"); log::error!("gallery: db error: {e}");
set_list(weak, Vec::new(), format!("db error: {e}"), false); set_list(weak, Vec::new(), format!("db error: {e}"));
return; return;
} }
}; };
@ -467,7 +602,7 @@ fn scan_thread(weak: &slint::Weak<Main>) {
let db_warm = !cached.is_empty(); let db_warm = !cached.is_empty();
if db_warm { if db_warm {
let n0 = cached.len(); let n0 = cached.len();
set_list(weak, cached, format!("{n0} files (cached)"), true); set_list(weak, cached, format!("{n0} files (cached)"));
} }
// 2. Background rescan: detect new/changed/removed files. // 2. Background rescan: detect new/changed/removed files.
@ -483,7 +618,7 @@ fn scan_thread(weak: &slint::Weak<Main>) {
} else { } else {
let mut sorted = batch.to_vec(); let mut sorted = batch.to_vec();
sorted.sort_by(|a, b| b.mtime.cmp(&a.mtime).then_with(|| a.path.cmp(&b.path))); sorted.sort_by(|a, b| b.mtime.cmp(&a.mtime).then_with(|| a.path.cmp(&b.path)));
set_list(weak, sorted, format!("Scanning… {n}"), false); set_list(weak, sorted, format!("Scanning… {n}"));
} }
}); });
@ -496,9 +631,9 @@ fn scan_thread(weak: &slint::Weak<Main>) {
let ms = t0.elapsed().as_millis(); let ms = t0.elapsed().as_millis();
log::info!("gallery: scan {n} files ({ins}+ {upd}~ {del}-) in {ms} ms"); log::info!("gallery: scan {n} files ({ins}+ {upd}~ {del}-) in {ms} ms");
// Final list: set_list bumps the generation + re-prefetches the top rows. // Final list: set_list bumps the generation + re-anchors the viewport;
// (Warm: those are disk-cache hits, so the repaint is nearly free.) // the next tick fills it (warm start: disk-cache hits, nearly free).
set_list(weak, entries, format!("{n} files ({ins}+ {upd}~ {del}–) in {ms} ms"), true); set_list(weak, entries, format!("{n} files ({ins}+ {upd}~ {del}–) in {ms} ms"));
} }
/// A clone of `AndroidApp` kept for the app's whole lifetime so the Java side /// A clone of `AndroidApp` kept for the app's whole lifetime so the Java side

View File

@ -24,6 +24,7 @@
use std::collections::VecDeque; use std::collections::VecDeque;
use std::fs; use std::fs;
use std::ops::Range;
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::sync::mpsc; use std::sync::mpsc;
@ -41,6 +42,12 @@ use crate::images;
pub const GRID_TARGET: u32 = 500; pub const GRID_TARGET: u32 = 500;
const JPEG_QUALITY: u8 = 85; const JPEG_QUALITY: u8 = 85;
/// Per-worker queue cap. A burst (one tick's submissions) is never bigger
/// than the viewport (~45 cells), so fresh viewport work ALWAYS survives the
/// cap; overflow is dropped from the TAIL (oldest transit work). 96 ≈ two
/// screens per worker of headroom.
const MAX_WORKER_QUEUE: usize = 96;
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct Job { pub struct Job {
/// Row index in the UI model at submit time. /// Row index in the UI model at submit time.
@ -52,12 +59,32 @@ pub struct Job {
pub gen: u64, pub gen: u64,
} }
/// One worker's job queue: Mutex<VecDeque> + Condvar, NOT mpsc — mpsc is
/// FIFO-only, but the point of this design is that a new batch (the user's
/// current intent) must be placed AHEAD of older queued work. The
/// dispatcher prepends each burst in order; the worker pops the front.
struct WorkerQueue {
q: Mutex<VecDeque<Job>>,
cv: Condvar,
}
impl WorkerQueue {
fn new() -> Self {
Self { q: Mutex::new(VecDeque::new()), cv: Condvar::new() }
}
}
#[derive(Debug)] #[derive(Debug)]
pub struct JobResult { pub struct JobResult {
pub id: usize, pub id: usize,
pub gen: u64, pub gen: u64,
pub path: String, pub path: String,
pub ok: bool, pub ok: bool,
/// The job was DROPPED before being decoded (worker-queue overflow:
/// it was stale transit work behind newer viewport work). The UI clears
/// its in-flight marker; the cell is NOT failed — a re-entry re-submits
/// it (disk-cache hit if it was ever decoded, otherwise a fresh decode).
pub skipped: bool,
/// RGB8, orientation applied, fitted to the engine's target size. /// RGB8, orientation applied, fitted to the engine's target size.
pub rgb: Vec<u8>, pub rgb: Vec<u8>,
pub width: u32, pub width: u32,
@ -113,8 +140,10 @@ pub fn cache_path(cache_dir: &Path, path: &str, mtime: i64, size: i64, target: u
impl Engine { impl Engine {
/// Spawn the dispatcher + `workers` worker threads. All jobs render at /// Spawn the dispatcher + `workers` worker threads. All jobs render at
/// `target` px (longest side). /// `target` px (longest side). `max_worker_queue` bounds each worker's
pub fn new(cache_dir: PathBuf, workers: usize, target: u32) -> Result<Self, String> { /// pending queue (stale transit work beyond the cap is dropped with a
/// `skipped` result); pass `usize::MAX` for uncapped (tests).
pub fn new(cache_dir: PathBuf, workers: usize, target: u32, max_worker_queue: usize) -> Result<Self, String> {
fs::create_dir_all(&cache_dir).map_err(|e| format!("mkdir {}: {e}", cache_dir.display()))?; fs::create_dir_all(&cache_dir).map_err(|e| format!("mkdir {}: {e}", cache_dir.display()))?;
let jobs = Arc::new(Mutex::new(VecDeque::<Job>::new())); let jobs = Arc::new(Mutex::new(VecDeque::<Job>::new()));
let job_cv = Arc::new(Condvar::new()); let job_cv = Arc::new(Condvar::new());
@ -128,51 +157,37 @@ impl Engine {
errs: Arc::new(AtomicU64::new(0)), errs: Arc::new(AtomicU64::new(0)),
}; };
let mut workers_tx = Vec::with_capacity(workers); let mut wqs: Vec<Arc<WorkerQueue>> = Vec::with_capacity(workers);
for i in 0..workers { for i in 0..workers {
let (wtx, wrx) = mpsc::channel::<Job>(); let wq = Arc::new(WorkerQueue::new());
let (res_tx, gen, in_flight) = (res_tx.clone(), gen.clone(), in_flight.clone()); let (wq_t, res_tx, gen, in_flight, closed_w) =
(wq.clone(), res_tx.clone(), gen.clone(), in_flight.clone(), closed.clone());
let (done, hits, errs) = (stats.done.clone(), stats.hits.clone(), stats.errs.clone()); let (done, hits, errs) = (stats.done.clone(), stats.hits.clone(), stats.errs.clone());
let cache_dir = cache_dir.clone(); let cache_dir = cache_dir.clone();
std::thread::Builder::new() std::thread::Builder::new()
.name(format!("thumb-{i}")) .name(format!("thumb-{i}"))
.spawn(move || { .spawn(move || {
worker_loop(wrx, res_tx, gen, in_flight, done, hits, errs, cache_dir, target) worker_loop(
&wq_t, res_tx, gen, in_flight, done, hits, errs, cache_dir, target, closed_w,
)
}) })
.map_err(|e| e.to_string())?; .map_err(|e| e.to_string())?;
workers_tx.push(wtx); wqs.push(wq);
} }
drop(res_tx);
// Dispatcher: pops the job queue (front = most recent intent) and // Dispatcher: drains the central queue as priority-ordered bursts
// round-robins to workers. Exits when the engine is dropped and the // and PREPENDS each burst to the worker queues (the cap drops the
// queue has drained. // oldest tail work with `skipped` results). Exits when the engine is
// dropped and the queue has drained.
let (d_jobs, d_cv, d_closed) = (jobs.clone(), job_cv.clone(), closed.clone()); let (d_jobs, d_cv, d_closed) = (jobs.clone(), job_cv.clone(), closed.clone());
let (d_res, d_inflight) = (res_tx.clone(), in_flight.clone());
std::thread::Builder::new() std::thread::Builder::new()
.name("thumb-dispatch".into()) .name("thumb-dispatch".into())
.spawn(move || { .spawn(move || {
let mut next = 0usize; dispatch_loop(d_jobs, d_cv, d_closed, wqs, d_res, d_inflight, max_worker_queue)
loop {
let job = {
let mut dq = d_jobs.lock().unwrap();
while dq.is_empty() {
if d_closed.load(Ordering::SeqCst) {
return; // engine dropped, queue drained
}
dq = d_cv.wait(dq).unwrap();
}
dq.pop_front().unwrap()
};
for _ in 0..workers_tx.len() {
if workers_tx[next % workers_tx.len()].send(job.clone()).is_ok() {
next += 1;
break;
}
next += 1;
}
}
}) })
.map_err(|e| e.to_string())?; .map_err(|e| e.to_string())?;
drop(res_tx);
Ok(Self { Ok(Self {
jobs, jobs,
@ -223,6 +238,29 @@ impl Engine {
self.job_cv.notify_one(); self.job_cv.notify_one();
} }
/// Drop QUEUED (not yet started) jobs outside `range`, returning the
/// number dropped. Queued work for cells the user has scrolled away from
/// is low value: if they come back, the disk cache serves it in ~10 ms.
/// This bounds the queue to the retention band, so viewport work never
/// waits behind stale transit work (in-flight jobs finish; their
/// out-of-viewport results are dropped at consumption).
pub fn prune_outside(&self, range: &Range<usize>) -> Vec<Job> {
let mut dq = self.jobs.lock().unwrap();
let mut dropped: Vec<Job> = Vec::new();
dq.retain(|j| {
if range.contains(&j.id) {
true
} else {
dropped.push(j.clone());
false
}
});
if !dropped.is_empty() {
self.in_flight.fetch_sub(dropped.len(), Ordering::Relaxed);
}
dropped
}
/// Non-blocking drain (UI thread). Stale-generation results are dropped. /// Non-blocking drain (UI thread). Stale-generation results are dropped.
pub fn drain(&self) -> Vec<JobResult> { pub fn drain(&self) -> Vec<JobResult> {
let mut out = Vec::new(); let mut out = Vec::new();
@ -259,8 +297,97 @@ struct Stats {
errs: Arc<AtomicU64>, errs: Arc<AtomicU64>,
} }
/// The dispatcher (one thread): drains the central queue as a single
/// burst — the deque is already in priority order (newest batch first, and
/// center-first within a batch) — splits it round-robin over the workers,
/// and PREPENDS each worker's share. Overflow beyond `max_worker_queue`
/// is dropped from the TAIL (the oldest transit work) with a `skipped`
/// result so the UI can clear its in-flight marker.
///
/// This is what makes "the viewport is first" true at EVERY level of the
/// pipeline: no matter how much stale transit work is queued — or in
/// flight — a fresh submission (the current viewport) is placed ahead of
/// it. A burst never exceeds the viewport (~45) < the cap (96), so fresh
/// viewport work is never itself dropped.
fn dispatch_loop(
jobs: Arc<Mutex<VecDeque<Job>>>,
cv: Arc<Condvar>,
closed: Arc<AtomicBool>,
wqs: Vec<Arc<WorkerQueue>>,
res_tx: mpsc::Sender<JobResult>,
in_flight: Arc<AtomicUsize>,
max_worker_queue: usize,
) {
loop {
// Drain the whole central deque as one burst (priority order).
let burst: Vec<Job> = {
let mut dq = jobs.lock().unwrap();
while dq.is_empty() {
if closed.load(Ordering::SeqCst) {
return; // engine dropped, queue drained
}
dq = cv.wait(dq).unwrap();
}
dq.drain(..).collect()
};
// Round-robin split, preserving per-worker order.
let n = wqs.len();
let mut per: Vec<Vec<Job>> = vec![Vec::new(); n];
for (i, j) in burst.into_iter().enumerate() {
per[i % n].push(j);
}
for w in 0..n {
if per[w].is_empty() {
continue;
}
let dropped: Vec<Job> = {
let mut g = wqs[w].q.lock().unwrap();
let mut dropped = Vec::new();
// Cap: prepend the burst; drop overflow from the tail
// (oldest first: existing tail, then the burst's own tail).
let overflow = (per[w].len() + g.len()).saturating_sub(max_worker_queue);
for _ in 0..overflow {
match g.pop_back() {
Some(j) => dropped.push(j),
None => break,
}
}
while per[w].len().saturating_sub(g.len()) > max_worker_queue {
match per[w].pop() {
Some(j) => dropped.push(j),
None => break,
}
}
// Prepend in order (last-first): the front stays the newest.
for j in per[w].drain(..).rev() {
g.push_front(j);
}
dropped
};
wqs[w].cv.notify_one();
for j in dropped {
in_flight.fetch_sub(1, Ordering::Relaxed);
let _ = res_tx.send(JobResult {
id: j.id,
gen: j.gen,
path: j.path,
ok: false,
skipped: true,
rgb: Vec::new(),
width: 0,
height: 0,
from_cache: false,
elapsed_ms: 0,
});
}
}
}
}
fn worker_loop( fn worker_loop(
rx: mpsc::Receiver<Job>, wq: &Arc<WorkerQueue>,
res_tx: mpsc::Sender<JobResult>, res_tx: mpsc::Sender<JobResult>,
gen: Arc<AtomicU64>, gen: Arc<AtomicU64>,
in_flight: Arc<AtomicUsize>, in_flight: Arc<AtomicUsize>,
@ -269,6 +396,7 @@ fn worker_loop(
errs: Arc<AtomicU64>, errs: Arc<AtomicU64>,
cache_dir: PathBuf, cache_dir: PathBuf,
target: u32, target: u32,
closed: Arc<AtomicBool>,
) { ) {
// One WAL connection per worker; the scan thread keeps its own. // One WAL connection per worker; the scan thread keeps its own.
let mut conn = match db::open(&db::db_path()) { let mut conn = match db::open(&db::db_path()) {
@ -280,9 +408,15 @@ fn worker_loop(
}; };
loop { loop {
let job = match rx.recv() { let job = {
Ok(j) => j, let mut g = wq.q.lock().unwrap();
Err(_) => return, // channel closed while g.is_empty() {
if closed.load(Ordering::SeqCst) {
return; // engine dropped, queue drained
}
g = wq.cv.wait(g).unwrap();
}
g.pop_front().unwrap()
}; };
let t0 = Instant::now(); let t0 = Instant::now();
let (ok, thumb, from_cache) = process(&job, &cache_dir, target, &hits, &errs); let (ok, thumb, from_cache) = process(&job, &cache_dir, target, &hits, &errs);
@ -331,6 +465,7 @@ fn worker_loop(
gen: job.gen, gen: job.gen,
path: job.path, path: job.path,
ok, ok,
skipped: false,
rgb, rgb,
width, width,
height, height,
@ -420,7 +555,7 @@ mod tests {
}).collect::<Vec<_>>(); }).collect::<Vec<_>>();
// 2. Cold run. // 2. Cold run.
let eng = Engine::new(cache.clone(), 4, GRID_TARGET).unwrap(); let eng = Engine::new(cache.clone(), 4, GRID_TARGET, usize::MAX).unwrap();
let gen = eng.generation(); let gen = eng.generation();
let t0 = Instant::now(); let t0 = Instant::now();
eng.submit(&entries.iter().enumerate().map(|(i, e)| Job { eng.submit(&entries.iter().enumerate().map(|(i, e)| Job {
@ -449,7 +584,7 @@ mod tests {
assert_eq!(std::fs::read_dir(&cache).unwrap().count(), N, "cache holds all entries"); assert_eq!(std::fs::read_dir(&cache).unwrap().count(), N, "cache holds all entries");
// 3. Warm run: everything must come from the disk cache. // 3. Warm run: everything must come from the disk cache.
let eng2 = Engine::new(cache.clone(), 4, GRID_TARGET).unwrap(); let eng2 = Engine::new(cache.clone(), 4, GRID_TARGET, usize::MAX).unwrap();
let gen2 = eng2.generation(); let gen2 = eng2.generation();
let t0 = Instant::now(); let t0 = Instant::now();
eng2.submit(&entries.iter().enumerate().map(|(i, e)| Job { eng2.submit(&entries.iter().enumerate().map(|(i, e)| Job {
@ -471,7 +606,7 @@ mod tests {
assert!(warm_total.as_secs_f64() < cold_total.as_secs_f64()); assert!(warm_total.as_secs_f64() < cold_total.as_secs_f64());
// 4. Cancellation: submit, bump generation before draining -> dropped. // 4. Cancellation: submit, bump generation before draining -> dropped.
let eng3 = Engine::new(cache.clone(), 4, GRID_TARGET).unwrap(); let eng3 = Engine::new(cache.clone(), 4, GRID_TARGET, usize::MAX).unwrap();
let stale = eng3.generation(); let stale = eng3.generation();
let p = &entries[0]; let p = &entries[0];
// Use a fresh path (mtime bump) so it is a cache miss and takes time. // Use a fresh path (mtime bump) so it is a cache miss and takes time.
@ -522,7 +657,7 @@ mod tests {
let top: Vec<Entry> = (0..60).map(|i| mk(&format!("top{i:03}.jpg"))).collect(); let top: Vec<Entry> = (0..60).map(|i| mk(&format!("top{i:03}.jpg"))).collect();
let far: Vec<Entry> = (0..180).map(|i| mk(&format!("far{i:03}.jpg"))).collect(); let far: Vec<Entry> = (0..180).map(|i| mk(&format!("far{i:03}.jpg"))).collect();
let eng = Engine::new(cache.clone(), 4, GRID_TARGET).unwrap(); let eng = Engine::new(cache.clone(), 4, GRID_TARGET, 96).unwrap();
let gen = eng.generation(); let gen = eng.generation();
let jobs = |es: &[Entry], base: usize| { let jobs = |es: &[Entry], base: usize| {
es.iter() es.iter()
@ -575,6 +710,83 @@ mod tests {
); );
} }
/// Worker-level priority (regression): the old design put every batch
/// into the workers' FIFO mpsc, so a fresh viewport batch had to wait
/// behind ALL stale transit work already queued in a worker. A new
/// batch must now be placed AHEAD of it: with one worker, a fresh job
/// submitted while a slow batch is queued must finish within the first
/// two results (never after the whole old queue).
#[test]
fn fresh_batch_preempts_stale_worker_queue() {
let dir = tempfile::tempdir().unwrap();
let cache = dir.path().join("thumbs");
std::env::set_var("GALLERY_DB_PATH", dir.path().join("gallery.db"));
// Detailed noise JPEGs with real decode cost.
let mk = |name: &str| -> Entry {
let (w, h) = (1600u32, 1200u32);
let mut rgb = Vec::with_capacity(3 * (w * h) as usize);
let mut x: u64 = 0x243f6a8885a308d3;
for _ in 0..(w * h) {
x = x
.wrapping_mul(6364136223846793005)
.wrapping_add(1442695040888963407);
rgb.extend_from_slice(&[(x >> 33) as u8, (x >> 17) as u8, (x >> 1) as u8]);
}
let p = dir.path().join(name);
std::fs::write(&p, images::encode_jpeg(&rgb, w, h, 80).unwrap()).unwrap();
Entry {
path: p.to_str().unwrap().into(),
mtime: 1,
size: std::fs::metadata(&p).unwrap().len() as i64,
}
};
let stale: Vec<Entry> = (0..10).map(|i| mk(&format!("stale{i}.jpg"))).collect();
let fresh = mk("fresh_viewport.jpg");
// ONE worker: no round-robin rescue — the ordering is purely the
// queue's.
let eng = Engine::new(cache, 1, GRID_TARGET, 96).unwrap();
let gen = eng.generation();
let stale_jobs: Vec<Job> = stale
.iter()
.enumerate()
.map(|(i, e)| Job { id: i, path: e.path.clone(), mtime: e.mtime, size: e.size, gen })
.collect();
eng.submit(&stale_jobs);
// Wait until the worker has started (first result = stale job 0
// finished; the rest sit in the worker queue).
let mut got = 0;
while got < 1 {
std::thread::sleep(std::time::Duration::from_millis(5));
got += eng.drain().len();
}
// Fresh viewport submission: must jump AHEAD of the 9 queued stale
// jobs. It may finish just before/after the one stale job already
// in flight (job 1) — but it can NEVER be behind jobs 2..9.
let fresh_job = Job {
id: 100,
path: fresh.path.clone(),
mtime: fresh.mtime,
size: fresh.size,
gen,
};
eng.submit(&[fresh_job.clone()]);
let mut seen: Vec<usize> = Vec::new();
while seen.iter().position(|id| *id == 100).is_none() {
std::thread::sleep(std::time::Duration::from_millis(5));
for r in eng.drain() {
seen.push(r.id);
}
assert!(seen.len() < 8, "fresh job arrived behind the stale queue: {seen:?}");
}
let pos = seen.iter().position(|id| *id == 100).unwrap();
// At most the in-flight stale job (id 1) may precede it.
assert!(pos <= 1, "fresh job arrived in position {pos}: {seen:?}");
}
#[test] #[test]
fn cache_key_stable_and_distinct() { fn cache_key_stable_and_distinct() {
let d = Path::new("/tmp/x"); let d = Path::new("/tmp/x");