mroot/yazi
1
0
Fork 0
mirror of https://github.com/sxyazi/yazi.git synced 2026-07-25 08:41:05 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

feat: finer control over concurrent workers

Browse source
This commit is contained in:
sxyazi 2026-02-06 21:53:42 +08:00
parent 6757fed5aa
commit 30fea9be2f
No known key found for this signature in database
52 changed files with 1075 additions and 783 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
yazi-config/preset/yazi-default.toml
View file

@ -87,8 +87,11 @@ rules = [
]
[tasks]
micro_workers = 10
macro_workers = 10
file_workers = 3
plugin_workers = 5
fetch_workers = 5
preload_workers = 2
process_workers = 5
bizarre_retry = 3
image_alloc = 536870912 # 512MB
image_bound = [ 10000, 10000 ]

3
yazi-config/src/plugin/mod.rs
View file

@ -1,3 +1,4 @@
yazi_macro::mod_flat!(fetcher plugin preloader previewer spotter);
pub const MAX_PREWORKERS: u8 = 32;
pub const MAX_FETCHERS: u8 = 16;
pub const MAX_PRELOADERS: u8 = 16;

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

@ -1,4 +1,4 @@
use anyhow::Result;
use anyhow::{Result, bail};
use hashbrown::HashSet;
use serde::Deserialize;
use tracing::warn;
@ -6,7 +6,7 @@ use yazi_codegen::DeserializeOver2;
use yazi_fs::File;
use super::{Fetcher, Preloader, Previewer, Spotter};
use crate::{Preset, plugin::MAX_PREWORKERS};
use crate::{Preset, plugin::{MAX_FETCHERS, MAX_PRELOADERS}};
#[derive(Default, Deserialize, DeserializeOver2)]
pub struct Plugin {
@ -52,7 +52,7 @@ impl Plugin {
}
pub fn mime_fetchers(&self, files: Vec<File>) -> impl Iterator<Item = (&Fetcher, Vec<File>)> {
let mut tasks: [Vec<_>; MAX_PREWORKERS as usize] = Default::default();
let mut tasks: [Vec<_>; MAX_FETCHERS as usize] = Default::default();
for f in files {
let found = self.fetchers.iter().find(|&g| g.id == "mime" && g.matches(&f, ""));
if let Some(g) = found {
@ -116,15 +116,17 @@ impl Plugin {
self.previewers =
Preset::mix(self.prepend_previewers, self.previewers, self.append_previewers).collect();
if self.fetchers.len() + self.preloaders.len() > MAX_PREWORKERS as usize {
panic!("Fetchers and preloaders exceed the limit of {MAX_PREWORKERS}");
if self.fetchers.len() > MAX_FETCHERS as usize {
bail!("Fetchers exceed the limit of {MAX_FETCHERS}");
} else if self.preloaders.len() > MAX_PRELOADERS as usize {
bail!("Preloaders exceed the limit of {MAX_PRELOADERS}");
}
for (i, p) in self.fetchers.iter_mut().enumerate() {
p.idx = i as u8;
}
for (i, p) in self.preloaders.iter_mut().enumerate() {
p.idx = self.fetchers.len() as u8 + i as u8;
p.idx = i as u8;
}
Ok(Self {

24
yazi-config/src/tasks/tasks.rs
View file

@ -4,8 +4,12 @@ use yazi_codegen::DeserializeOver2;
#[derive(Debug, Deserialize, DeserializeOver2)]
pub struct Tasks {
pub micro_workers: u8,
pub macro_workers: u8,
pub file_workers: u8,
pub plugin_workers: u8,
pub fetch_workers: u8,
pub preload_workers: u8,
pub process_workers: u8,
pub bizarre_retry: u8,
pub image_alloc: u32,
@ -16,12 +20,18 @@ pub struct Tasks {
impl Tasks {
pub(crate) fn reshape(self) -> Result<Self> {
if self.micro_workers < 1 {
bail!("[tasks].micro_workers must be at least 1.");
} else if self.macro_workers < 1 {
bail!("[tasks].macro_workers must be at least 1.");
if self.file_workers < 1 {
bail!("[tasks].file_workers must be at least 1.");
} else if self.plugin_workers < 1 {
bail!("[tasks].plugin_workers must be at least 1.");
} else if self.fetch_workers < 1 {
bail!("[tasks].fetch_workers must be at least 1.");
} else if self.preload_workers < 1 {
bail!("[tasks].preload_workers must be at least 1.");
} else if self.process_workers < 1 {
bail!("[tasks].process_workers must be at least 1.");
} else if self.bizarre_retry < 1 {
bail!("[tasks].bizarre_retry` must be at least 1.");
bail!("[tasks].bizarre_retry must be at least 1.");
}
Ok(self)
}

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

@ -1,4 +1,4 @@
use yazi_config::{YAZI, plugin::MAX_PREWORKERS};
use yazi_config::{YAZI, plugin::MAX_FETCHERS};
use yazi_fs::{File, Files, FsHash64, SortBy};
use super::Tasks;
@ -6,8 +6,8 @@ use crate::mgr::Mimetype;
impl Tasks {
pub fn fetch_paged(&self, paged: &[File], mimetype: &Mimetype) {
let mut loaded = self.scheduler.runner.prework.loaded.lock();
let mut tasks: [Vec<_>; MAX_PREWORKERS as usize] = Default::default();
let mut loaded = self.scheduler.fetch.loaded.lock();
let mut tasks: [Vec<_>; MAX_FETCHERS as usize] = Default::default();
for f in paged {
let hash = f.hash_u64();
for g in YAZI.plugin.fetchers(f, mimetype.get(&f.url).unwrap_or_default()) {
@ -29,7 +29,7 @@ impl Tasks {
}
pub fn preload_paged(&self, paged: &[File], mimetype: &Mimetype) {
let mut loaded = self.scheduler.runner.prework.loaded.lock();
let mut loaded = self.scheduler.preload.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.runner.prework.sizing.read();
let loading = self.scheduler.size.sizing.read();
targets
.iter()
.filter(|f| {
@ -62,7 +62,7 @@ impl Tasks {
return;
}
let mut loading = self.scheduler.runner.prework.sizing.write();
let mut loading = self.scheduler.size.sizing.write();
for &target in &targets {
loading.insert(target.clone());
}

45
yazi-plugin/src/runtime/runtime.rs
View file

@ -44,17 +44,17 @@ fn mgr() -> Composer<ComposerGet, ComposerSet> {
b"ratio" => lua.to_value_with(&m.ratio, SER_OPT)?,
b"sort_by" => lua.to_value_with(&m.sort_by, SER_OPT)?,
b"sort_sensitive" => lua.to_value_with(&m.sort_sensitive, SER_OPT)?,
b"sort_reverse" => lua.to_value_with(&m.sort_reverse, SER_OPT)?,
b"sort_dir_first" => lua.to_value_with(&m.sort_dir_first, SER_OPT)?,
b"sort_translit" => lua.to_value_with(&m.sort_translit, SER_OPT)?,
b"sort_sensitive" => m.sort_sensitive.get().into_lua(lua)?,
b"sort_reverse" => m.sort_reverse.get().into_lua(lua)?,
b"sort_dir_first" => m.sort_dir_first.get().into_lua(lua)?,
b"sort_translit" => m.sort_translit.get().into_lua(lua)?,
b"linemode" => lua.to_value_with(&m.linemode, SER_OPT)?,
b"show_hidden" => lua.to_value_with(&m.show_hidden, SER_OPT)?,
b"show_symlink" => lua.to_value_with(&m.show_symlink, SER_OPT)?,
b"scrolloff" => lua.to_value_with(&m.scrolloff, SER_OPT)?,
b"linemode" => lua.create_string(&m.linemode)?.into_lua(lua)?,
b"show_hidden" => m.show_hidden.get().into_lua(lua)?,
b"show_symlink" => m.show_symlink.get().into_lua(lua)?,
b"scrolloff" => m.scrolloff.get().into_lua(lua)?,
b"mouse_events" => lua.to_value_with(&m.mouse_events, SER_OPT)?,
b"title_format" => lua.to_value_with(&m.title_format, SER_OPT)?,
b"title_format" => lua.create_string(&m.title_format)?.into_lua(lua)?,
_ => return Ok(Value::Nil),
}
.into_lua(lua)
@ -79,17 +79,17 @@ fn preview() -> Composer<ComposerGet, ComposerSet> {
let p = &YAZI.preview;
match key {
b"wrap" => lua.to_value_with(&p.wrap, SER_OPT)?,
b"tab_size" => lua.to_value_with(&p.tab_size, SER_OPT)?,
b"max_width" => lua.to_value_with(&p.max_width, SER_OPT)?,
b"max_height" => lua.to_value_with(&p.max_height, SER_OPT)?,
b"tab_size" => p.tab_size.into_lua(lua)?,
b"max_width" => p.max_width.into_lua(lua)?,
b"max_height" => p.max_height.into_lua(lua)?,
b"cache_dir" => lua.to_value_with(&p.cache_dir, SER_OPT)?,
b"image_delay" => lua.to_value_with(&p.image_delay, SER_OPT)?,
b"image_filter" => lua.to_value_with(&p.image_filter, SER_OPT)?,
b"image_quality" => lua.to_value_with(&p.image_quality, SER_OPT)?,
b"image_delay" => p.image_delay.into_lua(lua)?,
b"image_filter" => lua.create_string(&p.image_filter)?.into_lua(lua)?,
b"image_quality" => p.image_quality.into_lua(lua)?,
b"ueberzug_scale" => lua.to_value_with(&p.ueberzug_scale, SER_OPT)?,
b"ueberzug_scale" => p.ueberzug_scale.into_lua(lua)?,
b"ueberzug_offset" => lua.to_value_with(&p.ueberzug_offset, SER_OPT)?,
_ => return Ok(Value::Nil),
}
@ -105,14 +105,17 @@ fn tasks() -> Composer<ComposerGet, ComposerSet> {
fn get(lua: &Lua, key: &[u8]) -> mlua::Result<Value> {
let t = &YAZI.tasks;
match key {
b"micro_workers" => lua.to_value_with(&t.micro_workers, SER_OPT)?,
b"macro_workers" => lua.to_value_with(&t.macro_workers, SER_OPT)?,
b"bizarre_retry" => lua.to_value_with(&t.bizarre_retry, SER_OPT)?,
b"file_workers" => t.file_workers.into_lua(lua)?,
b"plugin_workers" => t.plugin_workers.into_lua(lua)?,
b"fetch_workers" => t.fetch_workers.into_lua(lua)?,
b"preload_workers" => t.preload_workers.into_lua(lua)?,
b"process_workers" => t.process_workers.into_lua(lua)?,
b"bizarre_retry" => t.bizarre_retry.into_lua(lua)?,
b"image_alloc" => lua.to_value_with(&t.image_alloc, SER_OPT)?,
b"image_alloc" => t.image_alloc.into_lua(lua)?,
b"image_bound" => lua.to_value_with(&t.image_bound, SER_OPT)?,
b"suppress_preload" => lua.to_value_with(&t.suppress_preload, SER_OPT)?,
b"suppress_preload" => t.suppress_preload.into_lua(lua)?,
_ => return Ok(Value::Nil),
}
.into_lua(lua)

59
yazi-scheduler/src/fetch/fetch.rs Normal file
View file

@ -0,0 +1,59 @@
use std::num::NonZeroUsize;
use anyhow::Result;
use lru::LruCache;
use parking_lot::Mutex;
use tokio::sync::mpsc;
use tracing::error;
use yazi_config::Priority;
use yazi_fs::FsHash64;
use yazi_plugin::isolate;
use yazi_shared::event::CmdCow;
use crate::{HIGH, LOW, TaskOp, TaskOps, fetch::{FetchIn, FetchOutFetch}};
pub struct Fetch {
ops: TaskOps,
tx: async_priority_channel::Sender<FetchIn, u8>,
pub loaded: Mutex<LruCache<u64, u16>>,
}
impl Fetch {
pub(crate) fn new(
ops: &mpsc::UnboundedSender<TaskOp>,
tx: async_priority_channel::Sender<FetchIn, u8>,
) -> Self {
Self {
ops: ops.into(),
tx,
loaded: Mutex::new(LruCache::new(NonZeroUsize::new(4096).unwrap())),
}
}
pub(crate) async fn fetch(&self, task: FetchIn) -> Result<(), FetchOutFetch> {
let hashes: Vec<_> = task.targets.iter().map(|f| f.hash_u64()).collect();
let (state, err) = isolate::fetch(CmdCow::from(&task.plugin.run), task.targets).await?;
let mut loaded = self.loaded.lock();
for (_, h) in hashes.into_iter().enumerate().filter(|&(i, _)| !state.get(i)) {
loaded.get_mut(&h).map(|x| *x &= !(1 << task.plugin.idx));
}
if let Some(e) = err {
error!("Error when running fetcher `{}`:\n{e}", task.plugin.run.name);
}
Ok(self.ops.out(task.id, FetchOutFetch::Succ))
}
}
impl Fetch {
pub(crate) fn submit(&self, r#in: FetchIn) {
let priority = match r#in.plugin.prio {
Priority::Low => LOW,
Priority::Normal => HIGH,
Priority::High => HIGH,
};
_ = self.tx.try_send(r#in, priority);
}
}

13
yazi-scheduler/src/fetch/in.rs Normal file
View file

@ -0,0 +1,13 @@
use yazi_config::plugin::Fetcher;
use yazi_shared::Id;
#[derive(Debug)]
pub(crate) struct FetchIn {
pub(crate) id: Id,
pub(crate) plugin: &'static Fetcher,
pub(crate) targets: Vec<yazi_fs::File>,
}
impl FetchIn {
pub(crate) fn id(&self) -> Id { self.id }
}

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

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

26
yazi-scheduler/src/fetch/out.rs Normal file
View file

@ -0,0 +1,26 @@
use crate::{Task, TaskProg};
#[derive(Debug)]
pub(crate) enum FetchOutFetch {
Succ,
Fail(String),
}
impl From<mlua::Error> for FetchOutFetch {
fn from(value: mlua::Error) -> Self { Self::Fail(value.to_string()) }
}
impl FetchOutFetch {
pub(crate) fn reduce(self, task: &mut Task) {
let TaskProg::Fetch(prog) = &mut task.prog else { return };
match self {
Self::Succ => {
prog.state = Some(true);
}
Self::Fail(reason) => {
prog.state = Some(false);
task.log(reason);
}
}
}
}

32
yazi-scheduler/src/fetch/progress.rs Normal file
View file

@ -0,0 +1,32 @@
use serde::Serialize;
use yazi_parser::app::TaskSummary;
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize)]
pub struct FetchProg {
pub state: Option<bool>,
}
impl From<FetchProg> for TaskSummary {
fn from(value: FetchProg) -> Self {
Self {
total: 1,
success: (value.state == Some(true)) as u32,
failed: (value.state == Some(false)) as u32,
percent: value.percent().map(Into::into),
}
}
}
impl FetchProg {
pub fn cooked(self) -> bool { self.state == Some(true) }
pub fn running(self) -> bool { self.state.is_none() }
pub fn success(self) -> bool { self.cooked() }
pub fn failed(self) -> bool { self.state == Some(false) }
pub fn cleaned(self) -> Option<bool> { None }
pub fn percent(self) -> Option<f32> { None }
}

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

@ -9,19 +9,19 @@ use yazi_shared::{path::PathCow, url::{AsUrl, UrlCow, UrlLike}};
use yazi_vfs::{VfsCha, maybe_exists, provider::{self, DirEntry}, unique_file};
use super::{FileInCopy, FileInDelete, FileInHardlink, FileInLink, FileInTrash};
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, Transaction, Traverse}, hook::{HookInOutCopy, HookInOutCut}, ok_or_not_found, progress_or_break};
use crate::{LOW, NORMAL, TaskOp, TaskOps, ctx, file::{FileIn, FileInCut, FileInDownload, FileInUpload, FileOutCopy, FileOutCopyDo, FileOutCut, FileOutCutDo, FileOutDelete, FileOutDeleteDo, FileOutDownload, FileOutDownloadDo, FileOutHardlink, FileOutHardlinkDo, FileOutLink, FileOutTrash, FileOutUpload, FileOutUploadDo, Transaction, Traverse}, hook::{HookInOutCopy, HookInOutCut}, ok_or_not_found, progress_or_break};
pub(crate) struct File {
ops: TaskOps,
r#macro: async_priority_channel::Sender<TaskIn, u8>,
ops: TaskOps,
tx: async_priority_channel::Sender<FileIn, u8>,
}
impl File {
pub(crate) fn new(
ops: &mpsc::UnboundedSender<TaskOp>,
r#macro: &async_priority_channel::Sender<TaskIn, u8>,
tx: async_priority_channel::Sender<FileIn, u8>,
) -> Self {
Self { ops: ops.into(), r#macro: r#macro.clone() }
Self { ops: ops.into(), tx }
}
pub(crate) async fn copy(&self, mut task: FileInCopy) -> Result<(), FileOutCopy> {
@ -43,10 +43,10 @@ impl File {
async |task, cha| {
Ok(if cha.is_orphan() || (cha.is_link() && !task.follow) {
self.ops.out(id, FileOutCopy::New(0));
self.queue(task.into_link(), NORMAL);
self.requeue(task.into_link(), NORMAL);
} else {
self.ops.out(id, FileOutCopy::New(cha.len));
self.queue(task, LOW);
self.requeue(task, LOW);
})
},
|err| {
@ -79,7 +79,7 @@ impl File {
{
task.retry += 1;
self.ops.out(task.id, FileOutCopyDo::Log(format!("Retrying due to error: {e}")));
return Ok(self.queue(task, LOW));
return Ok(self.requeue(task, LOW));
}
Err(e) => ctx!(task, Err(e))?,
}
@ -115,10 +115,10 @@ impl File {
self.ops.out(id, FileOutCut::New(if nofollow { 0 } else { cha.len }));
if nofollow {
self.queue(task.into_link(), NORMAL);
self.requeue(task.into_link(), NORMAL);
} else {
match (cha.is_link(), reorder) {
(_, false) => self.queue(task, LOW),
(_, false) => self.requeue(task, LOW),
(true, true) => links.push(task),
(false, true) => files.push(task),
}
@ -135,14 +135,14 @@ impl File {
if !links.is_empty() {
let (tx, mut rx) = mpsc::channel(1);
for task in links {
self.queue(task.with_drop(&tx), LOW);
self.requeue(task.with_drop(&tx), LOW);
}
drop(tx);
while rx.recv().await.is_some() {}
}
for task in files {
self.queue(task, LOW);
self.requeue(task, LOW);
}
Ok(self.ops.out(id, FileOutCut::Succ))
@ -172,7 +172,7 @@ impl File {
{
task.retry += 1;
self.ops.out(task.id, FileOutCutDo::Log(format!("Retrying due to error: {e}")));
return Ok(self.queue(task, LOW));
return Ok(self.requeue(task, LOW));
}
Err(e) => ctx!(task, Err(e))?,
}
@ -186,7 +186,7 @@ impl File {
unique_file(task.to, false).await.context("Cannot determine unique destination name")?;
}
Ok(self.queue(task, NORMAL))
Ok(self.requeue(task, NORMAL))
}
pub(crate) async fn link_do(&self, task: FileInLink) -> Result<(), FileOutLink> {
@ -240,7 +240,7 @@ impl File {
},
async |task, _cha| {
self.ops.out(id, FileOutHardlink::New);
Ok(self.queue(task, NORMAL))
Ok(self.requeue(task, NORMAL))
},
|err| {
self.ops.out(id, FileOutHardlink::Deform(err));
@ -274,7 +274,7 @@ impl File {
async |_dir| Ok(()),
async |task, cha| {
self.ops.out(id, FileOutDelete::New(cha.len));
Ok(self.queue(task, NORMAL))
Ok(self.requeue(task, NORMAL))
},
|_err| {},
)
@ -294,7 +294,7 @@ impl File {
}
pub(crate) async fn trash(&self, task: FileInTrash) -> Result<(), FileOutTrash> {
Ok(self.queue(task, LOW))
Ok(self.requeue(task, LOW))
}
pub(crate) async fn trash_do(&self, task: FileInTrash) -> Result<(), FileOutTrash> {
@ -317,7 +317,7 @@ impl File {
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.requeue(task, LOW);
})
},
|err| {
@ -364,7 +364,7 @@ impl File {
{
task.retry += 1;
self.ops.out(task.id, FileOutDownloadDo::Log(format!("Retrying due to error: {e}")));
return Ok(self.queue(task, LOW));
return Ok(self.requeue(task, LOW));
}
Err(e) => ctx!(task, Err(e))?,
}
@ -385,7 +385,7 @@ impl File {
Ok(c) if c.mtime == cha.mtime => {}
Ok(c) => {
self.ops.out(id, FileOutUpload::New(c.len));
self.queue(task, LOW);
self.requeue(task, LOW);
}
Err(e) if e.kind() == NotFound => {}
Err(e) => ctx!(task, Err(e))?,
@ -464,7 +464,12 @@ impl File {
impl File {
#[inline]
fn queue(&self, r#in: impl Into<TaskIn>, priority: u8) {
_ = self.r#macro.try_send(r#in.into(), priority);
pub(crate) fn submit(&self, r#in: impl Into<FileIn>, priority: u8) {
_ = self.tx.try_send(r#in.into(), priority);
}
#[inline]
fn requeue(&self, r#in: impl Into<FileIn>, priority: u8) {
_ = self.tx.try_send(r#in.into().into_doable(), priority);
}
}

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

@ -4,6 +4,81 @@ use tokio::sync::mpsc;
use yazi_fs::cha::Cha;
use yazi_shared::{CompletionToken, Id, url::UrlBuf};
#[derive(Debug)]
pub(crate) enum FileIn {
Copy(FileInCopy),
CopyDo(FileInCopy),
Cut(FileInCut),
CutDo(FileInCut),
Link(FileInLink),
LinkDo(FileInLink),
Hardlink(FileInHardlink),
HardlinkDo(FileInHardlink),
Delete(FileInDelete),
DeleteDo(FileInDelete),
Trash(FileInTrash),
TrashDo(FileInTrash),
Download(FileInDownload),
DownloadDo(FileInDownload),
Upload(FileInUpload),
UploadDo(FileInUpload),
}
impl_from_in! {
Copy(FileInCopy),
Cut(FileInCut),
Link(FileInLink),
Hardlink(FileInHardlink),
Delete(FileInDelete),
Trash(FileInTrash),
Download(FileInDownload),
Upload(FileInUpload),
}
impl FileIn {
pub(crate) fn id(&self) -> Id {
match self {
Self::Copy(r#in) => r#in.id,
Self::CopyDo(r#in) => r#in.id,
Self::Cut(r#in) => r#in.id,
Self::CutDo(r#in) => r#in.id,
Self::Link(r#in) => r#in.id,
Self::LinkDo(r#in) => r#in.id,
Self::Hardlink(r#in) => r#in.id,
Self::HardlinkDo(r#in) => r#in.id,
Self::Delete(r#in) => r#in.id,
Self::DeleteDo(r#in) => r#in.id,
Self::Trash(r#in) => r#in.id,
Self::TrashDo(r#in) => r#in.id,
Self::Download(r#in) => r#in.id,
Self::DownloadDo(r#in) => r#in.id,
Self::Upload(r#in) => r#in.id,
Self::UploadDo(r#in) => r#in.id,
}
}
pub(crate) fn into_doable(self) -> Self {
match self {
Self::Copy(r#in) => Self::CopyDo(r#in),
Self::CopyDo(_) => self,
Self::Cut(r#in) => Self::CutDo(r#in),
Self::CutDo(_) => self,
Self::Link(r#in) => Self::LinkDo(r#in),
Self::LinkDo(_) => self,
Self::Hardlink(r#in) => Self::HardlinkDo(r#in),
Self::HardlinkDo(_) => self,
Self::Delete(r#in) => Self::DeleteDo(r#in),
Self::DeleteDo(_) => self,
Self::Trash(r#in) => Self::TrashDo(r#in),
Self::TrashDo(_) => self,
Self::Download(r#in) => Self::DownloadDo(r#in),
Self::DownloadDo(_) => self,
Self::Upload(r#in) => Self::UploadDo(r#in),
Self::UploadDo(_) => self,
}
}
}
// --- Copy
#[derive(Clone, Debug)]
pub(crate) struct FileInCopy {

9
yazi-scheduler/src/file/macros.rs Normal file
View file

@ -0,0 +1,9 @@
macro_rules! impl_from_in {
($($variant:ident($type:ty)),* $(,)?) => {
$(
impl From<$type> for $crate::file::FileIn {
fn from(value: $type) -> Self { Self::$variant(value) }
}
)*
};
}

3
yazi-scheduler/src/file/mod.rs
View file

@ -1 +1,4 @@
#[macro_use]
mod macros;
yazi_macro::mod_flat!(file out progress r#in transaction traverse);

18
yazi-scheduler/src/hook/hook.rs
View file

@ -7,16 +7,21 @@ use yazi_fs::ok_or_not_found;
use yazi_proxy::TasksProxy;
use yazi_vfs::provider;
use crate::{Ongoing, TaskOp, TaskOps, file::{FileOutCopy, FileOutCut, FileOutDelete, FileOutDownload, FileOutTrash, FileOutUpload}, hook::{HookInDelete, HookInDownload, HookInOutCopy, HookInOutCut, HookInTrash, HookInUpload}};
use crate::{Ongoing, TaskOp, TaskOps, file::{FileOutCopy, FileOutCut, FileOutDelete, FileOutDownload, FileOutTrash, FileOutUpload}, hook::{HookIn, HookInDelete, HookInDownload, HookInOutCopy, HookInOutCut, HookInTrash, HookInUpload}};
pub(crate) struct Hook {
ops: TaskOps,
ongoing: Arc<Mutex<Ongoing>>,
tx: async_priority_channel::Sender<HookIn, u8>,
}
impl Hook {
pub(crate) fn new(ops: &mpsc::UnboundedSender<TaskOp>, ongoing: &Arc<Mutex<Ongoing>>) -> Self {
Self { ops: ops.into(), ongoing: ongoing.clone() }
pub(crate) fn new(
ops: &mpsc::UnboundedSender<TaskOp>,
ongoing: &Arc<Mutex<Ongoing>>,
tx: async_priority_channel::Sender<HookIn, u8>,
) -> Self {
Self { ops: ops.into(), ongoing: ongoing.clone(), tx }
}
// --- File
@ -74,3 +79,10 @@ impl Hook {
self.ops.out(task.id, FileOutUpload::Clean);
}
}
impl Hook {
#[inline]
pub(crate) fn submit(&self, r#in: impl Into<HookIn>, priority: u8) {
_ = self.tx.try_send(r#in.into(), priority);
}
}

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

@ -2,6 +2,38 @@ use yazi_shared::{Id, url::UrlBuf};
use crate::{Task, TaskProg, file::{FileInCopy, FileInCut}};
#[derive(Debug)]
pub(crate) enum HookIn {
Copy(HookInOutCopy),
Cut(HookInOutCut),
Delete(HookInDelete),
Trash(HookInTrash),
Download(HookInDownload),
Upload(HookInUpload),
}
impl_from_in!(
Copy(HookInOutCopy),
Cut(HookInOutCut),
Delete(HookInDelete),
Trash(HookInTrash),
Download(HookInDownload),
Upload(HookInUpload),
);
impl HookIn {
pub(crate) fn id(&self) -> Id {
match self {
Self::Copy(r#in) => r#in.id,
Self::Cut(r#in) => r#in.id,
Self::Delete(r#in) => r#in.id,
Self::Trash(r#in) => r#in.id,
Self::Download(r#in) => r#in.id,
Self::Upload(r#in) => r#in.id,
}
}
}
// --- Copy
#[derive(Debug)]
pub(crate) struct HookInOutCopy {

9
yazi-scheduler/src/hook/macros.rs Normal file
View file

@ -0,0 +1,9 @@
macro_rules! impl_from_in {
($($variant:ident($type:ty)),* $(,)?) => {
$(
impl From<$type> for $crate::hook::HookIn {
fn from(value: $type) -> Self { Self::$variant(value) }
}
)*
};
}

3
yazi-scheduler/src/hook/mod.rs
View file

@ -1 +1,4 @@
#[macro_use]
mod macros;
yazi_macro::mod_flat!(hook r#in);

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

@ -1,110 +0,0 @@
use yazi_shared::Id;
use crate::{file::{FileInCopy, FileInCut, FileInDelete, FileInDownload, FileInHardlink, FileInLink, FileInTrash, FileInUpload}, hook::{HookInDelete, HookInDownload, HookInOutCopy, HookInOutCut, HookInTrash, HookInUpload}, impl_from_in, plugin::PluginInEntry, prework::{PreworkInFetch, PreworkInLoad, PreworkInSize}, process::{ProcessInBg, ProcessInBlock, ProcessInOrphan}};
#[derive(Debug)]
pub(crate) enum TaskIn {
// File
FileCopy(FileInCopy),
FileCut(FileInCut),
FileLink(FileInLink),
FileHardlink(FileInHardlink),
FileDelete(FileInDelete),
FileTrash(FileInTrash),
FileDownload(FileInDownload),
FileUpload(FileInUpload),
// Plugin
PluginEntry(PluginInEntry),
// Prework
PreworkFetch(PreworkInFetch),
PreworkLoad(PreworkInLoad),
PreworkSize(PreworkInSize),
// Process
ProcessBlock(ProcessInBlock),
ProcessOrphan(ProcessInOrphan),
ProcessBg(ProcessInBg),
// Hook
HookCopy(HookInOutCopy),
HookCut(HookInOutCut),
HookDelete(HookInDelete),
HookTrash(HookInTrash),
HookDownload(HookInDownload),
HookUpload(HookInUpload),
}
impl_from_in! {
// File
FileCopy(FileInCopy), FileCut(FileInCut), FileLink(FileInLink), FileHardlink(FileInHardlink), FileDelete(FileInDelete), FileTrash(FileInTrash), FileDownload(FileInDownload), FileUpload(FileInUpload),
// Plugin
PluginEntry(PluginInEntry),
// Prework
PreworkFetch(PreworkInFetch), PreworkLoad(PreworkInLoad), PreworkSize(PreworkInSize),
// Process
ProcessBlock(ProcessInBlock), ProcessOrphan(ProcessInOrphan), ProcessBg(ProcessInBg),
// Hook
HookCopy(HookInOutCopy), HookCut(HookInOutCut), HookDelete(HookInDelete), HookTrash(HookInTrash), HookDownload(HookInDownload), HookUpload(HookInUpload),
}
impl TaskIn {
pub fn id(&self) -> Id {
match self {
// File
Self::FileCopy(r#in) => r#in.id,
Self::FileCut(r#in) => r#in.id,
Self::FileLink(r#in) => r#in.id,
Self::FileHardlink(r#in) => r#in.id,
Self::FileDelete(r#in) => r#in.id,
Self::FileTrash(r#in) => r#in.id,
Self::FileDownload(r#in) => r#in.id,
Self::FileUpload(r#in) => r#in.id,
// Plugin
Self::PluginEntry(r#in) => r#in.id,
// Prework
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::HookCopy(r#in) => r#in.id,
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::HookUpload(r#in) => r#in.id,
}
}
pub fn is_hook(&self) -> bool {
match self {
// File
Self::FileCopy(_) => false,
Self::FileCut(_) => false,
Self::FileLink(_) => false,
Self::FileHardlink(_) => false,
Self::FileDelete(_) => false,
Self::FileTrash(_) => false,
Self::FileDownload(_) => false,
Self::FileUpload(_) => false,
// Plugin
Self::PluginEntry(_) => false,
// Prework
Self::PreworkFetch(_) => false,
Self::PreworkLoad(_) => false,
Self::PreworkSize(_) => false,
// Process
Self::ProcessBlock(_) => false,
Self::ProcessOrphan(_) => false,
Self::ProcessBg(_) => false,
// Hook
Self::HookCopy(_) => true,
Self::HookCut(_) => true,
Self::HookDelete(_) => true,
Self::HookTrash(_) => true,
Self::HookDownload(_) => true,
Self::HookUpload(_) => true,
}
}
}

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

@ -1,8 +1,8 @@
mod macros;
yazi_macro::mod_pub!(file hook plugin prework process);
yazi_macro::mod_pub!(fetch file hook plugin preload process size);
yazi_macro::mod_flat!(ongoing op out progress r#in runner scheduler snap task);
yazi_macro::mod_flat!(ongoing op out progress runner scheduler snap task);
const LOW: u8 = yazi_config::Priority::Low as u8;
const NORMAL: u8 = yazi_config::Priority::Normal as u8;

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

@ -39,17 +39,6 @@ macro_rules! progress_or_break {
};
}
#[macro_export]
macro_rules! impl_from_in {
($($variant:ident($type:ty)),* $(,)?) => {
$(
impl From<$type> for $crate::TaskIn {
fn from(value: $type) -> Self { Self::$variant(value) }
}
)*
};
}
#[macro_export]
macro_rules! impl_from_out {
($($variant:ident($type:ty)),* $(,)?) => {

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

@ -5,7 +5,7 @@ use yazi_parser::app::TaskSummary;
use yazi_shared::{CompletionToken, Id, Ids};
use super::Task;
use crate::{TaskIn, TaskProg};
use crate::{TaskProg, hook::HookIn};
#[derive(Default)]
pub struct Ongoing {
@ -23,7 +23,7 @@ impl Ongoing {
self.inner.entry(id).insert(Task::new::<T>(id, name)).into_mut()
}
pub(super) fn cancel(&mut self, id: Id) -> Option<TaskIn> {
pub(super) fn cancel(&mut self, id: Id) -> Option<HookIn> {
match self.inner.entry(id) {
Entry::Occupied(mut oe) => {
let task = oe.get_mut();

26
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}, hook::{HookInOutCopy, HookInOutCut}, impl_from_out, plugin::PluginOutEntry, prework::{PreworkOutFetch, PreworkOutLoad, PreworkOutSize}, process::{ProcessOutBg, ProcessOutBlock, ProcessOutOrphan}};
use crate::{Task, fetch::FetchOutFetch, file::{FileOutCopy, FileOutCopyDo, FileOutCut, FileOutCutDo, FileOutDelete, FileOutDeleteDo, FileOutDownload, FileOutDownloadDo, FileOutHardlink, FileOutHardlinkDo, FileOutLink, FileOutTrash, FileOutUpload, FileOutUploadDo}, hook::{HookInOutCopy, HookInOutCut}, impl_from_out, plugin::PluginOutEntry, preload::PreloadOut, process::{ProcessOutBg, ProcessOutBlock, ProcessOutOrphan}, size::SizeOut};
#[derive(Debug)]
pub(super) enum TaskOut {
@ -19,10 +19,12 @@ pub(super) enum TaskOut {
FileUploadDo(FileOutUploadDo),
// Plugin
PluginEntry(PluginOutEntry),
// Prework
PreworkFetch(PreworkOutFetch),
PreworkLoad(PreworkOutLoad),
PreworkSize(PreworkOutSize),
// Fetch
Fetch(FetchOutFetch),
// Preload
Preload(PreloadOut),
// Size
Size(SizeOut),
// Process
ProcessBlock(ProcessOutBlock),
ProcessOrphan(ProcessOutOrphan),
@ -37,8 +39,12 @@ impl_from_out! {
FileCopy(FileOutCopy), FileCopyDo(FileOutCopyDo), FileCut(FileOutCut), FileCutDo(FileOutCutDo), FileLink(FileOutLink), FileHardlink(FileOutHardlink), FileHardlinkDo(FileOutHardlinkDo), FileDelete(FileOutDelete), FileDeleteDo(FileOutDeleteDo), FileTrash(FileOutTrash), FileDownload(FileOutDownload), FileDownloadDo(FileOutDownloadDo), FileUpload(FileOutUpload), FileUploadDo(FileOutUploadDo),
// Plugin
PluginEntry(PluginOutEntry),
// Prework
PreworkFetch(PreworkOutFetch), PreworkLoad(PreworkOutLoad), PreworkSize(PreworkOutSize),
// Fetch
Fetch(FetchOutFetch),
// Preload
Preload(PreloadOut),
// Size
Size(SizeOut),
// Process
ProcessBlock(ProcessOutBlock), ProcessOrphan(ProcessOutOrphan), ProcessBg(ProcessOutBg),
// Hook
@ -66,9 +72,9 @@ impl TaskOut {
// Plugin
Self::PluginEntry(out) => out.reduce(task),
// Prework
Self::PreworkFetch(out) => out.reduce(task),
Self::PreworkLoad(out) => out.reduce(task),
Self::PreworkSize(out) => out.reduce(task),
Self::Fetch(out) => out.reduce(task),
Self::Preload(out) => out.reduce(task),
Self::Size(out) => out.reduce(task),
// Process
Self::ProcessBlock(out) => out.reduce(task),
Self::ProcessOrphan(out) => out.reduce(task),

16
yazi-scheduler/src/plugin/in.rs
View file

@ -1,6 +1,22 @@
use yazi_parser::app::PluginOpt;
use yazi_shared::Id;
#[derive(Debug)]
pub(crate) enum PluginIn {
Entry(PluginInEntry),
}
impl_from_in!(Entry(PluginInEntry));
impl PluginIn {
pub(crate) fn id(&self) -> Id {
match self {
Self::Entry(r#in) => r#in.id,
}
}
}
// --- Entry
#[derive(Debug)]
pub(crate) struct PluginInEntry {
pub(crate) id: Id,

9
yazi-scheduler/src/plugin/macros.rs Normal file
View file

@ -0,0 +1,9 @@
macro_rules! impl_from_in {
($($variant:ident($type:ty)),* $(,)?) => {
$(
impl From<$type> for $crate::plugin::PluginIn {
fn from(value: $type) -> Self { Self::$variant(value) }
}
)*
};
}

3
yazi-scheduler/src/plugin/mod.rs
View file

@ -1 +1,4 @@
#[macro_use]
mod macros;
yazi_macro::mod_flat!(out plugin progress r#in);

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

@ -3,26 +3,22 @@ use tokio::sync::mpsc;
use yazi_plugin::isolate;
use super::PluginInEntry;
use crate::{HIGH, TaskIn, TaskOp, TaskOps, plugin::PluginOutEntry};
use crate::{TaskOp, TaskOps, plugin::{PluginIn, PluginOutEntry}};
pub(crate) struct Plugin {
ops: TaskOps,
r#macro: async_priority_channel::Sender<TaskIn, u8>,
ops: TaskOps,
tx: async_priority_channel::Sender<PluginIn, u8>,
}
impl Plugin {
pub(crate) fn new(
ops: &mpsc::UnboundedSender<TaskOp>,
r#macro: &async_priority_channel::Sender<TaskIn, u8>,
tx: async_priority_channel::Sender<PluginIn, u8>,
) -> Self {
Self { ops: ops.into(), r#macro: r#macro.clone() }
Self { ops: ops.into(), tx }
}
pub(crate) async fn entry(&self, task: PluginInEntry) -> Result<(), PluginOutEntry> {
Ok(self.queue(task, HIGH))
}
pub(crate) async fn entry_do(&self, task: PluginInEntry) -> Result<(), PluginOutEntry> {
isolate::entry(task.opt).await?;
Ok(self.ops.out(task.id, PluginOutEntry::Succ))
}
@ -30,7 +26,7 @@ impl Plugin {
impl Plugin {
#[inline]
fn queue(&self, r#in: impl Into<TaskIn>, priority: u8) {
_ = self.r#macro.try_send(r#in.into(), priority);
pub(crate) fn submit(&self, r#in: impl Into<PluginIn>, priority: u8) {
_ = self.tx.try_send(r#in.into(), priority);
}
}

13
yazi-scheduler/src/preload/in.rs Normal file
View file

@ -0,0 +1,13 @@
use yazi_config::plugin::Preloader;
use yazi_shared::Id;
#[derive(Clone, Debug)]
pub(crate) struct PreloadIn {
pub(crate) id: Id,
pub(crate) plugin: &'static Preloader,
pub(crate) target: yazi_fs::File,
}
impl PreloadIn {
pub(crate) fn id(&self) -> Id { self.id }
}

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

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

26
yazi-scheduler/src/preload/out.rs Normal file
View file

@ -0,0 +1,26 @@
use crate::{Task, TaskProg};
#[derive(Debug)]
pub(crate) enum PreloadOut {
Succ,
Fail(String),
}
impl From<mlua::Error> for PreloadOut {
fn from(value: mlua::Error) -> Self { Self::Fail(value.to_string()) }
}
impl PreloadOut {
pub(crate) fn reduce(self, task: &mut Task) {
let TaskProg::Preload(prog) = &mut task.prog else { return };
match self {
Self::Succ => {
prog.state = Some(true);
}
Self::Fail(reason) => {
prog.state = Some(false);
task.log(reason);
}
}
}
}

67
yazi-scheduler/src/preload/preload.rs Normal file
View file

@ -0,0 +1,67 @@
use std::num::NonZeroUsize;
use anyhow::Result;
use lru::LruCache;
use parking_lot::Mutex;
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
use tracing::error;
use yazi_config::Priority;
use yazi_fs::FsHash64;
use yazi_plugin::isolate;
use crate::{HIGH, LOW, NORMAL, TaskOp, TaskOps, preload::{PreloadIn, PreloadOut}};
pub struct Preload {
ops: TaskOps,
tx: async_priority_channel::Sender<PreloadIn, u8>,
pub loaded: Mutex<LruCache<u64, u16>>,
pub loading: Mutex<LruCache<u64, CancellationToken>>,
}
impl Preload {
pub(crate) fn new(
ops: &mpsc::UnboundedSender<TaskOp>,
tx: async_priority_channel::Sender<PreloadIn, u8>,
) -> Self {
Self {
ops: ops.into(),
tx,
loaded: Mutex::new(LruCache::new(NonZeroUsize::new(4096).unwrap())),
loading: Mutex::new(LruCache::new(NonZeroUsize::new(256).unwrap())),
}
}
pub(crate) async fn preload(&self, task: PreloadIn) -> Result<(), PreloadOut> {
let ct = CancellationToken::new();
if let Some(ct) = self.loading.lock().put(task.target.url.hash_u64(), ct.clone()) {
ct.cancel();
}
let hash = task.target.hash_u64();
let (ok, err) = isolate::preload(&task.plugin.run, task.target, ct).await?;
if !ok {
self.loaded.lock().get_mut(&hash).map(|x| *x &= !(1 << task.plugin.idx));
}
if let Some(e) = err {
error!("Error when running preloader `{}`:\n{e}", task.plugin.run.name);
}
Ok(self.ops.out(task.id, PreloadOut::Succ))
}
}
impl Preload {
pub(crate) fn submit(&self, r#in: PreloadIn) {
let priority = match r#in.plugin.prio {
Priority::Low => LOW,
Priority::Normal => NORMAL,
Priority::High => HIGH,
};
_ = self.tx.try_send(r#in.into(), priority);
}
}

32
yazi-scheduler/src/preload/progress.rs Normal file
View file

@ -0,0 +1,32 @@
use serde::Serialize;
use yazi_parser::app::TaskSummary;
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize)]
pub struct PreloadProg {
pub state: Option<bool>,
}
impl From<PreloadProg> for TaskSummary {
fn from(value: PreloadProg) -> Self {
Self {
total: 1,
success: (value.state == Some(true)) as u32,
failed: (value.state == Some(false)) as u32,
percent: value.percent().map(Into::into),
}
}
}
impl PreloadProg {
pub fn cooked(self) -> bool { self.state == Some(true) }
pub fn running(self) -> bool { self.state.is_none() }
pub fn success(self) -> bool { self.cooked() }
pub fn failed(self) -> bool { self.state == Some(false) }
pub fn cleaned(self) -> Option<bool> { None }
pub fn percent(self) -> Option<f32> { None }
}

25
yazi-scheduler/src/prework/in.rs
View file

@ -1,25 +0,0 @@
use std::sync::Arc;
use yazi_config::plugin::{Fetcher, Preloader};
use yazi_shared::{Id, Throttle, url::UrlBuf};
#[derive(Debug)]
pub(crate) struct PreworkInFetch {
pub(crate) id: Id,
pub(crate) plugin: &'static Fetcher,
pub(crate) targets: Vec<yazi_fs::File>,
}
#[derive(Clone, Debug)]
pub(crate) struct PreworkInLoad {
pub(crate) id: Id,
pub(crate) plugin: &'static Preloader,
pub(crate) target: yazi_fs::File,
}
#[derive(Debug)]
pub(crate) struct PreworkInSize {
pub(crate) id: Id,
pub(crate) target: UrlBuf,
pub(crate) throttle: Arc<Throttle<(UrlBuf, u64)>>,
}

1
yazi-scheduler/src/prework/mod.rs
View file

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

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

@ -1,70 +0,0 @@
use crate::{Task, TaskProg};
// --- Fetch
#[derive(Debug)]
pub(crate) enum PreworkOutFetch {
Succ,
Fail(String),
}
impl From<mlua::Error> for PreworkOutFetch {
fn from(value: mlua::Error) -> Self { Self::Fail(value.to_string()) }
}
impl PreworkOutFetch {
pub(crate) fn reduce(self, task: &mut Task) {
let TaskProg::PreworkFetch(prog) = &mut task.prog else { return };
match self {
Self::Succ => {
prog.state = Some(true);
}
Self::Fail(reason) => {
prog.state = Some(false);
task.log(reason);
}
}
}
}
// --- Load
#[derive(Debug)]
pub(crate) enum PreworkOutLoad {
Succ,
Fail(String),
}
impl From<mlua::Error> for PreworkOutLoad {
fn from(value: mlua::Error) -> Self { Self::Fail(value.to_string()) }
}
impl PreworkOutLoad {
pub(crate) fn reduce(self, task: &mut Task) {
let TaskProg::PreworkLoad(prog) = &mut task.prog else { return };
match self {
Self::Succ => {
prog.state = Some(true);
}
Self::Fail(reason) => {
prog.state = Some(false);
task.log(reason);
}
}
}
}
// --- Size
#[derive(Debug)]
pub(crate) enum PreworkOutSize {
Done,
}
impl PreworkOutSize {
pub(crate) fn reduce(self, task: &mut Task) {
let TaskProg::PreworkSize(prog) = &mut task.prog else { return };
match self {
Self::Done => {
prog.done = true;
}
}
}
}

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

@ -1,123 +0,0 @@
use std::num::NonZeroUsize;
use anyhow::Result;
use hashbrown::{HashMap, HashSet};
use lru::LruCache;
use parking_lot::{Mutex, RwLock};
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
use tracing::error;
use yazi_config::Priority;
use yazi_fs::{FilesOp, FsHash64};
use yazi_plugin::isolate;
use yazi_shared::{event::CmdCow, url::{UrlBuf, UrlLike}};
use yazi_vfs::provider;
use super::{PreworkInFetch, PreworkInLoad, PreworkInSize};
use crate::{HIGH, NORMAL, TaskIn, TaskOp, TaskOps, prework::{PreworkOutFetch, PreworkOutLoad, PreworkOutSize}};
pub struct Prework {
ops: TaskOps,
r#macro: async_priority_channel::Sender<TaskIn, u8>,
pub loaded: Mutex<LruCache<u64, u32>>,
pub loading: Mutex<LruCache<u64, CancellationToken>>,
pub sizing: RwLock<HashSet<UrlBuf>>,
}
impl Prework {
pub(crate) fn new(
ops: &mpsc::UnboundedSender<TaskOp>,
r#macro: &async_priority_channel::Sender<TaskIn, u8>,
) -> Self {
Self {
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())),
sizing: Default::default(),
}
}
pub(crate) async fn fetch(&self, task: PreworkInFetch) -> Result<(), PreworkOutFetch> {
match task.plugin.prio {
Priority::Low => Ok(self.queue(task, NORMAL)),
Priority::Normal => Ok(self.queue(task, HIGH)),
Priority::High => self.fetch_do(task).await,
}
}
pub(crate) async fn fetch_do(&self, task: PreworkInFetch) -> Result<(), PreworkOutFetch> {
let hashes: Vec<_> = task.targets.iter().map(|f| f.hash_u64()).collect();
let (state, err) = isolate::fetch(CmdCow::from(&task.plugin.run), task.targets).await?;
let mut loaded = self.loaded.lock();
for (_, h) in hashes.into_iter().enumerate().filter(|&(i, _)| !state.get(i)) {
loaded.get_mut(&h).map(|x| *x &= !(1 << task.plugin.idx));
}
if let Some(e) = err {
error!("Error when running fetcher `{}`:\n{e}", task.plugin.run.name);
}
Ok(self.ops.out(task.id, PreworkOutFetch::Succ))
}
pub(crate) async fn load(&self, task: PreworkInLoad) -> Result<(), PreworkOutLoad> {
match task.plugin.prio {
Priority::Low => Ok(self.queue(task, NORMAL)),
Priority::Normal => Ok(self.queue(task, HIGH)),
Priority::High => self.load_do(task).await,
}
}
pub(crate) async fn load_do(&self, task: PreworkInLoad) -> Result<(), PreworkOutLoad> {
let ct = CancellationToken::new();
if let Some(ct) = self.loading.lock().put(task.target.url.hash_u64(), ct.clone()) {
ct.cancel();
}
let hash = task.target.hash_u64();
let (ok, err) = isolate::preload(&task.plugin.run, task.target, ct).await?;
if !ok {
self.loaded.lock().get_mut(&hash).map(|x| *x &= !(1 << task.plugin.idx));
}
if let Some(e) = err {
error!("Error when running preloader `{}`:\n{e}", task.plugin.run.name);
}
Ok(self.ops.out(task.id, PreworkOutLoad::Succ))
}
pub(crate) async fn size(&self, task: PreworkInSize) -> Result<(), PreworkOutSize> {
self.size_do(task).await
}
pub(crate) async fn size_do(&self, task: PreworkInSize) -> Result<(), PreworkOutSize> {
let length = provider::calculate(&task.target).await.unwrap_or(0);
task.throttle.done((task.target, length), |buf| {
{
let mut loading = self.sizing.write();
for (path, _) in &buf {
loading.remove(path);
}
}
let parent = buf[0].0.parent().unwrap();
FilesOp::Size(
parent.into(),
HashMap::from_iter(buf.into_iter().map(|(u, s)| (u.urn().into(), s))),
)
.emit();
});
Ok(self.ops.out(task.id, PreworkOutSize::Done))
}
}
impl Prework {
#[inline]
fn queue(&self, r#in: impl Into<TaskIn>, priority: u8) {
_ = self.r#macro.try_send(r#in.into(), priority);
}
}

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

@ -1,95 +0,0 @@
use serde::Serialize;
use yazi_parser::app::TaskSummary;
// --- Fetch
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize)]
pub struct PreworkProgFetch {
pub state: Option<bool>,
}
impl From<PreworkProgFetch> for TaskSummary {
fn from(value: PreworkProgFetch) -> Self {
Self {
total: 1,
success: (value.state == Some(true)) as u32,
failed: (value.state == Some(false)) as u32,
percent: value.percent().map(Into::into),
}
}
}
impl PreworkProgFetch {
pub fn cooked(self) -> bool { self.state == Some(true) }
pub fn running(self) -> bool { self.state.is_none() }
pub fn success(self) -> bool { self.cooked() }
pub fn failed(self) -> bool { self.state == Some(false) }
pub fn cleaned(self) -> Option<bool> { None }
pub fn percent(self) -> Option<f32> { None }
}
// --- Load
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize)]
pub struct PreworkProgLoad {
pub state: Option<bool>,
}
impl From<PreworkProgLoad> for TaskSummary {
fn from(value: PreworkProgLoad) -> Self {
Self {
total: 1,
success: (value.state == Some(true)) as u32,
failed: (value.state == Some(false)) as u32,
percent: value.percent().map(Into::into),
}
}
}
impl PreworkProgLoad {
pub fn cooked(self) -> bool { self.state == Some(true) }
pub fn running(self) -> bool { self.state.is_none() }
pub fn success(self) -> bool { self.cooked() }
pub fn failed(self) -> bool { self.state == Some(false) }
pub fn cleaned(self) -> Option<bool> { None }
pub fn percent(self) -> Option<f32> { None }
}
// --- Size
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize)]
pub struct PreworkProgSize {
pub done: bool,
}
impl From<PreworkProgSize> for TaskSummary {
fn from(value: PreworkProgSize) -> Self {
Self {
total: 1,
success: value.done as u32,
failed: 0,
percent: value.percent().map(Into::into),
}
}
}
impl PreworkProgSize {
pub fn cooked(self) -> bool { self.done }
pub fn running(self) -> bool { !self.done }
pub fn success(self) -> bool { self.cooked() }
pub fn failed(self) -> bool { false }
pub fn cleaned(self) -> Option<bool> { None }
pub fn percent(self) -> Option<f32> { None }
}

19
yazi-scheduler/src/process/in.rs
View file

@ -4,6 +4,25 @@ use yazi_shared::{CompletionToken, Id, url::UrlCow};
use super::ShellOpt;
#[derive(Debug)]
pub(crate) enum ProcessIn {
Block(ProcessInBlock),
Orphan(ProcessInOrphan),
Bg(ProcessInBg),
}
impl_from_in!(Block(ProcessInBlock), Orphan(ProcessInOrphan), Bg(ProcessInBg));
impl ProcessIn {
pub(crate) fn id(&self) -> Id {
match self {
Self::Block(r#in) => r#in.id,
Self::Orphan(r#in) => r#in.id,
Self::Bg(r#in) => r#in.id,
}
}
}
// --- Block
#[derive(Debug)]
pub(crate) struct ProcessInBlock {

9
yazi-scheduler/src/process/macros.rs Normal file
View file

@ -0,0 +1,9 @@
macro_rules! impl_from_in {
($($variant:ident($type:ty)),* $(,)?) => {
$(
impl From<$type> for $crate::process::ProcessIn {
fn from(value: $type) -> Self { Self::$variant(value) }
}
)*
};
}

3
yazi-scheduler/src/process/mod.rs
View file

@ -1 +1,4 @@
#[macro_use]
mod macros;
yazi_macro::mod_flat!(out process progress r#in shell);

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

@ -4,14 +4,20 @@ use yazi_binding::Permit;
use yazi_proxy::{AppProxy, HIDER, NotifyProxy};
use super::{ProcessInBg, ProcessInBlock, ProcessInOrphan, ShellOpt};
use crate::{TaskOp, TaskOps, process::{ProcessOutBg, ProcessOutBlock, ProcessOutOrphan}};
use crate::{TaskOp, TaskOps, process::{ProcessIn, ProcessOutBg, ProcessOutBlock, ProcessOutOrphan}};
pub(crate) struct Process {
ops: TaskOps,
tx: async_priority_channel::Sender<ProcessIn, u8>,
}
impl Process {
pub(crate) fn new(ops: &mpsc::UnboundedSender<TaskOp>) -> Self { Self { ops: ops.into() } }
pub(crate) fn new(
ops: &mpsc::UnboundedSender<TaskOp>,
tx: async_priority_channel::Sender<ProcessIn, u8>,
) -> Self {
Self { ops: ops.into(), tx }
}
pub(crate) async fn block(&self, task: ProcessInBlock) -> Result<(), ProcessOutBlock> {
let _permit = Permit::new(HIDER.acquire().await.unwrap(), AppProxy::resume());
@ -85,3 +91,10 @@ impl Process {
Ok(self.ops.out(task.id, ProcessOutBg::Succ))
}
}
impl Process {
#[inline]
pub(crate) fn submit(&self, r#in: impl Into<ProcessIn>, priority: u8) {
_ = self.tx.try_send(r#in.into(), priority);
}
}

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

@ -1,7 +1,7 @@
use serde::Serialize;
use yazi_parser::app::TaskSummary;
use crate::{file::{FileProgCopy, FileProgCut, FileProgDelete, FileProgDownload, FileProgHardlink, FileProgLink, FileProgTrash, FileProgUpload}, impl_from_prog, plugin::PluginProgEntry, prework::{PreworkProgFetch, PreworkProgLoad, PreworkProgSize}, process::{ProcessProgBg, ProcessProgBlock, ProcessProgOrphan}};
use crate::{fetch::FetchProg, file::{FileProgCopy, FileProgCut, FileProgDelete, FileProgDownload, FileProgHardlink, FileProgLink, FileProgTrash, FileProgUpload}, impl_from_prog, plugin::PluginProgEntry, preload::PreloadProg, process::{ProcessProgBg, ProcessProgBlock, ProcessProgOrphan}, size::SizeProg};
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(tag = "kind")]
@ -17,10 +17,12 @@ pub enum TaskProg {
FileUpload(FileProgUpload),
// Plugin
PluginEntry(PluginProgEntry),
// Prework
PreworkFetch(PreworkProgFetch),
PreworkLoad(PreworkProgLoad),
PreworkSize(PreworkProgSize),
// Fetch
Fetch(FetchProg),
// Preload
Preload(PreloadProg),
// Size
Size(SizeProg),
// Process
ProcessBlock(ProcessProgBlock),
ProcessOrphan(ProcessProgOrphan),
@ -32,8 +34,12 @@ impl_from_prog! {
FileCopy(FileProgCopy), FileCut(FileProgCut), FileLink(FileProgLink), FileHardlink(FileProgHardlink), FileDelete(FileProgDelete), FileTrash(FileProgTrash), FileDownload(FileProgDownload), FileUpload(FileProgUpload),
// Plugin
PluginEntry(PluginProgEntry),
// Prework
PreworkFetch(PreworkProgFetch), PreworkLoad(PreworkProgLoad), PreworkSize(PreworkProgSize),
// Fetch
Fetch(FetchProg),
// Preload
Preload(PreloadProg),
// Size
Size(SizeProg),
// Process
ProcessBlock(ProcessProgBlock), ProcessOrphan(ProcessProgOrphan), ProcessBg(ProcessProgBg),
}
@ -53,9 +59,9 @@ impl From<TaskProg> for TaskSummary {
// Plugin
TaskProg::PluginEntry(p) => p.into(),
// Prework
TaskProg::PreworkFetch(p) => p.into(),
TaskProg::PreworkLoad(p) => p.into(),
TaskProg::PreworkSize(p) => p.into(),
TaskProg::Fetch(p) => p.into(),
TaskProg::Preload(p) => p.into(),
TaskProg::Size(p) => p.into(),
// Process
TaskProg::ProcessBlock(p) => p.into(),
TaskProg::ProcessOrphan(p) => p.into(),
@ -79,9 +85,9 @@ impl TaskProg {
// Plugin
Self::PluginEntry(p) => p.cooked(),
// Prework
Self::PreworkFetch(p) => p.cooked(),
Self::PreworkLoad(p) => p.cooked(),
Self::PreworkSize(p) => p.cooked(),
Self::Fetch(p) => p.cooked(),
Self::Preload(p) => p.cooked(),
Self::Size(p) => p.cooked(),
// Process
Self::ProcessBlock(p) => p.cooked(),
Self::ProcessOrphan(p) => p.cooked(),
@ -103,9 +109,9 @@ impl TaskProg {
// Plugin
Self::PluginEntry(p) => p.running(),
// Prework
Self::PreworkFetch(p) => p.running(),
Self::PreworkLoad(p) => p.running(),
Self::PreworkSize(p) => p.running(),
Self::Fetch(p) => p.running(),
Self::Preload(p) => p.running(),
Self::Size(p) => p.running(),
// Process
Self::ProcessBlock(p) => p.running(),
Self::ProcessOrphan(p) => p.running(),
@ -127,9 +133,9 @@ impl TaskProg {
// Plugin
Self::PluginEntry(p) => p.success(),
// Prework
Self::PreworkFetch(p) => p.success(),
Self::PreworkLoad(p) => p.success(),
Self::PreworkSize(p) => p.success(),
Self::Fetch(p) => p.success(),
Self::Preload(p) => p.success(),
Self::Size(p) => p.success(),
// Process
Self::ProcessBlock(p) => p.success(),
Self::ProcessOrphan(p) => p.success(),
@ -151,9 +157,9 @@ impl TaskProg {
// Plugin
Self::PluginEntry(p) => p.failed(),
// Prework
Self::PreworkFetch(p) => p.failed(),
Self::PreworkLoad(p) => p.failed(),
Self::PreworkSize(p) => p.failed(),
Self::Fetch(p) => p.failed(),
Self::Preload(p) => p.failed(),
Self::Size(p) => p.failed(),
// Process
Self::ProcessBlock(p) => p.failed(),
Self::ProcessOrphan(p) => p.failed(),
@ -175,9 +181,9 @@ impl TaskProg {
// Plugin
Self::PluginEntry(p) => p.cleaned(),
// Prework
Self::PreworkFetch(p) => p.cleaned(),
Self::PreworkLoad(p) => p.cleaned(),
Self::PreworkSize(p) => p.cleaned(),
Self::Fetch(p) => p.cleaned(),
Self::Preload(p) => p.cleaned(),
Self::Size(p) => p.cleaned(),
// Process
Self::ProcessBlock(p) => p.cleaned(),
Self::ProcessOrphan(p) => p.cleaned(),
@ -199,9 +205,9 @@ impl TaskProg {
// Plugin
Self::PluginEntry(p) => p.percent(),
// Prework
Self::PreworkFetch(p) => p.percent(),
Self::PreworkLoad(p) => p.percent(),
Self::PreworkSize(p) => p.percent(),
Self::Fetch(p) => p.percent(),
Self::Preload(p) => p.percent(),
Self::Size(p) => p.percent(),
// Process
Self::ProcessBlock(p) => p.percent(),
Self::ProcessOrphan(p) => p.percent(),
@ -223,9 +229,9 @@ impl TaskProg {
// Plugin
Self::PluginEntry(_) => true,
// Prework
Self::PreworkFetch(_) => false,
Self::PreworkLoad(_) => false,
Self::PreworkSize(_) => false,
Self::Fetch(_) => false,
Self::Preload(_) => false,
Self::Size(_) => false,
// Process
Self::ProcessBlock(_) => true,
Self::ProcessOrphan(_) => true,

330
yazi-scheduler/src/runner.rs
View file

@ -1,76 +1,294 @@
use std::sync::Arc;
use crate::{TaskIn, TaskOut, file::File, hook::Hook, plugin::Plugin, prework::Prework, process::Process};
use parking_lot::Mutex;
use tokio::{select, sync::mpsc, task::JoinHandle};
use yazi_config::YAZI;
use crate::{LOW, Ongoing, TaskOp, TaskOps, TaskOut, fetch::{Fetch, FetchIn}, file::{File, FileIn}, hook::{Hook, HookIn}, plugin::{Plugin, PluginIn}, preload::{Preload, PreloadIn}, process::{Process, ProcessIn}, size::{Size, SizeIn}};
#[derive(Clone)]
pub struct Runner {
pub(super) file: Arc<File>,
pub(super) plugin: Arc<Plugin>,
pub prework: Arc<Prework>,
pub fetch: Arc<Fetch>,
pub preload: Arc<Preload>,
pub size: Arc<Size>,
pub(super) process: Arc<Process>,
pub(super) hook: Arc<Hook>,
pub ops: TaskOps,
pub ongoing: Arc<Mutex<Ongoing>>,
}
impl Runner {
pub(super) async fn micro(&self, r#in: TaskIn) -> Result<(), TaskOut> {
pub(super) fn make() -> (Self, Vec<JoinHandle<()>>) {
let (file_tx, file_rx) = async_priority_channel::unbounded();
let (plugin_tx, plugin_rx) = async_priority_channel::unbounded();
let (fetch_tx, fetch_rx) = async_priority_channel::unbounded();
let (preload_tx, preload_rx) = async_priority_channel::unbounded();
let (size_tx, size_rx) = async_priority_channel::unbounded();
let (process_tx, process_rx) = async_priority_channel::unbounded();
let (hook_tx, hook_rx) = async_priority_channel::unbounded();
let (op_tx, op_rx) = mpsc::unbounded_channel();
let ongoing = Arc::new(Mutex::new(Ongoing::default()));
let me = Self {
file: Arc::new(File::new(&op_tx, file_tx)),
plugin: Arc::new(Plugin::new(&op_tx, plugin_tx)),
fetch: Arc::new(Fetch::new(&op_tx, fetch_tx)),
preload: Arc::new(Preload::new(&op_tx, preload_tx)),
size: Arc::new(Size::new(&op_tx, size_tx)),
process: Arc::new(Process::new(&op_tx, process_tx)),
hook: Arc::new(Hook::new(&op_tx, &ongoing, hook_tx)),
ops: TaskOps(op_tx),
ongoing,
};
let handles = []
.into_iter()
.chain((0..YAZI.tasks.file_workers).map(|_| me.file(file_rx.clone())))
.chain((0..YAZI.tasks.plugin_workers).map(|_| me.plugin(plugin_rx.clone())))
.chain((0..YAZI.tasks.fetch_workers).map(|_| me.fetch(fetch_rx.clone())))
.chain((0..YAZI.tasks.preload_workers).map(|_| me.preload(preload_rx.clone())))
.chain((0..3).map(|_| me.size(size_rx.clone())))
.chain((0..YAZI.tasks.process_workers).map(|_| me.process(process_rx.clone())))
.chain((0..3).map(|_| me.hook(hook_rx.clone())))
.chain([me.op(op_rx)])
.collect();
(me, handles)
}
fn file(&self, rx: async_priority_channel::Receiver<FileIn, u8>) -> JoinHandle<()> {
let me = self.clone();
tokio::spawn(async move {
loop {
if let Ok((r#in, _)) = rx.recv().await {
let id = r#in.id();
let Some(token) = me.ongoing.lock().get_token(id) else {
continue;
};
let result = select! {
r = me.file_do(r#in) => r,
false = token.future() => Ok(())
};
if let Err(out) = result {
me.ops.out(id, out);
}
}
}
})
}
async fn file_do(&self, r#in: FileIn) -> 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::HookCopy(r#in) => Ok(self.hook.copy(r#in).await),
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::HookUpload(r#in) => Ok(self.hook.upload(r#in).await),
FileIn::Copy(r#in) => self.file.copy(r#in).await.map_err(Into::into),
FileIn::CopyDo(r#in) => self.file.copy_do(r#in).await.map_err(Into::into),
FileIn::Cut(r#in) => self.file.cut(r#in).await.map_err(Into::into),
FileIn::CutDo(r#in) => self.file.cut_do(r#in).await.map_err(Into::into),
FileIn::Link(r#in) => self.file.link(r#in).await.map_err(Into::into),
FileIn::LinkDo(r#in) => self.file.link_do(r#in).await.map_err(Into::into),
FileIn::Hardlink(r#in) => self.file.hardlink(r#in).await.map_err(Into::into),
FileIn::HardlinkDo(r#in) => self.file.hardlink_do(r#in).await.map_err(Into::into),
FileIn::Delete(r#in) => self.file.delete(r#in).await.map_err(Into::into),
FileIn::DeleteDo(r#in) => self.file.delete_do(r#in).await.map_err(Into::into),
FileIn::Trash(r#in) => self.file.trash(r#in).await.map_err(Into::into),
FileIn::TrashDo(r#in) => self.file.trash_do(r#in).await.map_err(Into::into),
FileIn::Download(r#in) => self.file.download(r#in).await.map_err(Into::into),
FileIn::DownloadDo(r#in) => self.file.download_do(r#in).await.map_err(Into::into),
FileIn::Upload(r#in) => self.file.upload(r#in).await.map_err(Into::into),
FileIn::UploadDo(r#in) => self.file.upload_do(r#in).await.map_err(Into::into),
}
}
pub(super) async fn r#macro(&self, r#in: TaskIn) -> Result<(), TaskOut> {
fn plugin(&self, rx: async_priority_channel::Receiver<PluginIn, u8>) -> JoinHandle<()> {
let me = self.clone();
tokio::spawn(async move {
loop {
if let Ok((r#in, _)) = rx.recv().await {
let id = r#in.id();
let Some(token) = me.ongoing.lock().get_token(id) else {
continue;
};
let result = select! {
r = me.plugin_do(r#in) => r,
false = token.future() => Ok(())
};
if let Err(out) = result {
me.ops.out(id, out);
}
}
}
})
}
async fn plugin_do(&self, r#in: PluginIn) -> 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::HookCopy(_in) => unreachable!(),
TaskIn::HookCut(_in) => unreachable!(),
TaskIn::HookDelete(_in) => unreachable!(),
TaskIn::HookTrash(_in) => unreachable!(),
TaskIn::HookDownload(_in) => unreachable!(),
TaskIn::HookUpload(_in) => unreachable!(),
PluginIn::Entry(r#in) => self.plugin.entry(r#in).await.map_err(Into::into),
}
}
fn fetch(&self, rx: async_priority_channel::Receiver<FetchIn, u8>) -> JoinHandle<()> {
let me = self.clone();
tokio::spawn(async move {
loop {
if let Ok((r#in, _)) = rx.recv().await {
let id = r#in.id();
let Some(token) = me.ongoing.lock().get_token(id) else {
continue;
};
let result = select! {
r = me.fetch_do(r#in) => r,
false = token.future() => Ok(())
};
if let Err(out) = result {
me.ops.out(id, out);
}
}
}
})
}
async fn fetch_do(&self, r#in: FetchIn) -> Result<(), TaskOut> {
self.fetch.fetch(r#in).await.map_err(Into::into)
}
fn preload(&self, rx: async_priority_channel::Receiver<PreloadIn, u8>) -> JoinHandle<()> {
let me = self.clone();
tokio::spawn(async move {
loop {
if let Ok((r#in, _)) = rx.recv().await {
let id = r#in.id();
let Some(token) = me.ongoing.lock().get_token(id) else {
continue;
};
let result = select! {
r = me.preload_do(r#in) => r,
false = token.future() => Ok(())
};
if let Err(out) = result {
me.ops.out(id, out);
}
}
}
})
}
async fn preload_do(&self, r#in: PreloadIn) -> Result<(), TaskOut> {
self.preload.preload(r#in).await.map_err(Into::into)
}
fn size(&self, rx: async_priority_channel::Receiver<SizeIn, u8>) -> JoinHandle<()> {
let me = self.clone();
tokio::spawn(async move {
loop {
if let Ok((r#in, _)) = rx.recv().await {
let id = r#in.id();
let Some(token) = me.ongoing.lock().get_token(id) else {
continue;
};
let result = select! {
r = me.size_do(r#in) => r,
false = token.future() => Ok(())
};
if let Err(out) = result {
me.ops.out(id, out);
}
}
}
})
}
async fn size_do(&self, r#in: SizeIn) -> Result<(), TaskOut> {
self.size.size(r#in).await.map_err(Into::into)
}
fn process(&self, rx: async_priority_channel::Receiver<ProcessIn, u8>) -> JoinHandle<()> {
let me = self.clone();
tokio::spawn(async move {
loop {
if let Ok((r#in, _)) = rx.recv().await {
let id = r#in.id();
let Some(token) = me.ongoing.lock().get_token(id) else {
continue;
};
let result = select! {
r = me.process_do(r#in) => r,
false = token.future() => Ok(())
};
if let Err(out) = result {
me.ops.out(id, out);
}
}
}
})
}
async fn process_do(&self, r#in: ProcessIn) -> Result<(), TaskOut> {
match r#in {
ProcessIn::Block(r#in) => self.process.block(r#in).await.map_err(Into::into),
ProcessIn::Orphan(r#in) => self.process.orphan(r#in).await.map_err(Into::into),
ProcessIn::Bg(r#in) => self.process.bg(r#in).await.map_err(Into::into),
}
}
fn hook(&self, rx: async_priority_channel::Receiver<HookIn, u8>) -> JoinHandle<()> {
let me = self.clone();
tokio::spawn(async move {
loop {
if let Ok((r#in, _)) = rx.recv().await {
let id = r#in.id();
if !me.ongoing.lock().exists(id) {
continue;
}
me.hook_do(r#in).await;
}
}
})
}
async fn hook_do(&self, r#in: HookIn) {
match r#in {
HookIn::Copy(r#in) => self.hook.copy(r#in).await,
HookIn::Cut(r#in) => self.hook.cut(r#in).await,
HookIn::Delete(r#in) => self.hook.delete(r#in).await,
HookIn::Trash(r#in) => self.hook.trash(r#in).await,
HookIn::Download(r#in) => self.hook.download(r#in).await,
HookIn::Upload(r#in) => self.hook.upload(r#in).await,
}
}
fn op(&self, mut rx: mpsc::UnboundedReceiver<TaskOp>) -> JoinHandle<()> {
let me = self.clone();
tokio::spawn(async move {
while let Some(op) = rx.recv().await {
let mut ongoing = me.ongoing.lock();
let Some(task) = ongoing.get_mut(op.id) else { continue };
op.out.reduce(task);
if !task.prog.cooked() && task.done.completed() != Some(false) {
continue; // Not cooked yet, also not canceled
} else if task.prog.cleaned() == Some(false) {
continue; // Failed to clean up
} else if let Some(hook) = task.hook.take() {
me.hook.submit(hook, LOW);
} else {
ongoing.fulfill(op.id);
}
}
})
}
}

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

@ -1,60 +1,32 @@
use std::{sync::Arc, time::Duration};
use std::{ops::Deref, sync::Arc, time::Duration};
use parking_lot::Mutex;
use tokio::{select, sync::mpsc::{self, UnboundedReceiver}, task::JoinHandle};
use tokio::task::JoinHandle;
use yazi_config::{YAZI, plugin::{Fetcher, Preloader}};
use yazi_parser::{app::PluginOpt, tasks::ProcessOpenOpt};
use yazi_shared::{CompletionToken, Id, Throttle, url::{UrlBuf, UrlLike}};
use super::{Ongoing, TaskOp};
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, HookInDelete, HookInDownload, HookInTrash, HookInUpload}, plugin::{Plugin, PluginInEntry, PluginProgEntry}, prework::{Prework, PreworkInFetch, PreworkInLoad, PreworkInSize, PreworkProgFetch, PreworkProgLoad, PreworkProgSize}, process::{Process, ProcessInBg, ProcessInBlock, ProcessInOrphan, ProcessProgBg, ProcessProgBlock, ProcessProgOrphan}};
use crate::{HIGH, LOW, NORMAL, Runner, fetch::{FetchIn, FetchProg}, file::{FileInCopy, FileInCut, FileInDelete, FileInDownload, FileInHardlink, FileInLink, FileInTrash, FileInUpload, FileOutCopy, FileOutCut, FileOutDownload, FileOutHardlink, FileOutUpload, FileProgCopy, FileProgCut, FileProgDelete, FileProgDownload, FileProgHardlink, FileProgLink, FileProgTrash, FileProgUpload}, hook::{HookInDelete, HookInDownload, HookInTrash, HookInUpload}, plugin::{PluginInEntry, PluginProgEntry}, preload::{PreloadIn, PreloadProg}, process::{ProcessInBg, ProcessInBlock, ProcessInOrphan, ProcessProgBg, ProcessProgBlock, ProcessProgOrphan}, size::{SizeIn, SizeProg}};
pub struct Scheduler {
ops: TaskOps,
pub runner: Runner,
micro: async_priority_channel::Sender<TaskIn, u8>,
handles: Vec<JoinHandle<()>>,
pub ongoing: Arc<Mutex<Ongoing>>,
pub runner: Runner,
handles: Vec<JoinHandle<()>>,
}
impl Deref for Scheduler {
type Target = Runner;
fn deref(&self) -> &Self::Target { &self.runner }
}
impl Scheduler {
pub fn serve() -> Self {
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 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)),
};
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,
};
for _ in 0..YAZI.tasks.micro_workers {
scheduler.handles.push(scheduler.schedule_micro(micro_rx.clone()));
}
for _ in 0..YAZI.tasks.macro_workers {
scheduler.handles.push(scheduler.schedule_macro(micro_rx.clone(), macro_rx.clone()));
}
scheduler.handle_ops(op_rx);
scheduler
let (runner, handles) = Runner::make();
Self { runner, handles }
}
pub fn cancel(&self, id: Id) -> bool {
if let Some(hook) = self.ongoing.lock().cancel(id) {
self.micro.try_send(hook, HIGH).ok();
self.hook.submit(hook, HIGH);
return false;
}
@ -78,7 +50,7 @@ impl Scheduler {
}
let follow = !from.scheme().covariant(to.scheme());
self.queue(
self.file.submit(
FileInCut {
id: task.id,
from,
@ -105,7 +77,7 @@ impl Scheduler {
}
let follow = follow || !from.scheme().covariant(to.scheme());
self.queue(
self.file.submit(
FileInCopy {
id: task.id,
from,
@ -124,7 +96,7 @@ impl Scheduler {
let mut ongoing = self.ongoing.lock();
let task = ongoing.add::<FileProgLink>(format!("Link {} to {}", from.display(), to.display()));
self.queue(
self.file.submit(
FileInLink {
id: task.id,
from,
@ -156,7 +128,7 @@ impl Scheduler {
.out(task.id, FileOutHardlink::Fail("Cannot hardlink directory into itself".to_owned()));
}
self.queue(FileInHardlink { id: task.id, from, to, force, cha: None, follow }, LOW);
self.file.submit(FileInHardlink { id: task.id, from, to, force, cha: None, follow }, LOW);
}
pub fn file_delete(&self, target: UrlBuf) {
@ -164,7 +136,7 @@ impl Scheduler {
let task = ongoing.add::<FileProgDelete>(format!("Delete {}", target.display()));
task.set_hook(HookInDelete { id: task.id, target: target.clone() });
self.queue(FileInDelete { id: task.id, target, cha: None }, LOW);
self.file.submit(FileInDelete { id: task.id, target, cha: None }, LOW);
}
pub fn file_trash(&self, target: UrlBuf) {
@ -172,7 +144,7 @@ impl Scheduler {
let task = ongoing.add::<FileProgTrash>(format!("Trash {}", target.display()));
task.set_hook(HookInTrash { id: task.id, target: target.clone() });
self.queue(FileInTrash { id: task.id, target }, LOW);
self.file.submit(FileInTrash { id: task.id, target }, LOW);
}
pub fn file_download(&self, url: UrlBuf) -> CompletionToken {
@ -181,7 +153,7 @@ impl Scheduler {
if url.kind().is_remote() {
task.set_hook(HookInDownload { id: task.id });
self.queue(
self.file.submit(
FileInDownload { id: task.id, url, cha: None, retry: 0, done: task.done.clone() },
LOW,
);
@ -203,7 +175,7 @@ impl Scheduler {
};
task.set_hook(HookInUpload { id: task.id, target: url.clone() });
self.queue(
self.file.submit(
FileInUpload { id: task.id, url, cha: None, cache: None, done: task.done.clone() },
LOW,
);
@ -213,7 +185,7 @@ impl Scheduler {
let mut ongoing = self.ongoing.lock();
let task = ongoing.add::<PluginProgEntry>(format!("Run micro plugin `{}`", opt.id));
self.queue(PluginInEntry { id: task.id, opt }, NORMAL);
self.plugin.submit(PluginInEntry { id: task.id, opt }, NORMAL);
}
pub fn fetch_paged(
@ -222,13 +194,13 @@ impl Scheduler {
targets: Vec<yazi_fs::File>,
) -> CompletionToken {
let mut ongoing = self.ongoing.lock();
let task = ongoing.add::<PreworkProgFetch>(format!(
let task = ongoing.add::<FetchProg>(format!(
"Run fetcher `{}` with {} target(s)",
fetcher.run.name,
targets.len()
));
self.queue(PreworkInFetch { id: task.id, plugin: fetcher, targets }, NORMAL);
self.fetch.submit(FetchIn { id: task.id, plugin: fetcher, targets });
task.done.clone()
}
@ -248,10 +220,10 @@ impl Scheduler {
pub fn preload_paged(&self, preloader: &'static Preloader, target: &yazi_fs::File) {
let mut ongoing = self.ongoing.lock();
let task = ongoing.add::<PreworkProgLoad>(format!("Run preloader `{}`", preloader.run.name));
let task = ongoing.add::<PreloadProg>(format!("Run preloader `{}`", preloader.run.name));
let target = target.clone();
self.queue(PreworkInLoad { id: task.id, plugin: preloader, target }, NORMAL);
self.preload.submit(PreloadIn { id: task.id, plugin: preloader, target });
}
pub fn prework_size(&self, targets: Vec<&UrlBuf>) {
@ -259,12 +231,11 @@ impl Scheduler {
let mut ongoing = self.ongoing.lock();
for target in targets {
let task =
ongoing.add::<PreworkProgSize>(format!("Calculate the size of {}", target.display()));
let task = ongoing.add::<SizeProg>(format!("Calculate the size of {}", target.display()));
let target = target.clone();
let throttle = throttle.clone();
self.queue(PreworkInSize { id: task.id, target, throttle }, NORMAL);
self.size.submit(SizeIn { id: task.id, target, throttle }, NORMAL);
}
}
@ -293,12 +264,15 @@ impl Scheduler {
if opt.block {
self
.queue(ProcessInBlock { id: task.id, cwd: opt.cwd, cmd: opt.cmd, args: opt.args }, NORMAL);
.process
.submit(ProcessInBlock { id: task.id, cwd: opt.cwd, cmd: opt.cmd, args: opt.args }, NORMAL);
} else if opt.orphan {
self
.queue(ProcessInOrphan { id: task.id, cwd: opt.cwd, cmd: opt.cmd, args: opt.args }, NORMAL);
self.process.submit(
ProcessInOrphan { id: task.id, cwd: opt.cwd, cmd: opt.cmd, args: opt.args },
NORMAL,
);
} else {
self.queue(
self.process.submit(
ProcessInBg {
id: task.id,
cwd: opt.cwd,
@ -310,104 +284,4 @@ impl Scheduler {
);
};
}
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((r#in, _)) = rx.recv().await {
let id = r#in.id();
let Some(token) = ongoing.lock().get_token(id) else {
continue;
};
let result = if r#in.is_hook() {
runner.micro(r#in).await
} else {
select! {
r = runner.micro(r#in) => r,
false = token.future() => Ok(())
}
};
if let Err(out) = result {
ops.out(id, out);
}
}
}
})
}
fn schedule_macro(
&self,
micro: async_priority_channel::Receiver<TaskIn, u8>,
r#macro: 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 {
let (r#in, micro) = select! {
Ok((r#in, _)) = micro.recv() => (r#in, true),
Ok((r#in, _)) = r#macro.recv() => (r#in, false),
};
let id = r#in.id();
let Some(token) = ongoing.lock().get_token(id) else {
continue;
};
let result = if r#in.is_hook() {
if micro { runner.micro(r#in).await } else { runner.r#macro(r#in).await }
} else if micro {
select! {
r = runner.micro(r#in) => r,
false = token.future() => Ok(()),
}
} else {
select! {
r = runner.r#macro(r#in) => r,
false = token.future() => Ok(()),
}
};
if let Err(out) = result {
ops.out(id, out);
}
}
})
}
fn handle_ops(&self, mut rx: UnboundedReceiver<TaskOp>) -> JoinHandle<()> {
let micro = self.micro.clone();
let ongoing = self.ongoing.clone();
tokio::spawn(async move {
while let Some(op) = rx.recv().await {
let mut ongoing = ongoing.lock();
let Some(task) = ongoing.get_mut(op.id) else { continue };
op.out.reduce(task);
if !task.prog.cooked() && task.done.completed() != Some(false) {
continue; // Not cooked yet, also not canceled
} else if task.prog.cleaned() == Some(false) {
continue; // Failed to clean up
} else if let Some(hook) = task.hook.take() {
micro.try_send(hook, LOW).ok();
} else {
ongoing.fulfill(op.id);
}
}
})
}
#[inline]
fn queue(&self, r#in: impl Into<TaskIn>, priority: u8) {
_ = self.micro.try_send(r#in.into(), priority);
}
}

14
yazi-scheduler/src/size/in.rs Normal file
View file

@ -0,0 +1,14 @@
use std::sync::Arc;
use yazi_shared::{Id, Throttle, url::UrlBuf};
#[derive(Debug)]
pub(crate) struct SizeIn {
pub(crate) id: Id,
pub(crate) target: UrlBuf,
pub(crate) throttle: Arc<Throttle<(UrlBuf, u64)>>,
}
impl SizeIn {
pub(crate) fn id(&self) -> Id { self.id }
}

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

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

17
yazi-scheduler/src/size/out.rs Normal file
View file

@ -0,0 +1,17 @@
use crate::{Task, TaskProg};
#[derive(Debug)]
pub(crate) enum SizeOut {
Done,
}
impl SizeOut {
pub(crate) fn reduce(self, task: &mut Task) {
let TaskProg::Size(prog) = &mut task.prog else { return };
match self {
Self::Done => {
prog.done = true;
}
}
}
}

32
yazi-scheduler/src/size/progress.rs Normal file
View file

@ -0,0 +1,32 @@
use serde::Serialize;
use yazi_parser::app::TaskSummary;
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize)]
pub struct SizeProg {
pub done: bool,
}
impl From<SizeProg> for TaskSummary {
fn from(value: SizeProg) -> Self {
Self {
total: 1,
success: value.done as u32,
failed: 0,
percent: value.percent().map(Into::into),
}
}
}
impl SizeProg {
pub fn cooked(self) -> bool { self.done }
pub fn running(self) -> bool { !self.done }
pub fn success(self) -> bool { self.cooked() }
pub fn failed(self) -> bool { false }
pub fn cleaned(self) -> Option<bool> { None }
pub fn percent(self) -> Option<f32> { None }
}

54
yazi-scheduler/src/size/size.rs Normal file
View file

@ -0,0 +1,54 @@
use anyhow::Result;
use hashbrown::{HashMap, HashSet};
use parking_lot::RwLock;
use tokio::sync::mpsc;
use yazi_fs::FilesOp;
use yazi_shared::url::{UrlBuf, UrlLike};
use yazi_vfs::provider;
use super::SizeIn;
use crate::{TaskOp, TaskOps, size::SizeOut};
pub struct Size {
ops: TaskOps,
tx: async_priority_channel::Sender<SizeIn, u8>,
pub sizing: RwLock<HashSet<UrlBuf>>,
}
impl Size {
pub(crate) fn new(
ops: &mpsc::UnboundedSender<TaskOp>,
tx: async_priority_channel::Sender<SizeIn, u8>,
) -> Self {
Self { ops: ops.into(), tx, sizing: Default::default() }
}
pub(crate) async fn size(&self, task: SizeIn) -> Result<(), SizeOut> {
let length = provider::calculate(&task.target).await.unwrap_or(0);
task.throttle.done((task.target, length), |buf| {
{
let mut loading = self.sizing.write();
for (path, _) in &buf {
loading.remove(path);
}
}
let parent = buf[0].0.parent().unwrap();
FilesOp::Size(
parent.into(),
HashMap::from_iter(buf.into_iter().map(|(u, s)| (u.urn().into(), s))),
)
.emit();
});
Ok(self.ops.out(task.id, SizeOut::Done))
}
}
impl Size {
#[inline]
pub(crate) fn submit(&self, r#in: impl Into<SizeIn>, priority: u8) {
_ = self.tx.try_send(r#in.into(), priority);
}
}

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

@ -1,14 +1,14 @@
use tokio::sync::mpsc;
use yazi_shared::{CompletionToken, Id};
use crate::{TaskIn, TaskProg};
use crate::{TaskProg, hook::HookIn};
#[derive(Debug)]
pub struct Task {
pub id: Id,
pub name: String,
pub(crate) prog: TaskProg,
pub(crate) hook: Option<TaskIn>,
pub(crate) hook: Option<HookIn>,
pub done: CompletionToken,
pub logs: String,
@ -41,5 +41,5 @@ impl Task {
}
}
pub(super) fn set_hook(&mut self, hook: impl Into<TaskIn>) { self.hook = Some(hook.into()); }
pub(super) fn set_hook(&mut self, hook: impl Into<HookIn>) { self.hook = Some(hook.into()); }
}

2
yazi-shared/src/event/cmd.rs
View file

@ -209,7 +209,7 @@ impl Cmd {
let key = parts.next().expect("at least one part");
let val = parts.next().map_or(Data::Boolean(true), Data::from);
Ok((DataKey::from(key.to_owned()), val))
Ok((key.to_owned().into(), val))
})
.collect()
}