diff --git a/Cargo.lock b/Cargo.lock index 1cb3aa75..d4de3cf3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3223,7 +3223,6 @@ dependencies = [ "regex", "tokio", "tokio-stream", - "tokio-util", "tracing", "trash", "yazi-adaptor", diff --git a/yazi-core/src/manager/manager.rs b/yazi-core/src/manager/manager.rs index f536185c..a6bf8f2e 100644 --- a/yazi-core/src/manager/manager.rs +++ b/yazi-core/src/manager/manager.rs @@ -19,7 +19,7 @@ impl Manager { tabs: Tabs::make(), yanked: Default::default(), - watcher: Watcher::start(), + watcher: Watcher::serve(), mimetype: Default::default(), } } diff --git a/yazi-core/src/manager/watcher.rs b/yazi-core/src/manager/watcher.rs index e712e038..d2f8837f 100644 --- a/yazi-core/src/manager/watcher.rs +++ b/yazi-core/src/manager/watcher.rs @@ -20,7 +20,7 @@ pub struct Watcher { } impl Watcher { - pub(super) fn start() -> Self { + pub(super) fn serve() -> Self { let (tx, rx) = mpsc::unbounded_channel(); let watcher = RecommendedWatcher::new( { diff --git a/yazi-core/src/tasks/tasks.rs b/yazi-core/src/tasks/tasks.rs index 69b01178..2b680d46 100644 --- a/yazi-core/src/tasks/tasks.rs +++ b/yazi-core/src/tasks/tasks.rs @@ -1,6 +1,6 @@ use std::{sync::Arc, time::Duration}; -use tokio::time::sleep; +use tokio::{task::JoinHandle, time::sleep}; use yazi_scheduler::{Scheduler, TaskSummary}; use yazi_shared::{emit, event::Cmd, term::Term, Layer}; @@ -8,6 +8,7 @@ use super::{TasksProgress, TASKS_BORDER, TASKS_PADDING, TASKS_PERCENT}; pub struct Tasks { pub(super) scheduler: Arc, + handle: JoinHandle<()>, pub visible: bool, pub cursor: usize, @@ -16,17 +17,11 @@ pub struct Tasks { } impl Tasks { - pub fn start() -> Self { - let tasks = Self { - scheduler: Arc::new(Scheduler::start()), - visible: false, - cursor: 0, - progress: Default::default(), - summaries: Default::default(), - }; + pub fn serve() -> Self { + let scheduler = Scheduler::serve(); + let ongoing = scheduler.ongoing.clone(); - let ongoing = tasks.scheduler.ongoing.clone(); - tokio::spawn(async move { + let handle = tokio::spawn(async move { let mut last = TasksProgress::default(); loop { sleep(Duration::from_millis(500)).await; @@ -39,7 +34,20 @@ impl Tasks { } }); - tasks + Self { + scheduler: Arc::new(scheduler), + handle, + + visible: false, + cursor: 0, + progress: Default::default(), + summaries: Default::default(), + } + } + + pub fn shutdown(&self) { + self.scheduler.shutdown(); + self.handle.abort(); } #[inline] diff --git a/yazi-fm/src/app/commands/quit.rs b/yazi-fm/src/app/commands/quit.rs index e4bbf4cb..ccecf0dc 100644 --- a/yazi-fm/src/app/commands/quit.rs +++ b/yazi-fm/src/app/commands/quit.rs @@ -7,6 +7,8 @@ use crate::app::App; impl App { pub(crate) fn quit(&mut self, opt: EventQuit) -> ! { + self.cx.tasks.shutdown(); + if !opt.no_cwd_file { self.cwd_to_file(); } diff --git a/yazi-fm/src/context.rs b/yazi-fm/src/context.rs index 30e6c882..3c12b7d5 100644 --- a/yazi-fm/src/context.rs +++ b/yazi-fm/src/context.rs @@ -17,7 +17,7 @@ impl Ctx { pub fn make() -> Self { Self { manager: Manager::make(), - tasks: Tasks::start(), + tasks: Tasks::serve(), select: Default::default(), input: Default::default(), help: Default::default(), diff --git a/yazi-proxy/src/select.rs b/yazi-proxy/src/select.rs index 5629485d..fd00aca9 100644 --- a/yazi-proxy/src/select.rs +++ b/yazi-proxy/src/select.rs @@ -1,6 +1,6 @@ use tokio::sync::oneshot; use yazi_config::popup::SelectCfg; -use yazi_shared::{emit, event::Cmd, term::Term, Layer}; +use yazi_shared::{emit, event::Cmd, Layer}; use crate::options::SelectOpt; @@ -11,6 +11,6 @@ impl SelectProxy { pub async fn show(cfg: SelectCfg) -> anyhow::Result { let (tx, rx) = oneshot::channel(); emit!(Call(Cmd::new("show").with_data(SelectOpt { cfg, tx }), Layer::Select)); - rx.await.unwrap_or_else(|_| Term::goodbye(|| false)) + rx.await? } } diff --git a/yazi-scheduler/Cargo.toml b/yazi-scheduler/Cargo.toml index 6b8a51db..2e5e0524 100644 --- a/yazi-scheduler/Cargo.toml +++ b/yazi-scheduler/Cargo.toml @@ -25,7 +25,6 @@ parking_lot = "0.12.1" regex = "1.10.3" tokio = { version = "1.36.0", features = [ "parking_lot", "rt-multi-thread" ] } tokio-stream = "0.1.15" -tokio-util = "0.7.10" # Logging tracing = { version = "0.1.40", features = [ "max_level_debug", "release_max_level_warn" ] } diff --git a/yazi-scheduler/src/process/op.rs b/yazi-scheduler/src/process/op.rs index b740e071..e00d39a5 100644 --- a/yazi-scheduler/src/process/op.rs +++ b/yazi-scheduler/src/process/op.rs @@ -1,6 +1,6 @@ use std::ffi::OsString; -use tokio_util::sync::CancellationToken; +use tokio::sync::mpsc; use super::ShellOpt; @@ -32,10 +32,10 @@ impl From for ShellOpt { #[derive(Debug)] pub struct ProcessOpBg { - pub id: usize, - pub cmd: OsString, - pub args: Vec, - pub ct: CancellationToken, + pub id: usize, + pub cmd: OsString, + pub args: Vec, + pub cancel: mpsc::Receiver<()>, } impl From for ShellOpt { diff --git a/yazi-scheduler/src/process/process.rs b/yazi-scheduler/src/process/process.rs index e868f3ae..0c2feb0f 100644 --- a/yazi-scheduler/src/process/process.rs +++ b/yazi-scheduler/src/process/process.rs @@ -57,10 +57,12 @@ impl Process { let mut stdout = BufReader::new(child.stdout.take().unwrap()).lines(); let mut stderr = BufReader::new(child.stderr.take().unwrap()).lines(); + let mut cancel = task.cancel; loop { select! { - _ = task.ct.cancelled() => { + _ = cancel.recv() => { child.start_kill().ok(); + cancel.close(); break; } Ok(Some(line)) = stdout.next_line() => { diff --git a/yazi-scheduler/src/scheduler.rs b/yazi-scheduler/src/scheduler.rs index 8cff33eb..8449d92b 100644 --- a/yazi-scheduler/src/scheduler.rs +++ b/yazi-scheduler/src/scheduler.rs @@ -2,8 +2,7 @@ use std::{borrow::Cow, ffi::OsString, sync::Arc, time::Duration}; use futures::{future::BoxFuture, FutureExt}; use parking_lot::Mutex; -use tokio::{fs, select, sync::{mpsc::{self, UnboundedReceiver}, oneshot}}; -use tokio_util::sync::CancellationToken; +use tokio::{fs, select, sync::{mpsc::{self, UnboundedReceiver}, oneshot}, task::JoinHandle}; use yazi_config::{open::Opener, plugin::PluginRule, TASKS}; use yazi_plugin::ValueSendable; use yazi_shared::{fs::{unique_path, Url}, Throttle}; @@ -19,16 +18,17 @@ pub struct Scheduler { micro: async_priority_channel::Sender, u8>, prog: mpsc::UnboundedSender, + handles: Vec>, pub ongoing: Arc>, } impl Scheduler { - pub fn start() -> Self { + pub fn serve() -> Self { 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 { + let mut scheduler = Self { 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())), @@ -36,132 +36,35 @@ impl Scheduler { micro: micro_tx, prog: prog_tx, + handles: Vec::with_capacity(TASKS.micro_workers as usize + TASKS.macro_workers as usize + 1), ongoing: Default::default(), }; for _ in 0..TASKS.micro_workers { - scheduler.schedule_micro(micro_rx.clone()); + scheduler.handles.push(scheduler.schedule_micro(micro_rx.clone())); } for _ in 0..TASKS.macro_workers { - scheduler.schedule_macro(micro_rx.clone(), macro_rx.clone()); + scheduler.handles.push(scheduler.schedule_macro(micro_rx.clone(), macro_rx.clone())); } scheduler.progress(prog_rx); scheduler } - fn schedule_micro(&self, rx: async_priority_channel::Receiver, u8>) { - tokio::spawn(async move { - loop { - if let Ok((fut, _)) = rx.recv().await { - fut.await; - } - } - }); - } - - 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 ongoing = self.ongoing.clone(); - - tokio::spawn(async move { - loop { - select! { - Ok((fut, _)) = micro.recv() => { - fut.await; - } - Ok((op, _)) = macro_.recv() => { - let id = op.id(); - if !ongoing.lock().exists(id) { - continue; - } - - 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(); - } - } - } - } - }); - } - - fn progress(&self, mut rx: UnboundedReceiver) { - let micro = self.micro.clone(); - let ongoing = self.ongoing.clone(); - - tokio::spawn(async move { - while let Some(op) = rx.recv().await { - match op { - TaskProg::New(id, size) => { - if let Some(task) = ongoing.lock().get_mut(id) { - task.total += 1; - task.found += size; - } - } - TaskProg::Adv(id, succ, processed) => { - let mut ongoing = ongoing.lock(); - if let Some(task) = ongoing.get_mut(id) { - task.succ += succ; - task.processed += processed; - } - if succ > 0 { - if let Some(fut) = ongoing.try_remove(id, TaskStage::Pending) { - micro.try_send(fut, NORMAL).ok(); - } - } - } - TaskProg::Succ(id) => { - if let Some(fut) = ongoing.lock().try_remove(id, TaskStage::Dispatched) { - micro.try_send(fut, NORMAL).ok(); - } - } - TaskProg::Fail(id, reason) => { - if let Some(task) = ongoing.lock().get_mut(id) { - task.fail += 1; - task.logs.push_str(&reason); - task.logs.push('\n'); - - if let Some(logger) = &task.logger { - logger.send(reason).ok(); - } - } - } - TaskProg::Log(id, line) => { - if let Some(task) = ongoing.lock().get_mut(id) { - task.logs.push_str(&line); - task.logs.push('\n'); - - if let Some(logger) = &task.logger { - logger.send(line).ok(); - } - } - } - } - } - }); - } - pub fn cancel(&self, id: usize) -> bool { let mut ongoing = self.ongoing.lock(); - let b = ongoing.all.remove(&id).is_some(); if let Some(hook) = ongoing.hooks.remove(&id) { self.micro.try_send(hook(true), HIGH).ok(); + return false; + } + + ongoing.all.remove(&id).is_some() + } + + pub fn shutdown(&self) { + for handle in &self.handles { + handle.abort(); } - b } pub fn file_cut(&self, from: Url, mut to: Url, force: bool) { @@ -285,7 +188,7 @@ impl Scheduler { plugin.micro(PluginOpEntry { id, name, args }).await.ok(); } .boxed(), - HIGH, + NORMAL, ); } @@ -309,7 +212,7 @@ impl Scheduler { preload.rule(PreloadOpRule { id, plugin, targets }).await.ok(); } .boxed(), - HIGH, + NORMAL, ); } @@ -328,7 +231,7 @@ impl Scheduler { preload.size(PreloadOpSize { id, target, throttle }).await.ok(); } .boxed(), - HIGH, + NORMAL, ); } } @@ -348,40 +251,149 @@ impl Scheduler { } }; - let ct = CancellationToken::new(); + let (cancel_tx, cancel_rx) = mpsc::channel(1); let mut ongoing = self.ongoing.lock(); let id = ongoing.add(TaskKind::User, name); ongoing.hooks.insert(id, { - let ct = ct.clone(); let ongoing = self.ongoing.clone(); Box::new(move |canceled: bool| { async move { - ongoing.lock().try_remove(id, TaskStage::Hooked); if canceled { - ct.cancel(); + cancel_tx.send(()).await.ok(); + cancel_tx.closed().await; } if let Some(tx) = done { tx.send(()).ok(); } + ongoing.lock().try_remove(id, TaskStage::Hooked); } .boxed() }) }); + let cmd = OsString::from(&opener.run); let process = self.process.clone(); _ = self.micro.try_send( async move { if opener.block { - process.block(ProcessOpBlock { id, cmd: OsString::from(&opener.run), args }).await.ok(); + process.block(ProcessOpBlock { id, cmd, args }).await.ok(); } else if opener.orphan { - process.orphan(ProcessOpOrphan { id, cmd: OsString::from(&opener.run), args }).await.ok(); + process.orphan(ProcessOpOrphan { id, cmd, args }).await.ok(); } else { - process.bg(ProcessOpBg { id, cmd: OsString::from(&opener.run), args, ct }).await.ok(); + process.bg(ProcessOpBg { id, cmd, args, cancel: cancel_rx }).await.ok(); } } .boxed(), - HIGH, + NORMAL, ); } + + fn schedule_micro( + &self, + rx: async_priority_channel::Receiver, u8>, + ) -> JoinHandle<()> { + tokio::spawn(async move { + loop { + if let Ok((fut, _)) = rx.recv().await { + fut.await; + } + } + }) + } + + fn schedule_macro( + &self, + micro: async_priority_channel::Receiver, u8>, + macro_: async_priority_channel::Receiver, + ) -> JoinHandle<()> { + let file = self.file.clone(); + let plugin = self.plugin.clone(); + let preload = self.preload.clone(); + + let prog = self.prog.clone(); + let ongoing = self.ongoing.clone(); + + tokio::spawn(async move { + loop { + select! { + Ok((fut, _)) = micro.recv() => { + fut.await; + } + Ok((op, _)) = macro_.recv() => { + let id = op.id(); + if !ongoing.lock().exists(id) { + continue; + } + + 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(); + } + } + } + } + }) + } + + fn progress(&self, mut rx: UnboundedReceiver) -> JoinHandle<()> { + let micro = self.micro.clone(); + let ongoing = self.ongoing.clone(); + + tokio::spawn(async move { + while let Some(op) = rx.recv().await { + match op { + TaskProg::New(id, size) => { + if let Some(task) = ongoing.lock().get_mut(id) { + task.total += 1; + task.found += size; + } + } + TaskProg::Adv(id, succ, processed) => { + let mut ongoing = ongoing.lock(); + if let Some(task) = ongoing.get_mut(id) { + task.succ += succ; + task.processed += processed; + } + if succ > 0 { + if let Some(fut) = ongoing.try_remove(id, TaskStage::Pending) { + micro.try_send(fut, LOW).ok(); + } + } + } + TaskProg::Succ(id) => { + if let Some(fut) = ongoing.lock().try_remove(id, TaskStage::Dispatched) { + micro.try_send(fut, LOW).ok(); + } + } + TaskProg::Fail(id, reason) => { + if let Some(task) = ongoing.lock().get_mut(id) { + task.fail += 1; + task.logs.push_str(&reason); + task.logs.push('\n'); + + if let Some(logger) = &task.logger { + logger.send(reason).ok(); + } + } + } + TaskProg::Log(id, line) => { + if let Some(task) = ongoing.lock().get_mut(id) { + task.logs.push_str(&line); + task.logs.push('\n'); + + if let Some(logger) = &task.logger { + logger.send(line).ok(); + } + } + } + } + } + }) + } } diff --git a/yazi-shared/src/event/event.rs b/yazi-shared/src/event/event.rs index 19006883..808d7a0d 100644 --- a/yazi-shared/src/event/event.rs +++ b/yazi-shared/src/event/event.rs @@ -1,10 +1,10 @@ use std::{collections::VecDeque, ffi::OsString}; use crossterm::event::KeyEvent; -use tokio::sync::{mpsc, oneshot}; +use tokio::sync::mpsc; use super::Cmd; -use crate::{term::Term, Layer, RoCell}; +use crate::{Layer, RoCell}; static TX: RoCell> = RoCell::new(); @@ -31,12 +31,6 @@ impl Event { #[inline] pub fn emit(self) { TX.send(self).ok(); } - - #[inline] - pub async fn wait(self, rx: oneshot::Receiver) -> T { - TX.send(self).ok(); - rx.await.unwrap_or_else(|_| Term::goodbye(|| false)) - } } #[macro_export]