From 2a3b69df3bada957804bd0d3039bf0126dc71fb5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= <55464917+jlucaso1@users.noreply.github.com> Date: Thu, 29 Oct 2026 20:20:32 -0300 Subject: [PATCH 1/6] perf(gui): recover cached avatars on bounded priority workers --- .github/workflows/avatar-restore.yml | 79 ++++++++++ crates/gui/src/session/avatar/mod.rs | 123 +++++++++++++++- crates/gui/src/session/avatar/native.rs | 182 +++++++++++++++++++++++ crates/gui/src/session/avatar/probe.rs | 185 ++++++++++++++++++++++++ crates/gui/src/session/avatar/web.rs | 7 + docs/gotchas.md | 13 ++ 6 files changed, 581 insertions(+), 8 deletions(-) create mode 100644 .github/workflows/avatar-restore.yml create mode 100644 crates/gui/src/session/avatar/probe.rs diff --git a/.github/workflows/avatar-restore.yml b/.github/workflows/avatar-restore.yml new file mode 100644 index 00000000..ed18481f --- /dev/null +++ b/.github/workflows/avatar-restore.yml @@ -0,0 +1,79 @@ +name: Avatar restore + +on: + pull_request: + paths: + - 'crates/gui/src/session/avatar/**' + - '.github/workflows/avatar-restore.yml' + +permissions: + contents: read + +concurrency: + group: avatar-restore-${{ github.ref }} + cancel-in-progress: true + +env: + CARGO_INCREMENTAL: '0' + CARGO_PROFILE_DEV_DEBUG: line-tables-only + +jobs: + compare: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 + with: + ref: ${{ github.event.pull_request.head.sha }} + fetch-depth: 0 + persist-credentials: false + - uses: dtolnay/rust-toolchain@4360b52568e2003a75bf9bc1d59f33a8e3fc893c + - name: Install system dependencies + run: | + sudo apt-get update + sudo apt-get install -y --no-install-recommends \ + libasound2-dev libxkbcommon-dev libxkbcommon-x11-dev \ + libwayland-dev libxcb1-dev libxcb-render0-dev libxcb-shape0-dev \ + libxcb-xfixes0-dev libfontconfig1-dev libfreetype6-dev protobuf-compiler + - uses: Swatinem/rust-cache@d1527c8f67ef9af357290475d4af4b0fab8f12e0 + with: + key: ${{ hashFiles('Cargo.toml') }} + save-if: 'false' + # Only avatar implementation files change between measurements. The + # decoder, dependencies, profile, runner and synthetic probe stay fixed. + - name: Measure base implementation + env: + BASE_SHA: ${{ github.event.pull_request.base.sha }} + run: | + base=$(git merge-base HEAD "$BASE_SHA") + echo "Comparing avatar implementation at $base with $(git rev-parse HEAD)" + for file in mod.rs native.rs web.rs; do + git show "$base:crates/gui/src/session/avatar/$file" > "crates/gui/src/session/avatar/$file" + done + python3 - <<'PY' + from pathlib import Path + root = Path('crates/gui/src/session/avatar') + mod = root / 'mod.rs' + text = mod.read_text() + marker = 'mod native;' + assert marker in text + mod.write_text(text.replace(marker, marker + '\n#[cfg(all(test, not(target_family = "wasm")))]\nmod probe;', 1)) + native = root / 'native.rs' + text = native.read_text() + marker = 'pub async fn read_persistent(key: &str) -> Option> {' + assert marker in text + text = text.replace(marker, marker + '\n #[cfg(test)]\n PERSISTENT_READS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);', 1) + text += '\n#[cfg(test)]\npub(super) static PERSISTENT_READS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);\n' + native.write_text(text) + PY + cargo test -p oxidezap-gui --bin oxidezap --locked native_restore_probe -- --ignored --nocapture + OXIDEZAP_AVATAR_PROBE_FORMAT=jpeg cargo test -p oxidezap-gui --bin oxidezap --locked native_restore_probe -- --ignored --nocapture + - name: Restore candidate implementation + if: always() + run: git restore --source=HEAD -- crates/gui/src/session/avatar/mod.rs crates/gui/src/session/avatar/native.rs crates/gui/src/session/avatar/web.rs + - name: Test candidate lifecycle and responsiveness + run: | + cargo test -p oxidezap-gui --bin oxidezap --locked session::avatar:: -- --test-threads=1 + - name: Measure candidate implementation + run: | + cargo test -p oxidezap-gui --bin oxidezap --locked native_restore_probe -- --ignored --nocapture + OXIDEZAP_AVATAR_PROBE_FORMAT=jpeg cargo test -p oxidezap-gui --bin oxidezap --locked native_restore_probe -- --ignored --nocapture diff --git a/crates/gui/src/session/avatar/mod.rs b/crates/gui/src/session/avatar/mod.rs index 854efcde..10729fbc 100644 --- a/crates/gui/src/session/avatar/mod.rs +++ b/crates/gui/src/session/avatar/mod.rs @@ -15,6 +15,9 @@ mod web; #[cfg(target_family = "wasm")] mod web_lock; +#[cfg(all(test, not(target_family = "wasm")))] +mod probe; + #[cfg(not(target_family = "wasm"))] use native as imp; #[cfg(target_family = "wasm")] @@ -27,9 +30,7 @@ use std::sync::{Arc, Mutex}; use oxidezap_core::{AccountId, AvatarDemand}; use wacore::time::Instant; -use crate::session::media::{ - clear_image_sources, decode_avatar, get_avatar_image, put_avatar_image, -}; +use crate::session::media::{clear_image_sources, get_avatar_image, put_avatar_image}; #[allow(unused_imports)] pub use imp::{ @@ -130,10 +131,12 @@ impl AvatarManager { /// Reset in-memory cache and tracking (e.g. on cache clear or account change). pub fn clear(&self) { - clear_image_sources(); self.generation.fetch_add(1, Ordering::Relaxed); self.bump_demand_epoch(); self.reset_connection(); + // Retire reads before clearing images. A worker publishing while + // reset_connection waits for its read lock is then cleared as well. + clear_image_sources(); } /// Clear in-flight, revalidated, and cooldown tracking on connection restart while @@ -181,8 +184,23 @@ impl AvatarManager { for chat in visible { self.demand_one(chat, DemandPriority::High); } - for chat in overscan { - self.demand_one(chat, DemandPriority::Low); + // A batch of local reads is scheduled asynchronously, so spawning + // overscan last does not ensure it runs last. Let visible disk hits + // reach RAM before admitting speculative local work. Completion + // advances demand_epoch and wakes the viewport to admit overscan. + let visible_loading = { + let reads = self + .persistent_in_flight + .lock() + .unwrap_or_else(|e| e.into_inner()); + visible + .iter() + .any(|chat| reads.keys().any(|(jid, _)| jid == &chat.jid)) + }; + if !visible_loading { + for chat in overscan { + self.demand_one(chat, DemandPriority::Low); + } } // Prune low-priority pending demands that are no longer in the active window @@ -320,8 +338,30 @@ impl AvatarManager { } if let Some(bytes) = persistent_bytes { - if let Some((source, decoded_size)) = decode_avatar(&bytes) { - put_avatar_image(key_clone.clone(), source, decoded_size); + let decoded = imp::decode_avatar(bytes, priority).await; + { + let mut p = persistent_in_flight + .lock() + .unwrap_or_else(|e| e.into_inner()); + // A decode can outlive replacement, cache clear or + // account change. Check again after the worker's + // await, holding the read reservation through RAM + // publication so retirement cannot overtake it. + if p.get(&read_key) != Some(&op_token) + || generation.load(Ordering::Relaxed) != born_gen + || *account_id.lock().unwrap_or_else(|e| e.into_inner()) + != born_account + { + if p.get(&read_key) == Some(&op_token) { + p.remove(&read_key); + } + return; + } + if let Some((source, decoded_size)) = &decoded { + put_avatar_image(key_clone.clone(), source.clone(), *decoded_size); + } + } + if decoded.is_some() { let mut p = persistent_in_flight .lock() .unwrap_or_else(|e| e.into_inner()); @@ -330,6 +370,11 @@ impl AvatarManager { } p.remove(&read_key); drop(p); + // Disk reads can all finish while decodes are + // queued. Releasing a decode also releases local + // admission, so a viewport refused at the read + // limit must get another fingerprint and retry. + demand_epoch.fetch_add(1, Ordering::Relaxed); // Notify UI that the avatar image is ready in RAM if let Some(session) = session_handle @@ -609,6 +654,7 @@ fn note_revalidation(revalidated: &mut indexmap::IndexSet, jid: &str) -> #[cfg(test)] mod tests { use super::*; + use crate::session::media::decode_avatar; // Run the real native storage and manager together in a separate process: // changing a process-wide runtime path in a parallel test would risk the @@ -755,6 +801,67 @@ mod tests { assert!(window.take_demands().is_empty()); assert_eq!(window.demand_epoch(), epoch, "recovery must settle"); } + + // Force a local decode to wait behind occupied workers, then + // retire it. Its late result must not resurrect cleared images, + // select an old picture, or send freshness for a departed account. + for retirement in ["clear", "replacement", "account"] { + clear_image_sources(); + let release = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let (started, receiving) = std::sync::mpsc::channel(); + let mut blockers = Vec::new(); + let completed = native::DECODE_COMPLETED.load(std::sync::atomic::Ordering::Relaxed); + for _ in 0..2 { + let release = release.clone(); + let started = started.clone(); + blockers.push(smol::spawn(native::decode_work( + DemandPriority::High, + move || { + started.send(()).unwrap(); + while !release.load(std::sync::atomic::Ordering::Relaxed) { + std::thread::sleep(std::time::Duration::from_millis(1)); + } + }, + ))); + } + for _ in 0..2 { + receiving + .recv_timeout(std::time::Duration::from_secs(5)) + .unwrap(); + } + let retired = AvatarManager::new(); + retired.demand_chats_windowed(&chats[..1], &chats[1..2]); + let deadline = Instant::now() + std::time::Duration::from_secs(5); + while native::queued_decodes() == 0 { + assert!(Instant::now() < deadline); + crate::platform::sleep(std::time::Duration::from_millis(1)).await; + } + assert!( + !retired.is_in_flight(&chats[1].jid), + "overscan must wait for visible local recovery" + ); + match retirement { + "clear" => retired.clear(), + "replacement" => retired.on_avatar_ready(&chats[0].jid, "new-picture"), + "account" => retired.set_session(None), + _ => unreachable!(), + } + release.store(true, std::sync::atomic::Ordering::Relaxed); + for blocker in blockers { + blocker.await; + } + while native::DECODE_COMPLETED.load(std::sync::atomic::Ordering::Relaxed) + < completed + 3 + { + assert!(Instant::now() < deadline); + crate::platform::sleep(std::time::Duration::from_millis(1)).await; + } + assert!( + get_avatar_image(chats[0].cache_key.as_ref().unwrap()).is_none(), + "late decode survived {retirement}" + ); + assert!(retired.take_demands().is_empty()); + } }); } diff --git a/crates/gui/src/session/avatar/native.rs b/crates/gui/src/session/avatar/native.rs index 5bd39425..341abb0f 100644 --- a/crates/gui/src/session/avatar/native.rs +++ b/crates/gui/src/session/avatar/native.rs @@ -8,6 +8,13 @@ use std::sync::RwLock; static ACTIVE_ACCOUNT: RwLock> = RwLock::new(None); +#[cfg(test)] +pub(super) static PERSISTENT_READS: std::sync::atomic::AtomicU64 = + std::sync::atomic::AtomicU64::new(0); +#[cfg(test)] +pub(super) static DECODE_COMPLETED: std::sync::atomic::AtomicU64 = + std::sync::atomic::AtomicU64::new(0); + pub fn set_active_account(account: Option) { if let Ok(mut lock) = ACTIVE_ACCOUNT.write() { *lock = account; @@ -31,9 +38,80 @@ pub fn resolve_account(key: &str) -> Option { pub async fn read_persistent(key: &str) -> Option> { let path = oxidezap_ipc::media_path(key)?; + #[cfg(test)] + PERSISTENT_READS.fetch_add(1, std::sync::atomic::Ordering::Relaxed); smol::unblock(move || std::fs::read(path).ok()).await } +// A restore can read dozens of pictures together. Decoding them on smol's +// single default executor thread prevents other futures from making progress; +// unbounded blocking jobs would instead compete with rendering for every CPU +// and retain one full-size decoded allocation per picture. +type DecodeJob = Box; +struct DecodeQueues { + high: smol::channel::Sender, + low: smol::channel::Sender, +} +static DECODE_JOBS: std::sync::LazyLock = std::sync::LazyLock::new(|| { + let (high, high_receive) = smol::channel::bounded::(super::MAX_PERSISTENT_READS); + let (low, low_receive) = smol::channel::bounded::(super::MAX_PERSISTENT_READS); + for index in 0..2 { + let high = high_receive.clone(); + let low = low_receive.clone(); + std::thread::Builder::new() + .name(format!("avatar-decode-{index}")) + .spawn(move || { + while let Ok(job) = high + .try_recv() + .or_else(|_| smol::block_on(smol::future::or(high.recv(), low.recv()))) + { + job(); + } + }) + .expect("start avatar decoder"); + } + DecodeQueues { high, low } +}); + +pub(super) async fn decode_avatar( + bytes: Vec, + priority: super::DemandPriority, +) -> Option<(std::sync::Arc, u64)> { + decode_work(priority, move || { + crate::session::media::decode_avatar(&bytes) + }) + .await +} + +pub(super) async fn decode_work( + priority: super::DemandPriority, + work: impl FnOnce() -> T + Send + 'static, +) -> T { + let (send, receive) = smol::channel::bounded(1); + let queue = match priority { + super::DemandPriority::High => &DECODE_JOBS.high, + super::DemandPriority::Low => &DECODE_JOBS.low, + }; + queue + .send(Box::new(move || { + let _ = send.try_send(work()); + })) + .await + .expect("avatar decoder queue stays open"); + let result = receive + .recv() + .await + .expect("avatar decoder returns a result"); + #[cfg(test)] + DECODE_COMPLETED.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + result +} + +#[cfg(test)] +pub(super) fn queued_decodes() -> usize { + DECODE_JOBS.high.len() + DECODE_JOBS.low.len() +} + pub async fn delete_persistent(key: &str) -> Result<(), String> { if let Some(path) = oxidezap_ipc::media_path(key) { let _ = smol::unblock(move || std::fs::remove_file(path)).await; @@ -76,3 +154,107 @@ pub async fn purge_legacy_caches() { pub fn spawn_task(fut: impl std::future::Future + Send + 'static) { smol::spawn(fut).detach(); } + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; + use std::time::Duration; + + #[test] + fn decoding_yields_the_executor_and_bounds_cpu_jobs() { + let release = Arc::new(AtomicBool::new(false)); + let active = Arc::new(AtomicUsize::new(0)); + let peak = Arc::new(AtomicUsize::new(0)); + let (started, receiving) = std::sync::mpsc::channel(); + let mut tasks = Vec::new(); + for _ in 0..3 { + let release = release.clone(); + let active = active.clone(); + let peak = peak.clone(); + let started = started.clone(); + tasks.push(smol::spawn(decode_work( + super::super::DemandPriority::High, + move || { + let count = active.fetch_add(1, Ordering::SeqCst) + 1; + peak.fetch_max(count, Ordering::SeqCst); + started.send(()).unwrap(); + while !release.load(Ordering::SeqCst) { + std::thread::sleep(Duration::from_millis(1)); + } + active.fetch_sub(1, Ordering::SeqCst); + }, + ))); + } + let both_started = receiving.recv_timeout(Duration::from_secs(5)).is_ok() + && receiving.recv_timeout(Duration::from_secs(5)).is_ok(); + let pulse = smol::block_on(smol::future::or( + async { smol::spawn(async { true }).await }, + async { + smol::Timer::after(Duration::from_secs(1)).await; + false + }, + )); + let third_started_early = receiving.try_recv().is_ok(); + release.store(true, Ordering::SeqCst); + smol::block_on(async { + for task in tasks { + task.await; + } + }); + assert!(both_started, "blocking workers must run concurrently"); + assert!(pulse, "decoding must leave the async executor runnable"); + assert!(!third_started_early, "the third job must wait for a slot"); + assert_eq!(peak.load(Ordering::SeqCst), 2); + + // Keep one worker occupied while the other drains a low-priority + // request queued before a visible one. Visible work must overtake it. + let (started, receiving) = std::sync::mpsc::channel(); + let mut blockers = Vec::new(); + let mut releases = Vec::new(); + for _ in 0..2 { + let release = Arc::new(AtomicBool::new(false)); + releases.push(release.clone()); + let started = started.clone(); + blockers.push(smol::spawn(decode_work( + super::super::DemandPriority::High, + move || { + started.send(()).unwrap(); + while !release.load(Ordering::SeqCst) { + std::thread::sleep(Duration::from_millis(1)); + } + }, + ))); + } + for _ in 0..2 { + receiving.recv_timeout(Duration::from_secs(5)).unwrap(); + } + let (order, received) = std::sync::mpsc::channel(); + let low_order = order.clone(); + let low = smol::spawn(decode_work(super::super::DemandPriority::Low, move || { + low_order.send("overscan").unwrap() + })); + let high = smol::spawn(decode_work(super::super::DemandPriority::High, move || { + order.send("visible").unwrap() + })); + let deadline = wacore::time::Instant::now() + Duration::from_secs(5); + while queued_decodes() < 2 { + assert!(wacore::time::Instant::now() < deadline); + std::thread::sleep(Duration::from_millis(1)); + } + releases[0].store(true, Ordering::SeqCst); + let first = received.recv_timeout(Duration::from_secs(5)); + let second = received.recv_timeout(Duration::from_secs(5)); + releases[1].store(true, Ordering::SeqCst); + smol::block_on(async { + low.await; + high.await; + for blocker in blockers { + blocker.await; + } + }); + assert_eq!(first.unwrap(), "visible"); + assert_eq!(second.unwrap(), "overscan"); + } +} diff --git a/crates/gui/src/session/avatar/probe.rs b/crates/gui/src/session/avatar/probe.rs new file mode 100644 index 00000000..3db49ccd --- /dev/null +++ b/crates/gui/src/session/avatar/probe.rs @@ -0,0 +1,185 @@ +//! Opt-in synthetic local-cache restore measurement. No account or network. + +use super::*; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering as StdOrdering}; +use std::time::Duration; + +fn cpu_ticks() -> u64 { + std::fs::read_to_string("/proc/self/stat") + .ok() + .and_then(|stat| { + let (_, fields) = stat.rsplit_once(") ")?; + let values: Vec<_> = fields.split_whitespace().collect(); + Some(values.get(11)?.parse::().ok()? + values.get(12)?.parse::().ok()?) + }) + .unwrap_or_default() +} + +#[test] +#[ignore = "synthetic restore measurement; run explicitly with --nocapture"] +fn native_restore_probe() { + const ROOT: &str = "OXIDEZAP_AVATAR_PROBE_ROOT"; + let Some(root) = std::env::var_os(ROOT) else { + let root = std::env::temp_dir().join(format!("avatar-probe-{}", std::process::id())); + std::fs::create_dir_all(&root).unwrap(); + for _ in 0..3 { + let status = std::process::Command::new(std::env::current_exe().unwrap()) + .args([ + "--exact", + "session::avatar::probe::native_restore_probe", + "--ignored", + "--nocapture", + ]) + .env(ROOT, &root) + .env("XDG_RUNTIME_DIR", &root) + .env("LOCALAPPDATA", &root) + .env("OXIDEZAP_ACCOUNT", "synthetic-avatar-probe") + .env("SMOL_THREADS", "1") + .status() + .unwrap(); + assert!(status.success()); + } + std::fs::remove_dir_all(root).unwrap(); + return; + }; + let root = std::path::PathBuf::from(root); + let mut bytes = std::io::Cursor::new(Vec::new()); + let jpeg = std::env::var("OXIDEZAP_AVATAR_PROBE_FORMAT").as_deref() == Ok("jpeg"); + let edge = if jpeg { 512 } else { 1024 }; + // A JPEG preview and a larger PNG stress the real decoder/thumbnail path. + let image = image::RgbaImage::from_fn(edge, edge, |x, y| { + image::Rgba([(x % 251) as u8, (y % 241) as u8, ((x ^ y) % 239) as u8, 255]) + }); + let image = if jpeg { + image::DynamicImage::ImageRgb8(image::DynamicImage::ImageRgba8(image).to_rgb8()) + } else { + image::DynamicImage::ImageRgba8(image) + }; + image + .write_to( + &mut bytes, + if jpeg { + image::ImageFormat::Jpeg + } else { + image::ImageFormat::Png + }, + ) + .unwrap(); + let bytes = bytes.into_inner(); + let chats: Vec<_> = (0..25) + .map(|i| { + let key = format!("synthetic-avatar-probe-{i}"); + let path = oxidezap_ipc::media_path(&key).unwrap(); + assert!(path.starts_with(&root)); + std::fs::create_dir_all(path.parent().unwrap()).unwrap(); + std::fs::write(path, &bytes).unwrap(); + ChatDemand { + jid: format!("synthetic-{i}@example.invalid"), + picture_id: Some("picture".into()), + cache_key: Some(key), + } + }) + .collect(); + clear_image_sources(); + let finished = Arc::new(AtomicBool::new(false)); + let gap_us = Arc::new(AtomicU64::new(0)); + let heartbeat = { + let finished = finished.clone(); + let gap_us = gap_us.clone(); + smol::spawn(async move { + let mut previous = Instant::now(); + while !finished.load(StdOrdering::Relaxed) { + smol::Timer::after(Duration::from_millis(1)).await; + let now = Instant::now(); + gap_us.fetch_max( + now.saturating_duration_since(previous).as_micros() as u64, + StdOrdering::Relaxed, + ); + previous = now; + } + }) + }; + // Warm the async executor before starting the measurement. + smol::block_on(smol::Timer::after(Duration::from_millis(10))); + let before_io = std::fs::read_to_string("/proc/self/io").unwrap_or_default(); + let before_cpu = cpu_ticks(); + let before_reads = native::PERSISTENT_READS.load(StdOrdering::Relaxed); + let manager = AvatarManager::new(); + let start = Instant::now(); + let mut epoch = manager.demand_epoch(); + manager.demand_chats_windowed(&chats[..15], &chats[15..]); + let mut first = None; + let mut visible = None; + loop { + // Simulate viewport demand after a cache-completion notification. + let next_epoch = manager.demand_epoch(); + if next_epoch != epoch { + manager.demand_chats_windowed(&chats[..15], &chats[15..]); + epoch = next_epoch; + } + let loaded: Vec<_> = chats + .iter() + .map(|c| get_avatar_image(c.cache_key.as_ref().unwrap()).is_some()) + .collect(); + let elapsed = start.elapsed(); + if first.is_none() && loaded[..15].iter().any(|x| *x) { + first = Some(elapsed); + } + if visible.is_none() && loaded[..15].iter().all(|x| *x) { + visible = Some(elapsed); + } + if loaded.iter().all(|x| *x) { + break; + } + assert!(elapsed < Duration::from_secs(60), "restore did not settle"); + std::thread::sleep(Duration::from_millis(1)); + } + let complete = start.elapsed(); + finished.store(true, StdOrdering::Relaxed); + smol::block_on(heartbeat); + // Publication follows insertion into RAM; allow that final task to drain. + smol::block_on(smol::Timer::after(Duration::from_millis(10))); + let demands = manager.take_demands(); + assert_eq!(demands.len(), 25); + assert!(demands.iter().all(|d| !d.need_bytes && !d.cache_miss)); + assert_eq!( + native::PERSISTENT_READS.load(StdOrdering::Relaxed) - before_reads, + 25 + ); + let warm = Instant::now(); + for _ in 0..10 { + manager.demand_chats_windowed(&chats[..15], &chats[15..]); + } + assert!(manager.take_demands().is_empty()); + assert_eq!( + native::PERSISTENT_READS.load(StdOrdering::Relaxed) - before_reads, + 25, + "warm viewport must not reread disk" + ); + println!( + "restore first_us={} visible_us={} all_us={} executor_gap_us={} warm_10_us={} freshness={} byte_demands=0", + first.unwrap().as_micros(), + visible.unwrap().as_micros(), + complete.as_micros(), + gap_us.load(StdOrdering::Relaxed), + warm.elapsed().as_micros(), + demands.len() + ); + println!("fixture={} edge={edge}", if jpeg { "jpeg" } else { "png" }); + println!( + "io_before={before_io:?} io_after={:?}", + std::fs::read_to_string("/proc/self/io").unwrap_or_default() + ); + println!( + "cpu_ticks={} persistent_reads=25", + cpu_ticks().saturating_sub(before_cpu) + ); + println!( + "process={:?}", + std::fs::read_to_string("/proc/self/status") + .unwrap_or_default() + .lines() + .filter(|l| l.starts_with("VmRSS:") || l.starts_with("VmHWM:")) + .collect::>() + ); +} diff --git a/crates/gui/src/session/avatar/web.rs b/crates/gui/src/session/avatar/web.rs index 7cb5c55b..c70ae6e0 100644 --- a/crates/gui/src/session/avatar/web.rs +++ b/crates/gui/src/session/avatar/web.rs @@ -27,6 +27,13 @@ static STORAGE_GENERATION: portable_atomic::AtomicU64 = portable_atomic::AtomicU static ACTIVE_ACCOUNT: RwLock> = RwLock::new(None); +pub(super) async fn decode_avatar( + bytes: Vec, + _priority: super::DemandPriority, +) -> Option<(std::sync::Arc, u64)> { + crate::session::media::decode_avatar(&bytes) +} + pub fn set_active_account(account: Option) { if let Ok(mut lock) = ACTIVE_ACCOUNT.write() { if *lock != account { diff --git a/docs/gotchas.md b/docs/gotchas.md index 6c596404..dd5cca4b 100644 --- a/docs/gotchas.md +++ b/docs/gotchas.md @@ -1870,6 +1870,19 @@ Non-obvious behaviour, and the reasoning behind it. Read the entry before changi rotates the account scope (128-bit hex); clearing media cache evicts blobs without destroying durable descriptors. + Tray Hide ends the desktop front-end process, so Open recovers decoded + avatars from the persistent tier, just like a first opening. Native local + recovery decodes on two dedicated workers, with bounded queues preferring + visible rows and the header over overscan. The workers sleep when idle; + using fixed workers also avoids spreading decoder allocations across the + general blocking pool's allocator arenas. A completed decode rechecks its + read reservation, account and generation before publishing into RAM. + `session::avatar::probe::native_restore_probe` is an opt-in synthetic probe + for this path: three fresh processes, 15 visible pictures and 10 overscan, + disk-read and byte-demand counts, executor heartbeat, CPU ticks and RSS on + Linux. Its timings end at RAM readiness, before GPU upload or painting; + they do not measure complete tray-to-window latency. + The browser's persistent avatar index is bounded by bytes and entry count. Web Locks serialize mutations across tabs; a persisted generation rejects writes captured before a clear. Failure to persist that generation leaves From 2eaab33a4253981b16272f2c3850216abd5227d0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= <55464917+jlucaso1@users.noreply.github.com> Date: Thu, 29 Oct 2026 20:24:26 -0300 Subject: [PATCH 2/6] fix(gui): skip retired avatar work before decoding --- crates/gui/src/session/avatar/mod.rs | 21 ++++++++++++++++++++- crates/gui/src/session/avatar/native.rs | 9 +++++++++ crates/gui/src/session/avatar/web.rs | 5 ++++- 3 files changed, 33 insertions(+), 2 deletions(-) diff --git a/crates/gui/src/session/avatar/mod.rs b/crates/gui/src/session/avatar/mod.rs index 10729fbc..aa21507d 100644 --- a/crates/gui/src/session/avatar/mod.rs +++ b/crates/gui/src/session/avatar/mod.rs @@ -338,7 +338,20 @@ impl AvatarManager { } if let Some(bytes) = persistent_bytes { - let decoded = imp::decode_avatar(bytes, priority).await; + let current = { + let pending = persistent_in_flight.clone(); + let generation = generation.clone(); + let account = account_id.clone(); + let key = read_key.clone(); + move || { + let pending = pending.lock().unwrap_or_else(|e| e.into_inner()); + pending.get(&key) == Some(&op_token) + && generation.load(Ordering::Relaxed) == born_gen + && *account.lock().unwrap_or_else(|e| e.into_inner()) + == born_account + } + }; + let decoded = imp::decode_avatar(bytes, priority, current).await; { let mut p = persistent_in_flight .lock() @@ -811,6 +824,7 @@ mod tests { let (started, receiving) = std::sync::mpsc::channel(); let mut blockers = Vec::new(); let completed = native::DECODE_COMPLETED.load(std::sync::atomic::Ordering::Relaxed); + let performed = native::DECODE_PERFORMED.load(std::sync::atomic::Ordering::Relaxed); for _ in 0..2 { let release = release.clone(); let started = started.clone(); @@ -861,6 +875,11 @@ mod tests { "late decode survived {retirement}" ); assert!(retired.take_demands().is_empty()); + assert_eq!( + native::DECODE_PERFORMED.load(std::sync::atomic::Ordering::Relaxed), + performed, + "retired queued work must skip decoding after {retirement}" + ); } }); } diff --git a/crates/gui/src/session/avatar/native.rs b/crates/gui/src/session/avatar/native.rs index 341abb0f..ad6131a8 100644 --- a/crates/gui/src/session/avatar/native.rs +++ b/crates/gui/src/session/avatar/native.rs @@ -14,6 +14,9 @@ pub(super) static PERSISTENT_READS: std::sync::atomic::AtomicU64 = #[cfg(test)] pub(super) static DECODE_COMPLETED: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); +#[cfg(test)] +pub(super) static DECODE_PERFORMED: std::sync::atomic::AtomicU64 = + std::sync::atomic::AtomicU64::new(0); pub fn set_active_account(account: Option) { if let Ok(mut lock) = ACTIVE_ACCOUNT.write() { @@ -76,8 +79,14 @@ static DECODE_JOBS: std::sync::LazyLock = std::sync::LazyLock::new pub(super) async fn decode_avatar( bytes: Vec, priority: super::DemandPriority, + current: impl FnOnce() -> bool + Send + 'static, ) -> Option<(std::sync::Arc, u64)> { decode_work(priority, move || { + if !current() { + return None; + } + #[cfg(test)] + DECODE_PERFORMED.fetch_add(1, std::sync::atomic::Ordering::Relaxed); crate::session::media::decode_avatar(&bytes) }) .await diff --git a/crates/gui/src/session/avatar/web.rs b/crates/gui/src/session/avatar/web.rs index c70ae6e0..0d762116 100644 --- a/crates/gui/src/session/avatar/web.rs +++ b/crates/gui/src/session/avatar/web.rs @@ -30,8 +30,11 @@ static ACTIVE_ACCOUNT: RwLock> = RwLock::new(No pub(super) async fn decode_avatar( bytes: Vec, _priority: super::DemandPriority, + current: impl FnOnce() -> bool + Send + 'static, ) -> Option<(std::sync::Arc, u64)> { - crate::session::media::decode_avatar(&bytes) + current() + .then(|| crate::session::media::decode_avatar(&bytes)) + .flatten() } pub fn set_active_account(account: Option) { From ff2208190f59c920d4a54e6ba2b526a0fd54a393 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= <55464917+jlucaso1@users.noreply.github.com> Date: Thu, 29 Oct 2026 20:29:45 -0300 Subject: [PATCH 3/6] fix(gui): wake failed local recovery and promote queued avatars --- crates/gui/src/session/avatar/mod.rs | 145 ++++++++++++++++++++---- crates/gui/src/session/avatar/native.rs | 121 +++++++++++++------- crates/gui/src/session/avatar/web.rs | 2 +- docs/gotchas.md | 5 +- 4 files changed, 208 insertions(+), 65 deletions(-) diff --git a/crates/gui/src/session/avatar/mod.rs b/crates/gui/src/session/avatar/mod.rs index aa21507d..d9d70699 100644 --- a/crates/gui/src/session/avatar/mod.rs +++ b/crates/gui/src/session/avatar/mod.rs @@ -76,6 +76,28 @@ pub enum DemandPriority { Low, } +type DecodePriority = Arc; + +fn decode_priority(priority: DemandPriority) -> DecodePriority { + Arc::new(portable_atomic::AtomicBool::new( + priority == DemandPriority::High, + )) +} + +struct PendingRead { + token: u64, + priority: DecodePriority, +} + +impl PendingRead { + fn new(token: u64, priority: DemandPriority) -> Self { + Self { + token, + priority: decode_priority(priority), + } + } +} + const MAX_TRACKED_AVATARS: usize = 2048; const MAX_PENDING_DEMANDS: usize = 256; const MAX_PERSISTENT_READS: usize = 64; @@ -86,7 +108,7 @@ static NEXT_OP_TOKEN: AtomicU64 = AtomicU64::new(1); /// and in-memory decoded image caching. #[derive(Clone, Default)] pub struct AvatarManager { - persistent_in_flight: Arc>>, + persistent_in_flight: Arc>>, network_in_flight: Arc>>, revalidated: Arc>>, failed_cooldowns: Arc>>, @@ -289,13 +311,21 @@ impl AvatarManager { .persistent_in_flight .lock() .unwrap_or_else(|e| e.into_inner()); + if let Some(read) = p_flight.get(&(jid.clone(), key.clone())) { + if priority == DemandPriority::High { + read.priority.store(true, Ordering::Relaxed); + } + return; + } if p_flight.len() >= MAX_PERSISTENT_READS { return; } if let std::collections::hash_map::Entry::Vacant(entry) = p_flight.entry((jid.clone(), key.clone())) { - entry.insert(op_token); + let reservation = PendingRead::new(op_token, priority); + let decode_priority = reservation.priority.clone(); + entry.insert(reservation); let key_clone = key.clone(); let jid_clone = jid.clone(); let pic_id = chat.picture_id.clone(); @@ -319,7 +349,8 @@ impl AvatarManager { .lock() .unwrap_or_else(|e| e.into_inner()) .get(&read_key) - != Some(&op_token) + .map(|read| read.token) + != Some(op_token) { return; } @@ -331,7 +362,7 @@ impl AvatarManager { let mut p = persistent_in_flight .lock() .unwrap_or_else(|e| e.into_inner()); - if p.get(&read_key) == Some(&op_token) { + if p.get(&read_key).map(|read| read.token) == Some(op_token) { p.remove(&read_key); } return; @@ -345,13 +376,13 @@ impl AvatarManager { let key = read_key.clone(); move || { let pending = pending.lock().unwrap_or_else(|e| e.into_inner()); - pending.get(&key) == Some(&op_token) + pending.get(&key).map(|read| read.token) == Some(op_token) && generation.load(Ordering::Relaxed) == born_gen && *account.lock().unwrap_or_else(|e| e.into_inner()) == born_account } }; - let decoded = imp::decode_avatar(bytes, priority, current).await; + let decoded = imp::decode_avatar(bytes, decode_priority, current).await; { let mut p = persistent_in_flight .lock() @@ -360,12 +391,12 @@ impl AvatarManager { // account change. Check again after the worker's // await, holding the read reservation through RAM // publication so retirement cannot overtake it. - if p.get(&read_key) != Some(&op_token) + if p.get(&read_key).map(|read| read.token) != Some(op_token) || generation.load(Ordering::Relaxed) != born_gen || *account_id.lock().unwrap_or_else(|e| e.into_inner()) != born_account { - if p.get(&read_key) == Some(&op_token) { + if p.get(&read_key).map(|read| read.token) == Some(op_token) { p.remove(&read_key); } return; @@ -378,7 +409,7 @@ impl AvatarManager { let mut p = persistent_in_flight .lock() .unwrap_or_else(|e| e.into_inner()); - if p.get(&read_key) != Some(&op_token) { + if p.get(&read_key).map(|read| read.token) != Some(op_token) { return; } p.remove(&read_key); @@ -434,7 +465,7 @@ impl AvatarManager { let mut p = persistent_in_flight .lock() .unwrap_or_else(|e| e.into_inner()); - if p.get(&read_key) == Some(&op_token) { + if p.get(&read_key).map(|read| read.token) == Some(op_token) { p.remove(&read_key); } return; @@ -446,16 +477,28 @@ impl AvatarManager { let mut p = persistent_in_flight .lock() .unwrap_or_else(|e| e.into_inner()); - if p.get(&read_key) != Some(&op_token) { + if p.get(&read_key).map(|read| read.token) != Some(op_token) { return; } p.remove(&read_key); } - let mut n_flight = network_in_flight.lock().unwrap_or_else(|e| e.into_inner()); - if n_flight.len() >= MAX_PENDING_DEMANDS { - return; + let reserved = { + let mut n_flight = + network_in_flight.lock().unwrap_or_else(|e| e.into_inner()); + n_flight.len() < MAX_PENDING_DEMANDS && n_flight.insert(jid_clone.clone()) + }; + // A miss also releases visible local admission. Wake the + // viewport even offline, so deferred overscan can recover + // without waiting for a successful network response. + demand_epoch.fetch_add(1, Ordering::Relaxed); + if let Some(session) = session_handle + .lock() + .unwrap_or_else(|e| e.into_inner()) + .as_ref() + { + session.notify_avatar_loaded(); } - if n_flight.insert(jid_clone.clone()) { + if reserved { let demand = AvatarDemand { jid: jid_clone, known_picture_id: None, @@ -733,6 +776,44 @@ mod tests { assert!(!demand.need_bytes, "disk hit needs only freshness metadata"); assert_eq!(demand.known_picture_id, chat.picture_id); + // Offline failures in visible local recovery must release and + // advance admission so valid overscan hits can still load. + for corrupt in [false, true] { + clear_image_sources(); + let missing = ChatDemand { + jid: "offline-miss@example.invalid".into(), + picture_id: Some("missing-picture".into()), + cache_key: Some("avatar-offline-miss".into()), + }; + let missing_path = + oxidezap_ipc::media_path(missing.cache_key.as_ref().unwrap()).unwrap(); + if corrupt { + std::fs::write(&missing_path, b"corrupt fixture").unwrap(); + } + let recovery = AvatarManager::new(); + let epoch = recovery.demand_epoch(); + recovery.demand_chats_windowed( + std::slice::from_ref(&missing), + std::slice::from_ref(&chat), + ); + assert!(next(&recovery).await.need_bytes); + assert_ne!( + recovery.demand_epoch(), + epoch, + "failed reads must invalidate the demand fingerprint" + ); + recovery.demand_chats_windowed( + std::slice::from_ref(&missing), + std::slice::from_ref(&chat), + ); + assert!( + !next(&recovery).await.need_bytes, + "offline overscan must recover from disk" + ); + assert!(get_avatar_image(key).is_some()); + assert!(!missing_path.exists()); + } + let freshness = AvatarManager::new(); freshness.demand_chats_windowed(&[], std::slice::from_ref(&chat)); assert!(freshness.is_revalidated(&chat.jid)); @@ -829,7 +910,7 @@ mod tests { let release = release.clone(); let started = started.clone(); blockers.push(smol::spawn(native::decode_work( - DemandPriority::High, + decode_priority(DemandPriority::High), move || { started.send(()).unwrap(); while !release.load(std::sync::atomic::Ordering::Relaxed) { @@ -844,12 +925,32 @@ mod tests { .unwrap(); } let retired = AvatarManager::new(); - retired.demand_chats_windowed(&chats[..1], &chats[1..2]); + let reads = native::PERSISTENT_READS.load(std::sync::atomic::Ordering::Relaxed); + if retirement == "replacement" { + retired.demand_one(&chats[0], DemandPriority::Low); + } else { + retired.demand_chats_windowed(&chats[..1], &chats[1..2]); + } let deadline = Instant::now() + std::time::Duration::from_secs(5); while native::queued_decodes() == 0 { assert!(Instant::now() < deadline); crate::platform::sleep(std::time::Duration::from_millis(1)).await; } + retired.demand_chat(&chats[0]); + assert_eq!( + native::PERSISTENT_READS.load(std::sync::atomic::Ordering::Relaxed), + reads + 1, + "promotion must not reread disk" + ); + assert!( + retired + .persistent_in_flight + .lock() + .unwrap() + .values() + .all(|read| read.priority.load(Ordering::Relaxed)), + "visible demand must promote queued recovery" + ); assert!( !retired.is_in_flight(&chats[1].jid), "overscan must wait for visible local recovery" @@ -1032,8 +1133,14 @@ mod tests { for key in ["new-picture", ""] { let manager = AvatarManager::new(); let mut pending = manager.persistent_in_flight.lock().unwrap(); - pending.insert(("peer@example.invalid".into(), "old-picture".into()), 1); - pending.insert(("other@example.invalid".into(), "other-picture".into()), 2); + pending.insert( + ("peer@example.invalid".into(), "old-picture".into()), + PendingRead::new(1, DemandPriority::Low), + ); + pending.insert( + ("other@example.invalid".into(), "other-picture".into()), + PendingRead::new(2, DemandPriority::Low), + ); drop(pending); manager.on_avatar_ready("peer@example.invalid", key); let pending = manager.persistent_in_flight.lock().unwrap(); diff --git a/crates/gui/src/session/avatar/native.rs b/crates/gui/src/session/avatar/native.rs index ad6131a8..c381e4c8 100644 --- a/crates/gui/src/session/avatar/native.rs +++ b/crates/gui/src/session/avatar/native.rs @@ -50,35 +50,54 @@ pub async fn read_persistent(key: &str) -> Option> { // single default executor thread prevents other futures from making progress; // unbounded blocking jobs would instead compete with rendering for every CPU // and retain one full-size decoded allocation per picture. -type DecodeJob = Box; -struct DecodeQueues { - high: smol::channel::Sender, - low: smol::channel::Sender, +struct DecodeJob { + priority: super::DecodePriority, + run: Box, } -static DECODE_JOBS: std::sync::LazyLock = std::sync::LazyLock::new(|| { - let (high, high_receive) = smol::channel::bounded::(super::MAX_PERSISTENT_READS); - let (low, low_receive) = smol::channel::bounded::(super::MAX_PERSISTENT_READS); - for index in 0..2 { - let high = high_receive.clone(); - let low = low_receive.clone(); - std::thread::Builder::new() - .name(format!("avatar-decode-{index}")) - .spawn(move || { - while let Ok(job) = high - .try_recv() - .or_else(|_| smol::block_on(smol::future::or(high.recv(), low.recv()))) - { - job(); - } - }) - .expect("start avatar decoder"); - } - DecodeQueues { high, low } -}); +struct DecodeQueue { + jobs: std::sync::Mutex>, + ready: std::sync::Condvar, + slots: tokio::sync::Semaphore, +} +static DECODE_JOBS: std::sync::LazyLock> = + std::sync::LazyLock::new(|| { + let queue = std::sync::Arc::new(DecodeQueue { + jobs: std::sync::Mutex::new(std::collections::VecDeque::new()), + ready: std::sync::Condvar::new(), + slots: tokio::sync::Semaphore::const_new(super::MAX_PERSISTENT_READS), + }); + for index in 0..2 { + let queue = queue.clone(); + std::thread::Builder::new() + .name(format!("avatar-decode-{index}")) + .spawn(move || { + loop { + let job = { + let mut jobs = queue.jobs.lock().unwrap_or_else(|e| e.into_inner()); + while jobs.is_empty() { + jobs = queue.ready.wait(jobs).unwrap_or_else(|e| e.into_inner()); + } + // Read priority when selecting work: scrolling can promote + // an existing reservation without rereading its file. + let index = jobs + .iter() + .position(|job| { + job.priority.load(portable_atomic::Ordering::Relaxed) + }) + .unwrap_or(0); + jobs.remove(index).expect("nonempty decode queue") + }; + (job.run)(); + } + }) + .expect("start avatar decoder"); + } + queue + }); pub(super) async fn decode_avatar( bytes: Vec, - priority: super::DemandPriority, + priority: super::DecodePriority, current: impl FnOnce() -> bool + Send + 'static, ) -> Option<(std::sync::Arc, u64)> { decode_work(priority, move || { @@ -93,20 +112,29 @@ pub(super) async fn decode_avatar( } pub(super) async fn decode_work( - priority: super::DemandPriority, + priority: super::DecodePriority, work: impl FnOnce() -> T + Send + 'static, ) -> T { - let (send, receive) = smol::channel::bounded(1); - let queue = match priority { - super::DemandPriority::High => &DECODE_JOBS.high, - super::DemandPriority::Low => &DECODE_JOBS.low, - }; - queue - .send(Box::new(move || { - let _ = send.try_send(work()); - })) + // Count running and queued jobs together. Producers await admission + // without blocking the async executor. + let permit = DECODE_JOBS + .slots + .acquire() .await - .expect("avatar decoder queue stays open"); + .expect("decode slots never close"); + let (send, receive) = smol::channel::bounded(1); + DECODE_JOBS + .jobs + .lock() + .unwrap_or_else(|e| e.into_inner()) + .push_back(DecodeJob { + priority, + run: Box::new(move || { + let _permit = permit; + let _ = send.try_send(work()); + }), + }); + DECODE_JOBS.ready.notify_one(); let result = receive .recv() .await @@ -118,7 +146,11 @@ pub(super) async fn decode_work( #[cfg(test)] pub(super) fn queued_decodes() -> usize { - DECODE_JOBS.high.len() + DECODE_JOBS.low.len() + DECODE_JOBS + .jobs + .lock() + .unwrap_or_else(|e| e.into_inner()) + .len() } pub async fn delete_persistent(key: &str) -> Result<(), String> { @@ -184,7 +216,7 @@ mod tests { let peak = peak.clone(); let started = started.clone(); tasks.push(smol::spawn(decode_work( - super::super::DemandPriority::High, + super::super::decode_priority(super::super::DemandPriority::High), move || { let count = active.fetch_add(1, Ordering::SeqCst) + 1; peak.fetch_max(count, Ordering::SeqCst); @@ -227,7 +259,7 @@ mod tests { releases.push(release.clone()); let started = started.clone(); blockers.push(smol::spawn(decode_work( - super::super::DemandPriority::High, + super::super::decode_priority(super::super::DemandPriority::High), move || { started.send(()).unwrap(); while !release.load(Ordering::SeqCst) { @@ -241,10 +273,12 @@ mod tests { } let (order, received) = std::sync::mpsc::channel(); let low_order = order.clone(); - let low = smol::spawn(decode_work(super::super::DemandPriority::Low, move || { - low_order.send("overscan").unwrap() - })); - let high = smol::spawn(decode_work(super::super::DemandPriority::High, move || { + let low = smol::spawn(decode_work( + super::super::decode_priority(super::super::DemandPriority::Low), + move || low_order.send("overscan").unwrap(), + )); + let promoted = super::super::decode_priority(super::super::DemandPriority::Low); + let high = smol::spawn(decode_work(promoted.clone(), move || { order.send("visible").unwrap() })); let deadline = wacore::time::Instant::now() + Duration::from_secs(5); @@ -252,6 +286,7 @@ mod tests { assert!(wacore::time::Instant::now() < deadline); std::thread::sleep(Duration::from_millis(1)); } + promoted.store(true, portable_atomic::Ordering::Relaxed); releases[0].store(true, Ordering::SeqCst); let first = received.recv_timeout(Duration::from_secs(5)); let second = received.recv_timeout(Duration::from_secs(5)); diff --git a/crates/gui/src/session/avatar/web.rs b/crates/gui/src/session/avatar/web.rs index 0d762116..759c9a7e 100644 --- a/crates/gui/src/session/avatar/web.rs +++ b/crates/gui/src/session/avatar/web.rs @@ -29,7 +29,7 @@ static ACTIVE_ACCOUNT: RwLock> = RwLock::new(No pub(super) async fn decode_avatar( bytes: Vec, - _priority: super::DemandPriority, + _priority: super::DecodePriority, current: impl FnOnce() -> bool + Send + 'static, ) -> Option<(std::sync::Arc, u64)> { current() diff --git a/docs/gotchas.md b/docs/gotchas.md index dd5cca4b..f76e236a 100644 --- a/docs/gotchas.md +++ b/docs/gotchas.md @@ -1872,8 +1872,9 @@ Non-obvious behaviour, and the reasoning behind it. Read the entry before changi Tray Hide ends the desktop front-end process, so Open recovers decoded avatars from the persistent tier, just like a first opening. Native local - recovery decodes on two dedicated workers, with bounded queues preferring - visible rows and the header over overscan. The workers sleep when idle; + recovery decodes on two dedicated workers, with a bounded queue preferring + visible rows and the header over overscan. Queued priorities follow viewport + demand without rereading files. The workers sleep when idle; using fixed workers also avoids spreading decoder allocations across the general blocking pool's allocator arenas. A completed decode rechecks its read reservation, account and generation before publishing into RAM. From 2bfe5dc2f554c60f0b754ed1e71e426148932b79 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= <55464917+jlucaso1@users.noreply.github.com> Date: Thu, 29 Oct 2026 20:38:00 -0300 Subject: [PATCH 4/6] fix(gui): defer known avatar misses while network admission is full --- crates/gui/src/session/avatar/mod.rs | 137 +++++++++++++++++++++++++++ 1 file changed, 137 insertions(+) diff --git a/crates/gui/src/session/avatar/mod.rs b/crates/gui/src/session/avatar/mod.rs index d9d70699..ed7d0a69 100644 --- a/crates/gui/src/session/avatar/mod.rs +++ b/crates/gui/src/session/avatar/mod.rs @@ -113,6 +113,7 @@ pub struct AvatarManager { revalidated: Arc>>, failed_cooldowns: Arc>>, pending_demands: Arc>>, + deferred_misses: Arc>>, generation: Arc, demand_epoch: Arc, account_id: Arc>>, @@ -164,6 +165,10 @@ impl AvatarManager { /// Clear in-flight, revalidated, and cooldown tracking on connection restart while /// preserving decoded images in memory. pub fn reset_connection(&self) { + self.deferred_misses + .lock() + .unwrap_or_else(|e| e.into_inner()) + .clear(); self.persistent_in_flight .lock() .unwrap_or_else(|e| e.into_inner()) @@ -305,6 +310,48 @@ impl AvatarManager { return; } + // A previous read already established that this exact blob is + // absent. Retry network admission without reading it again on + // every repaint while the network queue remains full. + let deferred = self + .deferred_misses + .lock() + .unwrap_or_else(|e| e.into_inner()) + .get(jid) + == Some(key); + if deferred { + let reserved = { + let mut network = self + .network_in_flight + .lock() + .unwrap_or_else(|e| e.into_inner()); + network.len() < MAX_PENDING_DEMANDS && network.insert(jid.clone()) + }; + if reserved { + if self.enqueue_demand( + AvatarDemand { + jid: jid.clone(), + known_picture_id: None, + cache_key: None, + need_bytes: true, + cache_miss: true, + }, + priority, + ) { + self.deferred_misses + .lock() + .unwrap_or_else(|e| e.into_inner()) + .shift_remove(jid); + } else { + self.network_in_flight + .lock() + .unwrap_or_else(|e| e.into_inner()) + .remove(jid); + } + } + return; + } + // Miss in RAM: check persistent store if not already in flight let op_token = NEXT_OP_TOKEN.fetch_add(1, Ordering::Relaxed); let mut p_flight = self @@ -338,6 +385,7 @@ impl AvatarManager { let generation = Arc::clone(&self.generation); let demand_epoch = Arc::clone(&self.demand_epoch); let account_id = Arc::clone(&self.account_id); + let deferred_misses = Arc::clone(&self.deferred_misses); let born_gen = generation.load(Ordering::Relaxed); let born_account = *account_id.lock().unwrap_or_else(|e| e.into_inner()); @@ -487,6 +535,16 @@ impl AvatarManager { network_in_flight.lock().unwrap_or_else(|e| e.into_inner()); n_flight.len() < MAX_PENDING_DEMANDS && n_flight.insert(jid_clone.clone()) }; + if !reserved { + let mut deferred = + deferred_misses.lock().unwrap_or_else(|e| e.into_inner()); + if deferred.len() >= MAX_TRACKED_AVATARS + && !deferred.contains_key(&jid_clone) + { + deferred.shift_remove_index(0); + } + deferred.insert(jid_clone.clone(), key_clone.clone()); + } // A miss also releases visible local admission. Wake the // viewport even offline, so deferred overscan can recover // without waiting for a successful network response. @@ -618,6 +676,10 @@ impl AvatarManager { /// Called when the daemon signals an avatar is ready (or cleared). pub fn on_avatar_ready(&self, jid: &str, key: &str) { + self.deferred_misses + .lock() + .unwrap_or_else(|e| e.into_inner()) + .shift_remove(jid); self.network_in_flight .lock() .unwrap_or_else(|e| e.into_inner()) @@ -636,6 +698,28 @@ impl AvatarManager { } else { revalidated.shift_remove(jid); } + drop(revalidated); + self.wake_deferred_misses(); + } + + fn wake_deferred_misses(&self) { + if self + .deferred_misses + .lock() + .unwrap_or_else(|e| e.into_inner()) + .is_empty() + { + return; + } + self.bump_demand_epoch(); + if let Some(session) = self + .session + .lock() + .unwrap_or_else(|e| e.into_inner()) + .as_ref() + { + session.notify_avatar_loaded(); + } } /// Called when an avatar lookup, download, or materialization fails. @@ -671,6 +755,8 @@ impl AvatarManager { } } cooldowns.insert(jid.to_string(), (retryable, until)); + drop(cooldowns); + self.wake_deferred_misses(); until } @@ -814,6 +900,57 @@ mod tests { assert!(!missing_path.exists()); } + // Saturated network admission must not turn its own completion + // wake into repeated filesystem reads. Capacity release retries + // the known miss directly, without another disk lookup. + let saturated = AvatarManager::new(); + saturated + .network_in_flight + .lock() + .unwrap() + .extend((0..MAX_PENDING_DEMANDS).map(|i| format!("saturated-{i}@example.invalid"))); + let missing = ChatDemand { + jid: "deferred-miss@example.invalid".into(), + picture_id: Some("missing-picture".into()), + cache_key: Some("avatar-deferred-miss".into()), + }; + let reads = native::PERSISTENT_READS.load(std::sync::atomic::Ordering::Relaxed); + saturated.demand_chat(&missing); + let deadline = Instant::now() + std::time::Duration::from_secs(5); + while !saturated + .deferred_misses + .lock() + .unwrap() + .contains_key(&missing.jid) + { + assert!(Instant::now() < deadline); + crate::platform::sleep(std::time::Duration::from_millis(1)).await; + } + for _ in 0..10 { + saturated.demand_chat(&missing); + } + assert!(saturated.take_demands().is_empty()); + assert_eq!( + native::PERSISTENT_READS.load(std::sync::atomic::Ordering::Relaxed), + reads + 1 + ); + let epoch = saturated.demand_epoch(); + saturated.on_avatar_failed("saturated-0@example.invalid", true); + assert_ne!( + saturated.demand_epoch(), + epoch, + "capacity release must wake deferred demand" + ); + saturated.demand_chat(&missing); + let retry = saturated.take_demands(); + assert_eq!(retry.len(), 1); + assert!(retry[0].need_bytes && retry[0].cache_miss); + assert_eq!( + native::PERSISTENT_READS.load(std::sync::atomic::Ordering::Relaxed), + reads + 1, + "capacity retry must not read disk again" + ); + let freshness = AvatarManager::new(); freshness.demand_chats_windowed(&[], std::slice::from_ref(&chat)); assert!(freshness.is_revalidated(&chat.jid)); From 533ae61b68e8920a984a0bf7bc651cbc559f208c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= <55464917+jlucaso1@users.noreply.github.com> Date: Thu, 29 Oct 2026 21:01:30 -0300 Subject: [PATCH 5/6] test(gui): use portable avatar probe counters --- .github/workflows/avatar-restore.yml | 2 +- crates/gui/src/session/avatar/native.rs | 9 +++------ crates/gui/src/session/avatar/probe.rs | 2 +- 3 files changed, 5 insertions(+), 8 deletions(-) diff --git a/.github/workflows/avatar-restore.yml b/.github/workflows/avatar-restore.yml index ed18481f..5ee33dbe 100644 --- a/.github/workflows/avatar-restore.yml +++ b/.github/workflows/avatar-restore.yml @@ -62,7 +62,7 @@ jobs: marker = 'pub async fn read_persistent(key: &str) -> Option> {' assert marker in text text = text.replace(marker, marker + '\n #[cfg(test)]\n PERSISTENT_READS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);', 1) - text += '\n#[cfg(test)]\npub(super) static PERSISTENT_READS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);\n' + text += '\n#[cfg(test)]\npub(super) static PERSISTENT_READS: portable_atomic::AtomicU64 = portable_atomic::AtomicU64::new(0);\n' native.write_text(text) PY cargo test -p oxidezap-gui --bin oxidezap --locked native_restore_probe -- --ignored --nocapture diff --git a/crates/gui/src/session/avatar/native.rs b/crates/gui/src/session/avatar/native.rs index c381e4c8..46db5f52 100644 --- a/crates/gui/src/session/avatar/native.rs +++ b/crates/gui/src/session/avatar/native.rs @@ -9,14 +9,11 @@ use std::sync::RwLock; static ACTIVE_ACCOUNT: RwLock> = RwLock::new(None); #[cfg(test)] -pub(super) static PERSISTENT_READS: std::sync::atomic::AtomicU64 = - std::sync::atomic::AtomicU64::new(0); +pub(super) static PERSISTENT_READS: portable_atomic::AtomicU64 = portable_atomic::AtomicU64::new(0); #[cfg(test)] -pub(super) static DECODE_COMPLETED: std::sync::atomic::AtomicU64 = - std::sync::atomic::AtomicU64::new(0); +pub(super) static DECODE_COMPLETED: portable_atomic::AtomicU64 = portable_atomic::AtomicU64::new(0); #[cfg(test)] -pub(super) static DECODE_PERFORMED: std::sync::atomic::AtomicU64 = - std::sync::atomic::AtomicU64::new(0); +pub(super) static DECODE_PERFORMED: portable_atomic::AtomicU64 = portable_atomic::AtomicU64::new(0); pub fn set_active_account(account: Option) { if let Ok(mut lock) = ACTIVE_ACCOUNT.write() { diff --git a/crates/gui/src/session/avatar/probe.rs b/crates/gui/src/session/avatar/probe.rs index 3db49ccd..3a3ee53f 100644 --- a/crates/gui/src/session/avatar/probe.rs +++ b/crates/gui/src/session/avatar/probe.rs @@ -1,7 +1,7 @@ //! Opt-in synthetic local-cache restore measurement. No account or network. use super::*; -use std::sync::atomic::{AtomicBool, AtomicU64, Ordering as StdOrdering}; +use portable_atomic::{AtomicBool, AtomicU64, Ordering as StdOrdering}; use std::time::Duration; fn cpu_ticks() -> u64 { From ef2bb7e0f74c69c6d62a87e6ae1ab85717be2a2e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= <55464917+jlucaso1@users.noreply.github.com> Date: Thu, 29 Oct 2026 21:51:20 -0300 Subject: [PATCH 6/6] fix(avatar): keep decoder workers alive after image panics --- crates/gui/src/session/avatar/native.rs | 31 +++++++++++++++++++------ 1 file changed, 24 insertions(+), 7 deletions(-) diff --git a/crates/gui/src/session/avatar/native.rs b/crates/gui/src/session/avatar/native.rs index 46db5f52..abe67024 100644 --- a/crates/gui/src/session/avatar/native.rs +++ b/crates/gui/src/session/avatar/native.rs @@ -84,7 +84,9 @@ static DECODE_JOBS: std::sync::LazyLock> = .unwrap_or(0); jobs.remove(index).expect("nonempty decode queue") }; - (job.run)(); + // A bad image must not permanently remove a fixed worker. + // Unwinding drops the sender and releases its admission permit. + let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(job.run)); } }) .expect("start avatar decoder"); @@ -106,12 +108,13 @@ pub(super) async fn decode_avatar( crate::session::media::decode_avatar(&bytes) }) .await + .flatten() } pub(super) async fn decode_work( priority: super::DecodePriority, work: impl FnOnce() -> T + Send + 'static, -) -> T { +) -> Option { // Count running and queued jobs together. Producers await admission // without blocking the async executor. let permit = DECODE_JOBS @@ -132,13 +135,10 @@ pub(super) async fn decode_work( }), }); DECODE_JOBS.ready.notify_one(); - let result = receive - .recv() - .await - .expect("avatar decoder returns a result"); + let result = receive.recv().await.ok()?; #[cfg(test)] DECODE_COMPLETED.fetch_add(1, std::sync::atomic::Ordering::Relaxed); - result + Some(result) } #[cfg(test)] @@ -202,6 +202,23 @@ mod tests { #[test] fn decoding_yields_the_executor_and_bounds_cpu_jobs() { + // More panics than workers must neither exhaust the pool nor leak slots. + for _ in 0..super::super::MAX_PERSISTENT_READS + 2 { + let result = smol::block_on(smol::future::or( + decode_work( + super::super::decode_priority(super::super::DemandPriority::High), + || panic!("synthetic decoder panic"), + ), + async { + smol::Timer::after(Duration::from_secs(5)).await; + Some(()) + }, + )); + assert_eq!( + result, None, + "panic must return failure and release admission" + ); + } let release = Arc::new(AtomicBool::new(false)); let active = Arc::new(AtomicUsize::new(0)); let peak = Arc::new(AtomicUsize::new(0));