..

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

yazi-scheduler/Cargo.toml

@@ -16,7 +16,7 @@ yazi-plugin  = { path = "../yazi-plugin", version = "0.1.5" }
 
 # External dependencies
 anyhow                 = "^1"
-async-channel = "^1"
+async-priority-channel = "^0"
 base64                 = "^0"
 crossterm              = "^0"
 futures                = "^0"

yazi-scheduler/src/file/file.rs

@@ -8,15 +8,18 @@ use yazi_config::TASKS;
 use yazi_shared::fs::{calculate_size, copy_with_progress, path_relative_to, Url};
 
 use super::{FileOp, FileOpDelete, FileOpLink, FileOpPaste, FileOpTrash};
-use crate::{TaskOp, TaskProg};
+use crate::{TaskOp, TaskProg, LOW, VERY_LOW};
 
 pub struct File {
-	macro_: async_channel::Sender<TaskOp>,
+	macro_: async_priority_channel::Sender<TaskOp, u8>,
 	prog:   mpsc::UnboundedSender<TaskProg>,
 }
 
 impl File {
-	pub fn new(macro_: async_channel::Sender<TaskOp>, prog: mpsc::UnboundedSender<TaskProg>) -> Self {
+	pub fn new(
+		macro_: async_priority_channel::Sender<TaskOp, u8>,
+		prog: mpsc::UnboundedSender<TaskProg>,
+	) -> Self {
 		Self { macro_, prog }
 	}
 
@@ -50,7 +53,7 @@ impl File {
 						{
 							self.log(task.id, format!("Paste task retry: {:?}", task))?;
 							task.retry += 1;
-							return Ok(self.macro_.send(FileOp::Paste(task).into()).await?);
+							return Ok(self.macro_.send(FileOp::Paste(task).into(), VERY_LOW).await?);
 						}
 						Err(e) => Err(e)?,
 					}
@@ -144,9 +147,9 @@ impl File {
 			self.prog.send(TaskProg::New(id, meta.len()))?;
 
 			if meta.is_file() {
-				self.macro_.send(FileOp::Paste(task).into()).await?;
+				self.macro_.send(FileOp::Paste(task).into(), VERY_LOW).await?;
 			} else if meta.is_symlink() {
-				self.macro_.send(FileOp::Link(task.to_link(meta)).into()).await?;
+				self.macro_.send(FileOp::Link(task.to_link(meta)).into(), LOW).await?;
 			}
 			return self.succ(id);
 		}
@@ -190,9 +193,9 @@ impl File {
 				self.prog.send(TaskProg::New(task.id, meta.len()))?;
 
 				if meta.is_file() {
-					self.macro_.send(FileOp::Paste(task.clone()).into()).await?;
+					self.macro_.send(FileOp::Paste(task.clone()).into(), VERY_LOW).await?;
 				} else if meta.is_symlink() {
-					self.macro_.send(FileOp::Link(task.to_link(meta)).into()).await?;
+					self.macro_.send(FileOp::Link(task.to_link(meta)).into(), LOW).await?;
 				}
 			}
 		}
@@ -206,7 +209,7 @@ impl File {
 		}
 
 		self.prog.send(TaskProg::New(id, task.meta.as_ref().unwrap().len()))?;
-		self.macro_.send(FileOp::Link(task).into()).await?;
+		self.macro_.send(FileOp::Link(task).into(), LOW).await?;
 		self.succ(id)
 	}
 
@@ -216,7 +219,7 @@ impl File {
 			let id = task.id;
 			task.length = meta.len();
 			self.prog.send(TaskProg::New(id, meta.len()))?;
-			self.macro_.send(FileOp::Delete(task).into()).await?;
+			self.macro_.send(FileOp::Delete(task).into(), LOW).await?;
 			return self.succ(id);
 		}
 
@@ -241,7 +244,7 @@ impl File {
 				task.target = Url::from(entry.path());
 				task.length = meta.len();
 				self.prog.send(TaskProg::New(task.id, meta.len()))?;
-				self.macro_.send(FileOp::Delete(task.clone()).into()).await?;
+				self.macro_.send(FileOp::Delete(task.clone()).into(), LOW).await?;
 			}
 		}
 		self.succ(task.id)
@@ -252,7 +255,7 @@ impl File {
 		task.length = calculate_size(&task.target).await;
 
 		self.prog.send(TaskProg::New(id, task.length))?;
-		self.macro_.send(FileOp::Trash(task).into()).await?;
+		self.macro_.send(FileOp::Trash(task).into(), VERY_LOW).await?;
 		self.succ(id)
 	}
 

yazi-scheduler/src/lib.rs

@@ -16,4 +16,10 @@ pub use running::*;
 pub use scheduler::*;
 pub use task::*;
 
+const VERY_LOW: u8 = 0;
+const LOW: u8 = 1;
+const NORMAL: u8 = 2;
+const HIGH: u8 = 3;
+const VERY_HIGH: u8 = 4;
+
 pub fn init() { init_blocker(); }

yazi-scheduler/src/plugin/plugin.rs

@@ -2,15 +2,18 @@ use anyhow::Result;
 use tokio::sync::mpsc;
 
 use super::{PluginOp, PluginOpEntry};
-use crate::{TaskOp, TaskProg};
+use crate::{TaskOp, TaskProg, NORMAL};
 
 pub struct Plugin {
-	macro_: async_channel::Sender<TaskOp>,
+	macro_: async_priority_channel::Sender<TaskOp, u8>,
 	prog:   mpsc::UnboundedSender<TaskProg>,
 }
 
 impl Plugin {
-	pub fn new(macro_: async_channel::Sender<TaskOp>, prog: mpsc::UnboundedSender<TaskProg>) -> Self {
+	pub fn new(
+		macro_: async_priority_channel::Sender<TaskOp, u8>,
+		prog: mpsc::UnboundedSender<TaskProg>,
+	) -> Self {
 		Self { macro_, prog }
 	}
 
@@ -39,7 +42,7 @@ impl Plugin {
 		let id = task.id;
 
 		self.prog.send(TaskProg::New(id, 0))?;
-		self.macro_.send_blocking(PluginOp::Entry(task).into())?;
+		self.macro_.try_send(PluginOp::Entry(task).into(), NORMAL)?;
 		self.succ(id)
 	}
 }

yazi-scheduler/src/preload/preload.rs

@@ -10,7 +10,7 @@ use super::{PreloadOp, PreloadOpRule, PreloadOpSize};
 use crate::{TaskOp, TaskProg};
 
 pub struct Preload {
-	macro_: async_channel::Sender<TaskOp>,
+	macro_: async_priority_channel::Sender<TaskOp, u8>,
 	prog:   mpsc::UnboundedSender<TaskProg>,
 
 	pub rule_loaded:  RwLock<HashMap<Url, u32>>,
@@ -18,7 +18,10 @@ pub struct Preload {
 }
 
 impl Preload {
-	pub fn new(macro_: async_channel::Sender<TaskOp>, prog: mpsc::UnboundedSender<TaskProg>) -> Self {
+	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() }
 	}
 

yazi-scheduler/src/scheduler.rs

@@ -7,7 +7,7 @@ use yazi_config::{open::Opener, plugin::PluginRule, TASKS};
 use yazi_shared::{emit, event::Exec, fs::{unique_path, Url}, Layer, Throttle};
 
 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};
+use crate::{file::{File, FileOpDelete, FileOpLink, FileOpPaste, FileOpTrash}, plugin::{Plugin, PluginOpEntry}, preload::{Preload, PreloadOpRule, PreloadOpSize}, process::{Process, ProcessOpOpen}, TaskKind, TaskOp, HIGH, NORMAL, VERY_HIGH, VERY_LOW};
 
 pub struct Scheduler {
 	pub file:    Arc<File>,
@@ -15,15 +15,15 @@ pub struct Scheduler {
 	pub preload: Arc<Preload>,
 	pub process: Arc<Process>,
 
-	micro:       async_channel::Sender<BoxFuture<'static, ()>>,
+	micro:       async_priority_channel::Sender<BoxFuture<'static, ()>, u8>,
 	prog:        mpsc::UnboundedSender<TaskProg>,
 	pub running: Arc<RwLock<Running>>,
 }
 
 impl Scheduler {
 	pub fn start() -> Self {
-		let (micro_tx, micro_rx) = async_channel::unbounded();
+		let (micro_tx, micro_rx) = async_priority_channel::unbounded();
-		let (macro_tx, macro_rx) = async_channel::unbounded();
+		let (macro_tx, macro_rx) = async_priority_channel::unbounded();
 		let (prog_tx, prog_rx) = mpsc::unbounded_channel();
 
 		let scheduler = Self {
@@ -47,10 +47,10 @@ impl Scheduler {
 		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;
 				}
 			}
@@ -59,8 +59,8 @@ impl Scheduler {
 
 	fn schedule_macro(
 		&self,
-		micro: async_channel::Receiver<BoxFuture<'static, ()>>,
+		micro: async_priority_channel::Receiver<BoxFuture<'static, ()>, u8>,
-		macro_: async_channel::Receiver<TaskOp>,
+		macro_: async_priority_channel::Receiver<TaskOp, u8>,
 	) {
 		let file = self.file.clone();
 		let plugin = self.plugin.clone();
@@ -72,10 +72,10 @@ impl Scheduler {
 		tokio::spawn(async move {
 			loop {
 				select! {
-					Ok(fut) = micro.recv() => {
+					Ok((fut, _)) = micro.recv() => {
 						fut.await;
 					}
-					Ok(op) = macro_.recv() => {
+					Ok((op, _)) = macro_.recv() => {
 						let id = op.id();
 						if !running.read().exists(id) {
 							continue;
@@ -117,13 +117,13 @@ impl Scheduler {
 						}
 						if succ > 0 {
 							if let Some(fut) = running.try_remove(id, TaskStage::Pending) {
-								micro.send_blocking(fut).ok();
+								micro.try_send(fut, HIGH).ok();
 							}
 						}
 					}
 					TaskProg::Succ(id) => {
 						if let Some(fut) = running.write().try_remove(id, TaskStage::Dispatched) {
-							micro.send_blocking(fut).ok();
+							micro.try_send(fut, HIGH).ok();
 						}
 					}
 					TaskProg::Fail(id, reason) => {
@@ -157,7 +157,7 @@ impl Scheduler {
 		let b = running.all.remove(&id).is_some();
 
 		if let Some(hook) = running.hooks.remove(&id) {
-			self.micro.send_blocking(hook(true)).ok();
+			self.micro.try_send(hook(true), HIGH).ok();
 		}
 		b
 	}
@@ -194,40 +194,42 @@ impl Scheduler {
 			})
 		});
 
-		_ = self.micro.send_blocking({
 		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(),
-		});
+			VERY_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.micro.send_blocking({
 		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(),
-		});
+			VERY_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.micro.send_blocking({
 		let file = self.file.clone();
+		_ = self.micro.try_send(
 			async move {
 				if !force {
 					to = unique_path(to).await;
@@ -237,8 +239,9 @@ impl Scheduler {
 					.await
 					.ok();
 			}
-			.boxed()
+			.boxed(),
-		});
+			VERY_LOW,
+		);
 	}
 
 	pub fn file_delete(&self, target: Url) {
@@ -260,38 +263,41 @@ impl Scheduler {
 			})
 		});
 
-		_ = self.micro.send_blocking({
 		let file = self.file.clone();
+		_ = self.micro.try_send(
 			async move {
 				file.delete(FileOpDelete { id, target, length: 0 }).await.ok();
 			}
-			.boxed()
+			.boxed(),
-		});
+			VERY_LOW,
+		);
 	}
 
 	pub fn file_trash(&self, target: Url) {
 		let name = format!("Trash {:?}", target);
 		let id = self.running.write().add(TaskKind::User, name);
 
-		_ = self.micro.send_blocking({
 		let file = self.file.clone();
+		_ = self.micro.try_send(
 			async move {
 				file.trash(FileOpTrash { id, target, length: 0 }).await.ok();
 			}
-			.boxed()
+			.boxed(),
-		});
+			VERY_LOW,
+		);
 	}
 
 	pub fn plugin_micro(&self, name: String) {
 		let id = self.running.write().add(TaskKind::User, format!("Run micro plugin `{name}`"));
 
-		_ = self.micro.send_blocking({
 		let plugin = self.plugin.clone();
+		_ = self.micro.try_send(
 			async move {
 				plugin.micro(PluginOpEntry { id, name }).await.ok();
 			}
-			.boxed()
+			.boxed(),
-		});
+			VERY_HIGH,
+		);
 	}
 
 	pub fn plugin_macro(&self, name: String) {
@@ -306,17 +312,18 @@ impl Scheduler {
 			format!("Run preloader `{}` with {} target(s)", rule.exec.cmd, targets.len()),
 		);
 
-		_ = self.micro.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 preload = self.preload.clone();
+		_ = self.micro.try_send(
 			async move {
 				preload.rule(PreloadOpRule { id, rule_id, rule_multi, plugin: cmd, targets }).await.ok();
 			}
-			.boxed()
+			.boxed(),
-		});
+			HIGH,
+		);
 	}
 
 	pub fn preload_size(&self, targets: Vec<&Url>) {
@@ -324,15 +331,17 @@ impl Scheduler {
 		let mut running = self.running.write();
 		for target in targets {
 			let id = running.add(TaskKind::Preload, format!("Calculate the size of {:?}", target));
-			_ = self.micro.send_blocking({
-				let preload = self.preload.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,
+			);
 		}
 	}