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/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/file/file.rs b/yazi-scheduler/src/file/file.rs index fc9f86d6..0d09e8b2 100644 --- a/yazi-scheduler/src/file/file.rs +++ b/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, + macro_: async_priority_channel::Sender, prog: mpsc::UnboundedSender, } impl File { - pub fn new(macro_: async_channel::Sender, prog: mpsc::UnboundedSender) -> Self { + pub fn new( + macro_: async_priority_channel::Sender, + prog: mpsc::UnboundedSender, + ) -> 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) } diff --git a/yazi-scheduler/src/lib.rs b/yazi-scheduler/src/lib.rs index b94b2087..297404c1 100644 --- a/yazi-scheduler/src/lib.rs +++ b/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(); } diff --git a/yazi-scheduler/src/plugin/plugin.rs b/yazi-scheduler/src/plugin/plugin.rs index c2992065..391b90ee 100644 --- a/yazi-scheduler/src/plugin/plugin.rs +++ b/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, + macro_: async_priority_channel::Sender, prog: mpsc::UnboundedSender, } impl Plugin { - pub fn new(macro_: async_channel::Sender, prog: mpsc::UnboundedSender) -> Self { + pub fn new( + macro_: async_priority_channel::Sender, + prog: mpsc::UnboundedSender, + ) -> 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) } } diff --git a/yazi-scheduler/src/preload/preload.rs b/yazi-scheduler/src/preload/preload.rs index 01dc4dcd..3ccb9661 100644 --- a/yazi-scheduler/src/preload/preload.rs +++ b/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, + macro_: async_priority_channel::Sender, prog: mpsc::UnboundedSender, pub rule_loaded: RwLock>, @@ -18,7 +18,10 @@ pub struct Preload { } impl Preload { - pub fn new(macro_: async_channel::Sender, prog: mpsc::UnboundedSender) -> Self { + pub fn new( + macro_: async_priority_channel::Sender, + prog: mpsc::UnboundedSender, + ) -> Self { Self { macro_, prog, rule_loaded: Default::default(), size_loading: Default::default() } } diff --git a/yazi-scheduler/src/scheduler.rs b/yazi-scheduler/src/scheduler.rs index d1584e1e..aafaf7db 100644 --- a/yazi-scheduler/src/scheduler.rs +++ b/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, @@ -15,15 +15,15 @@ pub struct Scheduler { pub preload: Arc, pub process: Arc, - micro: async_channel::Sender>, + micro: async_priority_channel::Sender, u8>, prog: mpsc::UnboundedSender, pub running: Arc>, } impl Scheduler { pub fn start() -> Self { - let (micro_tx, micro_rx) = async_channel::unbounded(); - let (macro_tx, macro_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 { @@ -47,10 +47,10 @@ impl Scheduler { 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; } } @@ -59,8 +59,8 @@ impl Scheduler { fn schedule_macro( &self, - micro: async_channel::Receiver>, - macro_: async_channel::Receiver, + micro: async_priority_channel::Receiver, u8>, + macro_: async_priority_channel::Receiver, ) { 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(); + 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(); + 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(); + 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(); + 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(); + 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(); + 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 (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 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, + ); } }