From 5fe58ba2b1b0999f92ef292991256c151c97b16f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=89=E5=92=B2=E9=9B=85=20misaki=20masa?= Date: Sat, 13 Dec 2025 19:43:29 +0800 Subject: [PATCH] feat: include context in error logs when a task fails (#3427) --- yazi-actor/src/mgr/download.rs | 2 +- yazi-core/src/tasks/plugin.rs | 4 +- yazi-core/src/tasks/prework.rs | 8 +- yazi-dds/src/pump.rs | 23 +- yazi-fm/src/app/commands/plugin.rs | 4 +- yazi-scheduler/src/file/file.rs | 165 ++++++---- yazi-scheduler/src/file/in.rs | 6 + yazi-scheduler/src/file/out.rs | 64 ++-- yazi-scheduler/src/file/progress.rs | 8 +- yazi-scheduler/src/file/traverse.rs | 37 ++- yazi-scheduler/src/hook/hook.rs | 89 ++++++ yazi-scheduler/src/hook/in.rs | 132 ++++++++ yazi-scheduler/src/hook/mod.rs | 1 + yazi-scheduler/src/hooks.rs | 59 ---- yazi-scheduler/src/in.rs | 32 +- yazi-scheduler/src/lib.rs | 4 +- yazi-scheduler/src/macros.rs | 26 ++ yazi-scheduler/src/ongoing.rs | 25 +- yazi-scheduler/src/out.rs | 25 +- yazi-scheduler/src/plugin/plugin.rs | 14 +- yazi-scheduler/src/prework/out.rs | 4 + yazi-scheduler/src/prework/prework.rs | 4 +- yazi-scheduler/src/prework/progress.rs | 5 +- yazi-scheduler/src/process/process.rs | 2 +- yazi-scheduler/src/runner.rs | 80 +++++ yazi-scheduler/src/scheduler.rs | 419 ++++++++++--------------- yazi-scheduler/src/task.rs | 9 +- 27 files changed, 769 insertions(+), 482 deletions(-) create mode 100644 yazi-scheduler/src/hook/hook.rs create mode 100644 yazi-scheduler/src/hook/in.rs create mode 100644 yazi-scheduler/src/hook/mod.rs delete mode 100644 yazi-scheduler/src/hooks.rs create mode 100644 yazi-scheduler/src/runner.rs diff --git a/yazi-actor/src/mgr/download.rs b/yazi-actor/src/mgr/download.rs index cbeb0086..1263f7d8 100644 --- a/yazi-actor/src/mgr/download.rs +++ b/yazi-actor/src/mgr/download.rs @@ -31,7 +31,7 @@ impl Actor for Download { for url in opt.urls { let (tx, rx) = oneshot::channel(); scheduler.file_download(url.to_owned(), Some(tx)); - wg1.push(async move { (rx.await == Ok(false), url) }); + wg1.push(async move { (rx.await == Ok(true), url) }); } let mut wg2 = vec![]; diff --git a/yazi-core/src/tasks/plugin.rs b/yazi-core/src/tasks/plugin.rs index b3f3558f..d8fd904f 100644 --- a/yazi-core/src/tasks/plugin.rs +++ b/yazi-core/src/tasks/plugin.rs @@ -3,7 +3,5 @@ use yazi_parser::app::PluginOpt; use super::Tasks; impl Tasks { - pub fn plugin_micro(&self, opt: PluginOpt) { self.scheduler.plugin_micro(opt); } - - pub fn plugin_macro(&self, opt: PluginOpt) { self.scheduler.plugin_macro(opt); } + pub fn plugin_entry(&self, opt: PluginOpt) { self.scheduler.plugin_entry(opt); } } diff --git a/yazi-core/src/tasks/prework.rs b/yazi-core/src/tasks/prework.rs index 096caf56..2fb6fc88 100644 --- a/yazi-core/src/tasks/prework.rs +++ b/yazi-core/src/tasks/prework.rs @@ -6,7 +6,7 @@ use crate::mgr::Mimetype; impl Tasks { pub fn fetch_paged(&self, paged: &[File], mimetype: &Mimetype) { - let mut loaded = self.scheduler.prework.loaded.lock(); + let mut loaded = self.scheduler.runner.prework.loaded.lock(); let mut tasks: [Vec<_>; MAX_PREWORKERS as usize] = Default::default(); for f in paged { let hash = f.hash_u64(); @@ -29,7 +29,7 @@ impl Tasks { } pub fn preload_paged(&self, paged: &[File], mimetype: &Mimetype) { - let mut loaded = self.scheduler.prework.loaded.lock(); + let mut loaded = self.scheduler.runner.prework.loaded.lock(); for f in paged { let hash = f.hash_u64(); for p in YAZI.plugin.preloaders(f, mimetype.get(&f.url).unwrap_or_default()) { @@ -49,7 +49,7 @@ impl Tasks { } let targets: Vec<_> = { - let loading = self.scheduler.prework.sizing.read(); + let loading = self.scheduler.runner.prework.sizing.read(); targets .iter() .filter(|f| { @@ -62,7 +62,7 @@ impl Tasks { return; } - let mut loading = self.scheduler.prework.sizing.write(); + let mut loading = self.scheduler.runner.prework.sizing.write(); for &target in &targets { loading.insert(target.clone()); } diff --git a/yazi-dds/src/pump.rs b/yazi-dds/src/pump.rs index 6550da09..d8298e9f 100644 --- a/yazi-dds/src/pump.rs +++ b/yazi-dds/src/pump.rs @@ -5,7 +5,7 @@ use tokio::{pin, select, sync::mpsc}; use tokio_stream::{StreamExt, wrappers::UnboundedReceiverStream}; use tokio_util::sync::CancellationToken; use yazi_macro::err; -use yazi_shared::{RoCell, url::UrlBuf}; +use yazi_shared::{RoCell, url::{AsUrl, UrlBuf}}; use crate::{Pubsub, ember::BodyMoveItem}; @@ -17,21 +17,30 @@ static DELETE_TX: Mutex>> = Mutex::new(None pub struct Pump; impl Pump { - pub fn push_move(from: UrlBuf, to: UrlBuf) { + pub fn push_move(from: U, to: U) + where + U: AsUrl, + { if let Some(tx) = &*MOVE_TX.lock() { - tx.send(BodyMoveItem { from, to }).ok(); + tx.send(BodyMoveItem { from: from.as_url().to_owned(), to: to.as_url().to_owned() }).ok(); } } - pub fn push_trash(target: UrlBuf) { + pub fn push_trash(target: U) + where + U: AsUrl, + { if let Some(tx) = &*TRASH_TX.lock() { - tx.send(target).ok(); + tx.send(target.as_url().to_owned()).ok(); } } - pub fn push_delete(target: UrlBuf) { + pub fn push_delete(target: U) + where + U: AsUrl, + { if let Some(tx) = &*DELETE_TX.lock() { - tx.send(target).ok(); + tx.send(target.as_url().to_owned()).ok(); } } diff --git a/yazi-fm/src/app/commands/plugin.rs b/yazi-fm/src/app/commands/plugin.rs index 48bc16ff..d3023b41 100644 --- a/yazi-fm/src/app/commands/plugin.rs +++ b/yazi-fm/src/app/commands/plugin.rs @@ -22,7 +22,7 @@ impl App { } if opt.mode == PluginMode::Async { - succ!(self.core.tasks.plugin_micro(opt)); + succ!(self.core.tasks.plugin_entry(opt)); } else if opt.mode == PluginMode::Sync && hits { return self.plugin_do(opt); } @@ -47,7 +47,7 @@ impl App { } if opt.mode.auto_then(chunk.sync_entry) != PluginMode::Sync { - succ!(self.core.tasks.plugin_micro(opt)); + succ!(self.core.tasks.plugin_entry(opt)); } runtime_mut!(LUA)?.push(&opt.id); diff --git a/yazi-scheduler/src/file/file.rs b/yazi-scheduler/src/file/file.rs index 8976a40a..f48d12af 100644 --- a/yazi-scheduler/src/file/file.rs +++ b/yazi-scheduler/src/file/file.rs @@ -1,16 +1,15 @@ -use std::hash::{BuildHasher, Hash, Hasher}; +use std::{hash::{BuildHasher, Hash, Hasher}, mem}; use anyhow::{Context, Result, anyhow}; -use tokio::{io::{self, ErrorKind::{AlreadyExists, NotFound}}, sync::mpsc}; +use tokio::{io::{self, ErrorKind::NotFound}, sync::mpsc}; use tracing::warn; use yazi_config::YAZI; use yazi_fs::{Cwd, FsHash128, FsUrl, cha::Cha, ok_or_not_found, path::path_relative_to, provider::{Attrs, FileHolder, Provider, local::Local}}; -use yazi_macro::ok_or_not_found; use yazi_shared::{path::PathCow, timestamp_us, url::{AsUrl, UrlBuf, UrlCow, UrlLike}}; -use yazi_vfs::{VfsCha, maybe_exists, provider::{self, DirEntry}, unique_name}; +use yazi_vfs::{VfsCha, maybe_exists, must_be_dir, provider::{self, DirEntry}, unique_name}; use super::{FileInCopy, FileInDelete, FileInHardlink, FileInLink, FileInTrash}; -use crate::{LOW, NORMAL, TaskIn, TaskOp, TaskOps, file::{FileInCut, FileInDownload, FileInUpload, FileOutCopy, FileOutCopyDo, FileOutCut, FileOutCutDo, FileOutDelete, FileOutDeleteDo, FileOutDownload, FileOutDownloadDo, FileOutHardlink, FileOutHardlinkDo, FileOutLink, FileOutTrash, FileOutUpload, FileOutUploadDo}}; +use crate::{LOW, NORMAL, TaskIn, TaskOp, TaskOps, ctx, file::{FileInCut, FileInDownload, FileInUpload, FileOutCopy, FileOutCopyDo, FileOutCut, FileOutCutDo, FileOutDelete, FileOutDeleteDo, FileOutDownload, FileOutDownloadDo, FileOutHardlink, FileOutHardlinkDo, FileOutLink, FileOutTrash, FileOutUpload, FileOutUploadDo}, hook::HookInOutCut, ok_or_not_found}; pub(crate) struct File { ops: TaskOps, @@ -19,19 +18,25 @@ pub(crate) struct File { impl File { pub(crate) fn new( - tx: &mpsc::UnboundedSender, + ops: &mpsc::UnboundedSender, r#macro: &async_priority_channel::Sender, ) -> Self { - Self { ops: tx.into(), r#macro: r#macro.clone() } + Self { ops: ops.into(), r#macro: r#macro.clone() } } - pub(crate) async fn copy(&self, task: FileInCopy) -> Result<(), FileOutCopy> { + pub(crate) async fn copy(&self, mut task: FileInCopy) -> Result<(), FileOutCopy> { let id = task.id; + if !task.force { + task.to = unique_name(task.to, must_be_dir(&task.from)) + .await + .context("Cannot determine unique destination name")?; + } + super::traverse::( task, async |dir| match provider::create_dir(dir).await { - Err(e) if e.kind() != AlreadyExists => Err(e)?, + Err(e) if e.kind() != io::ErrorKind::AlreadyExists => Err(e)?, _ => Ok(()), }, async |task, cha| { @@ -53,8 +58,9 @@ impl File { } pub(crate) async fn copy_do(&self, mut task: FileInCopy) -> Result<(), FileOutCopyDo> { - ok_or_not_found!(provider::remove_file(&task.to).await); - let mut it = provider::copy_with_progress(&task.from, &task.to, task.cha.unwrap()).await?; + ok_or_not_found!(task, provider::remove_file(&task.to).await); + let mut it = + ctx!(task, provider::copy_with_progress(&task.from, &task.to, task.cha.unwrap()).await)?; while let Some(res) = it.recv().await { match res { @@ -74,26 +80,33 @@ impl File { self.ops.out(task.id, FileOutCopyDo::Log(format!("Retrying due to error: {e}"))); return Ok(self.queue(task, LOW)); } - Err(e) => Err(e)?, + Err(e) => ctx!(task, Err(e))?, } } Ok(self.ops.out(task.id, FileOutCopyDo::Succ)) } - pub(crate) async fn cut(&self, task: FileInCut) -> Result<(), FileOutCut> { + pub(crate) async fn cut(&self, mut task: FileInCut) -> Result<(), FileOutCut> { let id = task.id; + if !task.force { + task.to = unique_name(mem::take(&mut task.to), must_be_dir(&task.from)) + .await + .context("Cannot determine unique destination name")?; + } + + self.ops.out(id, HookInOutCut::from(&task)); if !task.follow && ok_or_not_found(provider::rename(&task.from, &task.to).await).is_ok() { return Ok(self.ops.out(id, FileOutCut::Succ)); } let (mut links, mut files) = (vec![], vec![]); - let reorder = task.follow && provider::capabilities(&task.from).await?.symlink; + let reorder = task.follow && ctx!(task, provider::capabilities(&task.from).await)?.symlink; super::traverse::( task, async |dir| match provider::create_dir(dir).await { - Err(e) if e.kind() != AlreadyExists => Err(e)?, + Err(e) if e.kind() != io::ErrorKind::AlreadyExists => Err(e)?, _ => Ok(()), }, |task, cha| { @@ -135,8 +148,9 @@ impl File { } pub(crate) async fn cut_do(&self, mut task: FileInCut) -> Result<(), FileOutCutDo> { - ok_or_not_found!(provider::remove_file(&task.to).await); - let mut it = provider::copy_with_progress(&task.from, &task.to, task.cha.unwrap()).await?; + ok_or_not_found!(task, provider::remove_file(&task.to).await); + let mut it = + ctx!(task, provider::copy_with_progress(&task.from, &task.to, task.cha.unwrap()).await)?; while let Some(res) = it.recv().await { match res { @@ -159,19 +173,26 @@ impl File { self.ops.out(task.id, FileOutCutDo::Log(format!("Retrying due to error: {e}"))); return Ok(self.queue(task, LOW)); } - Err(e) => Err(e)?, + Err(e) => ctx!(task, Err(e))?, } } Ok(self.ops.out(task.id, FileOutCutDo::Succ)) } - pub(crate) fn link(&self, task: FileInLink) -> Result<(), FileOutLink> { + pub(crate) async fn link(&self, mut task: FileInLink) -> Result<(), FileOutLink> { + if !task.force { + task.to = unique_name(task.to, must_be_dir(&task.from)) + .await + .context("Cannot determine unique destination name")?; + } + Ok(self.queue(task, NORMAL)) } pub(crate) async fn link_do(&self, task: FileInLink) -> Result<(), FileOutLink> { let mut src: PathCow = if task.resolve { ok_or_not_found!( + task, provider::read_link(&task.from).await, return Ok(self.ops.out(task.id, FileOutLink::Succ)) ) @@ -181,18 +202,21 @@ impl File { }; if task.relative { - let canon = provider::canonicalize(task.to.parent().unwrap()).await?; - src = path_relative_to(canon.loc(), src)?; + let canon = ctx!(task, provider::canonicalize(task.to.parent().unwrap()).await)?; + src = ctx!(task, path_relative_to(canon.loc(), src))?; } - ok_or_not_found!(provider::remove_file(&task.to).await); - provider::symlink(task.to, src, async || { - Ok(match task.cha { - Some(cha) => cha.is_dir(), - None => Self::cha(&task.from, task.resolve, None).await?.is_dir(), + ok_or_not_found!(task, provider::remove_file(&task.to).await); + ctx!( + task, + provider::symlink(&task.to, src, async || { + Ok(match task.cha { + Some(cha) => cha.is_dir(), + None => Self::cha(&task.from, task.resolve, None).await?.is_dir(), + }) }) - }) - .await?; + .await + )?; if task.delete { provider::remove_file(&task.from).await.ok(); @@ -200,13 +224,19 @@ impl File { Ok(self.ops.out(task.id, FileOutLink::Succ)) } - pub(crate) async fn hardlink(&self, task: FileInHardlink) -> Result<(), FileOutHardlink> { + pub(crate) async fn hardlink(&self, mut task: FileInHardlink) -> Result<(), FileOutHardlink> { let id = task.id; + if !task.force { + task.to = unique_name(task.to, must_be_dir(&task.from)) + .await + .context("Cannot determine unique destination name")?; + } + super::traverse::( task, async |dir| match provider::create_dir(dir).await { - Err(e) if e.kind() != AlreadyExists => Err(e)?, + Err(e) if e.kind() != io::ErrorKind::AlreadyExists => Err(e)?, _ => Ok(()), }, async |task, _cha| { @@ -231,8 +261,8 @@ impl File { UrlCow::from(&task.from) }; - ok_or_not_found!(provider::remove_file(&task.to).await); - ok_or_not_found!(provider::hard_link(&src, &task.to).await); + ok_or_not_found!(task, provider::remove_file(&task.to).await); + ok_or_not_found!(task, provider::hard_link(&src, &task.to).await); Ok(self.ops.out(task.id, FileOutHardlinkDo::Succ)) } @@ -259,17 +289,17 @@ impl File { Ok(()) => {} Err(e) if e.kind() == NotFound => {} Err(_) if !maybe_exists(&task.target).await => {} - Err(e) => Err(e)?, + Err(e) => ctx!(task, Err(e))?, } Ok(self.ops.out(task.id, FileOutDeleteDo::Succ(task.cha.unwrap().len))) } - pub(crate) fn trash(&self, task: FileInTrash) -> Result<(), FileOutTrash> { + pub(crate) async fn trash(&self, task: FileInTrash) -> Result<(), FileOutTrash> { Ok(self.queue(task, LOW)) } pub(crate) async fn trash_do(&self, task: FileInTrash) -> Result<(), FileOutTrash> { - provider::trash(&task.target).await?; + ctx!(task, provider::trash(&task.target).await)?; Ok(self.ops.out(task.id, FileOutTrash::Succ)) } @@ -285,10 +315,10 @@ impl File { }, async |task, cha| { Ok(if cha.is_orphan() { - Err(io::Error::new(NotFound, "Source of symlink doesn't exist"))?; + Err(anyhow!("Failed to work on {task:?}: source of symlink doesn't exist"))? } else { self.ops.out(id, FileOutDownload::New(cha.len)); - self.queue(task, LOW) + self.queue(task, LOW); }) }, |err| { @@ -306,19 +336,19 @@ impl File { ) -> Result<(), FileOutDownloadDo> { let cha = task.cha.unwrap(); - let cache = task.url.cache().context("Cannot determine cache path")?; - let cache_tmp = Self::tmp(&cache).await.context("Cannot determine temporary download cache")?; + let cache = ctx!(task, task.url.cache(), "Cannot determine cache path")?; + let cache_tmp = ctx!(task, Self::tmp(&cache).await, "Cannot determine download cache")?; - let mut it = provider::copy_with_progress(&task.url, &cache_tmp, cha).await?; + let mut it = ctx!(task, provider::copy_with_progress(&task.url, &cache_tmp, cha).await)?; while let Some(res) = it.recv().await { match res { Ok(0) => { Local::regular(&cache).remove_dir_all().await.ok(); - provider::rename(cache_tmp, cache).await.context("Cannot persist downloaded file")?; + ctx!(task, provider::rename(cache_tmp, cache).await, "Cannot persist downloaded file")?; - let lock = task.url.cache_lock().context("Cannot determine cache lock")?; + let lock = ctx!(task, task.url.cache_lock(), "Cannot determine cache lock")?; let hash = format!("{:x}", cha.hash_u128()); - Local::regular(&lock).write(hash).await.context("Cannot lock cache")?; + ctx!(task, Local::regular(&lock).write(hash).await, "Cannot lock cache")?; break; } @@ -337,7 +367,7 @@ impl File { self.ops.out(task.id, FileOutDownloadDo::Log(format!("Retrying due to error: {e}"))); return Ok(self.queue(task, LOW)); } - Err(e) => Err(e)?, + Err(e) => ctx!(task, Err(e))?, } } Ok(self.ops.out(task.id, FileOutDownloadDo::Succ)) @@ -350,7 +380,7 @@ impl File { task, async |_dir| Ok(()), async |task, cha| { - let cache = task.cache.as_ref().context("Cannot determine cache path")?; + let cache = ctx!(task, task.cache.as_ref(), "Cannot determine cache path")?; Ok(match Self::cha(cache, true, None).await { Ok(c) if c.mtime == cha.mtime => {} @@ -359,7 +389,7 @@ impl File { self.queue(task, LOW); } Err(e) if e.kind() == NotFound => {} - Err(e) => Err(e)?, + Err(e) => ctx!(task, Err(e))?, }) }, |err| { @@ -373,42 +403,47 @@ impl File { pub(crate) async fn upload_do(&self, task: FileInUpload) -> Result<(), FileOutUploadDo> { let cha = task.cha.unwrap(); - let cache = task.cache.context("Cannot determine cache path")?; - let lock = task.url.cache_lock().context("Cannot determine cache lock")?; + let cache = ctx!(task, task.cache.as_ref(), "Cannot determine cache path")?; + let lock = ctx!(task, task.url.cache_lock(), "Cannot determine cache lock")?; - let hash = Local::regular(&lock).read_to_string().await.context("Cannot read cache lock")?; - let hash = u128::from_str_radix(&hash, 16).context("Cannot parse hash from lock")?; + let hash = ctx!(task, Local::regular(&lock).read_to_string().await, "Cannot read cache lock")?; + let hash = ctx!(task, u128::from_str_radix(&hash, 16), "Cannot parse hash from lock")?; if hash != cha.hash_u128() { - Err(anyhow!("Remote file has changed since last download"))?; + Err(anyhow!("Failed to work on: {task:?}: remote file has changed since last download"))?; } - let tmp = Self::tmp(&task.url).await.context("Cannot determine temporary upload path")?; - let mut it = provider::copy_with_progress(&cache, &tmp, Attrs { - mode: Some(cha.mode), - atime: None, - btime: None, - mtime: None, - }) - .await?; + let tmp = ctx!(task, Self::tmp(&task.url).await, "Cannot determine temporary upload path")?; + let mut it = ctx!( + task, + provider::copy_with_progress(cache, &tmp, Attrs { + mode: Some(cha.mode), + atime: None, + btime: None, + mtime: None, + }) + .await + )?; while let Some(res) = it.recv().await { match res { Ok(0) => { - let cha = Self::cha(&task.url, true, None).await.context("Cannot stat original file")?; + let cha = + ctx!(task, Self::cha(&task.url, true, None).await, "Cannot stat original file")?; if hash != cha.hash_u128() { - Err(anyhow!("Remote file has changed during upload"))?; + Err(anyhow!("Failed to work on: {task:?}: remote file has changed during upload"))?; } - provider::rename(&tmp, &task.url).await.context("Cannot persist uploaded file")?; + ctx!(task, provider::rename(&tmp, &task.url).await, "Cannot persist uploaded file")?; - let cha = Self::cha(&task.url, true, None).await.context("Cannot stat uploaded file")?; + let cha = + ctx!(task, Self::cha(&task.url, true, None).await, "Cannot stat uploaded file")?; let hash = format!("{:x}", cha.hash_u128()); - Local::regular(&lock).write(hash).await.context("Cannot lock cache")?; + ctx!(task, Local::regular(&lock).write(hash).await, "Cannot lock cache")?; break; } Ok(n) => self.ops.out(task.id, FileOutUploadDo::Adv(n)), - Err(e) => Err(e)?, + Err(e) => ctx!(task, Err(e))?, } } Ok(self.ops.out(task.id, FileOutUploadDo::Succ)) diff --git a/yazi-scheduler/src/file/in.rs b/yazi-scheduler/src/file/in.rs index bee53877..77ccce5a 100644 --- a/yazi-scheduler/src/file/in.rs +++ b/yazi-scheduler/src/file/in.rs @@ -10,6 +10,7 @@ pub(crate) struct FileInCopy { pub(crate) id: Id, pub(crate) from: UrlBuf, pub(crate) to: UrlBuf, + pub(crate) force: bool, pub(crate) cha: Option, pub(crate) follow: bool, pub(crate) retry: u8, @@ -21,6 +22,7 @@ impl FileInCopy { id: self.id, from: self.from, to: self.to, + force: true, cha: self.cha, resolve: true, relative: false, @@ -35,6 +37,7 @@ pub(crate) struct FileInCut { pub(crate) id: Id, pub(crate) from: UrlBuf, pub(crate) to: UrlBuf, + pub(crate) force: bool, pub(crate) cha: Option, pub(crate) follow: bool, pub(crate) retry: u8, @@ -51,6 +54,7 @@ impl FileInCut { id: self.id, from: mem::take(&mut self.from), to: mem::take(&mut self.to), + force: true, cha: self.cha, resolve: true, relative: false, @@ -70,6 +74,7 @@ pub(crate) struct FileInLink { pub(crate) id: Id, pub(crate) from: UrlBuf, pub(crate) to: UrlBuf, + pub(crate) force: bool, pub(crate) cha: Option, pub(crate) resolve: bool, pub(crate) relative: bool, @@ -82,6 +87,7 @@ pub(crate) struct FileInHardlink { pub(crate) id: Id, pub(crate) from: UrlBuf, pub(crate) to: UrlBuf, + pub(crate) force: bool, pub(crate) cha: Option, pub(crate) follow: bool, } diff --git a/yazi-scheduler/src/file/out.rs b/yazi-scheduler/src/file/out.rs index 6e0cdb1d..41273ca9 100644 --- a/yazi-scheduler/src/file/out.rs +++ b/yazi-scheduler/src/file/out.rs @@ -9,6 +9,10 @@ pub(crate) enum FileOutCopy { Fail(String), } +impl From for FileOutCopy { + fn from(value: anyhow::Error) -> Self { Self::Fail(format!("{value:?}")) } +} + impl From for FileOutCopy { fn from(value: std::io::Error) -> Self { Self::Fail(format!("{value:?}")) } } @@ -46,8 +50,8 @@ pub(crate) enum FileOutCopyDo { Fail(String), } -impl From for FileOutCopyDo { - fn from(value: std::io::Error) -> Self { Self::Fail(format!("{value:?}")) } +impl From for FileOutCopyDo { + fn from(value: anyhow::Error) -> Self { Self::Fail(format!("{value:?}")) } } impl FileOutCopyDo { @@ -81,6 +85,10 @@ pub(crate) enum FileOutCut { Clean, } +impl From for FileOutCut { + fn from(value: anyhow::Error) -> Self { Self::Fail(format!("{value:?}")) } +} + impl From for FileOutCut { fn from(value: std::io::Error) -> Self { Self::Fail(format!("{value:?}")) } } @@ -121,8 +129,8 @@ pub(crate) enum FileOutCutDo { Fail(String), } -impl From for FileOutCutDo { - fn from(value: std::io::Error) -> Self { Self::Fail(format!("{value:?}")) } +impl From for FileOutCutDo { + fn from(value: anyhow::Error) -> Self { Self::Fail(format!("{value:?}")) } } impl FileOutCutDo { @@ -157,10 +165,6 @@ impl From for FileOutLink { fn from(value: anyhow::Error) -> Self { Self::Fail(format!("{value:?}")) } } -impl From for FileOutLink { - fn from(value: std::io::Error) -> Self { Self::Fail(format!("{value:?}")) } -} - impl FileOutLink { pub(crate) fn reduce(self, task: &mut Task) { if let TaskProg::FileLink(prog) = &mut task.prog { @@ -206,6 +210,10 @@ pub(crate) enum FileOutHardlink { Fail(String), } +impl From for FileOutHardlink { + fn from(value: anyhow::Error) -> Self { Self::Fail(format!("{value:?}")) } +} + impl From for FileOutHardlink { fn from(value: std::io::Error) -> Self { Self::Fail(format!("{value:?}")) } } @@ -240,8 +248,8 @@ pub(crate) enum FileOutHardlinkDo { Fail(String), } -impl From for FileOutHardlinkDo { - fn from(value: std::io::Error) -> Self { Self::Fail(format!("{value:?}")) } +impl From for FileOutHardlinkDo { + fn from(value: anyhow::Error) -> Self { Self::Fail(format!("{value:?}")) } } impl FileOutHardlinkDo { @@ -268,8 +276,8 @@ pub(crate) enum FileOutDelete { Clean, } -impl From for FileOutDelete { - fn from(value: std::io::Error) -> Self { Self::Fail(format!("{value:?}")) } +impl From for FileOutDelete { + fn from(value: anyhow::Error) -> Self { Self::Fail(format!("{value:?}")) } } impl FileOutDelete { @@ -301,8 +309,8 @@ pub(crate) enum FileOutDeleteDo { Fail(String), } -impl From for FileOutDeleteDo { - fn from(value: std::io::Error) -> Self { Self::Fail(format!("{value:?}")) } +impl From for FileOutDeleteDo { + fn from(value: anyhow::Error) -> Self { Self::Fail(format!("{value:?}")) } } impl FileOutDeleteDo { @@ -326,10 +334,11 @@ impl FileOutDeleteDo { pub(crate) enum FileOutTrash { Succ, Fail(String), + Clean, } -impl From for FileOutTrash { - fn from(value: std::io::Error) -> Self { Self::Fail(format!("{value:?}")) } +impl From for FileOutTrash { + fn from(value: anyhow::Error) -> Self { Self::Fail(format!("{value:?}")) } } impl FileOutTrash { @@ -343,6 +352,9 @@ impl FileOutTrash { prog.state = Some(false); task.log(reason); } + Self::Clean => { + prog.cleaned = true; + } } } } @@ -354,10 +366,11 @@ pub(crate) enum FileOutDownload { Deform(String), Succ, Fail(String), + Clean, } -impl From for FileOutDownload { - fn from(value: std::io::Error) -> Self { Self::Fail(format!("{value:?}")) } +impl From for FileOutDownload { + fn from(value: anyhow::Error) -> Self { Self::Fail(format!("{value:?}")) } } impl FileOutDownload { @@ -380,6 +393,9 @@ impl FileOutDownload { prog.collected = Some(false); task.log(reason); } + Self::Clean => { + prog.cleaned = true; + } } } } @@ -393,10 +409,6 @@ pub(crate) enum FileOutDownloadDo { Fail(String), } -impl From for FileOutDownloadDo { - fn from(value: std::io::Error) -> Self { Self::Fail(format!("{value:?}")) } -} - impl From for FileOutDownloadDo { fn from(value: anyhow::Error) -> Self { Self::Fail(format!("{value:?}")) } } @@ -431,10 +443,6 @@ pub(crate) enum FileOutUpload { Fail(String), } -impl From for FileOutUpload { - fn from(value: std::io::Error) -> Self { Self::Fail(format!("{value:?}")) } -} - impl From for FileOutUpload { fn from(value: anyhow::Error) -> Self { Self::Fail(format!("{value:?}")) } } @@ -471,10 +479,6 @@ pub(crate) enum FileOutUploadDo { Fail(String), } -impl From for FileOutUploadDo { - fn from(value: std::io::Error) -> Self { Self::Fail(format!("{value:?}")) } -} - impl From for FileOutUploadDo { fn from(value: anyhow::Error) -> Self { Self::Fail(format!("{value:?}")) } } diff --git a/yazi-scheduler/src/file/progress.rs b/yazi-scheduler/src/file/progress.rs index 709ece50..0950d8c3 100644 --- a/yazi-scheduler/src/file/progress.rs +++ b/yazi-scheduler/src/file/progress.rs @@ -213,7 +213,8 @@ impl FileProgDelete { // --- Trash #[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize)] pub struct FileProgTrash { - pub state: Option, + pub state: Option, + pub cleaned: bool, } impl From for TaskSummary { @@ -234,7 +235,7 @@ impl FileProgTrash { pub fn failed(self) -> bool { self.state == Some(false) } - pub fn cleaned(self) -> bool { false } + pub fn cleaned(self) -> bool { self.cleaned } pub fn percent(self) -> Option { None } } @@ -248,6 +249,7 @@ pub struct FileProgDownload { pub total_bytes: u64, pub processed_bytes: u64, pub collected: Option, + pub cleaned: bool, } impl From for TaskSummary { @@ -272,7 +274,7 @@ impl FileProgDownload { pub fn failed(self) -> bool { self.collected == Some(false) } - pub fn cleaned(self) -> bool { false } + pub fn cleaned(self) -> bool { self.cleaned } pub fn percent(self) -> Option { Some(if self.success() { diff --git a/yazi-scheduler/src/file/traverse.rs b/yazi-scheduler/src/file/traverse.rs index bd30db80..019f58da 100644 --- a/yazi-scheduler/src/file/traverse.rs +++ b/yazi-scheduler/src/file/traverse.rs @@ -1,10 +1,10 @@ -use std::{collections::VecDeque, fmt::Debug, io}; +use std::{collections::VecDeque, fmt::Debug}; use yazi_fs::{FsUrl, cha::Cha, path::skip_url, provider::{DirReader, FileHolder}}; use yazi_shared::{strand::StrandLike, url::{AsUrl, Url, UrlBuf, UrlLike}}; use yazi_vfs::provider::{self}; -use crate::file::{FileInCopy, FileInCut, FileInDelete, FileInDownload, FileInHardlink, FileInUpload}; +use crate::{ctx, file::{FileInCopy, FileInCut, FileInDelete, FileInDownload, FileInHardlink, FileInUpload}}; trait Traverse { fn cha(&mut self) -> &mut Option; @@ -13,7 +13,7 @@ trait Traverse { fn from(&self) -> Url<'_>; - async fn init(&mut self) -> io::Result { + async fn init(&mut self) -> anyhow::Result { if self.cha().is_none() { *self.cha() = Some(super::File::cha(self.from(), self.follow(), None).await?) } @@ -37,6 +37,7 @@ impl Traverse for FileInCopy { id: self.id, from, to: to.unwrap(), + force: self.force, cha: Some(cha), follow: self.follow, retry: self.retry, @@ -58,6 +59,7 @@ impl Traverse for FileInCut { id: self.id, from, to: to.unwrap(), + force: self.force, cha: Some(cha), follow: self.follow, retry: self.retry, @@ -76,7 +78,14 @@ impl Traverse for FileInHardlink { fn from(&self) -> Url<'_> { self.from.as_url() } fn spawn(&self, from: UrlBuf, to: Option, cha: Cha) -> Self { - Self { id: self.id, from, to: to.unwrap(), cha: Some(cha), follow: self.follow } + Self { + id: self.id, + from, + to: to.unwrap(), + force: self.force, + cha: Some(cha), + follow: self.follow, + } } fn to(&self) -> Option> { Some(self.to.as_url()) } @@ -117,7 +126,7 @@ impl Traverse for FileInUpload { fn from(&self) -> Url<'_> { self.url.as_url() } - async fn init(&mut self) -> io::Result { + async fn init(&mut self) -> anyhow::Result { if self.cha.is_none() { self.cha = Some(super::File::cha(self.from(), self.follow(), None).await?) } @@ -135,21 +144,21 @@ impl Traverse for FileInUpload { } #[allow(private_bounds)] -pub(super) async fn traverse( - mut task: T, +pub(super) async fn traverse( + mut task: I, on_dir: D, mut on_file: FC, on_error: E, -) -> Result<(), R> +) -> Result<(), O> where - R: Debug + From, - T: Traverse, - D: AsyncFn(Url) -> Result<(), R>, - FC: FnMut(T, Cha) -> FR, - FR: Future>, + O: Debug + From, + I: Debug + Traverse, + D: AsyncFn(Url) -> Result<(), O>, + FC: FnMut(I, Cha) -> FR, + FR: Future>, E: Fn(String), { - let cha = task.init().await?; + let cha = ctx!(task, task.init().await)?; if !cha.is_dir() { return on_file(task, cha).await; } diff --git a/yazi-scheduler/src/hook/hook.rs b/yazi-scheduler/src/hook/hook.rs new file mode 100644 index 00000000..91f4bc8f --- /dev/null +++ b/yazi-scheduler/src/hook/hook.rs @@ -0,0 +1,89 @@ +use std::sync::Arc; + +use parking_lot::Mutex; +use tokio::sync::mpsc; +use yazi_dds::Pump; +use yazi_proxy::TasksProxy; +use yazi_vfs::provider; + +use crate::{Ongoing, TaskOp, TaskOps, file::{FileOutCut, FileOutDelete, FileOutDownload, FileOutTrash}, hook::{HookInOutBg, HookInOutBlock, HookInOutCut, HookInOutDelete, HookInOutDownload, HookInOutFetch, HookInOutOrphan, HookInOutTrash}, prework::PreworkOutFetch, process::{ProcessOutBg, ProcessOutBlock, ProcessOutOrphan}}; + +pub(crate) struct Hook { + ops: TaskOps, + ongoing: Arc>, +} + +impl Hook { + pub(crate) fn new(ops: &mpsc::UnboundedSender, ongoing: &Arc>) -> Self { + Self { ops: ops.into(), ongoing: ongoing.clone() } + } + + // --- File + pub(crate) async fn cut(&self, task: HookInOutCut) { + let intact = self.ongoing.lock().intact(task.id); + if intact { + provider::remove_dir_clean(&task.from).await.ok(); + Pump::push_move(&task.from, &task.to); + } + self.ops.out(task.id, FileOutCut::Clean); + } + + pub(crate) async fn delete(&self, task: HookInOutDelete) { + let intact = self.ongoing.lock().intact(task.id); + if intact { + provider::remove_dir_all(&task.target).await.ok(); + TasksProxy::update_succeed(&task.target); + Pump::push_delete(&task.target); + } + self.ops.out(task.id, FileOutDelete::Clean); + } + + pub(crate) async fn trash(&self, task: HookInOutTrash) { + let intact = self.ongoing.lock().intact(task.id); + if intact { + TasksProxy::update_succeed(&task.target); + Pump::push_trash(&task.target); + } + self.ops.out(task.id, FileOutTrash::Clean); + } + + pub(crate) async fn download(&self, task: HookInOutDownload) { + let intact = self.ongoing.lock().intact(task.id); + task.done.send(intact).ok(); + self.ops.out(task.id, FileOutDownload::Clean); + } + + // --- Process + pub(crate) async fn block(&self, task: HookInOutBlock) { + if let Some(tx) = task.done { + tx.send(()).ok(); + } + self.ops.out(task.id, ProcessOutBlock::Clean); + } + + pub(crate) async fn orphan(&self, task: HookInOutOrphan) { + if let Some(tx) = task.done { + tx.send(()).ok(); + } + self.ops.out(task.id, ProcessOutOrphan::Clean); + } + + pub(crate) async fn bg(&self, task: HookInOutBg) { + let intact = self.ongoing.lock().intact(task.id); + if !intact { + task.cancel.send(()).await.ok(); + task.cancel.closed().await; + } + if let Some(tx) = task.done { + tx.send(()).ok(); + } + self.ops.out(task.id, ProcessOutBg::Clean); + } + + // --- Prework + pub(crate) async fn fetch(&self, task: HookInOutFetch) { + let intact = self.ongoing.lock().intact(task.id); + task.done.send(intact).ok(); + self.ops.out(task.id, PreworkOutFetch::Clean); + } +} diff --git a/yazi-scheduler/src/hook/in.rs b/yazi-scheduler/src/hook/in.rs new file mode 100644 index 00000000..e5549d85 --- /dev/null +++ b/yazi-scheduler/src/hook/in.rs @@ -0,0 +1,132 @@ +use tokio::sync::{mpsc, oneshot}; +use yazi_shared::{Id, url::UrlBuf}; + +use crate::{Task, TaskProg, file::FileInCut}; + +// --- Cut +#[derive(Debug)] +pub(crate) struct HookInOutCut { + pub(crate) id: Id, + pub(crate) from: UrlBuf, + pub(crate) to: UrlBuf, +} + +impl From<&FileInCut> for HookInOutCut { + fn from(value: &FileInCut) -> Self { + Self { id: value.id, from: value.from.clone(), to: value.to.clone() } + } +} + +impl HookInOutCut { + pub(crate) fn reduce(self, task: &mut Task) { + if let TaskProg::FileCut(_) = &task.prog { + task.hook = Some(self.into()); + } + } +} + +// --- Delete +#[derive(Debug)] +pub(crate) struct HookInOutDelete { + pub(crate) id: Id, + pub(crate) target: UrlBuf, +} + +impl HookInOutDelete { + pub(crate) fn reduce(self, task: &mut Task) { + if let TaskProg::FileDelete(_) = &task.prog { + task.hook = Some(self.into()); + } + } +} + +// --- Trash +#[derive(Debug)] +pub(crate) struct HookInOutTrash { + pub(crate) id: Id, + pub(crate) target: UrlBuf, +} + +impl HookInOutTrash { + pub(crate) fn reduce(self, task: &mut Task) { + if let TaskProg::FileTrash(_) = &task.prog { + task.hook = Some(self.into()); + } + } +} + +// --- Download +#[derive(Debug)] +pub(crate) struct HookInOutDownload { + pub(crate) id: Id, + pub(crate) done: oneshot::Sender, +} + +impl HookInOutDownload { + pub(crate) fn reduce(self, task: &mut Task) { + if let TaskProg::FileDownload(_) = &task.prog { + task.hook = Some(self.into()); + } + } +} + +// --- Fetch +#[derive(Debug)] +pub(crate) struct HookInOutFetch { + pub(crate) id: Id, + pub(crate) done: oneshot::Sender, +} + +impl HookInOutFetch { + pub(crate) fn reduce(self, task: &mut Task) { + if let TaskProg::PreworkFetch(_) = &task.prog { + task.hook = Some(self.into()); + } + } +} + +// --- Block +#[derive(Debug)] +pub(crate) struct HookInOutBlock { + pub(crate) id: Id, + pub(crate) done: Option>, +} + +impl HookInOutBlock { + pub(crate) fn reduce(self, task: &mut Task) { + if let TaskProg::ProcessBlock(_) = &task.prog { + task.hook = Some(self.into()); + } + } +} + +// --- Orphan +#[derive(Debug)] +pub(crate) struct HookInOutOrphan { + pub(crate) id: Id, + pub(crate) done: Option>, +} + +impl HookInOutOrphan { + pub(crate) fn reduce(self, task: &mut Task) { + if let TaskProg::ProcessOrphan(_) = &task.prog { + task.hook = Some(self.into()); + } + } +} + +// --- Bg +#[derive(Debug)] +pub(crate) struct HookInOutBg { + pub(crate) id: Id, + pub(crate) cancel: mpsc::Sender<()>, + pub(crate) done: Option>, +} + +impl HookInOutBg { + pub(crate) fn reduce(self, task: &mut Task) { + if let TaskProg::ProcessBg(_) = &task.prog { + task.hook = Some(self.into()); + } + } +} diff --git a/yazi-scheduler/src/hook/mod.rs b/yazi-scheduler/src/hook/mod.rs new file mode 100644 index 00000000..716fed04 --- /dev/null +++ b/yazi-scheduler/src/hook/mod.rs @@ -0,0 +1 @@ +yazi_macro::mod_flat!(hook r#in); diff --git a/yazi-scheduler/src/hooks.rs b/yazi-scheduler/src/hooks.rs deleted file mode 100644 index 994a2147..00000000 --- a/yazi-scheduler/src/hooks.rs +++ /dev/null @@ -1,59 +0,0 @@ -use futures::{FutureExt, future::BoxFuture}; -use hashbrown::HashMap; -use yazi_shared::Id; - -#[derive(Default)] -pub(super) struct Hooks { - inner: HashMap>, -} - -impl Hooks { - #[inline] - pub(super) fn add_sync(&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(&mut self, id: Id, f: F) - where - F: FnOnce(bool) -> Fut + Send + 'static, - Fut: Future + Send + 'static, - { - self.inner.insert(id, Box::new(AsyncHook(f))); - } - - #[inline] - pub(super) fn pop(&mut self, id: Id) -> Option> { self.inner.remove(&id) } -} - -// --- Hook -pub(super) trait Hook: Send { - fn call(self: Box, cancel: bool) -> Option>; -} - -struct SyncHook(F); - -impl Hook for SyncHook -where - F: FnOnce(bool) + Send, -{ - fn call(self: Box, cancel: bool) -> Option> { - (self.0)(cancel); - None - } -} - -struct AsyncHook(F); - -impl Hook for AsyncHook -where - F: FnOnce(bool) -> Fut + Send, - Fut: Future + Send + 'static, -{ - fn call(self: Box, cancel: bool) -> Option> { - Some((self.0)(cancel).boxed()) - } -} diff --git a/yazi-scheduler/src/in.rs b/yazi-scheduler/src/in.rs index d008d1c8..09bb7b8c 100644 --- a/yazi-scheduler/src/in.rs +++ b/yazi-scheduler/src/in.rs @@ -1,6 +1,6 @@ use yazi_shared::Id; -use crate::{file::{FileInCopy, FileInCut, FileInDelete, FileInDownload, FileInHardlink, FileInLink, FileInTrash, FileInUpload}, impl_from_in, plugin::PluginInEntry, prework::{PreworkInFetch, PreworkInLoad, PreworkInSize}}; +use crate::{file::{FileInCopy, FileInCut, FileInDelete, FileInDownload, FileInHardlink, FileInLink, FileInTrash, FileInUpload}, hook::{HookInOutBg, HookInOutBlock, HookInOutCut, HookInOutDelete, HookInOutDownload, HookInOutFetch, HookInOutOrphan, HookInOutTrash}, impl_from_in, plugin::PluginInEntry, prework::{PreworkInFetch, PreworkInLoad, PreworkInSize}, process::{ProcessInBg, ProcessInBlock, ProcessInOrphan}}; #[derive(Debug)] pub(crate) enum TaskIn { @@ -19,6 +19,19 @@ pub(crate) enum TaskIn { PreworkFetch(PreworkInFetch), PreworkLoad(PreworkInLoad), PreworkSize(PreworkInSize), + // Process + ProcessBlock(ProcessInBlock), + ProcessOrphan(ProcessInOrphan), + ProcessBg(ProcessInBg), + // Hook + HookCut(HookInOutCut), + HookDelete(HookInOutDelete), + HookTrash(HookInOutTrash), + HookDownload(HookInOutDownload), + HookBlock(HookInOutBlock), + HookOrphan(HookInOutOrphan), + HookBg(HookInOutBg), + HookFetch(HookInOutFetch), } impl_from_in! { @@ -28,6 +41,10 @@ impl_from_in! { PluginEntry(PluginInEntry), // Prework PreworkFetch(PreworkInFetch), PreworkLoad(PreworkInLoad), PreworkSize(PreworkInSize), + // Process + ProcessBlock(ProcessInBlock), ProcessOrphan(ProcessInOrphan), ProcessBg(ProcessInBg), + // Hook + HookCut(HookInOutCut), HookDelete(HookInOutDelete), HookTrash(HookInOutTrash), HookDownload(HookInOutDownload), HookBlock(HookInOutBlock), HookOrphan(HookInOutOrphan), HookBg(HookInOutBg), HookFetch(HookInOutFetch), } impl TaskIn { @@ -48,6 +65,19 @@ impl TaskIn { Self::PreworkFetch(r#in) => r#in.id, Self::PreworkLoad(r#in) => r#in.id, Self::PreworkSize(r#in) => r#in.id, + // Process + Self::ProcessBlock(r#in) => r#in.id, + Self::ProcessOrphan(r#in) => r#in.id, + Self::ProcessBg(r#in) => r#in.id, + // Hook + Self::HookCut(r#in) => r#in.id, + Self::HookDelete(r#in) => r#in.id, + Self::HookTrash(r#in) => r#in.id, + Self::HookDownload(r#in) => r#in.id, + Self::HookBlock(r#in) => r#in.id, + Self::HookOrphan(r#in) => r#in.id, + Self::HookBg(r#in) => r#in.id, + Self::HookFetch(r#in) => r#in.id, } } } diff --git a/yazi-scheduler/src/lib.rs b/yazi-scheduler/src/lib.rs index 4120f1a8..268f6a11 100644 --- a/yazi-scheduler/src/lib.rs +++ b/yazi-scheduler/src/lib.rs @@ -1,8 +1,8 @@ mod macros; -yazi_macro::mod_pub!(file plugin prework process); +yazi_macro::mod_pub!(file hook plugin prework process); -yazi_macro::mod_flat!(hooks ongoing op out progress r#in scheduler snap task); +yazi_macro::mod_flat!(ongoing op out progress r#in runner scheduler snap 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/macros.rs b/yazi-scheduler/src/macros.rs index 8d23fdd4..61ee88f3 100644 --- a/yazi-scheduler/src/macros.rs +++ b/yazi-scheduler/src/macros.rs @@ -1,3 +1,29 @@ +#[macro_export] +macro_rules! ctx { + ($task:ident, $result:expr) => {{ + use anyhow::Context; + $result.with_context(|| format!("Failed to work on {:?}", $task)) + }}; + ($task:ident, $result:expr, $($args:tt)*) => {{ + use anyhow::Context; + $result.with_context(|| format!("Failed to work on {:?}: {}", $task, format_args!($($args)*))) + }}; +} + +#[macro_export] +macro_rules! ok_or_not_found { + ($task:ident, $result:expr, $not_found:expr) => { + match $result { + Ok(v) => v, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => $not_found, + Err(e) => ctx!($task, Err(e))?, + } + }; + ($task:ident, $result:expr) => { + ok_or_not_found!($task, $result, Default::default()) + }; +} + #[macro_export] macro_rules! impl_from_in { ($($variant:ident($type:ty)),* $(,)?) => { diff --git a/yazi-scheduler/src/ongoing.rs b/yazi-scheduler/src/ongoing.rs index 285632db..9fec36b0 100644 --- a/yazi-scheduler/src/ongoing.rs +++ b/yazi-scheduler/src/ongoing.rs @@ -5,47 +5,48 @@ use yazi_parser::app::TaskSummary; use yazi_shared::{Id, Ids}; use super::Task; -use crate::{Hooks, TaskProg}; +use crate::TaskProg; #[derive(Default)] pub struct Ongoing { - pub(super) hooks: Hooks, - pub(super) all: HashMap, + pub(super) inner: HashMap, } impl Ongoing { - pub(super) fn add(&mut self, name: String) -> Id + pub(super) fn add(&mut self, name: String) -> &mut Task where T: Into + Default, { static IDS: Ids = Ids::new(); let id = IDS.next(); - self.all.insert(id, Task::new::(id, name)); - id + self.inner.entry(id).insert(Task::new::(id, name)).into_mut() } #[inline] - pub fn get_mut(&mut self, id: Id) -> Option<&mut Task> { self.all.get_mut(&id) } + pub fn get_mut(&mut self, id: Id) -> Option<&mut Task> { self.inner.get_mut(&id) } pub fn get_id(&self, idx: usize) -> Option { self.values().nth(idx).map(|t| t.id) } pub fn len(&self) -> usize { if YAZI.tasks.suppress_preload { - self.all.values().filter(|&t| t.prog.is_user()).count() + self.inner.values().filter(|&t| t.prog.is_user()).count() } else { - self.all.len() + self.inner.len() } } #[inline] - pub fn exists(&self, id: Id) -> bool { self.all.contains_key(&id) } + pub fn exists(&self, id: Id) -> bool { self.inner.contains_key(&id) } + + #[inline] + pub fn intact(&self, id: Id) -> bool { self.inner.get(&id).is_some_and(|t| !t.canceled) } pub fn values(&self) -> Box + '_> { if YAZI.tasks.suppress_preload { - Box::new(self.all.values().filter(|&t| t.prog.is_user())) + Box::new(self.inner.values().filter(|&t| t.prog.is_user())) } else { - Box::new(self.all.values()) + Box::new(self.inner.values()) } } diff --git a/yazi-scheduler/src/out.rs b/yazi-scheduler/src/out.rs index b428a7e3..de4cf878 100644 --- a/yazi-scheduler/src/out.rs +++ b/yazi-scheduler/src/out.rs @@ -1,4 +1,4 @@ -use crate::{Task, file::{FileOutCopy, FileOutCopyDo, FileOutCut, FileOutCutDo, FileOutDelete, FileOutDeleteDo, FileOutDownload, FileOutDownloadDo, FileOutHardlink, FileOutHardlinkDo, FileOutLink, FileOutTrash, FileOutUpload, FileOutUploadDo}, impl_from_out, plugin::PluginOutEntry, prework::{PreworkOutFetch, PreworkOutLoad, PreworkOutSize}, process::{ProcessOutBg, ProcessOutBlock, ProcessOutOrphan}}; +use crate::{Task, file::{FileOutCopy, FileOutCopyDo, FileOutCut, FileOutCutDo, FileOutDelete, FileOutDeleteDo, FileOutDownload, FileOutDownloadDo, FileOutHardlink, FileOutHardlinkDo, FileOutLink, FileOutTrash, FileOutUpload, FileOutUploadDo}, hook::{HookInOutBg, HookInOutBlock, HookInOutCut, HookInOutDelete, HookInOutDownload, HookInOutFetch, HookInOutOrphan, HookInOutTrash}, impl_from_out, plugin::PluginOutEntry, prework::{PreworkOutFetch, PreworkOutLoad, PreworkOutSize}, process::{ProcessOutBg, ProcessOutBlock, ProcessOutOrphan}}; #[derive(Debug)] pub(super) enum TaskOut { @@ -17,19 +17,25 @@ pub(super) enum TaskOut { FileDownloadDo(FileOutDownloadDo), FileUpload(FileOutUpload), FileUploadDo(FileOutUploadDo), - // Plugin PluginEntry(PluginOutEntry), - // Prework PreworkFetch(PreworkOutFetch), PreworkLoad(PreworkOutLoad), PreworkSize(PreworkOutSize), - // Process ProcessBlock(ProcessOutBlock), ProcessOrphan(ProcessOutOrphan), ProcessBg(ProcessOutBg), + // Hook + HookCut(HookInOutCut), + HookDelete(HookInOutDelete), + HookTrash(HookInOutTrash), + HookDownload(HookInOutDownload), + HookBlock(HookInOutBlock), + HookOrphan(HookInOutOrphan), + HookBg(HookInOutBg), + HookFetch(HookInOutFetch), } impl_from_out! { @@ -41,6 +47,8 @@ impl_from_out! { PreworkFetch(PreworkOutFetch), PreworkLoad(PreworkOutLoad), PreworkSize(PreworkOutSize), // Process ProcessBlock(ProcessOutBlock), ProcessOrphan(ProcessOutOrphan), ProcessBg(ProcessOutBg), + // Hook + HookCut(HookInOutCut), HookDelete(HookInOutDelete), HookTrash(HookInOutTrash), HookDownload(HookInOutDownload), HookBlock(HookInOutBlock), HookOrphan(HookInOutOrphan), HookBg(HookInOutBg), HookFetch(HookInOutFetch), } impl TaskOut { @@ -71,6 +79,15 @@ impl TaskOut { Self::ProcessBlock(out) => out.reduce(task), Self::ProcessOrphan(out) => out.reduce(task), Self::ProcessBg(out) => out.reduce(task), + // Hook + Self::HookCut(out) => out.reduce(task), + Self::HookDelete(out) => out.reduce(task), + Self::HookTrash(out) => out.reduce(task), + Self::HookDownload(out) => out.reduce(task), + Self::HookBlock(out) => out.reduce(task), + Self::HookOrphan(out) => out.reduce(task), + Self::HookBg(out) => out.reduce(task), + Self::HookFetch(out) => out.reduce(task), } } } diff --git a/yazi-scheduler/src/plugin/plugin.rs b/yazi-scheduler/src/plugin/plugin.rs index f9d9a0d1..49cf3db0 100644 --- a/yazi-scheduler/src/plugin/plugin.rs +++ b/yazi-scheduler/src/plugin/plugin.rs @@ -12,23 +12,17 @@ pub(crate) struct Plugin { impl Plugin { pub(crate) fn new( - tx: &mpsc::UnboundedSender, + ops: &mpsc::UnboundedSender, r#macro: &async_priority_channel::Sender, ) -> Self { - Self { ops: tx.into(), r#macro: r#macro.clone() } + Self { ops: ops.into(), r#macro: r#macro.clone() } } - pub(crate) async fn micro(&self, task: PluginInEntry) -> Result<(), PluginOutEntry> { - isolate::entry(task.opt).await?; - self.ops.out(task.id, PluginOutEntry::Succ); - Ok(()) - } - - pub(crate) fn r#macro(&self, task: PluginInEntry) -> Result<(), PluginOutEntry> { + pub(crate) async fn entry(&self, task: PluginInEntry) -> Result<(), PluginOutEntry> { Ok(self.queue(task, HIGH)) } - pub(crate) async fn macro_do(&self, task: PluginInEntry) -> Result<(), PluginOutEntry> { + pub(crate) async fn entry_do(&self, task: PluginInEntry) -> Result<(), PluginOutEntry> { isolate::entry(task.opt).await?; Ok(self.ops.out(task.id, PluginOutEntry::Succ)) } diff --git a/yazi-scheduler/src/prework/out.rs b/yazi-scheduler/src/prework/out.rs index 762ae06d..13724bc6 100644 --- a/yazi-scheduler/src/prework/out.rs +++ b/yazi-scheduler/src/prework/out.rs @@ -5,6 +5,7 @@ use crate::{Task, TaskProg}; pub(crate) enum PreworkOutFetch { Succ, Fail(String), + Clean, } impl From for PreworkOutFetch { @@ -22,6 +23,9 @@ impl PreworkOutFetch { prog.state = Some(false); task.log(reason); } + Self::Clean => { + prog.cleaned = true; + } } } } diff --git a/yazi-scheduler/src/prework/prework.rs b/yazi-scheduler/src/prework/prework.rs index e87eabe5..2c1cf697 100644 --- a/yazi-scheduler/src/prework/prework.rs +++ b/yazi-scheduler/src/prework/prework.rs @@ -27,11 +27,11 @@ pub struct Prework { impl Prework { pub(crate) fn new( - tx: &mpsc::UnboundedSender, + ops: &mpsc::UnboundedSender, r#macro: &async_priority_channel::Sender, ) -> Self { Self { - ops: tx.into(), + ops: ops.into(), r#macro: r#macro.clone(), loaded: Mutex::new(LruCache::new(NonZeroUsize::new(4096).unwrap())), loading: Mutex::new(LruCache::new(NonZeroUsize::new(256).unwrap())), diff --git a/yazi-scheduler/src/prework/progress.rs b/yazi-scheduler/src/prework/progress.rs index 9174ef82..694abbee 100644 --- a/yazi-scheduler/src/prework/progress.rs +++ b/yazi-scheduler/src/prework/progress.rs @@ -4,7 +4,8 @@ use yazi_parser::app::TaskSummary; // --- Fetch #[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize)] pub struct PreworkProgFetch { - pub state: Option, + pub state: Option, + pub cleaned: bool, } impl From for TaskSummary { @@ -25,7 +26,7 @@ impl PreworkProgFetch { pub fn failed(self) -> bool { self.state == Some(false) } - pub fn cleaned(self) -> bool { false } + pub fn cleaned(self) -> bool { self.cleaned } pub fn percent(self) -> Option { None } } diff --git a/yazi-scheduler/src/process/process.rs b/yazi-scheduler/src/process/process.rs index 9a5d28f9..387472cf 100644 --- a/yazi-scheduler/src/process/process.rs +++ b/yazi-scheduler/src/process/process.rs @@ -11,7 +11,7 @@ pub(crate) struct Process { } impl Process { - pub(crate) fn new(tx: &mpsc::UnboundedSender) -> Self { Self { ops: tx.into() } } + pub(crate) fn new(ops: &mpsc::UnboundedSender) -> Self { Self { ops: ops.into() } } pub(crate) async fn block(&self, task: ProcessInBlock) -> Result<(), ProcessOutBlock> { let _permit = HIDER.acquire().await.unwrap(); diff --git a/yazi-scheduler/src/runner.rs b/yazi-scheduler/src/runner.rs new file mode 100644 index 00000000..23e9c613 --- /dev/null +++ b/yazi-scheduler/src/runner.rs @@ -0,0 +1,80 @@ +use std::sync::Arc; + +use crate::{TaskIn, TaskOut, file::File, hook::Hook, plugin::Plugin, prework::Prework, process::Process}; + +#[derive(Clone)] +pub struct Runner { + pub(super) file: Arc, + pub(super) plugin: Arc, + pub prework: Arc, + pub(super) process: Arc, + pub(super) hook: Arc, +} + +impl Runner { + pub(super) async fn micro(&self, r#in: TaskIn) -> Result<(), TaskOut> { + match r#in { + // File + TaskIn::FileCopy(r#in) => self.file.copy(r#in).await.map_err(Into::into), + TaskIn::FileCut(r#in) => self.file.cut(r#in).await.map_err(Into::into), + TaskIn::FileLink(r#in) => self.file.link(r#in).await.map_err(Into::into), + TaskIn::FileHardlink(r#in) => self.file.hardlink(r#in).await.map_err(Into::into), + TaskIn::FileDelete(r#in) => self.file.delete(r#in).await.map_err(Into::into), + TaskIn::FileTrash(r#in) => self.file.trash(r#in).await.map_err(Into::into), + TaskIn::FileDownload(r#in) => self.file.download(r#in).await.map_err(Into::into), + TaskIn::FileUpload(r#in) => self.file.upload(r#in).await.map_err(Into::into), + // Plugin + TaskIn::PluginEntry(r#in) => self.plugin.entry(r#in).await.map_err(Into::into), + // Prework + TaskIn::PreworkFetch(r#in) => self.prework.fetch(r#in).await.map_err(Into::into), + TaskIn::PreworkLoad(r#in) => self.prework.load(r#in).await.map_err(Into::into), + TaskIn::PreworkSize(r#in) => self.prework.size(r#in).await.map_err(Into::into), + // Process + TaskIn::ProcessBlock(r#in) => self.process.block(r#in).await.map_err(Into::into), + TaskIn::ProcessOrphan(r#in) => self.process.orphan(r#in).await.map_err(Into::into), + TaskIn::ProcessBg(r#in) => self.process.bg(r#in).await.map_err(Into::into), + // Hook + TaskIn::HookCut(r#in) => Ok(self.hook.cut(r#in).await), + TaskIn::HookDelete(r#in) => Ok(self.hook.delete(r#in).await), + TaskIn::HookTrash(r#in) => Ok(self.hook.trash(r#in).await), + TaskIn::HookDownload(r#in) => Ok(self.hook.download(r#in).await), + TaskIn::HookBlock(r#in) => Ok(self.hook.block(r#in).await), + TaskIn::HookOrphan(r#in) => Ok(self.hook.orphan(r#in).await), + TaskIn::HookBg(r#in) => Ok(self.hook.bg(r#in).await), + TaskIn::HookFetch(r#in) => Ok(self.hook.fetch(r#in).await), + } + } + + pub(super) async fn r#macro(&self, r#in: TaskIn) -> Result<(), TaskOut> { + match r#in { + // File + TaskIn::FileCopy(r#in) => self.file.copy_do(r#in).await.map_err(Into::into), + TaskIn::FileCut(r#in) => self.file.cut_do(r#in).await.map_err(Into::into), + TaskIn::FileLink(r#in) => self.file.link_do(r#in).await.map_err(Into::into), + TaskIn::FileHardlink(r#in) => self.file.hardlink_do(r#in).await.map_err(Into::into), + TaskIn::FileDelete(r#in) => self.file.delete_do(r#in).await.map_err(Into::into), + TaskIn::FileTrash(r#in) => self.file.trash_do(r#in).await.map_err(Into::into), + TaskIn::FileDownload(r#in) => self.file.download_do(r#in).await.map_err(Into::into), + TaskIn::FileUpload(r#in) => self.file.upload_do(r#in).await.map_err(Into::into), + // Plugin + TaskIn::PluginEntry(r#in) => self.plugin.entry_do(r#in).await.map_err(Into::into), + // Prework + TaskIn::PreworkFetch(r#in) => self.prework.fetch_do(r#in).await.map_err(Into::into), + TaskIn::PreworkLoad(r#in) => self.prework.load_do(r#in).await.map_err(Into::into), + TaskIn::PreworkSize(r#in) => self.prework.size_do(r#in).await.map_err(Into::into), + // Process + TaskIn::ProcessBlock(_in) => unreachable!(), + TaskIn::ProcessOrphan(_in) => unreachable!(), + TaskIn::ProcessBg(_in) => unreachable!(), + // Hook + TaskIn::HookCut(_in) => unreachable!(), + TaskIn::HookDelete(_in) => unreachable!(), + TaskIn::HookTrash(_in) => unreachable!(), + TaskIn::HookDownload(_in) => unreachable!(), + TaskIn::HookBlock(_in) => unreachable!(), + TaskIn::HookOrphan(_in) => unreachable!(), + TaskIn::HookBg(_in) => unreachable!(), + TaskIn::HookFetch(_in) => unreachable!(), + } + } +} diff --git a/yazi-scheduler/src/scheduler.rs b/yazi-scheduler/src/scheduler.rs index 1d489bc4..57af5561 100644 --- a/yazi-scheduler/src/scheduler.rs +++ b/yazi-scheduler/src/scheduler.rs @@ -1,27 +1,19 @@ -use std::{future::Future, sync::Arc, time::Duration}; +use std::{sync::Arc, time::Duration}; -use anyhow::Result; -use futures::{FutureExt, future::BoxFuture}; +use hashbrown::hash_map::RawEntryMut; use parking_lot::Mutex; use tokio::{select, sync::{mpsc::{self, UnboundedReceiver}, oneshot}, task::JoinHandle}; use yazi_config::{YAZI, plugin::{Fetcher, Preloader}}; -use yazi_dds::Pump; use yazi_parser::{app::PluginOpt, tasks::ProcessOpenOpt}; -use yazi_proxy::TasksProxy; use yazi_shared::{Id, Throttle, url::{UrlBuf, UrlLike}}; -use yazi_vfs::{must_be_dir, provider, unique_name}; use super::{Ongoing, TaskOp}; -use crate::{HIGH, LOW, NORMAL, TaskIn, TaskOps, TaskOut, file::{File, FileInCopy, FileInCut, FileInDelete, FileInDownload, FileInHardlink, FileInLink, FileInTrash, FileInUpload, FileOutCopy, FileOutCut, FileOutDelete, FileOutDownload, FileOutHardlink, FileOutUpload, FileProgCopy, FileProgCut, FileProgDelete, FileProgDownload, FileProgHardlink, FileProgLink, FileProgTrash, FileProgUpload}, plugin::{Plugin, PluginInEntry, PluginProgEntry}, prework::{Prework, PreworkInFetch, PreworkInLoad, PreworkInSize, PreworkProgFetch, PreworkProgLoad, PreworkProgSize}, process::{Process, ProcessInBg, ProcessInBlock, ProcessInOrphan, ProcessOutBg, ProcessOutBlock, ProcessOutOrphan, ProcessProgBg, ProcessProgBlock, ProcessProgOrphan}}; +use crate::{HIGH, LOW, NORMAL, Runner, TaskIn, TaskOps, file::{File, FileInCopy, FileInCut, FileInDelete, FileInDownload, FileInHardlink, FileInLink, FileInTrash, FileInUpload, FileOutCopy, FileOutCut, FileOutDownload, FileOutHardlink, FileOutUpload, FileProgCopy, FileProgCut, FileProgDelete, FileProgDownload, FileProgHardlink, FileProgLink, FileProgTrash, FileProgUpload}, hook::{Hook, HookInOutBg, HookInOutBlock, HookInOutDelete, HookInOutDownload, HookInOutFetch, HookInOutOrphan, HookInOutTrash}, plugin::{Plugin, PluginInEntry, PluginProgEntry}, prework::{Prework, PreworkInFetch, PreworkInLoad, PreworkInSize, PreworkProgFetch, PreworkProgLoad, PreworkProgSize}, process::{Process, ProcessInBg, ProcessInBlock, ProcessInOrphan, ProcessProgBg, ProcessProgBlock, ProcessProgOrphan}}; pub struct Scheduler { - file: Arc, - plugin: Arc, - pub prework: Arc, - process: Arc, - ops: TaskOps, - micro: async_priority_channel::Sender, u8>, + pub runner: Runner, + micro: async_priority_channel::Sender, handles: Vec>, pub ongoing: Arc>, } @@ -31,19 +23,24 @@ impl Scheduler { let (op_tx, op_rx) = mpsc::unbounded_channel(); let (micro_tx, micro_rx) = async_priority_channel::unbounded(); let (macro_tx, macro_rx) = async_priority_channel::unbounded(); + let ongoing = Arc::new(Mutex::new(Ongoing::default())); - let mut scheduler = Self { + let runner = Runner { file: Arc::new(File::new(&op_tx, ¯o_tx)), plugin: Arc::new(Plugin::new(&op_tx, ¯o_tx)), prework: Arc::new(Prework::new(&op_tx, ¯o_tx)), process: Arc::new(Process::new(&op_tx)), + hook: Arc::new(Hook::new(&op_tx, &ongoing)), + }; - ops: TaskOps(op_tx), - micro: micro_tx, + let mut scheduler = Self { + ops: TaskOps(op_tx), + runner, + micro: micro_tx, handles: Vec::with_capacity( YAZI.tasks.micro_workers as usize + YAZI.tasks.macro_workers as usize + 1, ), - ongoing: Default::default(), + ongoing, }; for _ in 0..YAZI.tasks.micro_workers { @@ -59,14 +56,20 @@ impl Scheduler { pub fn cancel(&self, id: Id) -> bool { let mut ongoing = self.ongoing.lock(); - if let Some(hook) = ongoing.hooks.pop(id) - && let Some(fut) = hook.call(true) - { - self.micro.try_send(fut, HIGH).ok(); - return false; + match ongoing.inner.raw_entry_mut().from_key(&id) { + RawEntryMut::Occupied(mut oe) => { + let task = oe.get_mut(); + if let Some(hook) = task.hook.take() { + task.canceled = true; + self.micro.try_send(hook, HIGH).ok(); + false + } else { + oe.remove(); + true + } + } + RawEntryMut::Vacant(_) => false, } - - ongoing.all.remove(&id).is_some() } pub fn shutdown(&self) { @@ -75,189 +78,127 @@ impl Scheduler { } } - pub fn file_cut(&self, from: UrlBuf, mut to: UrlBuf, force: bool) { + pub fn file_cut(&self, from: UrlBuf, to: UrlBuf, force: bool) { let mut ongoing = self.ongoing.lock(); - let id = ongoing.add::(format!("Cut {} to {}", from.display(), to.display())); + let task = ongoing.add::(format!("Cut {} to {}", from.display(), to.display())); if to.try_starts_with(&from).unwrap_or(false) && !to.covariant(&from) { - return self.ops.out(id, FileOutCut::Fail("Cannot cut directory into itself".to_owned())); + return self + .ops + .out(task.id, FileOutCut::Fail("Cannot cut directory into itself".to_owned())); } - ongoing.hooks.add_async(id, { - let ops = self.ops.clone(); - let (from, to) = (from.clone(), to.clone()); - - move |canceled| async move { - if !canceled { - provider::remove_dir_clean(&from).await.ok(); - Pump::push_move(from, to); - } - ops.out(id, FileOutCut::Clean); - } - }); - - let file = self.file.clone(); let follow = !from.scheme().covariant(to.scheme()); - self.send_micro(id, LOW, async move { - if !force { - to = unique_name(to, must_be_dir(&from)).await?; - } - file.cut(FileInCut { id, from, to, cha: None, follow, retry: 0, drop: None }).await - }); + self.queue( + FileInCut { id: task.id, from, to, force, cha: None, follow, retry: 0, drop: None }, + LOW, + ); } - pub fn file_copy(&self, from: UrlBuf, mut to: UrlBuf, force: bool, follow: bool) { - let id = self.ongoing.lock().add::(format!( - "Copy {} to {}", - from.display(), - to.display() - )); + pub fn file_copy(&self, from: UrlBuf, to: UrlBuf, force: bool, follow: bool) { + let mut ongoing = self.ongoing.lock(); + let task = ongoing.add::(format!("Copy {} to {}", from.display(), to.display())); if to.try_starts_with(&from).unwrap_or(false) && !to.covariant(&from) { - return self.ops.out(id, FileOutCopy::Fail("Cannot copy directory into itself".to_owned())); + return self + .ops + .out(task.id, FileOutCopy::Fail("Cannot copy directory into itself".to_owned())); } - let file = self.file.clone(); let follow = follow || !from.scheme().covariant(to.scheme()); - self.send_micro(id, LOW, async move { - if !force { - to = unique_name(to, must_be_dir(&from)).await?; - } - file.copy(FileInCopy { id, from, to, cha: None, follow, retry: 0 }).await - }); + self.queue(FileInCopy { id: task.id, from, to, force, cha: None, follow, retry: 0 }, LOW); } - pub fn file_link(&self, from: UrlBuf, mut to: UrlBuf, relative: bool, force: bool) { - let id = self.ongoing.lock().add::(format!( - "Link {} to {}", - from.display(), - to.display() - )); + pub fn file_link(&self, from: UrlBuf, to: UrlBuf, relative: bool, force: bool) { + let mut ongoing = self.ongoing.lock(); + let task = ongoing.add::(format!("Link {} to {}", from.display(), to.display())); - let file = self.file.clone(); - self.send_micro(id, LOW, async move { - if !force { - to = unique_name(to, must_be_dir(&from)).await?; - } - file.link(FileInLink { id, from, to, cha: None, resolve: false, relative, delete: false }) - }); + self.queue( + FileInLink { + id: task.id, + from, + to, + force, + cha: None, + resolve: false, + relative, + delete: false, + }, + LOW, + ); } - pub fn file_hardlink(&self, from: UrlBuf, mut to: UrlBuf, force: bool, follow: bool) { - let id = self.ongoing.lock().add::(format!( - "Hardlink {} to {}", - from.display(), - to.display() - )); + pub fn file_hardlink(&self, from: UrlBuf, to: UrlBuf, force: bool, follow: bool) { + let mut ongoing = self.ongoing.lock(); + let task = + ongoing.add::(format!("Hardlink {} to {}", from.display(), to.display())); if !from.scheme().covariant(to.scheme()) { return self .ops - .out(id, FileOutHardlink::Fail("Cannot hardlink cross filesystem".to_owned())); + .out(task.id, FileOutHardlink::Fail("Cannot hardlink cross filesystem".to_owned())); } if to.try_starts_with(&from).unwrap_or(false) && !to.covariant(&from) { return self .ops - .out(id, FileOutHardlink::Fail("Cannot hardlink directory into itself".to_owned())); + .out(task.id, FileOutHardlink::Fail("Cannot hardlink directory into itself".to_owned())); } - let file = self.file.clone(); - self.send_micro(id, LOW, async move { - if !force { - to = unique_name(to, must_be_dir(&from)).await?; - } - file.hardlink(FileInHardlink { id, from, to, cha: None, follow }).await - }); + self.queue(FileInHardlink { id: task.id, from, to, force, cha: None, follow }, LOW); } pub fn file_delete(&self, target: UrlBuf) { let mut ongoing = self.ongoing.lock(); - let id = ongoing.add::(format!("Delete {}", target.display())); + let task = ongoing.add::(format!("Delete {}", target.display())); - ongoing.hooks.add_async(id, { - let ops = self.ops.clone(); - let target = target.clone(); - - move |canceled| async move { - if !canceled { - provider::remove_dir_all(&target).await.ok(); - TasksProxy::update_succeed(&target); - Pump::push_delete(target); - } - ops.out(id, FileOutDelete::Clean); - } - }); - - let file = self.file.clone(); - self.send_micro( - id, - LOW, - async move { file.delete(FileInDelete { id, target, cha: None }).await }, - ); + task.set_hook(HookInOutDelete { id: task.id, target: target.clone() }); + self.queue(FileInDelete { id: task.id, target, cha: None }, LOW); } pub fn file_trash(&self, target: UrlBuf) { let mut ongoing = self.ongoing.lock(); - let id = ongoing.add::(format!("Trash {}", target.display())); + let task = ongoing.add::(format!("Trash {}", target.display())); - ongoing.hooks.add_sync(id, { - let target = target.clone(); - move |canceled| { - if !canceled { - TasksProxy::update_succeed(&target); - Pump::push_trash(target); - } - } - }); - - let file = self.file.clone(); - self.send_micro(id, LOW, async move { file.trash(FileInTrash { id, target }) }) + task.set_hook(HookInOutTrash { id: task.id, target: target.clone() }); + self.queue(FileInTrash { id: task.id, target }, LOW); } pub fn file_download(&self, url: UrlBuf, done: Option>) { let mut ongoing = self.ongoing.lock(); - let id = ongoing.add::(format!("Download {}", url.display())); - - if let Some(tx) = done { - ongoing.hooks.add_sync(id, move |canceled| _ = tx.send(canceled)); - } + let task = ongoing.add::(format!("Download {}", url.display())); if !url.kind().is_remote() { - return self.ops.out(id, FileOutDownload::Fail("Cannot download non-remote file".to_owned())); + return self + .ops + .out(task.id, FileOutDownload::Fail("Cannot download non-remote file".to_owned())); }; - let file = self.file.clone(); - self.send_micro(id, LOW, async move { - file.download(FileInDownload { id, url, cha: None, retry: 0 }).await - }); + if let Some(done) = done { + task.set_hook(HookInOutDownload { id: task.id, done }); + } + + self.queue(FileInDownload { id: task.id, url, cha: None, retry: 0 }, LOW); } pub fn file_upload(&self, url: UrlBuf) { let mut ongoing = self.ongoing.lock(); - let id = ongoing.add::(format!("Upload {}", url.display())); + let task = ongoing.add::(format!("Upload {}", url.display())); if !url.kind().is_remote() { - return self.ops.out(id, FileOutUpload::Fail("Cannot upload non-remote file".to_owned())); + return self + .ops + .out(task.id, FileOutUpload::Fail("Cannot upload non-remote file".to_owned())); }; - let file = self.file.clone(); - self.send_micro(id, LOW, async move { - file.upload(FileInUpload { id, url, cha: None, cache: None }).await - }); + self.queue(FileInUpload { id: task.id, url, cha: None, cache: None }, LOW); } - pub fn plugin_micro(&self, opt: PluginOpt) { - let id = self.ongoing.lock().add::(format!("Run micro plugin `{}`", opt.id)); + pub fn plugin_entry(&self, opt: PluginOpt) { + let mut ongoing = self.ongoing.lock(); + let task = ongoing.add::(format!("Run micro plugin `{}`", opt.id)); - let plugin = self.plugin.clone(); - self.send_micro(id, NORMAL, async move { plugin.micro(PluginInEntry { id, opt }).await }); - } - - pub fn plugin_macro(&self, opt: PluginOpt) { - let id = self.ongoing.lock().add::(format!("Run macro plugin `{}`", opt.id)); - - self.plugin.r#macro(PluginInEntry { id, opt }).ok(); + self.queue(PluginInEntry { id: task.id, opt }, NORMAL); } pub fn fetch_paged( @@ -267,20 +208,17 @@ impl Scheduler { done: Option>, ) { let mut ongoing = self.ongoing.lock(); - let id = ongoing.add::(format!( + let task = ongoing.add::(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)); + if let Some(done) = done { + task.set_hook(HookInOutFetch { id: task.id, done }); } - let prework = self.prework.clone(); - self.send_micro(id, NORMAL, async move { - prework.fetch(PreworkInFetch { id, plugin: fetcher, targets }).await - }); + self.queue(PreworkInFetch { id: task.id, plugin: fetcher, targets }, NORMAL); } pub async fn fetch_mimetype(&self, targets: Vec) -> bool { @@ -292,7 +230,7 @@ impl Scheduler { } for rx in wg { - if rx.await != Ok(false) { + if rx.await != Ok(true) { return false; // Canceled or error } } @@ -300,14 +238,11 @@ impl Scheduler { } pub fn preload_paged(&self, preloader: &'static Preloader, target: &yazi_fs::File) { - let id = - self.ongoing.lock().add::(format!("Run preloader `{}`", preloader.run.name)); + let mut ongoing = self.ongoing.lock(); + let task = ongoing.add::(format!("Run preloader `{}`", preloader.run.name)); let target = target.clone(); - let prework = self.prework.clone(); - self.send_micro(id, NORMAL, async move { - prework.load(PreworkInLoad { id, plugin: preloader, target }).await - }); + self.queue(PreworkInLoad { id: task.id, plugin: preloader, target }, NORMAL); } pub fn prework_size(&self, targets: Vec<&UrlBuf>) { @@ -315,15 +250,12 @@ impl Scheduler { let mut ongoing = self.ongoing.lock(); for target in targets { - let id = + let task = ongoing.add::(format!("Calculate the size of {}", target.display())); let target = target.clone(); let throttle = throttle.clone(); - let prework = self.prework.clone(); - self.send_micro(id, NORMAL, async move { - prework.size(PreworkInSize { id, target, throttle }).await - }); + self.queue(PreworkInSize { id: task.id, target, throttle }, NORMAL); } } @@ -338,50 +270,55 @@ impl Scheduler { }; let mut ongoing = self.ongoing.lock(); - let (id, clean): (_, TaskOut) = if opt.block { - (ongoing.add::(name), ProcessOutBlock::Clean.into()) + let task = if opt.block { + ongoing.add::(name) } else if opt.orphan { - (ongoing.add::(name), ProcessOutOrphan::Clean.into()) + ongoing.add::(name) } else { - (ongoing.add::(name), ProcessOutBg::Clean.into()) + ongoing.add::(name) }; - let ops = self.ops.clone(); - let (cancel_tx, cancel_rx) = mpsc::channel(1); - ongoing.hooks.add_async(id, move |canceled| async move { - if canceled { - cancel_tx.send(()).await.ok(); - cancel_tx.closed().await; - } - if let Some(tx) = opt.done { - tx.send(()).ok(); - } - ops.out(id, clean); - }); - - let process = self.process.clone(); - self.send_micro::<_, TaskOut>(id, NORMAL, async move { - if opt.block { - process.block(ProcessInBlock { id, cwd: opt.cwd, cmd: opt.cmd, args: opt.args }).await?; - } else if opt.orphan { - process.orphan(ProcessInOrphan { id, cwd: opt.cwd, cmd: opt.cmd, args: opt.args }).await?; - } else { - process - .bg(ProcessInBg { id, cwd: opt.cwd, cmd: opt.cmd, args: opt.args, cancel: cancel_rx }) - .await?; - } - Ok(()) - }); + if opt.block { + task.set_hook(HookInOutBlock { id: task.id, done: opt.done }); + self + .queue(ProcessInBlock { id: task.id, cwd: opt.cwd, cmd: opt.cmd, args: opt.args }, NORMAL); + } else if opt.orphan { + task.set_hook(HookInOutOrphan { id: task.id, done: opt.done }); + self + .queue(ProcessInOrphan { id: task.id, cwd: opt.cwd, cmd: opt.cmd, args: opt.args }, NORMAL); + } else { + let (cancel_tx, cancel_rx) = mpsc::channel(1); + task.set_hook(HookInOutBg { id: task.id, cancel: cancel_tx, done: opt.done }); + self.queue( + ProcessInBg { + id: task.id, + cwd: opt.cwd, + cmd: opt.cmd, + args: opt.args, + cancel: cancel_rx, + }, + NORMAL, + ); + }; } - fn schedule_micro( - &self, - rx: async_priority_channel::Receiver, u8>, - ) -> JoinHandle<()> { + fn schedule_micro(&self, rx: async_priority_channel::Receiver) -> JoinHandle<()> { + let ops = self.ops.clone(); + let runner = self.runner.clone(); + let ongoing = self.ongoing.clone(); + tokio::spawn(async move { loop { - if let Ok((fut, _)) = rx.recv().await { - fut.await; + if let Ok((r#in, _)) = rx.recv().await { + let id = r#in.id(); + if !ongoing.lock().exists(id) { + continue; + } + + let result = runner.micro(r#in).await; + if let Err(out) = result { + ops.out(id, out); + } } } }) @@ -389,50 +326,28 @@ impl Scheduler { fn schedule_macro( &self, - micro: async_priority_channel::Receiver, u8>, + micro: async_priority_channel::Receiver, r#macro: async_priority_channel::Receiver, ) -> JoinHandle<()> { - let file = self.file.clone(); - let plugin = self.plugin.clone(); - let prework = self.prework.clone(); - let ops = self.ops.clone(); + let runner = self.runner.clone(); let ongoing = self.ongoing.clone(); tokio::spawn(async move { loop { - select! { - Ok((fut, _)) = micro.recv() => { - fut.await; - } - Ok((r#in, _)) = r#macro.recv() => { - let id = r#in.id(); - if !ongoing.lock().exists(id) { - continue; - } + let (r#in, micro) = select! { + Ok((r#in, _)) = micro.recv() => (r#in, true), + Ok((r#in, _)) = r#macro.recv() => (r#in, false), + }; - let result: Result<_, TaskOut> = match r#in { - // File - TaskIn::FileCopy(r#in) => file.copy_do(r#in).await.map_err(Into::into), - TaskIn::FileCut(r#in) => file.cut_do(r#in).await.map_err(Into::into), - TaskIn::FileLink(r#in) => file.link_do(r#in).await.map_err(Into::into), - TaskIn::FileHardlink(r#in) => file.hardlink_do(r#in).await.map_err(Into::into), - TaskIn::FileDelete(r#in) => file.delete_do(r#in).await.map_err(Into::into), - TaskIn::FileTrash(r#in) => file.trash_do(r#in).await.map_err(Into::into), - TaskIn::FileDownload(r#in) => file.download_do(r#in).await.map_err(Into::into), - TaskIn::FileUpload(r#in) => file.upload_do(r#in).await.map_err(Into::into), - // Plugin - TaskIn::PluginEntry(r#in) => plugin.macro_do(r#in).await.map_err(Into::into), - // Prework - TaskIn::PreworkFetch(r#in) => prework.fetch_do(r#in).await.map_err(Into::into), - TaskIn::PreworkLoad(r#in) => prework.load_do(r#in).await.map_err(Into::into), - TaskIn::PreworkSize(r#in) => prework.size_do(r#in).await.map_err(Into::into), - }; + let id = r#in.id(); + if !ongoing.lock().exists(id) { + continue; + } - if let Err(out) = result { - ops.out(id, out); - } - } + let result = if micro { runner.micro(r#in).await } else { runner.r#macro(r#in).await }; + if let Err(out) = result { + ops.out(id, out); } } }) @@ -450,31 +365,17 @@ impl Scheduler { op.out.reduce(task); if !task.prog.success() && !task.prog.cleaned() { continue; - } else if let Some(hook) = ongoing.hooks.pop(op.id) - && let Some(fut) = hook.call(false) - { - micro.try_send(fut, LOW).ok(); + } else if let Some(hook) = task.hook.take() { + micro.try_send(hook, LOW).ok(); } else { - ongoing.all.remove(&op.id); + ongoing.inner.remove(&op.id); } } }) } - fn send_micro(&self, id: Id, priority: u8, f: F) - where - F: Future> + Send + 'static, - E: Into, - { - let ops = self.ops.clone(); - _ = self.micro.try_send( - async move { - if let Err(out) = f.await { - ops.out(id, out); - } - } - .boxed(), - priority, - ); + #[inline] + fn queue(&self, r#in: impl Into, priority: u8) { + _ = self.micro.try_send(r#in.into(), priority); } } diff --git a/yazi-scheduler/src/task.rs b/yazi-scheduler/src/task.rs index 1fcb67e6..cd94c54e 100644 --- a/yazi-scheduler/src/task.rs +++ b/yazi-scheduler/src/task.rs @@ -1,13 +1,15 @@ use tokio::sync::mpsc; use yazi_shared::Id; -use crate::TaskProg; +use crate::{TaskIn, TaskProg}; #[derive(Debug)] pub struct Task { pub id: Id, pub name: String, pub(crate) prog: TaskProg, + pub(crate) hook: Option, + pub canceled: bool, pub logs: String, pub logger: Option>, @@ -22,6 +24,9 @@ impl Task { id, name, prog: T::default().into(), + hook: None, + canceled: false, + logs: Default::default(), logger: Default::default(), } @@ -35,4 +40,6 @@ impl Task { logger.send(line).ok(); } } + + pub(super) fn set_hook(&mut self, hook: impl Into) { self.hook = Some(hook.into()); } }