diff --git a/Cargo.lock b/Cargo.lock index 2a5739cf..830ec6b4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -131,14 +131,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bddcadddf5e9015d310179a59bb28c4d4b9920ad0f11e8e14dbadf654890c9a6" [[package]] -name = "async-channel" -version = "1.9.0" +name = "async-priority-channel" +version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "81953c529336010edd6d8e358f886d9581267795c61b19475b71314bffa46d35" +checksum = "c21678992e1b21bebfe2bc53ab5f5f68c106eddab31b24e0bb06e9b715a86640" dependencies = [ - "concurrent-queue", "event-listener", - "futures-core", ] [[package]] @@ -373,15 +371,6 @@ version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "acbf1af155f9b9ef647e42cdc158db4b64a1b61f743629225fde6f3e0be2a7c7" -[[package]] -name = "concurrent-queue" -version = "2.4.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d16048cd947b08fa32c24458a22f5dc5e835264f689f4f5653210c69fd107363" -dependencies = [ - "crossbeam-utils", -] - [[package]] name = "console" version = "0.15.7" @@ -2745,7 +2734,7 @@ name = "yazi-scheduler" version = "0.1.5" dependencies = [ "anyhow", - "async-channel", + "async-priority-channel", "base64", "crossterm", "futures", diff --git a/cspell.json b/cspell.json index 07e982c0..933340ab 100644 --- a/cspell.json +++ b/cspell.json @@ -1 +1 @@ -{"language":"en","flagWords":[],"words":["Punct","KEYMAP","splitn","crossterm","YAZI","unar","peekable","ratatui","syntect","pbpaste","pbcopy","ffmpegthumbnailer","oneshot","Posix","Lsar","XADDOS","zoxide","cands","Deque","precache","imageops","IFBLK","IFCHR","IFDIR","IFIFO","IFLNK","IFMT","IFSOCK","IRGRP","IROTH","IRUSR","ISGID","ISUID","ISVTX","IWGRP","IWOTH","IWUSR","IXGRP","IXOTH","IXUSR","libc","winsize","TIOCGWINSZ","xpixel","ypixel","ioerr","appender","Catppuccin","macchiato","gitmodules","Dotfiles","bashprofile","vimrc","flac","webp","exiftool","mediainfo","ripgrep","nvim","indexmap","indexmap","unwatch","canonicalize","serde","fsevent","Ueberzug","iterm","wezterm","sixel","chafa","ueberzugpp","️ Überzug","️ Überzug","Konsole","Alacritty","Überzug","pkgs","paru","unarchiver","pdftoppm","poppler","prebuild","singlefile","jpegopt","EXIF","rustfmt","mktemp","nanos","xclip","xsel","natord","Mintty","nixos","nixpkgs","SIGTSTP","SIGCONT","SIGCONT","mlua","nonstatic","userdata","metatable","natsort","backstack","luajit","Succ","Succ","cand","fileencoding","foldmethod","lightgreen","darkgray","lightred","lightyellow","lightcyan","nushell","msvc","aarch","linemode","sxyazi","rsplit","ZELLIJ","bitflags","bitflags","USERPROFILE","Neovim","vergen","gitcl","Renderable","preloaders","prec","imagesize","Upserting"],"version":"0.2"} +{"language":"en","words":["Punct","KEYMAP","splitn","crossterm","YAZI","unar","peekable","ratatui","syntect","pbpaste","pbcopy","ffmpegthumbnailer","oneshot","Posix","Lsar","XADDOS","zoxide","cands","Deque","precache","imageops","IFBLK","IFCHR","IFDIR","IFIFO","IFLNK","IFMT","IFSOCK","IRGRP","IROTH","IRUSR","ISGID","ISUID","ISVTX","IWGRP","IWOTH","IWUSR","IXGRP","IXOTH","IXUSR","libc","winsize","TIOCGWINSZ","xpixel","ypixel","ioerr","appender","Catppuccin","macchiato","gitmodules","Dotfiles","bashprofile","vimrc","flac","webp","exiftool","mediainfo","ripgrep","nvim","indexmap","indexmap","unwatch","canonicalize","serde","fsevent","Ueberzug","iterm","wezterm","sixel","chafa","ueberzugpp","️ Überzug","️ Überzug","Konsole","Alacritty","Überzug","pkgs","paru","unarchiver","pdftoppm","poppler","prebuild","singlefile","jpegopt","EXIF","rustfmt","mktemp","nanos","xclip","xsel","natord","Mintty","nixos","nixpkgs","SIGTSTP","SIGCONT","SIGCONT","mlua","nonstatic","userdata","metatable","natsort","backstack","luajit","Succ","Succ","cand","fileencoding","foldmethod","lightgreen","darkgray","lightred","lightyellow","lightcyan","nushell","msvc","aarch","linemode","sxyazi","rsplit","ZELLIJ","bitflags","bitflags","USERPROFILE","Neovim","vergen","gitcl","Renderable","preloaders","prec","imagesize","Upserting","prio"],"version":"0.2","flagWords":[]} diff --git a/yazi-config/preset/keymap.toml b/yazi-config/preset/keymap.toml index 9675fd11..115aa716 100644 --- a/yazi-config/preset/keymap.toml +++ b/yazi-config/preset/keymap.toml @@ -33,10 +33,10 @@ keymap = [ { on = [ "H" ], exec = "back", desc = "Go back to the previous directory" }, { on = [ "L" ], exec = "forward", desc = "Go forward to the next directory" }, - { on = [ "" ], exec = "seek -5", desc = "Peek up 5 units in the preview" }, - { on = [ "" ], exec = "seek 5", desc = "Peek down 5 units in the preview" }, - { on = [ "" ], exec = "seek -5", desc = "Peek up 5 units in the preview" }, - { on = [ "" ], exec = "seek 5", desc = "Peek down 5 units in the preview" }, + { on = [ "" ], exec = "seek -5", desc = "Seek up 5 units in the preview" }, + { on = [ "" ], exec = "seek 5", desc = "Seek down 5 units in the preview" }, + { on = [ "" ], exec = "seek -5", desc = "Seek up 5 units in the preview" }, + { on = [ "" ], exec = "seek 5", desc = "Seek down 5 units in the preview" }, { on = [ "" ], exec = "arrow -1", desc = "Move cursor up" }, { on = [ "" ], exec = "arrow 1", desc = "Move cursor down" }, @@ -90,13 +90,13 @@ keymap = [ { on = [ "c", "n" ], exec = "copy name_without_ext", desc = "Copy the name of the file without the extension" }, # Filter - { on = [ "f" ], exec = "filter" }, + { on = [ "f" ], exec = "filter --smart", desc = "Filter the files" }, # Find - { on = [ "/" ], exec = "find --smart" }, - { on = [ "?" ], exec = "find --previous --smart" }, - { on = [ "n" ], exec = "find_arrow" }, - { on = [ "N" ], exec = "find_arrow --previous" }, + { on = [ "/" ], exec = "find --smart", desc = "Find next file" }, + { on = [ "?" ], exec = "find --previous --smart", desc = "Find previous file" }, + { on = [ "n" ], exec = "find_arrow", desc = "Go to next found file" }, + { on = [ "N" ], exec = "find_arrow --previous", desc = "Go to previous found file" }, # Sorting { on = [ ",", "m" ], exec = "sort modified --dir_first", desc = "Sort by modified time" }, diff --git a/yazi-config/preset/yazi.toml b/yazi-config/preset/yazi.toml index bd5b4439..8425278d 100644 --- a/yazi-config/preset/yazi.toml +++ b/yazi-config/preset/yazi.toml @@ -77,7 +77,7 @@ suppress_preload = false [plugin] preloaders = [ - { name = "*", cond = "!mime", exec = "mime.lua", multi = true }, + { name = "*", cond = "!mime", exec = "mime.lua", multi = true, prio = "high" }, # Image { mime = "image/vnd.djvu", exec = "noop.lua" }, { mime = "image/*", exec = "image.lua" }, diff --git a/yazi-config/src/lib.rs b/yazi-config/src/lib.rs index e71004f8..3347986c 100644 --- a/yazi-config/src/lib.rs +++ b/yazi-config/src/lib.rs @@ -13,6 +13,7 @@ pub mod plugin; pub mod popup; mod preset; pub mod preview; +mod priority; mod tasks; pub mod theme; mod validation; @@ -21,6 +22,7 @@ mod xdg; pub use layout::*; pub(crate) use pattern::*; pub(crate) use preset::*; +pub use priority::*; pub(crate) use xdg::*; pub static ARGS: RoCell = RoCell::new(); diff --git a/yazi-config/src/plugin/mod.rs b/yazi-config/src/plugin/mod.rs index af03fb09..ec375606 100644 --- a/yazi-config/src/plugin/mod.rs +++ b/yazi-config/src/plugin/mod.rs @@ -1,7 +1,9 @@ mod exec; mod plugin; +mod props; pub use exec::*; pub use plugin::*; +pub use props::*; pub const MAX_PRELOADERS: u8 = 32; diff --git a/yazi-config/src/plugin/plugin.rs b/yazi-config/src/plugin/plugin.rs index 83eb47ef..34cf4645 100644 --- a/yazi-config/src/plugin/plugin.rs +++ b/yazi-config/src/plugin/plugin.rs @@ -3,7 +3,7 @@ use std::path::Path; use serde::Deserialize; use yazi_shared::{event::Exec, Condition, MIME_DIR}; -use crate::{pattern::Pattern, plugin::MAX_PRELOADERS, MERGED_YAZI}; +use crate::{pattern::Pattern, plugin::MAX_PRELOADERS, Priority, MERGED_YAZI}; #[derive(Deserialize)] pub struct Plugin { @@ -24,6 +24,8 @@ pub struct PluginRule { pub sync: bool, #[serde(default)] pub multi: bool, + #[serde(default)] + pub prio: Priority, } impl Default for Plugin { diff --git a/yazi-config/src/plugin/props.rs b/yazi-config/src/plugin/props.rs new file mode 100644 index 00000000..80a8196c --- /dev/null +++ b/yazi-config/src/plugin/props.rs @@ -0,0 +1,16 @@ +use super::PluginRule; +use crate::Priority; + +#[derive(Debug, Clone)] +pub struct PluginProps { + pub id: u8, + pub cmd: String, + pub multi: bool, + pub prio: Priority, +} + +impl From<&PluginRule> for PluginProps { + fn from(rule: &PluginRule) -> Self { + Self { id: rule.id, cmd: rule.exec.cmd.to_owned(), multi: rule.multi, prio: rule.prio } + } +} diff --git a/yazi-config/src/priority.rs b/yazi-config/src/priority.rs new file mode 100644 index 00000000..50535297 --- /dev/null +++ b/yazi-config/src/priority.rs @@ -0,0 +1,32 @@ +use std::str::FromStr; + +use anyhow::anyhow; +use serde::Deserialize; + +#[derive(Default, Clone, Copy, Debug, Deserialize)] +#[serde(try_from = "String")] +pub enum Priority { + Low = 0, + #[default] + Normal = 1, + High = 2, +} + +impl FromStr for Priority { + type Err = anyhow::Error; + + fn from_str(s: &str) -> Result { + match s { + "low" => Ok(Self::Low), + "normal" => Ok(Self::Normal), + "high" => Ok(Self::High), + _ => Err(anyhow!("Invalid priority: {s}")), + } + } +} + +impl TryFrom for Priority { + type Error = anyhow::Error; + + fn try_from(s: String) -> Result { Self::from_str(&s) } +} diff --git a/yazi-core/src/tab/preview.rs b/yazi-core/src/tab/preview.rs index 54456592..0ab91c79 100644 --- a/yazi-core/src/tab/preview.rs +++ b/yazi-core/src/tab/preview.rs @@ -5,7 +5,7 @@ use tokio_stream::{wrappers::UnboundedReceiverStream, StreamExt}; use tokio_util::sync::CancellationToken; use yazi_adaptor::ADAPTOR; use yazi_config::PLUGIN; -use yazi_plugin::{external::Highlighter, utils::PreviewLock}; +use yazi_plugin::{external::Highlighter, isolate, utils::PreviewLock}; use yazi_shared::{fs::{Cha, File, FilesOp, Url}, MIME_DIR}; use crate::folder::Files; @@ -32,9 +32,9 @@ impl Preview { self.abort(); if previewer.sync { - yazi_plugin::isolate::peek_sync(&previewer.exec, file, self.skip); + isolate::peek_sync(&previewer.exec, file, self.skip); } else { - self.previewer_ct = Some(yazi_plugin::isolate::peek(&previewer.exec, file, self.skip)); + self.previewer_ct = Some(isolate::peek(&previewer.exec, file, self.skip)); } } diff --git a/yazi-scheduler/Cargo.toml b/yazi-scheduler/Cargo.toml index 7405e799..31898978 100644 --- a/yazi-scheduler/Cargo.toml +++ b/yazi-scheduler/Cargo.toml @@ -15,16 +15,16 @@ yazi-shared = { path = "../yazi-shared", version = "0.1.5" } yazi-plugin = { path = "../yazi-plugin", version = "0.1.5" } # External dependencies -anyhow = "^1" -async-channel = "^1" -base64 = "^0" -crossterm = "^0" -futures = "^0" -parking_lot = "^0" -regex = "^1" -tokio = { version = "^1", features = [ "parking_lot", "rt-multi-thread" ] } -tokio-stream = "^0" -trash = "^3" +anyhow = "^1" +async-priority-channel = "^0" +base64 = "^0" +crossterm = "^0" +futures = "^0" +parking_lot = "^0" +regex = "^1" +tokio = { version = "^1", features = [ "parking_lot", "rt-multi-thread" ] } +tokio-stream = "^0" +trash = "^3" # Logging tracing = { version = "^0", features = [ "max_level_debug", "release_max_level_warn" ] } diff --git a/yazi-scheduler/src/workers/file.rs b/yazi-scheduler/src/file/file.rs similarity index 70% rename from yazi-scheduler/src/workers/file.rs rename to yazi-scheduler/src/file/file.rs index d0b92796..dc869440 100644 --- a/yazi-scheduler/src/workers/file.rs +++ b/yazi-scheduler/src/file/file.rs @@ -7,77 +7,25 @@ use tracing::warn; use yazi_config::TASKS; use yazi_shared::fs::{calculate_size, copy_with_progress, path_relative_to, Url}; -use crate::TaskOp; +use super::{FileOp, FileOpDelete, FileOpLink, FileOpPaste, FileOpTrash}; +use crate::{TaskOp, TaskProg, LOW, NORMAL}; pub struct File { - tx: async_channel::Sender, - rx: async_channel::Receiver, - - sch: mpsc::UnboundedSender, -} - -#[derive(Debug)] -pub enum FileOp { - Paste(FileOpPaste), - Link(FileOpLink), - Delete(FileOpDelete), - Trash(FileOpTrash), -} - -#[derive(Clone, Debug)] -pub struct FileOpPaste { - pub id: usize, - pub from: Url, - pub to: Url, - pub cut: bool, - pub follow: bool, - pub retry: u8, -} - -#[derive(Clone, Debug)] -pub struct FileOpLink { - pub id: usize, - pub from: Url, - pub to: Url, - pub meta: Option, - pub resolve: bool, - pub relative: bool, - pub delete: bool, -} - -#[derive(Clone, Debug)] -pub struct FileOpDelete { - pub id: usize, - pub target: Url, - pub length: u64, -} - -#[derive(Clone, Debug)] -pub struct FileOpTrash { - pub id: usize, - pub target: Url, - pub length: u64, + macro_: async_priority_channel::Sender, + prog: mpsc::UnboundedSender, } impl File { - pub fn new(sch: mpsc::UnboundedSender) -> Self { - let (tx, rx) = async_channel::unbounded(); - Self { tx, rx, sch } + pub fn new( + macro_: async_priority_channel::Sender, + prog: mpsc::UnboundedSender, + ) -> Self { + Self { macro_, prog } } - #[inline] - pub async fn recv(&self) -> Result<(usize, FileOp)> { - Ok(match self.rx.recv().await? { - FileOp::Paste(t) => (t.id, FileOp::Paste(t)), - FileOp::Link(t) => (t.id, FileOp::Link(t)), - FileOp::Delete(t) => (t.id, FileOp::Delete(t)), - FileOp::Trash(t) => (t.id, FileOp::Trash(t)), - }) - } - - pub async fn work(&self, op: &mut FileOp) -> Result<()> { + pub async fn work(&self, op: FileOp) -> Result<()> { match op { - FileOp::Paste(task) => { + FileOp::Paste(mut task) => { match fs::remove_file(&task.to).await { Err(e) if e.kind() != NotFound => Err(e)?, _ => {} @@ -92,7 +40,7 @@ impl File { } break; } - Ok(n) => self.sch.send(TaskOp::Adv(task.id, 0, n))?, + Ok(n) => self.prog.send(TaskProg::Adv(task.id, 0, n))?, Err(e) if e.kind() == NotFound => { warn!("Paste task partially done: {:?}", task); break; @@ -105,12 +53,12 @@ impl File { { self.log(task.id, format!("Paste task retry: {:?}", task))?; task.retry += 1; - return Ok(self.tx.send(FileOp::Paste(task.clone())).await?); + return Ok(self.macro_.send(FileOp::Paste(task).into(), LOW).await?); } Err(e) => Err(e)?, } } - self.sch.send(TaskOp::Adv(task.id, 1, 0))?; + self.prog.send(TaskProg::Adv(task.id, 1, 0))?; } FileOp::Link(task) => { let meta = task.meta.as_ref().unwrap(); @@ -120,7 +68,7 @@ impl File { Ok(p) => Cow::Owned(p), Err(e) if e.kind() == NotFound => { self.log(task.id, format!("Link task partially done: {:?}", task))?; - return Ok(self.sch.send(TaskOp::Adv(task.id, 1, meta.len()))?); + return Ok(self.prog.send(TaskProg::Adv(task.id, 1, meta.len()))?); } Err(e) => Err(e)?, } @@ -155,7 +103,7 @@ impl File { if task.delete { fs::remove_file(&task.from).await.ok(); } - self.sch.send(TaskOp::Adv(task.id, 1, meta.len()))?; + self.prog.send(TaskProg::Adv(task.id, 1, meta.len()))?; } FileOp::Delete(task) => { if let Err(e) = fs::remove_file(&task.target).await { @@ -164,7 +112,7 @@ impl File { Err(e)? } } - self.sch.send(TaskOp::Adv(task.id, 1, task.length))? + self.prog.send(TaskProg::Adv(task.id, 1, task.length))? } FileOp::Trash(task) => { #[cfg(target_os = "macos")] @@ -178,7 +126,7 @@ impl File { { trash::delete(&task.target)?; } - self.sch.send(TaskOp::Adv(task.id, 1, task.length))?; + self.prog.send(TaskProg::Adv(task.id, 1, task.length))?; } } Ok(()) @@ -196,12 +144,12 @@ impl File { let meta = Self::metadata(&task.from, task.follow).await?; if !meta.is_dir() { let id = task.id; - self.sch.send(TaskOp::New(id, meta.len()))?; + self.prog.send(TaskProg::New(id, meta.len()))?; if meta.is_file() { - self.tx.send(FileOp::Paste(task)).await?; + self.macro_.send(FileOp::Paste(task).into(), LOW).await?; } else if meta.is_symlink() { - self.tx.send(FileOp::Link(task.to_link(meta))).await?; + self.macro_.send(FileOp::Link(task.to_link(meta)).into(), NORMAL).await?; } return self.succ(id); } @@ -211,7 +159,7 @@ impl File { match $result { Ok(v) => v, Err(e) => { - self.sch.send(TaskOp::New(task.id, 0))?; + self.prog.send(TaskProg::New(task.id, 0))?; self.fail(task.id, format!("An error occurred while pasting: {e}"))?; continue; } @@ -242,12 +190,12 @@ impl File { task.to = dest.join(src.file_name().unwrap()); task.from = src; - self.sch.send(TaskOp::New(task.id, meta.len()))?; + self.prog.send(TaskProg::New(task.id, meta.len()))?; if meta.is_file() { - self.tx.send(FileOp::Paste(task.clone())).await?; + self.macro_.send(FileOp::Paste(task.clone()).into(), LOW).await?; } else if meta.is_symlink() { - self.tx.send(FileOp::Link(task.to_link(meta))).await?; + self.macro_.send(FileOp::Link(task.to_link(meta)).into(), NORMAL).await?; } } } @@ -260,8 +208,8 @@ impl File { task.meta = Some(fs::symlink_metadata(&task.from).await?); } - self.sch.send(TaskOp::New(id, task.meta.as_ref().unwrap().len()))?; - self.tx.send(FileOp::Link(task)).await?; + self.prog.send(TaskProg::New(id, task.meta.as_ref().unwrap().len()))?; + self.macro_.send(FileOp::Link(task).into(), NORMAL).await?; self.succ(id) } @@ -270,8 +218,8 @@ impl File { if !meta.is_dir() { let id = task.id; task.length = meta.len(); - self.sch.send(TaskOp::New(id, meta.len()))?; - self.tx.send(FileOp::Delete(task)).await?; + self.prog.send(TaskProg::New(id, meta.len()))?; + self.macro_.send(FileOp::Delete(task).into(), NORMAL).await?; return self.succ(id); } @@ -295,8 +243,8 @@ impl File { task.target = Url::from(entry.path()); task.length = meta.len(); - self.sch.send(TaskOp::New(task.id, meta.len()))?; - self.tx.send(FileOp::Delete(task.clone())).await?; + self.prog.send(TaskProg::New(task.id, meta.len()))?; + self.macro_.send(FileOp::Delete(task.clone()).into(), NORMAL).await?; } } self.succ(task.id) @@ -306,8 +254,8 @@ impl File { let id = task.id; task.length = calculate_size(&task.target).await; - self.sch.send(TaskOp::New(id, task.length))?; - self.tx.send(FileOp::Trash(task)).await?; + self.prog.send(TaskProg::New(id, task.length))?; + self.macro_.send(FileOp::Trash(task).into(), LOW).await?; self.succ(id) } @@ -343,15 +291,17 @@ impl File { impl File { #[inline] - fn succ(&self, id: usize) -> Result<()> { Ok(self.sch.send(TaskOp::Succ(id))?) } + fn succ(&self, id: usize) -> Result<()> { Ok(self.prog.send(TaskProg::Succ(id))?) } #[inline] fn fail(&self, id: usize, reason: String) -> Result<()> { - Ok(self.sch.send(TaskOp::Fail(id, reason))?) + Ok(self.prog.send(TaskProg::Fail(id, reason))?) } #[inline] - fn log(&self, id: usize, line: String) -> Result<()> { Ok(self.sch.send(TaskOp::Log(id, line))?) } + fn log(&self, id: usize, line: String) -> Result<()> { + Ok(self.prog.send(TaskProg::Log(id, line))?) + } } impl FileOpPaste { diff --git a/yazi-scheduler/src/file/mod.rs b/yazi-scheduler/src/file/mod.rs new file mode 100644 index 00000000..ecb518b7 --- /dev/null +++ b/yazi-scheduler/src/file/mod.rs @@ -0,0 +1,7 @@ +#![allow(clippy::module_inception)] + +mod file; +mod op; + +pub use file::*; +pub use op::*; diff --git a/yazi-scheduler/src/file/op.rs b/yazi-scheduler/src/file/op.rs new file mode 100644 index 00000000..cb80abfe --- /dev/null +++ b/yazi-scheduler/src/file/op.rs @@ -0,0 +1,57 @@ +use std::fs::Metadata; + +use yazi_shared::fs::Url; + +#[derive(Debug)] +pub enum FileOp { + Paste(FileOpPaste), + Link(FileOpLink), + Delete(FileOpDelete), + Trash(FileOpTrash), +} + +impl FileOp { + pub fn id(&self) -> usize { + match self { + Self::Paste(op) => op.id, + Self::Link(op) => op.id, + Self::Delete(op) => op.id, + Self::Trash(op) => op.id, + } + } +} + +#[derive(Clone, Debug)] +pub struct FileOpPaste { + pub id: usize, + pub from: Url, + pub to: Url, + pub cut: bool, + pub follow: bool, + pub retry: u8, +} + +#[derive(Clone, Debug)] +pub struct FileOpLink { + pub id: usize, + pub from: Url, + pub to: Url, + pub meta: Option, + pub resolve: bool, + pub relative: bool, + pub delete: bool, +} + +#[derive(Clone, Debug)] +pub struct FileOpDelete { + pub id: usize, + pub target: Url, + pub length: u64, +} + +#[derive(Clone, Debug)] +pub struct FileOpTrash { + pub id: usize, + pub target: Url, + pub length: u64, +} diff --git a/yazi-scheduler/src/lib.rs b/yazi-scheduler/src/lib.rs index 6c163d0b..9c807796 100644 --- a/yazi-scheduler/src/lib.rs +++ b/yazi-scheduler/src/lib.rs @@ -1,14 +1,23 @@ #![allow(clippy::option_map_unit_fn, clippy::unit_arg)] mod blocker; +mod file; +mod op; +mod plugin; +mod preload; +mod process; mod running; mod scheduler; mod task; -pub mod workers; pub use blocker::*; +pub use op::*; pub use running::*; pub use scheduler::*; pub use task::*; +const LOW: u8 = yazi_config::Priority::Low as u8; +const NORMAL: u8 = yazi_config::Priority::Normal as u8; +const HIGH: u8 = yazi_config::Priority::High as u8; + pub fn init() { init_blocker(); } diff --git a/yazi-scheduler/src/op.rs b/yazi-scheduler/src/op.rs new file mode 100644 index 00000000..6afcbebe --- /dev/null +++ b/yazi-scheduler/src/op.rs @@ -0,0 +1,30 @@ +use crate::{file::FileOp, plugin::PluginOp, preload::PreloadOp}; + +#[derive(Debug)] +pub enum TaskOp { + File(Box), + Plugin(Box), + Preload(Box), +} + +impl TaskOp { + pub fn id(&self) -> usize { + match self { + TaskOp::File(op) => op.id(), + TaskOp::Plugin(op) => op.id(), + TaskOp::Preload(op) => op.id(), + } + } +} + +impl From for TaskOp { + fn from(op: FileOp) -> Self { Self::File(Box::new(op)) } +} + +impl From for TaskOp { + fn from(op: PluginOp) -> Self { Self::Plugin(Box::new(op)) } +} + +impl From for TaskOp { + fn from(op: PreloadOp) -> Self { Self::Preload(Box::new(op)) } +} diff --git a/yazi-scheduler/src/plugin/mod.rs b/yazi-scheduler/src/plugin/mod.rs new file mode 100644 index 00000000..4ef84332 --- /dev/null +++ b/yazi-scheduler/src/plugin/mod.rs @@ -0,0 +1,7 @@ +#![allow(clippy::module_inception)] + +mod op; +mod plugin; + +pub use op::*; +pub use plugin::*; diff --git a/yazi-scheduler/src/plugin/op.rs b/yazi-scheduler/src/plugin/op.rs new file mode 100644 index 00000000..729021d9 --- /dev/null +++ b/yazi-scheduler/src/plugin/op.rs @@ -0,0 +1,18 @@ +#[derive(Debug)] +pub enum PluginOp { + Entry(PluginOpEntry), +} + +impl PluginOp { + pub fn id(&self) -> usize { + match self { + Self::Entry(op) => op.id, + } + } +} + +#[derive(Clone, Debug)] +pub struct PluginOpEntry { + pub id: usize, + pub name: String, +} diff --git a/yazi-scheduler/src/plugin/plugin.rs b/yazi-scheduler/src/plugin/plugin.rs new file mode 100644 index 00000000..28ebae3d --- /dev/null +++ b/yazi-scheduler/src/plugin/plugin.rs @@ -0,0 +1,59 @@ +use anyhow::Result; +use tokio::sync::mpsc; +use yazi_plugin::isolate; + +use super::{PluginOp, PluginOpEntry}; +use crate::{TaskOp, TaskProg, HIGH}; + +pub struct Plugin { + macro_: async_priority_channel::Sender, + prog: mpsc::UnboundedSender, +} + +impl Plugin { + pub fn new( + macro_: async_priority_channel::Sender, + prog: mpsc::UnboundedSender, + ) -> Self { + Self { macro_, prog } + } + + pub async fn work(&self, op: PluginOp) -> Result<()> { + match op { + PluginOp::Entry(task) => { + isolate::entry(&task.name).await?; + } + } + Ok(()) + } + + pub async fn micro(&self, task: PluginOpEntry) -> Result<()> { + self.prog.send(TaskProg::New(task.id, 0))?; + + if let Err(e) = isolate::entry(&task.name).await { + self.fail(task.id, format!("Micro plugin failed:\n{e}"))?; + return Err(e.into()); + } + + self.prog.send(TaskProg::Adv(task.id, 1, 0))?; + self.succ(task.id) + } + + pub fn macro_(&self, task: PluginOpEntry) -> Result<()> { + let id = task.id; + + self.prog.send(TaskProg::New(id, 0))?; + self.macro_.try_send(PluginOp::Entry(task).into(), HIGH)?; + self.succ(id) + } +} + +impl Plugin { + #[inline] + fn succ(&self, id: usize) -> Result<()> { Ok(self.prog.send(TaskProg::Succ(id))?) } + + #[inline] + fn fail(&self, id: usize, reason: String) -> Result<()> { + Ok(self.prog.send(TaskProg::Fail(id, reason))?) + } +} diff --git a/yazi-scheduler/src/preload/mod.rs b/yazi-scheduler/src/preload/mod.rs new file mode 100644 index 00000000..829ef0e5 --- /dev/null +++ b/yazi-scheduler/src/preload/mod.rs @@ -0,0 +1,7 @@ +#![allow(clippy::module_inception)] + +mod op; +mod preload; + +pub use op::*; +pub use preload::*; diff --git a/yazi-scheduler/src/preload/op.rs b/yazi-scheduler/src/preload/op.rs new file mode 100644 index 00000000..6e4b4d01 --- /dev/null +++ b/yazi-scheduler/src/preload/op.rs @@ -0,0 +1,33 @@ +use std::sync::Arc; + +use yazi_config::plugin::PluginProps; +use yazi_shared::{fs::Url, Throttle}; + +#[derive(Debug)] +pub enum PreloadOp { + Rule(PreloadOpRule), + Size(PreloadOpSize), +} + +impl PreloadOp { + pub fn id(&self) -> usize { + match self { + Self::Rule(op) => op.id, + Self::Size(op) => op.id, + } + } +} + +#[derive(Clone, Debug)] +pub struct PreloadOpRule { + pub id: usize, + pub plugin: PluginProps, + pub targets: Vec, +} + +#[derive(Debug)] +pub struct PreloadOpSize { + pub id: usize, + pub target: Url, + pub throttle: Arc>, +} diff --git a/yazi-scheduler/src/preload/preload.rs b/yazi-scheduler/src/preload/preload.rs new file mode 100644 index 00000000..14fb7aae --- /dev/null +++ b/yazi-scheduler/src/preload/preload.rs @@ -0,0 +1,100 @@ +use std::collections::{BTreeMap, BTreeSet, HashMap}; + +use anyhow::Result; +use parking_lot::RwLock; +use tokio::sync::mpsc; +use tracing::error; +use yazi_config::Priority; +use yazi_plugin::isolate; +use yazi_shared::fs::{calculate_size, FilesOp, Url}; + +use super::{PreloadOp, PreloadOpRule, PreloadOpSize}; +use crate::{TaskOp, TaskProg, HIGH, NORMAL}; + +pub struct Preload { + macro_: async_priority_channel::Sender, + prog: mpsc::UnboundedSender, + + pub rule_loaded: RwLock>, + pub size_loading: RwLock>, +} + +impl Preload { + pub fn new( + macro_: async_priority_channel::Sender, + prog: mpsc::UnboundedSender, + ) -> Self { + Self { macro_, prog, rule_loaded: Default::default(), size_loading: Default::default() } + } + + pub async fn work(&self, op: PreloadOp) -> Result<()> { + match op { + PreloadOp::Rule(task) => { + let urls: Vec<_> = task.targets.iter().map(|f| f.url()).collect(); + let result = isolate::preload(task.plugin.cmd, task.targets, task.plugin.multi).await; + if let Err(e) = result { + self.fail(task.id, format!("Preload task failed:\n{e}"))?; + return Err(e.into()); + }; + + let code = result.unwrap(); + if code & 1 == 0 { + error!("Preload task returned {code}"); + } + if code >> 1 & 1 != 0 { + let mut loaded = self.rule_loaded.write(); + for url in urls { + loaded.get_mut(&url).map(|x| *x ^= 1 << task.plugin.id); + } + } + self.prog.send(TaskProg::Adv(task.id, 1, 0))?; + } + PreloadOp::Size(task) => { + let length = calculate_size(&task.target).await; + task.throttle.done((task.target, length), |buf| { + { + let mut loading = self.size_loading.write(); + for (path, _) in &buf { + loading.remove(path); + } + } + + let parent = buf[0].0.parent_url().unwrap(); + FilesOp::Size(parent, BTreeMap::from_iter(buf)).emit(); + }); + self.prog.send(TaskProg::Adv(task.id, 1, 0))?; + } + } + Ok(()) + } + + pub async fn rule(&self, task: PreloadOpRule) -> Result<()> { + let id = task.id; + self.prog.send(TaskProg::New(id, 0))?; + + match task.plugin.prio { + Priority::Low => self.macro_.send(PreloadOp::Rule(task).into(), NORMAL).await?, + Priority::Normal => self.macro_.send(PreloadOp::Rule(task).into(), HIGH).await?, + Priority::High => self.work(PreloadOp::Rule(task)).await?, + } + self.succ(id) + } + + pub async fn size(&self, task: PreloadOpSize) -> Result<()> { + let id = task.id; + + self.prog.send(TaskProg::New(id, 0))?; + self.work(PreloadOp::Size(task)).await?; + self.succ(id) + } +} + +impl Preload { + #[inline] + fn succ(&self, id: usize) -> Result<()> { Ok(self.prog.send(TaskProg::Succ(id))?) } + + #[inline] + fn fail(&self, id: usize, reason: String) -> Result<()> { + Ok(self.prog.send(TaskProg::Fail(id, reason))?) + } +} diff --git a/yazi-scheduler/src/process/mod.rs b/yazi-scheduler/src/process/mod.rs new file mode 100644 index 00000000..1d72a7b1 --- /dev/null +++ b/yazi-scheduler/src/process/mod.rs @@ -0,0 +1,7 @@ +#![allow(clippy::module_inception)] + +mod op; +mod process; + +pub use op::*; +pub use process::*; diff --git a/yazi-scheduler/src/process/op.rs b/yazi-scheduler/src/process/op.rs new file mode 100644 index 00000000..8436f37a --- /dev/null +++ b/yazi-scheduler/src/process/op.rs @@ -0,0 +1,25 @@ +use std::{ffi::OsString, mem}; + +use tokio::sync::oneshot; +use yazi_plugin::external::ShellOpt; + +#[derive(Debug)] +pub struct ProcessOpOpen { + pub id: usize, + pub cmd: OsString, + pub args: Vec, + pub block: bool, + pub orphan: bool, + pub cancel: oneshot::Sender<()>, +} + +impl From<&mut ProcessOpOpen> for ShellOpt { + fn from(op: &mut ProcessOpOpen) -> Self { + Self { + cmd: mem::take(&mut op.cmd), + args: mem::take(&mut op.args), + piped: false, + orphan: op.orphan, + } + } +} diff --git a/yazi-scheduler/src/workers/process.rs b/yazi-scheduler/src/process/process.rs similarity index 61% rename from yazi-scheduler/src/workers/process.rs rename to yazi-scheduler/src/process/process.rs index 75ce27fb..2431db80 100644 --- a/yazi-scheduler/src/workers/process.rs +++ b/yazi-scheduler/src/process/process.rs @@ -1,38 +1,16 @@ -use std::{ffi::OsString, mem}; - use anyhow::Result; -use tokio::{io::{AsyncBufReadExt, BufReader}, select, sync::{mpsc, oneshot}}; +use tokio::{io::{AsyncBufReadExt, BufReader}, select, sync::mpsc}; use yazi_plugin::external::{self, ShellOpt}; -use crate::{Scheduler, TaskOp, BLOCKER}; +use super::ProcessOpOpen; +use crate::{Scheduler, TaskProg, BLOCKER}; pub struct Process { - sch: mpsc::UnboundedSender, -} - -#[derive(Debug)] -pub struct ProcessOpOpen { - pub id: usize, - pub cmd: OsString, - pub args: Vec, - pub block: bool, - pub orphan: bool, - pub cancel: oneshot::Sender<()>, -} - -impl From<&mut ProcessOpOpen> for ShellOpt { - fn from(value: &mut ProcessOpOpen) -> Self { - Self { - cmd: mem::take(&mut value.cmd), - args: mem::take(&mut value.args), - piped: false, - orphan: value.orphan, - } - } + prog: mpsc::UnboundedSender, } impl Process { - pub fn new(sch: mpsc::UnboundedSender) -> Self { Self { sch } } + pub fn new(prog: mpsc::UnboundedSender) -> Self { Self { prog } } pub async fn open(&self, mut task: ProcessOpOpen) -> Result<()> { let opt = ShellOpt::from(&mut task); @@ -46,7 +24,7 @@ impl Process { self.succ(task.id)?; } Err(e) => { - self.sch.send(TaskOp::New(task.id, 0))?; + self.prog.send(TaskProg::New(task.id, 0))?; self.fail(task.id, format!("Failed to spawn process: {e}"))?; } } @@ -57,14 +35,14 @@ impl Process { match external::shell(opt) { Ok(_) => self.succ(task.id)?, Err(e) => { - self.sch.send(TaskOp::New(task.id, 0))?; + self.prog.send(TaskProg::New(task.id, 0))?; self.fail(task.id, format!("Failed to spawn process: {e}"))?; } } return Ok(()); } - self.sch.send(TaskOp::New(task.id, 0))?; + self.prog.send(TaskProg::New(task.id, 0))?; let mut child = external::shell(opt.with_piped())?; let mut stdout = BufReader::new(child.stdout.take().unwrap()).lines(); @@ -94,20 +72,22 @@ impl Process { } } - self.sch.send(TaskOp::Adv(task.id, 1, 0))?; + self.prog.send(TaskProg::Adv(task.id, 1, 0))?; self.succ(task.id) } } impl Process { #[inline] - fn succ(&self, id: usize) -> Result<()> { Ok(self.sch.send(TaskOp::Succ(id))?) } + fn succ(&self, id: usize) -> Result<()> { Ok(self.prog.send(TaskProg::Succ(id))?) } #[inline] fn fail(&self, id: usize, reason: String) -> Result<()> { - Ok(self.sch.send(TaskOp::Fail(id, reason))?) + Ok(self.prog.send(TaskProg::Fail(id, reason))?) } #[inline] - fn log(&self, id: usize, line: String) -> Result<()> { Ok(self.sch.send(TaskOp::Log(id, line))?) } + fn log(&self, id: usize, line: String) -> Result<()> { + Ok(self.prog.send(TaskProg::Log(id, line))?) + } } diff --git a/yazi-scheduler/src/scheduler.rs b/yazi-scheduler/src/scheduler.rs index dffda18a..ad0111f9 100644 --- a/yazi-scheduler/src/scheduler.rs +++ b/yazi-scheduler/src/scheduler.rs @@ -6,8 +6,8 @@ use tokio::{fs, select, sync::{mpsc::{self, UnboundedReceiver}, oneshot}}; use yazi_config::{open::Opener, plugin::PluginRule, TASKS}; use yazi_shared::{emit, event::Exec, fs::{unique_path, Url}, Layer, Throttle}; -use super::{Running, TaskOp, TaskStage}; -use crate::{workers::{File, FileOpDelete, FileOpLink, FileOpPaste, FileOpTrash, Plugin, PluginOpEntry, Preload, PreloadOpRule, PreloadOpSize, Process, ProcessOpOpen}, TaskKind}; +use super::{Running, TaskProg, TaskStage}; +use crate::{file::{File, FileOpDelete, FileOpLink, FileOpPaste, FileOpTrash}, plugin::{Plugin, PluginOpEntry}, preload::{Preload, PreloadOpRule, PreloadOpSize}, process::{Process, ProcessOpOpen}, TaskKind, TaskOp, HIGH, LOW, NORMAL}; pub struct Scheduler { pub file: Arc, @@ -15,79 +15,80 @@ pub struct Scheduler { pub preload: Arc, pub process: Arc, - todo: async_channel::Sender>, - prog: mpsc::UnboundedSender, + micro: async_priority_channel::Sender, u8>, + prog: mpsc::UnboundedSender, pub running: Arc>, } impl Scheduler { pub fn start() -> Self { - let (todo_tx, todo_rx) = async_channel::unbounded(); + let (micro_tx, micro_rx) = async_priority_channel::unbounded(); + let (macro_tx, macro_rx) = async_priority_channel::unbounded(); let (prog_tx, prog_rx) = mpsc::unbounded_channel(); let scheduler = Self { - file: Arc::new(File::new(prog_tx.clone())), - plugin: Arc::new(Plugin::new(prog_tx.clone())), - preload: Arc::new(Preload::new(prog_tx.clone())), + file: Arc::new(File::new(macro_tx.clone(), prog_tx.clone())), + plugin: Arc::new(Plugin::new(macro_tx.clone(), prog_tx.clone())), + preload: Arc::new(Preload::new(macro_tx.clone(), prog_tx.clone())), process: Arc::new(Process::new(prog_tx.clone())), - todo: todo_tx, + micro: micro_tx, prog: prog_tx, running: Default::default(), }; for _ in 0..TASKS.micro_workers { - scheduler.schedule_micro(todo_rx.clone()); + scheduler.schedule_micro(micro_rx.clone()); } for _ in 0..TASKS.macro_workers { - scheduler.schedule_macro(todo_rx.clone()); + scheduler.schedule_macro(micro_rx.clone(), macro_rx.clone()); } scheduler.progress(prog_rx); scheduler } - fn schedule_micro(&self, rx: async_channel::Receiver>) { + fn schedule_micro(&self, rx: async_priority_channel::Receiver, u8>) { tokio::spawn(async move { loop { - if let Ok(fut) = rx.recv().await { + if let Ok((fut, _)) = rx.recv().await { fut.await; } } }); } - fn schedule_macro(&self, rx: async_channel::Receiver>) { + fn schedule_macro( + &self, + micro: async_priority_channel::Receiver, u8>, + macro_: async_priority_channel::Receiver, + ) { let file = self.file.clone(); let plugin = self.plugin.clone(); + let preload = self.preload.clone(); let prog = self.prog.clone(); let running = self.running.clone(); tokio::spawn(async move { loop { - if let Ok(fut) = rx.try_recv() { - fut.await; - continue; - } - select! { - Ok(fut) = rx.recv() => { + Ok((fut, _)) = micro.recv() => { fut.await; } - Ok((id, mut op)) = file.recv() => { + Ok((op, _)) = macro_.recv() => { + let id = op.id(); if !running.read().exists(id) { continue; } - if let Err(e) = file.work(&mut op).await { - prog.send(TaskOp::Fail(id, format!("Failed to work on this task: {:?}", e))).ok(); - } - } - Ok((id, mut op)) = plugin.recv() => { - if !running.read().exists(id) { - continue; - } - if let Err(e) = plugin.work(&mut op).await { - prog.send(TaskOp::Fail(id, format!("Failed to work on this task: {:?}", e))).ok(); + + let result = match op { + TaskOp::File(op) => file.work(*op).await, + TaskOp::Plugin(op) => plugin.work(*op).await, + TaskOp::Preload(op) => preload.work(*op).await, + }; + + if let Err(e) = result { + prog.send(TaskProg::Fail(id, format!("Failed to work on this task: {:?}", e))).ok(); } } } @@ -95,20 +96,20 @@ impl Scheduler { }); } - fn progress(&self, mut rx: UnboundedReceiver) { - let todo = self.todo.clone(); + fn progress(&self, mut rx: UnboundedReceiver) { + let micro = self.micro.clone(); let running = self.running.clone(); tokio::spawn(async move { while let Some(op) = rx.recv().await { match op { - TaskOp::New(id, size) => { + TaskProg::New(id, size) => { if let Some(task) = running.write().get_mut(id) { task.total += 1; task.found += size; } } - TaskOp::Adv(id, succ, processed) => { + TaskProg::Adv(id, succ, processed) => { let mut running = running.write(); if let Some(task) = running.get_mut(id) { task.succ += succ; @@ -116,16 +117,16 @@ impl Scheduler { } if succ > 0 { if let Some(fut) = running.try_remove(id, TaskStage::Pending) { - todo.send_blocking(fut).ok(); + micro.try_send(fut, NORMAL).ok(); } } } - TaskOp::Succ(id) => { + TaskProg::Succ(id) => { if let Some(fut) = running.write().try_remove(id, TaskStage::Dispatched) { - todo.send_blocking(fut).ok(); + micro.try_send(fut, NORMAL).ok(); } } - TaskOp::Fail(id, reason) => { + TaskProg::Fail(id, reason) => { if let Some(task) = running.write().get_mut(id) { task.fail += 1; task.logs.push_str(&reason); @@ -136,7 +137,7 @@ impl Scheduler { } } } - TaskOp::Log(id, line) => { + TaskProg::Log(id, line) => { if let Some(task) = running.write().get_mut(id) { task.logs.push_str(&line); task.logs.push('\n'); @@ -156,7 +157,7 @@ impl Scheduler { let b = running.all.remove(&id).is_some(); if let Some(hook) = running.hooks.remove(&id) { - self.todo.send_blocking(hook(true)).ok(); + self.micro.try_send(hook(true), HIGH).ok(); } b } @@ -193,40 +194,42 @@ impl Scheduler { }) }); - _ = self.todo.send_blocking({ - let file = self.file.clone(); + let file = self.file.clone(); + _ = self.micro.try_send( async move { if !force { to = unique_path(to).await; } file.paste(FileOpPaste { id, from, to, cut: true, follow: false, retry: 0 }).await.ok(); } - .boxed() - }); + .boxed(), + LOW, + ); } pub fn file_copy(&self, from: Url, mut to: Url, force: bool) { let name = format!("Copy {:?} to {:?}", from, to); let id = self.running.write().add(TaskKind::User, name); - _ = self.todo.send_blocking({ - let file = self.file.clone(); + let file = self.file.clone(); + _ = self.micro.try_send( async move { if !force { to = unique_path(to).await; } file.paste(FileOpPaste { id, from, to, cut: false, follow: true, retry: 0 }).await.ok(); } - .boxed() - }); + .boxed(), + LOW, + ); } pub fn file_link(&self, from: Url, mut to: Url, relative: bool, force: bool) { let name = format!("Link {from:?} to {to:?}"); let id = self.running.write().add(TaskKind::User, name); - _ = self.todo.send_blocking({ - let file = self.file.clone(); + let file = self.file.clone(); + _ = self.micro.try_send( async move { if !force { to = unique_path(to).await; @@ -236,8 +239,9 @@ impl Scheduler { .await .ok(); } - .boxed() - }); + .boxed(), + LOW, + ); } pub fn file_delete(&self, target: Url) { @@ -259,38 +263,41 @@ impl Scheduler { }) }); - _ = self.todo.send_blocking({ - let file = self.file.clone(); + let file = self.file.clone(); + _ = self.micro.try_send( async move { file.delete(FileOpDelete { id, target, length: 0 }).await.ok(); } - .boxed() - }); + .boxed(), + LOW, + ); } pub fn file_trash(&self, target: Url) { let name = format!("Trash {:?}", target); let id = self.running.write().add(TaskKind::User, name); - _ = self.todo.send_blocking({ - let file = self.file.clone(); + let file = self.file.clone(); + _ = self.micro.try_send( async move { file.trash(FileOpTrash { id, target, length: 0 }).await.ok(); } - .boxed() - }); + .boxed(), + LOW, + ); } pub fn plugin_micro(&self, name: String) { let id = self.running.write().add(TaskKind::User, format!("Run micro plugin `{name}`")); - _ = self.todo.send_blocking({ - let plugin = self.plugin.clone(); + let plugin = self.plugin.clone(); + _ = self.micro.try_send( async move { plugin.micro(PluginOpEntry { id, name }).await.ok(); } - .boxed() - }); + .boxed(), + HIGH, + ); } pub fn plugin_macro(&self, name: String) { @@ -305,33 +312,35 @@ impl Scheduler { format!("Run preloader `{}` with {} target(s)", rule.exec.cmd, targets.len()), ); - _ = self.todo.send_blocking({ - let preload = self.preload.clone(); - - let (rule_id, rule_multi) = (rule.id, rule.multi); - let cmd = rule.exec.cmd.clone(); - let targets = targets.into_iter().cloned().collect(); + let plugin = rule.into(); + let targets = targets.into_iter().cloned().collect(); + let preload = self.preload.clone(); + _ = self.micro.try_send( async move { - preload.rule(PreloadOpRule { id, rule_id, rule_multi, plugin: cmd, targets }).await.ok(); + preload.rule(PreloadOpRule { id, plugin, targets }).await.ok(); } - .boxed() - }); + .boxed(), + HIGH, + ); } pub fn preload_size(&self, targets: Vec<&Url>) { let throttle = Arc::new(Throttle::new(targets.len(), Duration::from_millis(300))); let mut running = self.running.write(); + for target in targets { let id = running.add(TaskKind::Preload, format!("Calculate the size of {:?}", target)); - _ = self.todo.send_blocking({ - let preload = self.preload.clone(); - let target = target.clone(); - let throttle = throttle.clone(); + let target = target.clone(); + let throttle = throttle.clone(); + + let preload = self.preload.clone(); + _ = self.micro.try_send( async move { preload.size(PreloadOpSize { id, target, throttle }).await.ok(); } - .boxed() - }); + .boxed(), + HIGH, + ); } } diff --git a/yazi-scheduler/src/task.rs b/yazi-scheduler/src/task.rs index 6c404453..27baceea 100644 --- a/yazi-scheduler/src/task.rs +++ b/yazi-scheduler/src/task.rs @@ -59,7 +59,7 @@ impl From<&Task> for TaskSummary { } #[derive(Debug)] -pub enum TaskOp { +pub enum TaskProg { // id, size New(usize, u64), // id, processed, size diff --git a/yazi-scheduler/src/workers/mod.rs b/yazi-scheduler/src/workers/mod.rs deleted file mode 100644 index 01cec209..00000000 --- a/yazi-scheduler/src/workers/mod.rs +++ /dev/null @@ -1,9 +0,0 @@ -mod file; -mod plugin; -mod preload; -mod process; - -pub use file::*; -pub use plugin::*; -pub use preload::*; -pub use process::*; diff --git a/yazi-scheduler/src/workers/plugin.rs b/yazi-scheduler/src/workers/plugin.rs deleted file mode 100644 index 5f8eb2d9..00000000 --- a/yazi-scheduler/src/workers/plugin.rs +++ /dev/null @@ -1,75 +0,0 @@ -use anyhow::Result; -use tokio::sync::mpsc; - -use crate::TaskOp; - -pub struct Plugin { - tx: async_channel::Sender, - rx: async_channel::Receiver, - - sch: mpsc::UnboundedSender, -} - -#[derive(Debug)] -pub enum PluginOp { - Entry(PluginOpEntry), -} - -#[derive(Clone, Debug)] -pub struct PluginOpEntry { - pub id: usize, - pub name: String, -} - -impl Plugin { - pub fn new(sch: mpsc::UnboundedSender) -> Self { - let (tx, rx) = async_channel::unbounded(); - Self { tx, rx, sch } - } - - #[inline] - pub async fn recv(&self) -> Result<(usize, PluginOp)> { - Ok(match self.rx.recv().await? { - PluginOp::Entry(t) => (t.id, PluginOp::Entry(t)), - }) - } - - pub async fn work(&self, op: &mut PluginOp) -> Result<()> { - match op { - PluginOp::Entry(task) => { - yazi_plugin::isolate::entry(&task.name).await?; - } - } - Ok(()) - } - - pub async fn micro(&self, task: PluginOpEntry) -> Result<()> { - self.sch.send(TaskOp::New(task.id, 0))?; - - if let Err(e) = yazi_plugin::isolate::entry(&task.name).await { - self.fail(task.id, format!("Micro plugin failed:\n{e}"))?; - return Err(e.into()); - } - - self.sch.send(TaskOp::Adv(task.id, 1, 0))?; - self.succ(task.id) - } - - pub fn macro_(&self, task: PluginOpEntry) -> Result<()> { - let id = task.id; - - self.sch.send(TaskOp::New(id, 0))?; - self.tx.send_blocking(PluginOp::Entry(task))?; - self.succ(id) - } -} - -impl Plugin { - #[inline] - fn succ(&self, id: usize) -> Result<()> { Ok(self.sch.send(TaskOp::Succ(id))?) } - - #[inline] - fn fail(&self, id: usize, reason: String) -> Result<()> { - Ok(self.sch.send(TaskOp::Fail(id, reason))?) - } -} diff --git a/yazi-scheduler/src/workers/preload.rs b/yazi-scheduler/src/workers/preload.rs deleted file mode 100644 index 610d0eca..00000000 --- a/yazi-scheduler/src/workers/preload.rs +++ /dev/null @@ -1,93 +0,0 @@ -use std::{collections::{BTreeMap, BTreeSet, HashMap}, sync::Arc}; - -use anyhow::Result; -use parking_lot::RwLock; -use tokio::sync::mpsc; -use tracing::error; -use yazi_shared::{fs::{calculate_size, FilesOp, Url}, Throttle}; - -use crate::TaskOp; - -pub struct Preload { - sch: mpsc::UnboundedSender, - - pub rule_loaded: RwLock>, - pub size_loading: RwLock>, -} - -#[derive(Debug)] -pub struct PreloadOpSize { - pub id: usize, - pub target: Url, - pub throttle: Arc>, -} - -#[derive(Clone, Debug)] -pub struct PreloadOpRule { - pub id: usize, - pub rule_id: u8, - pub rule_multi: bool, - pub plugin: String, - pub targets: Vec, -} - -impl Preload { - pub fn new(sch: mpsc::UnboundedSender) -> Self { - Self { sch, rule_loaded: Default::default(), size_loading: Default::default() } - } - - pub async fn rule(&self, task: PreloadOpRule) -> Result<()> { - self.sch.send(TaskOp::New(task.id, 0))?; - - let urls: Vec<_> = task.targets.iter().map(|f| f.url()).collect(); - let result = yazi_plugin::isolate::preload(task.plugin, task.targets, task.rule_multi).await; - if let Err(e) = result { - self.fail(task.id, format!("Preload task failed:\n{e}"))?; - return Err(e.into()); - }; - - let code = result.unwrap(); - if code & 1 == 0 { - error!("Preload task returned {code}"); - } - if code >> 1 & 1 != 0 { - let mut loaded = self.rule_loaded.write(); - for url in urls { - loaded.get_mut(&url).map(|x| *x ^= 1 << task.rule_id); - } - } - - self.sch.send(TaskOp::Adv(task.id, 1, 0))?; - self.succ(task.id) - } - - pub async fn size(&self, task: PreloadOpSize) -> Result<()> { - self.sch.send(TaskOp::New(task.id, 0))?; - - let length = calculate_size(&task.target).await; - task.throttle.done((task.target, length), |buf| { - { - let mut loading = self.size_loading.write(); - for (path, _) in &buf { - loading.remove(path); - } - } - - let parent = buf[0].0.parent_url().unwrap(); - FilesOp::Size(parent, BTreeMap::from_iter(buf)).emit(); - }); - - self.sch.send(TaskOp::Adv(task.id, 1, 0))?; - self.succ(task.id) - } -} - -impl Preload { - #[inline] - fn succ(&self, id: usize) -> Result<()> { Ok(self.sch.send(TaskOp::Succ(id))?) } - - #[inline] - fn fail(&self, id: usize, reason: String) -> Result<()> { - Ok(self.sch.send(TaskOp::Fail(id, reason))?) - } -}