From 917fee939f23b00a6ea6180f74f7f0e76afc56f9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=89=E5=92=B2=E9=9B=85=20misaki=20masa?= Date: Sat, 14 Jun 2025 16:13:48 +0800 Subject: [PATCH] perf: make preload tasks discardable (#2875) --- Cargo.lock | 6 ++- Cargo.toml | 2 +- yazi-core/src/mgr/watcher.rs | 2 +- yazi-core/src/tasks/preload.rs | 8 ++-- yazi-fs/src/file.rs | 2 +- yazi-plugin/src/isolate/preload.rs | 55 ++++++++++++++++++++------- yazi-plugin/src/isolate/spot.rs | 15 +++----- yazi-plugin/src/macros.rs | 2 +- yazi-scheduler/Cargo.toml | 1 + yazi-scheduler/src/hooks.rs | 43 +++++++++++++++++++++ yazi-scheduler/src/lib.rs | 2 +- yazi-scheduler/src/ongoing.rs | 8 ++-- yazi-scheduler/src/prework/prework.rs | 28 ++++++++------ yazi-scheduler/src/scheduler.rs | 28 +++++++------- yazi-shared/Cargo.toml | 1 + yazi-shared/src/url/url.rs | 5 ++- 16 files changed, 145 insertions(+), 63 deletions(-) create mode 100644 yazi-scheduler/src/hooks.rs diff --git a/Cargo.lock b/Cargo.lock index 050cc56d..713ba8df 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1308,9 +1308,9 @@ checksum = "03087c2bad5e1034e8cace5926dec053fb3790248370865f5117a7d0213354c8" [[package]] name = "libc" -version = "0.2.172" +version = "0.2.173" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d750af042f7ef4f724306de029d18836c26c1765a54a6a3f094cbd23a7267ffa" +checksum = "d8cfeafaffdbc32176b64fb251369d52ea9f0a8fbc6f8759edffef7b525d64bb" [[package]] name = "libfuzzer-sys" @@ -3735,6 +3735,7 @@ dependencies = [ "parking_lot", "scopeguard", "tokio", + "tokio-util", "tracing", "trash", "yazi-config", @@ -3752,6 +3753,7 @@ version = "25.6.11" dependencies = [ "anyhow", "crossterm 0.29.0", + "foldhash", "futures", "libc", "memchr", diff --git a/Cargo.toml b/Cargo.toml index b21c226d..9e35bad3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -32,7 +32,7 @@ foldhash = "0.1.5" futures = "0.3.31" globset = "0.4.16" indexmap = { version = "2.9.0", features = [ "serde" ] } -libc = "0.2.172" +libc = "0.2.173" lru = "0.14.0" mlua = { version = "0.10.5", features = [ "anyhow", "async", "error-send", "lua54", "macros", "serialize" ] } objc = "0.2.7" diff --git a/yazi-core/src/mgr/watcher.rs b/yazi-core/src/mgr/watcher.rs index ff6187fc..a7475387 100644 --- a/yazi-core/src/mgr/watcher.rs +++ b/yazi-core/src/mgr/watcher.rs @@ -122,7 +122,7 @@ impl Watcher { async fn fan_out(rx: UnboundedReceiver) { // TODO: revert this once a new notification is implemented - let rx = UnboundedReceiverStream::new(rx).chunks_timeout(1000, Duration::from_millis(350)); + let rx = UnboundedReceiverStream::new(rx).chunks_timeout(1000, Duration::from_millis(250)); pin!(rx); while let Some(chunk) = rx.next().await { diff --git a/yazi-core/src/tasks/preload.rs b/yazi-core/src/tasks/preload.rs index 2254a780..a1745f4b 100644 --- a/yazi-core/src/tasks/preload.rs +++ b/yazi-core/src/tasks/preload.rs @@ -9,7 +9,7 @@ impl Tasks { let mut loaded = self.scheduler.prework.loaded.lock(); let mut tasks: [Vec<_>; MAX_PREWORKERS as usize] = Default::default(); for f in paged { - let hash = f.hash(); + let hash = f.hash_u64(); for g in YAZI.plugin.fetchers(&f.url, mimetype.by_file(f).unwrap_or_default()) { match loaded.get_mut(&hash) { Some(n) if *n & (1 << g.idx) != 0 => continue, @@ -31,7 +31,7 @@ impl Tasks { pub fn preload_paged(&self, paged: &[File], mimetype: &Mimetype) { let mut loaded = self.scheduler.prework.loaded.lock(); for f in paged { - let hash = f.hash(); + let hash = f.hash_u64(); for p in YAZI.plugin.preloaders(&f.url, mimetype.by_file(f).unwrap_or_default()) { match loaded.get_mut(&hash) { Some(n) if *n & (1 << p.idx) != 0 => continue, @@ -49,7 +49,7 @@ impl Tasks { } let targets: Vec<_> = { - let loading = self.scheduler.prework.size_loading.read(); + let loading = self.scheduler.prework.sizing.read(); targets .iter() .filter(|f| f.is_dir() && !targets.sizes.contains_key(f.urn()) && !loading.contains(&f.url)) @@ -60,7 +60,7 @@ impl Tasks { return; } - let mut loading = self.scheduler.prework.size_loading.write(); + let mut loading = self.scheduler.prework.sizing.write(); for &target in &targets { loading.insert(target.clone()); } diff --git a/yazi-fs/src/file.rs b/yazi-fs/src/file.rs index e26fa30f..469a5952 100644 --- a/yazi-fs/src/file.rs +++ b/yazi-fs/src/file.rs @@ -45,7 +45,7 @@ impl File { } #[inline] - pub fn hash(&self) -> u64 { + pub fn hash_u64(&self) -> u64 { let mut h = foldhash::fast::FixedState::default().build_hasher(); self.url.hash(&mut h); h.write_u8(0); diff --git a/yazi-plugin/src/isolate/preload.rs b/yazi-plugin/src/isolate/preload.rs index d39f6784..97d0f6b3 100644 --- a/yazi-plugin/src/isolate/preload.rs +++ b/yazi-plugin/src/isolate/preload.rs @@ -1,5 +1,6 @@ -use mlua::{ExternalResult, IntoLua, ObjectLike}; -use tokio::runtime::Handle; +use mlua::{ExternalError, ExternalResult, HookTriggers, IntoLua, ObjectLike, VmState}; +use tokio::{runtime::Handle, select}; +use tokio_util::sync::CancellationToken; use yazi_binding::Error; use yazi_config::LAYOUT; use yazi_dds::Sendable; @@ -11,21 +12,49 @@ use crate::{elements::Rect, file::File, loader::LOADER}; pub async fn preload( cmd: &'static Cmd, file: yazi_fs::File, + ct: CancellationToken, ) -> mlua::Result<(bool, Option)> { - LOADER.ensure(&cmd.name, |_| ()).await.into_lua_err()?; - + let ct_ = ct.clone(); tokio::task::spawn_blocking(move || { - let lua = slim_lua(&cmd.name)?; - let plugin = LOADER.load_once(&lua, &cmd.name)?; + let future = async { + LOADER.ensure(&cmd.name, |_| ()).await.into_lua_err()?; - let job = lua.create_table_from([ - ("area", Rect::from(LAYOUT.get().preview).into_lua(&lua)?), - ("args", Sendable::args_to_table_ref(&lua, &cmd.args)?.into_lua(&lua)?), - ("file", File::new(file).into_lua(&lua)?), - ("skip", 0.into_lua(&lua)?), - ])?; + let lua = slim_lua(&cmd.name)?; + lua.set_hook( + HookTriggers::new().on_calls().on_returns().every_nth_instruction(2000), + move |_, dbg| { + if ct.is_cancelled() && dbg.source().what != "C" { + Err("Preload task cancelled".into_lua_err()) + } else { + Ok(VmState::Continue) + } + }, + ); - Handle::current().block_on(plugin.call_async_method("preload", job)) + let plugin = LOADER.load_once(&lua, &cmd.name)?; + let job = lua.create_table_from([ + ("area", Rect::from(LAYOUT.get().preview).into_lua(&lua)?), + ("args", Sendable::args_to_table_ref(&lua, &cmd.args)?.into_lua(&lua)?), + ("file", File::new(file).into_lua(&lua)?), + ("skip", 0.into_lua(&lua)?), + ])?; + + if ct_.is_cancelled() { + Ok((false, None)) + } else { + plugin.call_async_method("preload", job).await + } + }; + + Handle::current().block_on(async { + select! { + _ = ct_.cancelled() => Ok((false, None)), + r = future => match r { + Err(e) if e.to_string().contains("Preload task cancelled") => Ok((false, None)), + Ok(_) | Err(_) => r, + }, + } + }) }) .await .into_lua_err()? diff --git a/yazi-plugin/src/isolate/spot.rs b/yazi-plugin/src/isolate/spot.rs index 85831e17..9f03ad66 100644 --- a/yazi-plugin/src/isolate/spot.rs +++ b/yazi-plugin/src/isolate/spot.rs @@ -50,18 +50,15 @@ pub fn spot( if ct2.is_cancelled() { Ok(()) } else { plugin.call_async_method("spot", job).await } }; - let result = Handle::current().block_on(async { + Handle::current().block_on(async { select! { - _ = ct2.cancelled() => Ok(()), - r = future => r, + _ = ct2.cancelled() => {}, + Err(e) = future => if !e.to_string().contains("Spot task cancelled") { + error!("{e}"); + }, + else => {} } }); - - if let Err(e) = result { - if !e.to_string().contains("Spot task cancelled") { - error!("{e}"); - } - } }); ct diff --git a/yazi-plugin/src/macros.rs b/yazi-plugin/src/macros.rs index b2710b5c..3834d44d 100644 --- a/yazi-plugin/src/macros.rs +++ b/yazi-plugin/src/macros.rs @@ -101,7 +101,7 @@ macro_rules! impl_file_fields { #[macro_export] macro_rules! impl_file_methods { ($methods:ident) => { - $methods.add_method("hash", |_, me, ()| Ok(me.hash())); + $methods.add_method("hash", |_, me, ()| Ok(me.hash_u64())); $methods.add_method("icon", |_, me, ()| { use yazi_shared::theme::IconCache; diff --git a/yazi-scheduler/Cargo.toml b/yazi-scheduler/Cargo.toml index 1cfc2e5b..2e10ba35 100644 --- a/yazi-scheduler/Cargo.toml +++ b/yazi-scheduler/Cargo.toml @@ -25,6 +25,7 @@ lru = { workspace = true } parking_lot = { workspace = true } scopeguard = { workspace = true } tokio = { workspace = true } +tokio-util = { workspace = true } tracing = { workspace = true } [target."cfg(unix)".dependencies] diff --git a/yazi-scheduler/src/hooks.rs b/yazi-scheduler/src/hooks.rs new file mode 100644 index 00000000..9ed7cf2f --- /dev/null +++ b/yazi-scheduler/src/hooks.rs @@ -0,0 +1,43 @@ +use std::collections::HashMap; + +use futures::future::BoxFuture; +use yazi_shared::Id; + +#[derive(Default)] +pub(super) struct Hooks { + inner: HashMap, +} + +impl Hooks { + #[allow(dead_code)] + pub(super) fn add_sync(&mut self, id: Id, f: F) + where + F: FnOnce(bool) + Send + Sync + 'static, + { + self.inner.insert(id, Hook::Sync(Box::new(f))); + } + + pub(super) fn add_async(&mut self, id: Id, f: F) + where + F: FnOnce(bool) -> BoxFuture<'static, ()> + Send + Sync + 'static, + { + self.inner.insert(id, Hook::Async(Box::new(f))); + } + + pub(super) fn run_or_pop(&mut self, id: Id, cancel: bool) -> Option> { + match self.inner.remove(&id)? { + Hook::Sync(f) => { + f(cancel); + None + } + Hook::Async(f) => Some(f(cancel)), + } + } +} + +// --- Hook +// TODO: remove Send + Sync bounds when not needed +pub(super) enum Hook { + Sync(Box), + Async(Box BoxFuture<'static, ()>) + Send + Sync>), +} diff --git a/yazi-scheduler/src/lib.rs b/yazi-scheduler/src/lib.rs index 5d1cedee..e34c1889 100644 --- a/yazi-scheduler/src/lib.rs +++ b/yazi-scheduler/src/lib.rs @@ -2,7 +2,7 @@ yazi_macro::mod_pub!(file plugin prework process); -yazi_macro::mod_flat!(ongoing out r#in scheduler task); +yazi_macro::mod_flat!(hooks ongoing out r#in scheduler task); const LOW: u8 = yazi_config::Priority::Low as u8; const NORMAL: u8 = yazi_config::Priority::Normal as u8; diff --git a/yazi-scheduler/src/ongoing.rs b/yazi-scheduler/src/ongoing.rs index 9eb5b848..b37928d3 100644 --- a/yazi-scheduler/src/ongoing.rs +++ b/yazi-scheduler/src/ongoing.rs @@ -5,11 +5,11 @@ use yazi_config::YAZI; use yazi_shared::{Id, Ids}; use super::{Task, TaskStage}; -use crate::TaskKind; +use crate::{Hooks, TaskKind}; #[derive(Default)] pub struct Ongoing { - pub(super) hooks: HashMap BoxFuture<'static, ()>) + Send + Sync>>, + pub(super) hooks: Hooks, pub(super) all: HashMap, } @@ -67,8 +67,8 @@ impl Ongoing { if task.succ < task.total { return None; } - if let Some(hook) = self.hooks.remove(&id) { - return Some(hook(false)); + if let Some(fut) = self.hooks.run_or_pop(id, false) { + return Some(fut); } } TaskStage::Hooked => {} diff --git a/yazi-scheduler/src/prework/prework.rs b/yazi-scheduler/src/prework/prework.rs index 856e7e3a..915bf972 100644 --- a/yazi-scheduler/src/prework/prework.rs +++ b/yazi-scheduler/src/prework/prework.rs @@ -4,6 +4,7 @@ use anyhow::{Result, anyhow}; use lru::LruCache; use parking_lot::{Mutex, RwLock}; use tokio::sync::mpsc; +use tokio_util::sync::CancellationToken; use tracing::error; use yazi_config::Priority; use yazi_fs::{FilesOp, SizeCalculator}; @@ -17,8 +18,9 @@ pub struct Prework { r#macro: async_priority_channel::Sender, prog: mpsc::UnboundedSender, - pub loaded: Mutex>, - pub size_loading: RwLock>, + pub loaded: Mutex>, + pub loading: Mutex>, + pub sizing: RwLock>, } impl Prework { @@ -30,14 +32,16 @@ impl Prework { r#macro, prog, loaded: Mutex::new(LruCache::new(NonZeroUsize::new(4096).unwrap())), - size_loading: Default::default(), + loading: Mutex::new(LruCache::new(NonZeroUsize::new(256).unwrap())), + sizing: Default::default(), } } pub async fn work(&self, r#in: PreworkIn) -> Result<()> { + let id = r#in.id(); match r#in { PreworkIn::Fetch(task) => { - let hashes: Vec<_> = task.targets.iter().map(|f| f.hash()).collect(); + let hashes: Vec<_> = task.targets.iter().map(|f| f.hash_u64()).collect(); let result = isolate::fetch(CmdCow::from(&task.plugin.run), task.targets).await; if let Err(e) = result { self.fail(task.id, format!("Failed to run fetcher `{}`:\n{e}", task.plugin.run.name))?; @@ -52,11 +56,15 @@ impl Prework { if let Some(e) = err { error!("Error when running fetcher `{}`:\n{e}", task.plugin.run.name); } - self.prog.send(TaskProg::Adv(task.id, 1, 0))?; } PreworkIn::Load(task) => { - let hash = task.target.hash(); - let result = isolate::preload(&task.plugin.run, task.target).await; + let ct = CancellationToken::new(); + if let Some(ct) = self.loading.lock().put(task.target.url.hash_u64(), ct.clone()) { + ct.cancel(); + } + + let hash = task.target.hash_u64(); + let result = isolate::preload(&task.plugin.run, task.target, ct).await; if let Err(e) = result { self .fail(task.id, format!("Failed to run preloader `{}`:\n{e}", task.plugin.run.name))?; @@ -70,13 +78,12 @@ impl Prework { if let Some(e) = err { error!("Error when running preloader `{}`:\n{e}", task.plugin.run.name); } - self.prog.send(TaskProg::Adv(task.id, 1, 0))?; } PreworkIn::Size(task) => { let length = SizeCalculator::total(&task.target).await.unwrap_or(0); task.throttle.done((task.target, length), |buf| { { - let mut loading = self.size_loading.write(); + let mut loading = self.sizing.write(); for (path, _) in &buf { loading.remove(path); } @@ -89,10 +96,9 @@ impl Prework { ) .emit(); }); - self.prog.send(TaskProg::Adv(task.id, 1, 0))?; } } - Ok(()) + Ok(self.prog.send(TaskProg::Adv(id, 1, 0))?) } pub async fn fetch(&self, task: PreworkInFetch) -> Result<()> { diff --git a/yazi-scheduler/src/scheduler.rs b/yazi-scheduler/src/scheduler.rs index b83fd821..0a5b5fa6 100644 --- a/yazi-scheduler/src/scheduler.rs +++ b/yazi-scheduler/src/scheduler.rs @@ -58,8 +58,8 @@ impl Scheduler { pub fn cancel(&self, id: Id) -> bool { let mut ongoing = self.ongoing.lock(); - if let Some(hook) = ongoing.hooks.remove(&id) { - self.micro.try_send(hook(true), HIGH).ok(); + if let Some(fut) = ongoing.hooks.run_or_pop(id, true) { + self.micro.try_send(fut, HIGH).ok(); return false; } @@ -81,11 +81,11 @@ impl Scheduler { return; } - ongoing.hooks.insert(id, { + ongoing.hooks.add_async(id, { let ongoing = self.ongoing.clone(); let (from, to) = (from.clone(), to.clone()); - Box::new(move |canceled: bool| { + move |canceled: bool| { async move { if !canceled { remove_dir_clean(&from).await; @@ -94,7 +94,7 @@ impl Scheduler { ongoing.lock().try_remove(id, TaskStage::Hooked); } .boxed() - }) + } }); let file = self.file.clone(); @@ -158,11 +158,11 @@ impl Scheduler { let mut ongoing = self.ongoing.lock(); let id = ongoing.add(TaskKind::User, format!("Delete {target}")); - ongoing.hooks.insert(id, { + ongoing.hooks.add_async(id, { let target = target.clone(); let ongoing = self.ongoing.clone(); - Box::new(move |canceled: bool| { + move |canceled: bool| { async move { if !canceled { fs::remove_dir_all(&target).await.ok(); @@ -172,7 +172,7 @@ impl Scheduler { ongoing.lock().try_remove(id, TaskStage::Hooked); } .boxed() - }) + } }); let file = self.file.clone(); @@ -187,11 +187,11 @@ impl Scheduler { let mut ongoing = self.ongoing.lock(); let id = ongoing.add(TaskKind::User, format!("Trash {target}")); - ongoing.hooks.insert(id, { + ongoing.hooks.add_async(id, { let target = target.clone(); let ongoing = self.ongoing.clone(); - Box::new(move |canceled: bool| { + move |canceled: bool| { async move { if !canceled { MgrProxy::update_tasks(&target); @@ -200,7 +200,7 @@ impl Scheduler { ongoing.lock().try_remove(id, TaskStage::Hooked); } .boxed() - }) + } }); let file = self.file.clone(); @@ -275,9 +275,9 @@ impl Scheduler { let mut ongoing = self.ongoing.lock(); let id = ongoing.add(TaskKind::User, name); - ongoing.hooks.insert(id, { + ongoing.hooks.add_async(id, { let ongoing = self.ongoing.clone(); - Box::new(move |canceled: bool| { + move |canceled: bool| { async move { if canceled { cancel_tx.send(()).await.ok(); @@ -289,7 +289,7 @@ impl Scheduler { ongoing.lock().try_remove(id, TaskStage::Hooked); } .boxed() - }) + } }); let cmd = OsString::from(&opener.run); diff --git a/yazi-shared/Cargo.toml b/yazi-shared/Cargo.toml index f6c01b85..329f8b1e 100644 --- a/yazi-shared/Cargo.toml +++ b/yazi-shared/Cargo.toml @@ -15,6 +15,7 @@ yazi-macro = { path = "../yazi-macro", version = "25.6.11" } # External dependencies anyhow = { workspace = true } crossterm = { workspace = true } +foldhash = { workspace = true } futures = { workspace = true } memchr = "2.7.5" parking_lot = { workspace = true } diff --git a/yazi-shared/src/url/url.rs b/yazi-shared/src/url/url.rs index 6cf63350..48bbd0ee 100644 --- a/yazi-shared/src/url/url.rs +++ b/yazi-shared/src/url/url.rs @@ -1,4 +1,4 @@ -use std::{ffi::OsStr, fmt::{Debug, Display, Formatter}, hash::{Hash, Hasher}, ops::Deref, path::{Path, PathBuf}}; +use std::{ffi::OsStr, fmt::{Debug, Display, Formatter}, hash::{BuildHasher, Hash, Hasher}, ops::Deref, path::{Path, PathBuf}}; use percent_encoding::{AsciiSet, CONTROLS, percent_decode_str, percent_encode}; use serde::{Deserialize, Serialize}; @@ -171,6 +171,9 @@ impl Url { #[inline] pub fn pair(&self) -> Option<(Self, UrnBuf)> { Some((self.parent_url()?, self.loc.urn_owned())) } + #[inline] + pub fn hash_u64(&self) -> u64 { foldhash::fast::FixedState::default().hash_one(self) } + #[inline] pub fn rebase(&self, parent: &Path) -> Self { debug_assert!(self.is_regular());