feat: offload mimetype fetching on opening files to the task scheduler (#3141)

This commit is contained in:
三咲雅 misaki masa 2025-09-06 00:18:31 +08:00 committed by GitHub
parent 3c42195bac
commit e795d6cf42
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
17 changed files with 190 additions and 83 deletions

View file

@ -32,6 +32,7 @@ impl UserData for TaskSnap {
fields.add_field_method_get("running", |_, me| Ok(me.prog.running()));
fields.add_field_method_get("success", |_, me| Ok(me.prog.success()));
fields.add_field_method_get("cleaning", |_, me| Ok(me.prog.cleaning()));
fields.add_field_method_get("percent", |_, me| Ok(me.prog.percent()));
}
}

View file

@ -38,9 +38,9 @@ impl UserData for Yanked {
fn add_methods<M: UserDataMethods<Self>>(methods: &mut M) {
methods.add_meta_method(MetaMethod::Len, |_, me, ()| Ok(me.len()));
methods.add_meta_function(MetaMethod::Pairs, |lua, ud: AnyUserData| {
let me = ud.borrow::<Self>()?;
get_metatable(lua, &me.iter)?.call_function::<MultiValue>(MetaMethod::Pairs.name(), ud)
methods.add_meta_method(MetaMethod::Pairs, |lua, me, ()| {
get_metatable(lua, &me.iter)?
.call_function::<MultiValue>(MetaMethod::Pairs.name(), me.iter.clone())
});
}
}

View file

@ -1,15 +1,14 @@
use std::{borrow::Cow, iter};
use anyhow::Result;
use tracing::error;
use tokio::sync::oneshot;
use yazi_config::{YAZI, popup::PickCfg};
use yazi_core::tab::Folder;
use yazi_fs::File;
use yazi_macro::{act, succ};
use yazi_parser::mgr::{OpenDoOpt, OpenOpt};
use yazi_plugin::isolate;
use yazi_proxy::{MgrProxy, PickProxy, TasksProxy};
use yazi_shared::{MIME_DIR, event::{CmdCow, Data}, url::UrlBuf};
use yazi_shared::{MIME_DIR, event::Data, url::UrlBuf};
use crate::{Actor, Ctx, mgr::Quit};
@ -52,6 +51,7 @@ impl Actor for Open {
return act!(mgr:open_do, cx, OpenDoOpt { cwd, hovered, targets, interactive: opt.interactive });
}
let scheduler = cx.tasks.scheduler.clone();
tokio::spawn(async move {
let mut files = Vec::with_capacity(todo.len());
for i in todo {
@ -60,9 +60,16 @@ impl Actor for Open {
}
}
let mut wg = vec![];
for (fetcher, files) in YAZI.plugin.mime_fetchers(files) {
if let Err(e) = isolate::fetch(CmdCow::from(&fetcher.run), files).await {
error!("Fetch mime failed on opening: {e}");
let (tx, rx) = oneshot::channel();
scheduler.fetch_paged(fetcher, files, Some(tx));
wg.push(rx);
}
for rx in wg {
if !rx.await.is_ok_and(|canceled| !canceled) {
return;
}
}

View file

@ -55,7 +55,7 @@ impl Actor for Inspect {
execute!(TTY.writer(), crossterm::style::Print(line), crossterm::style::Print("\r\n")).ok();
}
_ = time::sleep(time::Duration::from_millis(500)) => {
if ongoing.lock().get(id).is_none() {
if !ongoing.lock().exists(id) {
execute!(TTY.writer(), crossterm::style::Print("Task finished, press `q` to quit\r\n")).ok();
break;
}

View file

@ -23,7 +23,7 @@ impl Tasks {
drop(loaded);
for (i, tasks) in tasks.into_iter().enumerate() {
if !tasks.is_empty() {
self.scheduler.fetch_paged(&YAZI.plugin.fetchers[i], tasks);
self.scheduler.fetch_paged(&YAZI.plugin.fetchers[i], tasks, None);
}
}
}

View file

@ -73,9 +73,9 @@ impl UserData for UpdateYankedIter {
fn add_methods<M: UserDataMethods<Self>>(methods: &mut M) {
methods.add_meta_method(MetaMethod::Len, |_, me, ()| Ok(me.len));
methods.add_meta_function(MetaMethod::Pairs, |lua, ud: AnyUserData| {
let me = ud.borrow::<Self>()?;
get_metatable(lua, &me.inner)?.call_function::<MultiValue>(MetaMethod::Pairs.name(), ud)
methods.add_meta_method(MetaMethod::Pairs, |lua, me, ()| {
get_metatable(lua, &me.inner)?
.call_function::<MultiValue>(MetaMethod::Pairs.name(), me.inner.clone())
});
}
}

View file

@ -74,9 +74,16 @@ end
function Tasks:progress_redraw(snap, y)
local kind = snap.prog.kind
if kind == "FilePaste" or kind == "FileDelete" then
local percent
if snap.cleaning then
percent = "Cleaning…"
else
percent = string.format("%3d%%", math.floor(snap.percent))
end
local label = string.format(
"%3d%% - %s / %s",
math.floor(snap.percent),
"%s - %s / %s",
percent,
ya.readable_size(snap.prog.processed_bytes),
ya.readable_size(snap.prog.total_bytes)
)
@ -103,7 +110,7 @@ function Tasks:progress_redraw(snap, y)
if snap.running then
text = "Running…"
elseif snap.success then
text = "Completing…"
text = "Cleaning…"
else
text = "Failed, press Enter to view log…"
end

View file

@ -4,8 +4,9 @@ use crate::{Task, TaskProg};
#[derive(Debug)]
pub(crate) enum FileOutPaste {
New(u64),
Init,
Deform(String),
Init,
Clean,
}
impl From<std::io::Error> for FileOutPaste {
@ -20,14 +21,17 @@ impl FileOutPaste {
prog.total_files += 1;
prog.total_bytes += bytes;
}
Self::Init => {
prog.collected = true;
}
Self::Deform(reason) => {
prog.total_files += 1;
prog.failed_files += 1;
task.log(reason);
}
Self::Init => {
prog.collected = true;
}
Self::Clean => {
prog.cleaned = true;
}
}
}
}
@ -111,8 +115,8 @@ impl FileOutLink {
#[derive(Debug)]
pub(crate) enum FileOutHardlink {
New,
Init,
Deform(String),
Init,
}
impl From<std::io::Error> for FileOutHardlink {
@ -126,14 +130,14 @@ impl FileOutHardlink {
Self::New => {
prog.total += 1;
}
Self::Init => {
prog.collected = true;
}
Self::Deform(reason) => {
prog.total += 1;
prog.failed += 1;
task.log(reason);
}
Self::Init => {
prog.collected = true;
}
}
}
}
@ -168,8 +172,9 @@ impl FileOutHardlinkDo {
#[derive(Debug)]
pub(crate) enum FileOutDelete {
New(u64),
Init,
Deform(String),
Init,
Clean,
}
impl From<std::io::Error> for FileOutDelete {
@ -184,14 +189,17 @@ impl FileOutDelete {
prog.total_files += 1;
prog.total_bytes += size;
}
Self::Init => {
prog.collected = true;
}
Self::Deform(reason) => {
prog.total_files += 1;
prog.failed_files += 1;
task.log(reason);
}
Self::Init => {
prog.collected = true;
}
Self::Clean => {
prog.cleaned = true;
}
}
}
}

View file

@ -10,6 +10,7 @@ pub struct FileProgPaste {
pub total_bytes: u64,
pub processed_bytes: u64,
pub collected: bool,
pub cleaned: bool,
}
impl From<FileProgPaste> for TaskSummary {
@ -30,6 +31,8 @@ impl FileProgPaste {
pub fn success(self) -> bool { self.collected && self.success_files == self.total_files }
pub fn cleaning(self) -> bool { !self.cleaned && self.success() }
pub fn percent(self) -> Option<f32> {
Some(if self.success() {
100.0
@ -63,6 +66,8 @@ impl FileProgLink {
pub fn success(self) -> bool { self.state == Some(true) }
pub fn cleaning(self) -> bool { self.success() }
pub fn percent(self) -> Option<f32> { None }
}
@ -91,6 +96,8 @@ impl FileProgHardlink {
pub fn success(self) -> bool { self.collected && self.success == self.total }
pub fn cleaning(self) -> bool { self.success() }
pub fn percent(self) -> Option<f32> { None }
}
@ -103,6 +110,7 @@ pub struct FileProgDelete {
pub total_bytes: u64,
pub processed_bytes: u64,
pub collected: bool,
pub cleaned: bool,
}
impl From<FileProgDelete> for TaskSummary {
@ -123,6 +131,8 @@ impl FileProgDelete {
pub fn success(self) -> bool { self.collected && self.success_files == self.total_files }
pub fn cleaning(self) -> bool { !self.cleaned && self.success() }
pub fn percent(self) -> Option<f32> {
Some(if self.success() {
100.0
@ -156,5 +166,7 @@ impl FileProgTrash {
pub fn success(self) -> bool { self.state == Some(true) }
pub fn cleaning(self) -> bool { self.success() }
pub fn percent(self) -> Option<f32> { None }
}

View file

@ -1,4 +1,4 @@
use futures::future::BoxFuture;
use futures::{FutureExt, future::BoxFuture};
use hashbrown::HashMap;
use yazi_shared::Id;
@ -8,13 +8,21 @@ pub(super) struct Hooks {
}
impl Hooks {
#[inline]
pub(super) fn add_sync<F>(&mut self, id: Id, f: F)
where
F: FnOnce(bool) + Send + 'static,
{
self.inner.insert(id, Box::new(SyncHook(f)));
}
#[inline]
pub(super) fn add_async<F, Fut>(&mut self, id: Id, f: F)
where
F: FnOnce(bool) -> Fut + Send + 'static,
Fut: Future<Output = ()> + Send + 'static,
{
self.inner.insert(id, Box::new(f));
self.inner.insert(id, Box::new(AsyncHook(f)));
}
#[inline]
@ -22,14 +30,30 @@ impl Hooks {
}
// --- Hook
pub trait Hook: Send {
fn call(self: Box<Self>, cancel: bool) -> BoxFuture<'static, ()>;
pub(super) trait Hook: Send {
fn call(self: Box<Self>, cancel: bool) -> Option<BoxFuture<'static, ()>>;
}
impl<F, Fut> Hook for F
struct SyncHook<F>(F);
impl<F> Hook for SyncHook<F>
where
F: FnOnce(bool) + Send,
{
fn call(self: Box<Self>, cancel: bool) -> Option<BoxFuture<'static, ()>> {
(self.0)(cancel);
None
}
}
struct AsyncHook<F>(F);
impl<F, Fut> Hook for AsyncHook<F>
where
F: FnOnce(bool) -> Fut + Send,
Fut: Future<Output = ()> + Send + 'static,
{
fn call(self: Box<Self>, cancel: bool) -> BoxFuture<'static, ()> { Box::pin((*self)(cancel)) }
fn call(self: Box<Self>, cancel: bool) -> Option<BoxFuture<'static, ()>> {
Some((self.0)(cancel).boxed())
}
}

View file

@ -25,16 +25,11 @@ impl Ongoing {
id
}
#[inline]
pub fn get(&self, id: Id) -> Option<&Task> { self.all.get(&id) }
#[inline]
pub fn get_mut(&mut self, id: Id) -> Option<&mut Task> { self.all.get_mut(&id) }
#[inline]
pub fn get_id(&self, idx: usize) -> Option<Id> { self.values().nth(idx).map(|t| t.id) }
#[inline]
pub fn len(&self) -> usize {
if YAZI.tasks.suppress_preload {
self.all.values().filter(|&t| t.prog.is_user()).count()
@ -46,7 +41,6 @@ impl Ongoing {
#[inline]
pub fn exists(&self, id: Id) -> bool { self.all.contains_key(&id) }
#[inline]
pub fn values(&self) -> Box<dyn Iterator<Item = &Task> + '_> {
if YAZI.tasks.suppress_preload {
Box::new(self.all.values().filter(|&t| t.prog.is_user()))
@ -58,11 +52,6 @@ impl Ongoing {
#[inline]
pub fn is_empty(&self) -> bool { self.len() == 0 }
pub fn remove(&mut self, id: Id) {
self.hooks.pop(id);
self.all.remove(&id);
}
pub fn summary(&self) -> TaskSummary {
let mut summary = TaskSummary::default();
let mut percent_sum = 0.0f64;

View file

@ -23,5 +23,7 @@ impl PluginProgEntry {
pub fn success(self) -> bool { self.state == Some(true) }
pub fn cleaning(self) -> bool { self.success() }
pub fn percent(self) -> Option<f32> { None }
}

View file

@ -23,6 +23,8 @@ impl PreworkProgFetch {
pub fn success(self) -> bool { self.state == Some(true) }
pub fn cleaning(self) -> bool { self.success() }
pub fn percent(self) -> Option<f32> { None }
}
@ -48,6 +50,8 @@ impl PreworkProgLoad {
pub fn success(self) -> bool { self.state == Some(true) }
pub fn cleaning(self) -> bool { self.success() }
pub fn percent(self) -> Option<f32> { None }
}
@ -73,5 +77,7 @@ impl PreworkProgSize {
pub fn success(self) -> bool { self.done }
pub fn cleaning(self) -> bool { self.success() }
pub fn percent(self) -> Option<f32> { None }
}

View file

@ -4,6 +4,7 @@ use crate::{Task, TaskProg};
pub(crate) enum ProcessOutBlock {
Succ,
Fail(String),
Clean,
}
impl From<std::io::Error> for ProcessOutBlock {
@ -21,6 +22,9 @@ impl ProcessOutBlock {
prog.state = Some(false);
task.log(reason);
}
Self::Clean => {
prog.cleaned = true;
}
}
}
}
@ -30,6 +34,7 @@ impl ProcessOutBlock {
pub(crate) enum ProcessOutOrphan {
Succ,
Fail(String),
Clean,
}
impl From<anyhow::Error> for ProcessOutOrphan {
@ -47,6 +52,9 @@ impl ProcessOutOrphan {
prog.state = Some(false);
task.log(reason);
}
Self::Clean => {
prog.cleaned = true;
}
}
}
}
@ -57,6 +65,7 @@ pub(crate) enum ProcessOutBg {
Log(String),
Succ,
Fail(String),
Clean,
}
impl From<anyhow::Error> for ProcessOutBg {
@ -77,6 +86,9 @@ impl ProcessOutBg {
prog.state = Some(false);
task.log(reason);
}
Self::Clean => {
prog.cleaned = true;
}
}
}
}

View file

@ -4,7 +4,8 @@ use yazi_parser::app::TaskSummary;
// --- Block
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize)]
pub struct ProcessProgBlock {
pub state: Option<bool>,
pub state: Option<bool>,
pub cleaned: bool,
}
impl From<ProcessProgBlock> for TaskSummary {
@ -23,13 +24,16 @@ impl ProcessProgBlock {
pub fn success(self) -> bool { self.state == Some(true) }
pub fn cleaning(self) -> bool { !self.cleaned && self.success() }
pub fn percent(self) -> Option<f32> { None }
}
// --- Orphan
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize)]
pub struct ProcessProgOrphan {
pub state: Option<bool>,
pub state: Option<bool>,
pub cleaned: bool,
}
impl From<ProcessProgOrphan> for TaskSummary {
@ -48,13 +52,16 @@ impl ProcessProgOrphan {
pub fn success(self) -> bool { self.state == Some(true) }
pub fn cleaning(self) -> bool { !self.cleaned && self.success() }
pub fn percent(self) -> Option<f32> { None }
}
// --- Bg
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize)]
pub struct ProcessProgBg {
pub state: Option<bool>,
pub state: Option<bool>,
pub cleaned: bool,
}
impl From<ProcessProgBg> for TaskSummary {
@ -73,5 +80,7 @@ impl ProcessProgBg {
pub fn success(self) -> bool { self.state == Some(true) }
pub fn cleaning(self) -> bool { !self.cleaned && self.success() }
pub fn percent(self) -> Option<f32> { None }
}

View file

@ -101,6 +101,27 @@ impl TaskProg {
}
}
pub fn cleaning(self) -> bool {
match self {
// File
Self::FilePaste(p) => p.cleaning(),
Self::FileLink(p) => p.cleaning(),
Self::FileHardlink(p) => p.cleaning(),
Self::FileDelete(p) => p.cleaning(),
Self::FileTrash(p) => p.cleaning(),
// Plugin
Self::PluginEntry(p) => p.cleaning(),
// Prework
Self::PreworkFetch(p) => p.cleaning(),
Self::PreworkLoad(p) => p.cleaning(),
Self::PreworkSize(p) => p.cleaning(),
// Process
Self::ProcessBlock(p) => p.cleaning(),
Self::ProcessOrphan(p) => p.cleaning(),
Self::ProcessBg(p) => p.cleaning(),
}
}
pub fn percent(self) -> Option<f32> {
match self {
// File

View file

@ -3,7 +3,7 @@ use std::{ffi::OsString, future::Future, sync::Arc, time::Duration};
use anyhow::Result;
use futures::{FutureExt, future::BoxFuture};
use parking_lot::Mutex;
use tokio::{select, sync::mpsc::{self, UnboundedReceiver}, task::JoinHandle};
use tokio::{select, sync::{mpsc::{self, UnboundedReceiver}, oneshot}, task::JoinHandle};
use yazi_config::{YAZI, plugin::{Fetcher, Preloader}};
use yazi_dds::Pump;
use yazi_fs::{must_be_dir, path::unique_name, provider, remove_dir_clean};
@ -12,7 +12,7 @@ use yazi_proxy::TasksProxy;
use yazi_shared::{Id, Throttle, url::UrlBuf};
use super::{Ongoing, TaskOp};
use crate::{HIGH, LOW, NORMAL, TaskIn, TaskOps, TaskOut, file::{File, FileInDelete, FileInHardlink, FileInLink, FileInPaste, FileInTrash, FileOutHardlink, FileOutPaste, FileProgDelete, FileProgHardlink, FileProgLink, FileProgPaste, FileProgTrash}, plugin::{Plugin, PluginInEntry, PluginProgEntry}, prework::{Prework, PreworkInFetch, PreworkInLoad, PreworkInSize, PreworkProgFetch, PreworkProgLoad, PreworkProgSize}, process::{Process, ProcessInBg, ProcessInBlock, ProcessInOrphan, ProcessProgBg, ProcessProgBlock, ProcessProgOrphan}};
use crate::{HIGH, LOW, NORMAL, TaskIn, TaskOps, TaskOut, file::{File, FileInDelete, FileInHardlink, FileInLink, FileInPaste, FileInTrash, FileOutDelete, FileOutHardlink, FileOutPaste, FileProgDelete, FileProgHardlink, FileProgLink, FileProgPaste, FileProgTrash}, plugin::{Plugin, PluginInEntry, PluginProgEntry}, prework::{Prework, PreworkInFetch, PreworkInLoad, PreworkInSize, PreworkProgFetch, PreworkProgLoad, PreworkProgSize}, process::{Process, ProcessInBg, ProcessInBlock, ProcessInOrphan, ProcessOutBg, ProcessOutBlock, ProcessOutOrphan, ProcessProgBg, ProcessProgBlock, ProcessProgOrphan}};
pub struct Scheduler {
file: Arc<File>,
@ -59,8 +59,10 @@ impl Scheduler {
pub fn cancel(&self, id: Id) -> bool {
let mut ongoing = self.ongoing.lock();
if let Some(hook) = ongoing.hooks.pop(id) {
self.micro.try_send(hook.call(true), HIGH).ok();
if let Some(hook) = ongoing.hooks.pop(id)
&& let Some(fut) = hook.call(true)
{
self.micro.try_send(fut, HIGH).ok();
return false;
}
@ -82,15 +84,15 @@ impl Scheduler {
}
ongoing.hooks.add_async(id, {
let ongoing = self.ongoing.clone();
let ops = self.ops.clone();
let (from, to) = (from.clone(), to.clone());
move |canceled: bool| async move {
move |canceled| async move {
if !canceled {
remove_dir_clean(&from).await;
Pump::push_move(from, to);
}
ongoing.lock().remove(id);
ops.out(id, FileOutPaste::Clean);
}
});
@ -168,16 +170,16 @@ impl Scheduler {
let id = ongoing.add::<FileProgDelete>(format!("Delete {}", target.display()));
ongoing.hooks.add_async(id, {
let ops = self.ops.clone();
let target = target.clone();
let ongoing = self.ongoing.clone();
move |canceled: bool| async move {
move |canceled| async move {
if !canceled {
provider::remove_dir_all(&target).await.ok();
TasksProxy::update_succeed(&target);
Pump::push_delete(target);
}
ongoing.lock().remove(id);
ops.out(id, FileOutDelete::Clean);
}
});
@ -193,21 +195,18 @@ impl Scheduler {
let mut ongoing = self.ongoing.lock();
let id = ongoing.add::<FileProgTrash>(format!("Trash {}", target.display()));
ongoing.hooks.add_async(id, {
ongoing.hooks.add_sync(id, {
let target = target.clone();
let ongoing = self.ongoing.clone();
move |canceled: bool| async move {
move |canceled| {
if !canceled {
TasksProxy::update_succeed(&target);
Pump::push_trash(target);
}
ongoing.lock().remove(id);
}
});
let file = self.file.clone();
self.send_micro(id, LOW, async move { file.trash(FileInTrash { id, target: target.clone() }) })
self.send_micro(id, LOW, async move { file.trash(FileInTrash { id, target }) })
}
pub fn plugin_micro(&self, opt: PluginOpt) {
@ -223,13 +222,23 @@ impl Scheduler {
self.plugin.r#macro(PluginInEntry { id, opt }).ok();
}
pub fn fetch_paged(&self, fetcher: &'static Fetcher, targets: Vec<yazi_fs::File>) {
let id = self.ongoing.lock().add::<PreworkProgFetch>(format!(
pub fn fetch_paged(
&self,
fetcher: &'static Fetcher,
targets: Vec<yazi_fs::File>,
done: Option<oneshot::Sender<bool>>,
) {
let mut ongoing = self.ongoing.lock();
let id = ongoing.add::<PreworkProgFetch>(format!(
"Run fetcher `{}` with {} target(s)",
fetcher.run.name,
targets.len()
));
if let Some(tx) = done {
ongoing.hooks.add_sync(id, move |canceled| _ = tx.send(canceled));
}
let prework = self.prework.clone();
self.send_micro(id, NORMAL, async move {
prework.fetch(PreworkInFetch { id, plugin: fetcher, targets }).await
@ -275,27 +284,25 @@ impl Scheduler {
};
let mut ongoing = self.ongoing.lock();
let id = if opener.block {
ongoing.add::<ProcessProgBlock>(name)
let (id, clean): (_, TaskOut) = if opener.block {
(ongoing.add::<ProcessProgBlock>(name), ProcessOutBlock::Clean.into())
} else if opener.orphan {
ongoing.add::<ProcessProgOrphan>(name)
(ongoing.add::<ProcessProgOrphan>(name), ProcessOutOrphan::Clean.into())
} else {
ongoing.add::<ProcessProgBg>(name)
(ongoing.add::<ProcessProgBg>(name), ProcessOutBg::Clean.into())
};
let ops = self.ops.clone();
let (cancel_tx, cancel_rx) = mpsc::channel(1);
ongoing.hooks.add_async(id, {
let ongoing = self.ongoing.clone();
move |canceled: bool| async move {
if canceled {
cancel_tx.send(()).await.ok();
cancel_tx.closed().await;
}
if let Some(tx) = done {
tx.send(()).ok();
}
ongoing.lock().remove(id);
ongoing.hooks.add_async(id, move |canceled| async move {
if canceled {
cancel_tx.send(()).await.ok();
cancel_tx.closed().await;
}
if let Some(tx) = done {
tx.send(()).ok();
}
ops.out(id, clean);
});
let cmd = OsString::from(&opener.run);
@ -385,8 +392,10 @@ impl Scheduler {
op.out.reduce(task);
if !task.prog.success() {
continue;
} else if let Some(hook) = ongoing.hooks.pop(op.id) {
micro.try_send(hook.call(false), LOW).ok();
} else if let Some(hook) = ongoing.hooks.pop(op.id)
&& let Some(fut) = hook.call(false)
{
micro.try_send(fut, LOW).ok();
} else {
ongoing.all.remove(&op.id);
}