mroot/yazi
1
0
Fork 0
mirror of https://github.com/sxyazi/yazi.git synced 2026-07-21 23:01:05 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

feat: include context in error logs when a task fails (#3427)

Browse source
This commit is contained in:
三咲雅 misaki masa 2025-12-13 19:43:29 +08:00 • committed by GitHub
parent ac7718c226
commit 5fe58ba2b1
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
27 changed files with 769 additions and 482 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
yazi-actor/src/mgr/download.rs
View file

@ -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![];

4
yazi-core/src/tasks/plugin.rs
View file

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

8
yazi-core/src/tasks/prework.rs
View file

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

23
yazi-dds/src/pump.rs
View file

@ -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<Option<mpsc::UnboundedSender<UrlBuf>>> = Mutex::new(None
pub struct Pump;
impl Pump {
pub fn push_move(from: UrlBuf, to: UrlBuf) {
pub fn push_move<U>(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<U>(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<U>(target: U)
where
U: AsUrl,
{
if let Some(tx) = &*DELETE_TX.lock() {
tx.send(target).ok();
tx.send(target.as_url().to_owned()).ok();
}
}

4
yazi-fm/src/app/commands/plugin.rs
View file

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

165
yazi-scheduler/src/file/file.rs
View file

@ -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<TaskOp>,
ops: &mpsc::UnboundedSender<TaskOp>,
r#macro: &async_priority_channel::Sender<TaskIn, u8>,
) -> 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::<FileOutCopy, _, _, _, _, _>(
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::<FileOutCut, _, _, _, _, _>(
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::<FileOutHardlink, _, _, _, _, _>(
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))

6
yazi-scheduler/src/file/in.rs
View file

@ -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<Cha>,
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<Cha>,
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<Cha>,
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<Cha>,
pub(crate) follow: bool,
}

64
yazi-scheduler/src/file/out.rs
View file

@ -9,6 +9,10 @@ pub(crate) enum FileOutCopy {
Fail(String),
}
impl From<anyhow::Error> for FileOutCopy {
fn from(value: anyhow::Error) -> Self { Self::Fail(format!("{value:?}")) }
}
impl From<std::io::Error> 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<std::io::Error> for FileOutCopyDo {
fn from(value: std::io::Error) -> Self { Self::Fail(format!("{value:?}")) }
impl From<anyhow::Error> 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<anyhow::Error> for FileOutCut {
fn from(value: anyhow::Error) -> Self { Self::Fail(format!("{value:?}")) }
}
impl From<std::io::Error> 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<std::io::Error> for FileOutCutDo {
fn from(value: std::io::Error) -> Self { Self::Fail(format!("{value:?}")) }
impl From<anyhow::Error> for FileOutCutDo {
fn from(value: anyhow::Error) -> Self { Self::Fail(format!("{value:?}")) }
}
impl FileOutCutDo {
@ -157,10 +165,6 @@ impl From<anyhow::Error> for FileOutLink {
fn from(value: anyhow::Error) -> Self { Self::Fail(format!("{value:?}")) }
}
impl From<std::io::Error> 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<anyhow::Error> for FileOutHardlink {
fn from(value: anyhow::Error) -> Self { Self::Fail(format!("{value:?}")) }
}
impl From<std::io::Error> 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<std::io::Error> for FileOutHardlinkDo {
fn from(value: std::io::Error) -> Self { Self::Fail(format!("{value:?}")) }
impl From<anyhow::Error> 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<std::io::Error> for FileOutDelete {
fn from(value: std::io::Error) -> Self { Self::Fail(format!("{value:?}")) }
impl From<anyhow::Error> 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<std::io::Error> for FileOutDeleteDo {
fn from(value: std::io::Error) -> Self { Self::Fail(format!("{value:?}")) }
impl From<anyhow::Error> 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<std::io::Error> for FileOutTrash {
fn from(value: std::io::Error) -> Self { Self::Fail(format!("{value:?}")) }
impl From<anyhow::Error> 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<std::io::Error> for FileOutDownload {
fn from(value: std::io::Error) -> Self { Self::Fail(format!("{value:?}")) }
impl From<anyhow::Error> 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<std::io::Error> for FileOutDownloadDo {
fn from(value: std::io::Error) -> Self { Self::Fail(format!("{value:?}")) }
}
impl From<anyhow::Error> for FileOutDownloadDo {
fn from(value: anyhow::Error) -> Self { Self::Fail(format!("{value:?}")) }
}
@ -431,10 +443,6 @@ pub(crate) enum FileOutUpload {
Fail(String),
}
impl From<std::io::Error> for FileOutUpload {
fn from(value: std::io::Error) -> Self { Self::Fail(format!("{value:?}")) }
}
impl From<anyhow::Error> for FileOutUpload {
fn from(value: anyhow::Error) -> Self { Self::Fail(format!("{value:?}")) }
}
@ -471,10 +479,6 @@ pub(crate) enum FileOutUploadDo {
Fail(String),
}
impl From<std::io::Error> for FileOutUploadDo {
fn from(value: std::io::Error) -> Self { Self::Fail(format!("{value:?}")) }
}
impl From<anyhow::Error> for FileOutUploadDo {
fn from(value: anyhow::Error) -> Self { Self::Fail(format!("{value:?}")) }
}

8
yazi-scheduler/src/file/progress.rs
View file

@ -213,7 +213,8 @@ impl FileProgDelete {
// --- Trash
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize)]
pub struct FileProgTrash {
pub state: Option<bool>,
pub state: Option<bool>,
pub cleaned: bool,
}
impl From<FileProgTrash> 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<f32> { None }
}
@ -248,6 +249,7 @@ pub struct FileProgDownload {
pub total_bytes: u64,
pub processed_bytes: u64,
pub collected: Option<bool>,
pub cleaned: bool,
}
impl From<FileProgDownload> 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<f32> {
Some(if self.success() {

37
yazi-scheduler/src/file/traverse.rs
View file

@ -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<Cha>;
@ -13,7 +13,7 @@ trait Traverse {
fn from(&self) -> Url<'_>;
async fn init(&mut self) -> io::Result<Cha> {
async fn init(&mut self) -> anyhow::Result<Cha> {
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<UrlBuf>, 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<Url<'_>> { 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<Cha> {
async fn init(&mut self) -> anyhow::Result<Cha> {
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<R, T, D, FC, FR, E>(
mut task: T,
pub(super) async fn traverse<O, I, D, FC, FR, E>(
mut task: I,
on_dir: D,
mut on_file: FC,
on_error: E,
) -> Result<(), R>
) -> Result<(), O>
where
R: Debug + From<io::Error>,
T: Traverse,
D: AsyncFn(Url) -> Result<(), R>,
FC: FnMut(T, Cha) -> FR,
FR: Future<Output = Result<(), R>>,
O: Debug + From<anyhow::Error>,
I: Debug + Traverse,
D: AsyncFn(Url) -> Result<(), O>,
FC: FnMut(I, Cha) -> FR,
FR: Future<Output = Result<(), O>>,
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;
}

89
yazi-scheduler/src/hook/hook.rs Normal file
View file

@ -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<Mutex<Ongoing>>,
}
impl Hook {
pub(crate) fn new(ops: &mpsc::UnboundedSender<TaskOp>, ongoing: &Arc<Mutex<Ongoing>>) -> 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);
}
}

132
yazi-scheduler/src/hook/in.rs Normal file
View file

@ -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<bool>,
}
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<bool>,
}
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<oneshot::Sender<()>>,
}
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<oneshot::Sender<()>>,
}
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<oneshot::Sender<()>>,
}
impl HookInOutBg {
pub(crate) fn reduce(self, task: &mut Task) {
if let TaskProg::ProcessBg(_) = &task.prog {
task.hook = Some(self.into());
}
}
}

1
yazi-scheduler/src/hook/mod.rs Normal file
View file

@ -0,0 +1 @@
yazi_macro::mod_flat!(hook r#in);

59
yazi-scheduler/src/hooks.rs
View file

@ -1,59 +0,0 @@
use futures::{FutureExt, future::BoxFuture};
use hashbrown::HashMap;
use yazi_shared::Id;
#[derive(Default)]
pub(super) struct Hooks {
inner: HashMap<Id, Box<dyn Hook>>,
}
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(AsyncHook(f)));
}
#[inline]
pub(super) fn pop(&mut self, id: Id) -> Option<Box<dyn Hook>> { self.inner.remove(&id) }
}
// --- Hook
pub(super) trait Hook: Send {
fn call(self: Box<Self>, cancel: bool) -> Option<BoxFuture<'static, ()>>;
}
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) -> Option<BoxFuture<'static, ()>> {
Some((self.0)(cancel).boxed())
}
}

32
yazi-scheduler/src/in.rs
View file

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

4
yazi-scheduler/src/lib.rs
View file

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

26
yazi-scheduler/src/macros.rs
View file

@ -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)),* $(,)?) => {

25
yazi-scheduler/src/ongoing.rs
View file

@ -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<Id, Task>,
pub(super) inner: HashMap<Id, Task>,
}
impl Ongoing {
pub(super) fn add<T>(&mut self, name: String) -> Id
pub(super) fn add<T>(&mut self, name: String) -> &mut Task
where
T: Into<TaskProg> + Default,
{
static IDS: Ids = Ids::new();
let id = IDS.next();
self.all.insert(id, Task::new::<T>(id, name));
id
self.inner.entry(id).insert(Task::new::<T>(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<Id> { 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<dyn Iterator<Item = &Task> + '_> {
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())
}
}

25
yazi-scheduler/src/out.rs
View file

@ -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),
}
}
}

14
yazi-scheduler/src/plugin/plugin.rs
View file

@ -12,23 +12,17 @@ pub(crate) struct Plugin {
impl Plugin {
pub(crate) fn new(
tx: &mpsc::UnboundedSender<TaskOp>,
ops: &mpsc::UnboundedSender<TaskOp>,
r#macro: &async_priority_channel::Sender<TaskIn, u8>,
) -> 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))
}

4
yazi-scheduler/src/prework/out.rs
View file

@ -5,6 +5,7 @@ use crate::{Task, TaskProg};
pub(crate) enum PreworkOutFetch {
Succ,
Fail(String),
Clean,
}
impl From<mlua::Error> for PreworkOutFetch {
@ -22,6 +23,9 @@ impl PreworkOutFetch {
prog.state = Some(false);
task.log(reason);
}
Self::Clean => {
prog.cleaned = true;
}
}
}
}

4
yazi-scheduler/src/prework/prework.rs
View file

@ -27,11 +27,11 @@ pub struct Prework {
impl Prework {
pub(crate) fn new(
tx: &mpsc::UnboundedSender<TaskOp>,
ops: &mpsc::UnboundedSender<TaskOp>,
r#macro: &async_priority_channel::Sender<TaskIn, u8>,
) -> 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())),

5
yazi-scheduler/src/prework/progress.rs
View file

@ -4,7 +4,8 @@ use yazi_parser::app::TaskSummary;
// --- Fetch
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize)]
pub struct PreworkProgFetch {
pub state: Option<bool>,
pub state: Option<bool>,
pub cleaned: bool,
}
impl From<PreworkProgFetch> 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<f32> { None }
}

2
yazi-scheduler/src/process/process.rs
View file

@ -11,7 +11,7 @@ pub(crate) struct Process {
}
impl Process {
pub(crate) fn new(tx: &mpsc::UnboundedSender<TaskOp>) -> Self { Self { ops: tx.into() } }
pub(crate) fn new(ops: &mpsc::UnboundedSender<TaskOp>) -> Self { Self { ops: ops.into() } }
pub(crate) async fn block(&self, task: ProcessInBlock) -> Result<(), ProcessOutBlock> {
let _permit = HIDER.acquire().await.unwrap();

80
yazi-scheduler/src/runner.rs Normal file
View file

@ -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<File>,
pub(super) plugin: Arc<Plugin>,
pub prework: Arc<Prework>,
pub(super) process: Arc<Process>,
pub(super) hook: Arc<Hook>,
}
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!(),
}
}
}

419
yazi-scheduler/src/scheduler.rs
View file

@ -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<File>,
plugin: Arc<Plugin>,
pub prework: Arc<Prework>,
process: Arc<Process>,
ops: TaskOps,
micro: async_priority_channel::Sender<BoxFuture<'static, ()>, u8>,
pub runner: Runner,
micro: async_priority_channel::Sender<TaskIn, u8>,
handles: Vec<JoinHandle<()>>,
pub ongoing: Arc<Mutex<Ongoing>>,
}
@ -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, &macro_tx)),
plugin: Arc::new(Plugin::new(&op_tx, &macro_tx)),
prework: Arc::new(Prework::new(&op_tx, &macro_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::<FileProgCut>(format!("Cut {} to {}", from.display(), to.display()));
let task = ongoing.add::<FileProgCut>(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::<FileProgCopy>(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::<FileProgCopy>(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::<FileProgLink>(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::<FileProgLink>(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::<FileProgHardlink>(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::<FileProgHardlink>(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::<FileProgDelete>(format!("Delete {}", target.display()));
let task = ongoing.add::<FileProgDelete>(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::<FileProgTrash>(format!("Trash {}", target.display()));
let task = ongoing.add::<FileProgTrash>(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<oneshot::Sender<bool>>) {
let mut ongoing = self.ongoing.lock();
let id = ongoing.add::<FileProgDownload>(format!("Download {}", url.display()));
if let Some(tx) = done {
ongoing.hooks.add_sync(id, move |canceled| _ = tx.send(canceled));
}
let task = ongoing.add::<FileProgDownload>(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::<FileProgUpload>(format!("Upload {}", url.display()));
let task = ongoing.add::<FileProgUpload>(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::<PluginProgEntry>(format!("Run micro plugin `{}`", opt.id));
pub fn plugin_entry(&self, opt: PluginOpt) {
let mut ongoing = self.ongoing.lock();
let task = ongoing.add::<PluginProgEntry>(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::<PluginProgEntry>(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<oneshot::Sender<bool>>,
) {
let mut ongoing = self.ongoing.lock();
let id = ongoing.add::<PreworkProgFetch>(format!(
let task = 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));
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<yazi_fs::File>) -> 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::<PreworkProgLoad>(format!("Run preloader `{}`", preloader.run.name));
let mut ongoing = self.ongoing.lock();
let task = ongoing.add::<PreworkProgLoad>(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::<PreworkProgSize>(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::<ProcessProgBlock>(name), ProcessOutBlock::Clean.into())
let task = if opt.block {
ongoing.add::<ProcessProgBlock>(name)
} else if opt.orphan {
(ongoing.add::<ProcessProgOrphan>(name), ProcessOutOrphan::Clean.into())
ongoing.add::<ProcessProgOrphan>(name)
} else {
(ongoing.add::<ProcessProgBg>(name), ProcessOutBg::Clean.into())
ongoing.add::<ProcessProgBg>(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<BoxFuture<'static, ()>, u8>,
) -> JoinHandle<()> {
fn schedule_micro(&self, rx: async_priority_channel::Receiver<TaskIn, u8>) -> 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<BoxFuture<'static, ()>, u8>,
micro: async_priority_channel::Receiver<TaskIn, u8>,
r#macro: async_priority_channel::Receiver<TaskIn, u8>,
) -> 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<F, E>(&self, id: Id, priority: u8, f: F)
where
F: Future<Output = Result<(), E>> + Send + 'static,
E: Into<TaskOut>,
{
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<TaskIn>, priority: u8) {
_ = self.micro.try_send(r#in.into(), priority);
}
}

9
yazi-scheduler/src/task.rs
View file

@ -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<TaskIn>,
pub canceled: bool,
pub logs: String,
pub logger: Option<mpsc::UnboundedSender<String>>,
@ -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<TaskIn>) { self.hook = Some(hook.into()); }
}