perf: make preload tasks discardable (#2875)

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",

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"

yazi-core/src/mgr/watcher.rs

@@ -122,7 +122,7 @@ impl Watcher {
 
 	async fn fan_out(rx: UnboundedReceiver<Url>) {
 		// 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 {

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());
 		}

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);

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<Error>)> {
-	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()?

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

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;

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]

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<Id, Hook>,
+}
+
+impl Hooks {
+	#[allow(dead_code)]
+	pub(super) fn add_sync<F>(&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<F>(&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<BoxFuture<'static, ()>> {
+		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<dyn FnOnce(bool) + Send + Sync>),
+	Async(Box<dyn (FnOnce(bool) -> BoxFuture<'static, ()>) + Send + Sync>),
+}

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;

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<Id, Box<dyn (FnOnce(bool) -> BoxFuture<'static, ()>) + Send + Sync>>,
+	pub(super) hooks: Hooks,
 	pub(super) all:   HashMap<Id, Task>,
 }
 
@@ -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 => {}

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<TaskOp, u8>,
 	prog:    mpsc::UnboundedSender<TaskProg>,
 
-	pub loaded:       Mutex<LruCache<u64, u32>>,
-	pub size_loading: RwLock<HashSet<Url>>,
+	pub loaded:  Mutex<LruCache<u64, u32>>,
+	pub loading: Mutex<LruCache<u64, CancellationToken>>,
+	pub sizing:  RwLock<HashSet<Url>>,
 }
 
 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<()> {

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);

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 }

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());