This commit is contained in:
sxyazi 2025-06-14 15:34:13 +08:00
parent 0698a7fe28
commit 0ea8111168
No known key found for this signature in database
3 changed files with 65 additions and 33 deletions

View file

@ -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()?

View file

@ -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

View file

@ -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<()> {