feat: fine-grained scheduling priority (#462)

Cargo.lock

@@ -131,14 +131,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bddcadddf5e9015d310179a59bb28c4d4b9920ad0f11e8e14dbadf654890c9a6"
 
 [[package]]
-name = "async-channel"
+name = "async-priority-channel"
-version = "1.9.0"
+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",

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":[]}

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 = [ "<A-k>" ], exec = "seek -5", desc = "Peek up 5 units in the preview" },
+	{ on = [ "<A-k>" ], exec = "seek -5", desc = "Seek up 5 units in the preview" },
-	{ on = [ "<A-j>" ], exec = "seek 5",  desc = "Peek down 5 units in the preview" },
+	{ on = [ "<A-j>" ], exec = "seek 5",  desc = "Seek down 5 units in the preview" },
-	{ on = [ "<A-PageUp>" ],   exec = "seek -5", desc = "Peek up 5 units in the preview" },
+	{ on = [ "<A-PageUp>" ],   exec = "seek -5", desc = "Seek up 5 units in the preview" },
-	{ on = [ "<A-PageDown>" ], exec = "seek 5",  desc = "Peek down 5 units in the preview" },
+	{ on = [ "<A-PageDown>" ], exec = "seek 5",  desc = "Seek down 5 units in the preview" },
 
 	{ on = [ "<Up>" ],    exec = "arrow -1", desc = "Move cursor up" },
 	{ on = [ "<Down>" ],  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 --smart",            desc = "Find next file" },
-	{ on = [ "?" ], exec = "find --previous --smart" },
+	{ on = [ "?" ], exec = "find --previous --smart", desc = "Find previous file" },
-	{ on = [ "n" ], exec = "find_arrow" },
+	{ on = [ "n" ], exec = "find_arrow",              desc = "Go to next found file" },
-	{ on = [ "N" ], exec = "find_arrow --previous" },
+	{ 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" },

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

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<boot::Args> = RoCell::new();

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;

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 {

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 }
+	}
+}

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<Self, Self::Err> {
+		match s {
+			"low" => Ok(Self::Low),
+			"normal" => Ok(Self::Normal),
+			"high" => Ok(Self::High),
+			_ => Err(anyhow!("Invalid priority: {s}")),
+		}
+	}
+}
+
+impl TryFrom<String> for Priority {
+	type Error = anyhow::Error;
+
+	fn try_from(s: String) -> Result<Self, Self::Error> { Self::from_str(&s) }
+}

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

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"
+anyhow                 = "^1"
-async-channel = "^1"
+async-priority-channel = "^0"
-base64        = "^0"
+base64                 = "^0"
-crossterm     = "^0"
+crossterm              = "^0"
-futures       = "^0"
+futures                = "^0"
-parking_lot   = "^0"
+parking_lot            = "^0"
-regex         = "^1"
+regex                  = "^1"
-tokio         = { version = "^1", features = [ "parking_lot", "rt-multi-thread" ] }
+tokio                  = { version = "^1", features = [ "parking_lot", "rt-multi-thread" ] }
-tokio-stream  = "^0"
+tokio-stream           = "^0"
-trash         = "^3"
+trash                  = "^3"
 
 # Logging
 tracing = { version = "^0", features = [ "max_level_debug", "release_max_level_warn" ] }

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<FileOp>,
+	macro_: async_priority_channel::Sender<TaskOp, u8>,
-	rx: async_channel::Receiver<FileOp>,
+	prog:   mpsc::UnboundedSender<TaskProg>,
-
-	sch: mpsc::UnboundedSender<TaskOp>,
-}
-
-#[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<Metadata>,
-	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,
 }
 
 impl File {
-	pub fn new(sch: mpsc::UnboundedSender<TaskOp>) -> Self {
+	pub fn new(
-		let (tx, rx) = async_channel::unbounded();
+		macro_: async_priority_channel::Sender<TaskOp, u8>,
-		Self { tx, rx, sch }
+		prog: mpsc::UnboundedSender<TaskProg>,
+	) -> Self {
+		Self { macro_, prog }
 	}
 
-	#[inline]
+	pub async fn work(&self, op: FileOp) -> Result<()> {
-	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<()> {
 		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.prog.send(TaskProg::New(id, task.meta.as_ref().unwrap().len()))?;
-		self.tx.send(FileOp::Link(task)).await?;
+		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.prog.send(TaskProg::New(id, meta.len()))?;
-			self.tx.send(FileOp::Delete(task)).await?;
+			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.prog.send(TaskProg::New(task.id, meta.len()))?;
-				self.tx.send(FileOp::Delete(task.clone())).await?;
+				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.prog.send(TaskProg::New(id, task.length))?;
-		self.tx.send(FileOp::Trash(task)).await?;
+		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 {

yazi-scheduler/src/file/mod.rs

@@ -0,0 +1,7 @@
+#![allow(clippy::module_inception)]
+
+mod file;
+mod op;
+
+pub use file::*;
+pub use op::*;

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<Metadata>,
+	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,
+}

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

yazi-scheduler/src/op.rs

@@ -0,0 +1,30 @@
+use crate::{file::FileOp, plugin::PluginOp, preload::PreloadOp};
+
+#[derive(Debug)]
+pub enum TaskOp {
+	File(Box<FileOp>),
+	Plugin(Box<PluginOp>),
+	Preload(Box<PreloadOp>),
+}
+
+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<FileOp> for TaskOp {
+	fn from(op: FileOp) -> Self { Self::File(Box::new(op)) }
+}
+
+impl From<PluginOp> for TaskOp {
+	fn from(op: PluginOp) -> Self { Self::Plugin(Box::new(op)) }
+}
+
+impl From<PreloadOp> for TaskOp {
+	fn from(op: PreloadOp) -> Self { Self::Preload(Box::new(op)) }
+}

yazi-scheduler/src/plugin/mod.rs

@@ -0,0 +1,7 @@
+#![allow(clippy::module_inception)]
+
+mod op;
+mod plugin;
+
+pub use op::*;
+pub use plugin::*;

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

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<TaskOp, u8>,
+	prog:   mpsc::UnboundedSender<TaskProg>,
+}
+
+impl Plugin {
+	pub fn new(
+		macro_: async_priority_channel::Sender<TaskOp, u8>,
+		prog: mpsc::UnboundedSender<TaskProg>,
+	) -> 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))?)
+	}
+}

yazi-scheduler/src/preload/mod.rs

@@ -0,0 +1,7 @@
+#![allow(clippy::module_inception)]
+
+mod op;
+mod preload;
+
+pub use op::*;
+pub use preload::*;

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<yazi_shared::fs::File>,
+}
+
+#[derive(Debug)]
+pub struct PreloadOpSize {
+	pub id:       usize,
+	pub target:   Url,
+	pub throttle: Arc<Throttle<(Url, u64)>>,
+}

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<TaskOp, u8>,
+	prog:   mpsc::UnboundedSender<TaskProg>,
+
+	pub rule_loaded:  RwLock<HashMap<Url, u32>>,
+	pub size_loading: RwLock<BTreeSet<Url>>,
+}
+
+impl Preload {
+	pub fn new(
+		macro_: async_priority_channel::Sender<TaskOp, u8>,
+		prog: mpsc::UnboundedSender<TaskProg>,
+	) -> 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))?)
+	}
+}

yazi-scheduler/src/process/mod.rs

@@ -0,0 +1,7 @@
+#![allow(clippy::module_inception)]
+
+mod op;
+mod process;
+
+pub use op::*;
+pub use process::*;

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<OsString>,
+	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,
+		}
+	}
+}

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<TaskOp>,
+	prog: mpsc::UnboundedSender<TaskProg>,
-}
-
-#[derive(Debug)]
-pub struct ProcessOpOpen {
-	pub id:     usize,
-	pub cmd:    OsString,
-	pub args:   Vec<OsString>,
-	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,
-		}
-	}
 }
 
 impl Process {
-	pub fn new(sch: mpsc::UnboundedSender<TaskOp>) -> Self { Self { sch } }
+	pub fn new(prog: mpsc::UnboundedSender<TaskProg>) -> 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))?)
+	}
 }

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 super::{Running, TaskProg, TaskStage};
-use crate::{workers::{File, FileOpDelete, FileOpLink, FileOpPaste, FileOpTrash, Plugin, PluginOpEntry, Preload, PreloadOpRule, PreloadOpSize, Process, ProcessOpOpen}, TaskKind};
+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<File>,
@@ -15,79 +15,80 @@ pub struct Scheduler {
 	pub preload: Arc<Preload>,
 	pub process: Arc<Process>,
 
-	todo:        async_channel::Sender<BoxFuture<'static, ()>>,
+	micro:       async_priority_channel::Sender<BoxFuture<'static, ()>, u8>,
-	prog:        mpsc::UnboundedSender<TaskOp>,
+	prog:        mpsc::UnboundedSender<TaskProg>,
 	pub running: Arc<RwLock<Running>>,
 }
 
 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())),
+			file:    Arc::new(File::new(macro_tx.clone(), prog_tx.clone())),
-			plugin:  Arc::new(Plugin::new(prog_tx.clone())),
+			plugin:  Arc::new(Plugin::new(macro_tx.clone(), prog_tx.clone())),
-			preload: Arc::new(Preload::new(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<BoxFuture<'static, ()>>) {
+	fn schedule_micro(&self, rx: async_priority_channel::Receiver<BoxFuture<'static, ()>, 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<BoxFuture<'static, ()>>) {
+	fn schedule_macro(
+		&self,
+		micro: async_priority_channel::Receiver<BoxFuture<'static, ()>, u8>,
+		macro_: async_priority_channel::Receiver<TaskOp, u8>,
+	) {
 		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();
+						let result = match op {
-						}
+							TaskOp::File(op) => file.work(*op).await,
-					}
+							TaskOp::Plugin(op) => plugin.work(*op).await,
-					Ok((id, mut op)) = plugin.recv() => {
+							TaskOp::Preload(op) => preload.work(*op).await,
-						if !running.read().exists(id) {
+						};
-							continue;
+
-						}
+						if let Err(e) = result {
-						if let Err(e) = plugin.work(&mut op).await {
+							prog.send(TaskProg::Fail(id, format!("Failed to work on this task: {:?}", e))).ok();
-							prog.send(TaskOp::Fail(id, format!("Failed to work on this task: {:?}", e))).ok();
 						}
 					}
 				}
@@ -95,20 +96,20 @@ impl Scheduler {
 		});
 	}
 
-	fn progress(&self, mut rx: UnboundedReceiver<TaskOp>) {
+	fn progress(&self, mut rx: UnboundedReceiver<TaskProg>) {
-		let todo = self.todo.clone();
+		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 plugin = rule.into();
-			let preload = self.preload.clone();
+		let targets = targets.into_iter().cloned().collect();
-
+		let preload = self.preload.clone();
-			let (rule_id, rule_multi) = (rule.id, rule.multi);
+		_ = self.micro.try_send(
-			let cmd = rule.exec.cmd.clone();
-			let targets = targets.into_iter().cloned().collect();
 			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 target = target.clone();
-				let preload = self.preload.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,
+			);
 		}
 	}
 

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

yazi-scheduler/src/workers/mod.rs

@@ -1,9 +0,0 @@
-mod file;
-mod plugin;
-mod preload;
-mod process;
-
-pub use file::*;
-pub use plugin::*;
-pub use preload::*;
-pub use process::*;

yazi-scheduler/src/workers/plugin.rs

@@ -1,75 +0,0 @@
-use anyhow::Result;
-use tokio::sync::mpsc;
-
-use crate::TaskOp;
-
-pub struct Plugin {
-	tx: async_channel::Sender<PluginOp>,
-	rx: async_channel::Receiver<PluginOp>,
-
-	sch: mpsc::UnboundedSender<TaskOp>,
-}
-
-#[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<TaskOp>) -> 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))?)
-	}
-}

yazi-scheduler/src/workers/preload.rs

@@ -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<TaskOp>,
-
-	pub rule_loaded:  RwLock<HashMap<Url, u32>>,
-	pub size_loading: RwLock<BTreeSet<Url>>,
-}
-
-#[derive(Debug)]
-pub struct PreloadOpSize {
-	pub id:       usize,
-	pub target:   Url,
-	pub throttle: Arc<Throttle<(Url, u64)>>,
-}
-
-#[derive(Clone, Debug)]
-pub struct PreloadOpRule {
-	pub id:         usize,
-	pub rule_id:    u8,
-	pub rule_multi: bool,
-	pub plugin:     String,
-	pub targets:    Vec<yazi_shared::fs::File>,
-}
-
-impl Preload {
-	pub fn new(sch: mpsc::UnboundedSender<TaskOp>) -> 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))?)
-	}
-}