From 85f09e0103240bd33efe5d7370ffaa1ccd07dfbc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=89=E5=92=B2=E9=9B=85=20=C2=B7=20Misaki=20Masa?= Date: Thu, 3 Aug 2023 18:55:40 +0800 Subject: [PATCH] feat: show the output of running tasks in real-time (#17) --- Cargo.lock | 12 +- README.md | 2 +- app/src/app.rs | 1 + app/src/executor.rs | 2 + config/preset/keymap.toml | 1 + core/Cargo.toml | 2 +- core/src/manager/tab.rs | 8 +- core/src/tasks/mod.rs | 13 +-- core/src/tasks/process.rs | 92 ---------------- core/src/tasks/running.rs | 73 ++++++++++++ core/src/tasks/scheduler.rs | 134 +++++++---------------- core/src/tasks/task.rs | 81 ++++++++++++++ core/src/tasks/tasks.rs | 111 +++++++++++-------- core/src/tasks/{ => workers}/file.rs | 48 ++++---- core/src/tasks/workers/mod.rs | 7 ++ core/src/tasks/{ => workers}/precache.rs | 31 +++--- core/src/tasks/workers/process.rs | 84 ++++++++++++++ cspell.json | 2 +- 18 files changed, 415 insertions(+), 289 deletions(-) delete mode 100644 core/src/tasks/process.rs create mode 100644 core/src/tasks/running.rs create mode 100644 core/src/tasks/task.rs rename core/src/tasks/{ => workers}/file.rs (85%) create mode 100644 core/src/tasks/workers/mod.rs rename core/src/tasks/{ => workers}/precache.rs (78%) create mode 100644 core/src/tasks/workers/process.rs diff --git a/Cargo.lock b/Cargo.lock index 98e737fd..69470a19 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -184,9 +184,9 @@ checksum = "df8670b8c7b9dae1793364eafadf7239c40d669904660c5960d74cfd80b46a53" [[package]] name = "cc" -version = "1.0.80" +version = "1.0.81" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "51f1226cd9da55587234753d1245dd5b132343ea240f26b6a9003d68706141ba" +checksum = "6c6b2562119bf28c3439f7f02db99faf0aa1a8cdfe5772a2ee155d32227239f0" dependencies = [ "libc", ] @@ -357,9 +357,9 @@ checksum = "7a81dae078cea95a014a339291cec439d2f232ebe854a9d672b796c6afafa9b7" [[package]] name = "deranged" -version = "0.3.6" +version = "0.3.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8810e7e2cf385b1e9b50d68264908ec367ba642c96d02edfe61c39e88e2a3c01" +checksum = "7684a49fb1af197853ef7b2ee694bc1f5b4179556f1e5710e1760c5db6f5e929" [[package]] name = "either" @@ -1449,9 +1449,9 @@ dependencies = [ [[package]] name = "time" -version = "0.3.24" +version = "0.3.25" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b79eabcd964882a646b3584543ccabeae7869e9ac32a46f6f22b7a5bd405308b" +checksum = "b0fdd63d58b18d663fbdf70e049f00a22c8e42be082203be7f26589213cd75ea" dependencies = [ "deranged", "itoa", diff --git a/README.md b/README.md index 2703bddd..4756fd01 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ ## Yazi - ⚡️ Blazing Fast Terminal File Manager -Yazi ("duck" in Chinese) is a terminal file manager written in Rust, based on non-blocking async I/O. It aims to provide an efficient, user-friendly, and configurable file management experience. +Yazi ("duck" in Chinese) is a terminal file manager written in Rust, based on non-blocking async I/O. It aims to provide an efficient, user-friendly, and customizable file management experience. https://github.com/sxyazi/yazi/assets/17523360/740a41f4-3d24-4287-952c-3aec51520a32 diff --git a/app/src/app.rs b/app/src/app.rs index 5ab54b91..cd0993d2 100644 --- a/app/src/app.rs +++ b/app/src/app.rs @@ -81,6 +81,7 @@ impl App { self.term = Some(Term::start().unwrap()); self.signals.stop_term(false); self.cx.manager.preview(self.cx.image_layer()); + emit!(Render); emit!(Hover); } if let Some(tx) = tx { diff --git a/app/src/executor.rs b/app/src/executor.rs index 607c34a5..40da2e69 100644 --- a/app/src/executor.rs +++ b/app/src/executor.rs @@ -168,6 +168,8 @@ impl Executor { if step > 0 { cx.tasks.next() } else { cx.tasks.prev() } } + "inspect" => cx.tasks.inspect(), + "cancel" => cx.tasks.cancel(), _ => false, } diff --git a/config/preset/keymap.toml b/config/preset/keymap.toml index 059f6ae4..34ced965 100644 --- a/config/preset/keymap.toml +++ b/config/preset/keymap.toml @@ -110,6 +110,7 @@ keymap = [ { on = [ "" ], exec = "arrow -1" }, { on = [ "" ], exec = "arrow 1" }, + { on = [ "" ], exec = "inspect" }, { on = [ "x" ], exec = "cancel" }, ] diff --git a/core/Cargo.toml b/core/Cargo.toml index ffb73f4d..fb767613 100644 --- a/core/Cargo.toml +++ b/core/Cargo.toml @@ -21,7 +21,7 @@ ratatui = "^0" serde = "^1" serde_json = "^1" syntect = "^5" -tokio = { version = "^1", features = [ "parking_lot", "macros", "rt-multi-thread", "sync", "time", "fs", "process", "io-util" ] } +tokio = { version = "^1", features = [ "parking_lot", "macros", "rt-multi-thread", "sync", "time", "fs", "process", "io-std", "io-util" ] } tracing = "^0" trash = "^3" unicode-width = "^0" diff --git a/core/src/manager/tab.rs b/core/src/manager/tab.rs index dbb2cfa6..59f14261 100644 --- a/core/src/manager/tab.rs +++ b/core/src/manager/tab.rs @@ -216,14 +216,14 @@ impl Tab { handle.abort(); } if self.current.in_search { + self.preview_reset_image(); + let cwd = self.current.cwd.clone(); let rep = self.history_new(&cwd); drop(mem::replace(&mut self.current, rep)); + emit!(Refresh); } - - self.preview_reset_image(); - emit!(Refresh); - true + false } pub fn jump(&self, global: bool) -> bool { diff --git a/core/src/tasks/mod.rs b/core/src/tasks/mod.rs index 82037064..b79f612a 100644 --- a/core/src/tasks/mod.rs +++ b/core/src/tasks/mod.rs @@ -1,13 +1,12 @@ -mod file; -mod precache; -mod process; +mod running; mod scheduler; +mod task; mod tasks; +mod workers; -use file::*; -pub use precache::*; -use process::*; -pub use scheduler::*; +use running::*; +use scheduler::*; +use task::*; pub use tasks::*; pub const TASKS_PADDING: u16 = 2; diff --git a/core/src/tasks/process.rs b/core/src/tasks/process.rs deleted file mode 100644 index e53f5c10..00000000 --- a/core/src/tasks/process.rs +++ /dev/null @@ -1,92 +0,0 @@ -use std::{ffi::OsString, process::Stdio}; - -use anyhow::Result; -use tokio::{process::Command, select, sync::{mpsc, oneshot}}; -use tracing::trace; - -use super::TaskOp; -use crate::{emit, BLOCKER}; - -pub(super) struct Process { - rx: async_channel::Receiver, - tx: async_channel::Sender, - - sch: mpsc::UnboundedSender, -} - -#[derive(Debug)] -pub(super) enum ProcessOp { - Open(ProcessOpOpen), -} - -#[derive(Debug)] -pub(super) struct ProcessOpOpen { - pub id: usize, - pub cmd: String, - pub args: Vec, - pub block: bool, - pub cancel: oneshot::Sender<()>, -} - -impl Process { - pub(super) fn new(sch: mpsc::UnboundedSender) -> Self { - let (tx, rx) = async_channel::unbounded(); - Self { tx, rx, sch } - } - - #[inline] - pub(super) async fn recv(&self) -> Result<(usize, ProcessOp)> { - Ok(match self.rx.recv().await? { - ProcessOp::Open(t) => (t.id, ProcessOp::Open(t)), - }) - } - - pub(super) async fn work(&self, task: &mut ProcessOp) -> Result<()> { - match task { - ProcessOp::Open(task) => { - trace!("Open task: {:?}", task); - if !task.block { - let status = Command::new(&task.cmd) - .args(&task.args) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .kill_on_drop(true) - .status(); - - select! { - _ = task.cancel.closed() => {}, - Ok(status) = status => { - trace!("{} exited with {:?}", task.cmd, status); - } - } - return Ok(self.sch.send(TaskOp::Adv(task.id, 1, 0))?); - } - - let _guard = BLOCKER.acquire().await.unwrap(); - emit!(Stop(true)).await; - - match Command::new(&task.cmd).args(&task.args).kill_on_drop(true).spawn() { - Ok(mut child) => { - child.wait().await.ok(); - } - Err(e) => { - trace!("Failed to spawn {}: {}", task.cmd, e); - } - } - - emit!(Stop(false)).await; - self.sch.send(TaskOp::Adv(task.id, 1, 0))?; - } - } - Ok(()) - } - - fn done(&self, id: usize) -> Result<()> { Ok(self.sch.send(TaskOp::Done(id))?) } - - pub(super) async fn open(&self, task: ProcessOpOpen) -> Result<()> { - let id = task.id; - self.sch.send(TaskOp::New(id, 0))?; - self.tx.send(ProcessOp::Open(task)).await?; - self.done(id) - } -} diff --git a/core/src/tasks/running.rs b/core/src/tasks/running.rs new file mode 100644 index 00000000..5b068bfd --- /dev/null +++ b/core/src/tasks/running.rs @@ -0,0 +1,73 @@ +use std::collections::BTreeMap; + +use futures::future::BoxFuture; + +use super::{Task, TaskStage}; + +#[derive(Default)] +pub(super) struct Running { + incr: usize, + + pub(super) hooks: + BTreeMap BoxFuture<'static, ()>) + Send + Sync>>, + pub(super) all: BTreeMap, +} + +impl Running { + pub(super) fn add(&mut self, name: String) -> usize { + self.incr += 1; + self.all.insert(self.incr, Task::new(self.incr, name)); + self.incr + } + + #[inline] + pub(super) fn get(&self, id: usize) -> Option<&Task> { self.all.get(&id) } + + #[inline] + pub(super) fn get_mut(&mut self, id: usize) -> Option<&mut Task> { self.all.get_mut(&id) } + + #[inline] + pub(super) fn get_id(&self, idx: usize) -> Option { + self.values().skip(idx).next().map(|t| t.id) + } + + #[inline] + pub(super) fn len(&self) -> usize { self.all.len() } + + #[inline] + pub(super) fn exists(&self, id: usize) -> bool { self.all.contains_key(&id) } + + #[inline] + pub(super) fn values(&self) -> impl Iterator { self.all.values() } + + #[inline] + pub(super) fn is_empty(&self) -> bool { self.all.is_empty() } + + pub(super) fn try_remove( + &mut self, + id: usize, + stage: TaskStage, + ) -> Option> { + if let Some(task) = self.get_mut(id) { + if stage > task.stage { + task.stage = stage; + } + + match task.stage { + TaskStage::Pending => return None, + TaskStage::Dispatched => { + if task.processed < task.found { + return None; + } + if let Some(hook) = self.hooks.remove(&id) { + return Some(hook(false)); + } + } + TaskStage::Hooked => {} + } + + self.all.remove(&id); + } + None + } +} diff --git a/core/src/tasks/scheduler.rs b/core/src/tasks/scheduler.rs index 7607748d..54d2fc95 100644 --- a/core/src/tasks/scheduler.rs +++ b/core/src/tasks/scheduler.rs @@ -1,4 +1,4 @@ -use std::{collections::BTreeMap, ffi::{OsStr, OsString}, path::PathBuf, sync::Arc, time::Duration}; +use std::{ffi::{OsStr, OsString}, path::PathBuf, sync::Arc, time::Duration}; use async_channel::{Receiver, Sender}; use config::open::Opener; @@ -8,64 +8,9 @@ use shared::{unique_path, Throttle}; use tokio::{fs, select, sync::{mpsc::{self, UnboundedReceiver}, oneshot}, time::sleep}; use tracing::{info, trace}; -use super::{File, FileOpDelete, FileOpPaste, FileOpTrash, Precache, PrecacheOpMime, PrecacheOpSize, Process, ProcessOpOpen, Task, TaskOp, TaskStage}; +use super::{workers::{File, FileOpDelete, FileOpPaste, FileOpTrash, Precache, PrecacheOpMime, PrecacheOpSize, Process, ProcessOpOpen}, Running, TaskOp, TaskStage}; use crate::emit; -#[derive(Default)] -pub(super) struct Running { - incr: usize, - - hooks: BTreeMap BoxFuture<'static, ()>) + Send + Sync>>, - all: BTreeMap, -} - -impl Running { - fn add(&mut self, name: String) -> usize { - self.incr += 1; - self.all.insert(self.incr, Task::new(self.incr, name)); - self.incr - } - - #[inline] - fn get(&mut self, id: usize) -> Option<&mut Task> { self.all.get_mut(&id) } - - #[inline] - pub(super) fn len(&self) -> usize { self.all.len() } - - #[inline] - fn exists(&self, id: usize) -> bool { self.all.contains_key(&id) } - - #[inline] - pub(super) fn values(&self) -> impl Iterator { self.all.values() } - - #[inline] - fn is_empty(&self) -> bool { self.all.is_empty() } - - fn try_remove(&mut self, id: usize, stage: TaskStage) -> Option> { - if let Some(task) = self.get(id) { - if stage > task.stage { - task.stage = stage; - } - - match task.stage { - TaskStage::Pending => return None, - TaskStage::Dispatched => { - if task.processed < task.found { - return None; - } - if let Some(hook) = self.hooks.remove(&id) { - return Some(hook(false)); - } - } - TaskStage::Hooked => {} - } - - self.all.remove(&id); - } - None - } -} - pub struct Scheduler { file: Arc, precache: Arc, @@ -112,7 +57,6 @@ impl Scheduler { fn schedule_macro(&self, rx: Receiver>) { let file = self.file.clone(); let precache = self.precache.clone(); - let process = self.process.clone(); let running = self.running.clone(); tokio::spawn(async move { @@ -123,42 +67,31 @@ impl Scheduler { } select! { - Ok(fut) = rx.recv() => { - fut.await; + Ok(fut) = rx.recv() => { + fut.await; + } + Ok((id, mut op)) = file.recv() => { + if !running.read().exists(id) { + trace!("Skipping task {:?} as it was removed", op); + continue; } - Ok((id, mut task)) = file.recv() => { - if !running.read().exists(id) { - trace!("Skipping task {:?} as it was removed", task); - continue; - } - if let Err(e) = file.work(&mut task).await { - info!("Failed to work on task {:?}: {}", task, e); - } else { - trace!("Finished task {:?}", task); - } + if let Err(e) = file.work(&mut op).await { + info!("Failed to work on task {:?}: {}", op, e); + } else { + trace!("Finished task {:?}", op); } - Ok((id, mut task)) = precache.recv() => { - if !running.read().exists(id) { - trace!("Skipping task {:?} as it was removed", task); - continue; - } - if let Err(e) = precache.work(&mut task).await { - info!("Failed to work on task {:?}: {}", task, e); - } else { - trace!("Finished task {:?}", task); - } + } + Ok((id, mut op)) = precache.recv() => { + if !running.read().exists(id) { + trace!("Skipping task {:?} as it was removed", op); + continue; } - Ok((id, mut task)) = process.recv() => { - if !running.read().exists(id) { - trace!("Skipping task {:?} as it was removed", task); - continue; - } - if let Err(e) = process.work(&mut task).await { - info!("Failed to work on task {:?}: {}", task, e); - } else { - trace!("Finished task {:?}", task); - } + if let Err(e) = precache.work(&mut op).await { + info!("Failed to work on task {:?}: {}", op, e); + } else { + trace!("Finished task {:?}", op); } + } } } }); @@ -169,17 +102,27 @@ impl Scheduler { let running = self.running.clone(); tokio::spawn(async move { - while let Some(task) = rx.recv().await { - match task { + while let Some(op) = rx.recv().await { + match op { TaskOp::New(id, size) => { - if let Some(task) = running.write().get(id) { + if let Some(task) = running.write().get_mut(id) { task.found += 1; task.todo += size; } } + TaskOp::Log(id, line) => { + if let Some(task) = running.write().get_mut(id) { + task.logs.push_str(&line); + task.logs.push('\n'); + + if let Some(logger) = &task.logger { + logger.send(line).ok(); + } + } + } TaskOp::Adv(id, processed, size) => { let mut running = running.write(); - if let Some(task) = running.get(id) { + if let Some(task) = running.get_mut(id) { task.processed += processed; task.done += size; } @@ -370,7 +313,7 @@ impl Scheduler { }) }); - let _ = self.todo.send_blocking({ + tokio::spawn({ let process = self.process.clone(); let opener = opener.clone(); async move { @@ -379,7 +322,6 @@ impl Scheduler { .await .ok(); } - .boxed() }); } diff --git a/core/src/tasks/task.rs b/core/src/tasks/task.rs new file mode 100644 index 00000000..fdee3349 --- /dev/null +++ b/core/src/tasks/task.rs @@ -0,0 +1,81 @@ +use tokio::sync::mpsc; + +#[derive(Debug)] +pub struct Task { + pub id: usize, + pub name: String, + pub stage: TaskStage, + + pub found: u32, + pub processed: u32, + + pub todo: u64, + pub done: u64, + + pub logs: String, + pub logger: Option>, +} + +#[derive(Debug)] +pub struct TaskSummary { + pub name: String, + + pub found: u32, + pub processed: u32, + + pub todo: u64, + pub done: u64, +} + +impl Task { + pub fn new(id: usize, name: String) -> Self { + Self { + id, + name, + stage: Default::default(), + + found: 0, + processed: 0, + + todo: 0, + done: 0, + + logs: Default::default(), + logger: Default::default(), + } + } +} + +impl Into for &Task { + fn into(self) -> TaskSummary { + TaskSummary { + name: self.name.clone(), + + found: self.found, + processed: self.processed, + + todo: self.todo, + done: self.done, + } + } +} + +#[derive(Debug)] +pub enum TaskOp { + // task_id, size + New(usize, u64), + // task_id, line + Log(usize, String), + // task_id, processed, size + Adv(usize, u32, u64), + // task_id + Done(usize), +} + +#[derive(Clone, Debug, Default, Eq, PartialEq, Ord, PartialOrd)] +pub enum TaskStage { + #[default] + Pending, + Dispatched, + Hooked, +} diff --git a/core/src/tasks/tasks.rs b/core/src/tasks/tasks.rs index 23ef5817..0fd10e6c 100644 --- a/core/src/tasks/tasks.rs +++ b/core/src/tasks/tasks.rs @@ -1,48 +1,13 @@ -use std::{collections::{BTreeMap, HashMap, HashSet}, ffi::OsStr, path::{Path, PathBuf}, sync::Arc}; +use std::{collections::{BTreeMap, HashMap, HashSet}, ffi::OsStr, io::{stdout, Write}, path::{Path, PathBuf}, sync::Arc}; use config::{manager::SortBy, open::Opener, OPEN}; +use crossterm::terminal::{disable_raw_mode, enable_raw_mode}; use shared::{tty_size, MimeKind}; +use tokio::{io::AsyncReadExt, select, sync::mpsc, time}; use tracing::trace; -use super::{Scheduler, TASKS_PADDING, TASKS_PERCENT}; -use crate::{emit, files::{File, Files}, input::InputOpt, Position}; - -#[derive(Clone, Debug)] -pub struct Task { - pub id: usize, - pub name: String, - pub stage: TaskStage, - - pub found: u32, - pub processed: u32, - - pub todo: u64, - pub done: u64, -} - -impl Task { - pub fn new(id: usize, name: String) -> Self { - Self { id, name, stage: Default::default(), found: 0, processed: 0, todo: 0, done: 0 } - } -} - -#[derive(Debug)] -pub enum TaskOp { - // task_id, size - New(usize, u64), - // task_id, processed, size - Adv(usize, u32, u64), - // task_id - Done(usize), -} - -#[derive(Clone, Debug, Default, Eq, PartialEq, Ord, PartialOrd)] -pub enum TaskStage { - #[default] - Pending, - Dispatched, - Hooked, -} +use super::{task::TaskSummary, Scheduler, TASKS_PADDING, TASKS_PERCENT}; +use crate::{emit, files::{File, Files}, input::InputOpt, Position, BLOCKER}; pub struct Tasks { scheduler: Arc, @@ -88,13 +53,73 @@ impl Tasks { old != self.cursor } - pub fn paginate(&self) -> Vec { + pub fn paginate(&self) -> Vec { let running = self.scheduler.running.read(); - running.values().take(Self::limit()).cloned().collect::>() + running.values().take(Self::limit()).map(|t| t.into()).collect() + } + + pub fn inspect(&self) -> bool { + let id = if let Some(id) = self.scheduler.running.read().get_id(self.cursor) { + id + } else { + return false; + }; + + let scheduler = self.scheduler.clone(); + tokio::spawn(async move { + let _guard = BLOCKER.acquire().await.unwrap(); + let (tx, mut rx) = mpsc::unbounded_channel(); + + let buffered = { + let mut running = scheduler.running.write(); + let task = if let Some(task) = running.get_mut(id) { task } else { return }; + + task.logger = Some(tx); + task.logs.clone() + }; + + emit!(Stop(true)).await; + stdout().write_all("\n".repeat(tty_size().ws_row as usize).as_bytes()).ok(); + stdout().write_all(buffered.as_bytes()).ok(); + enable_raw_mode().ok(); + + let mut stdin = tokio::io::stdin(); + let mut quit = [0; 1]; + loop { + select! { + Some(line) = rx.recv() => { + stdout().write_all(line.as_bytes()).ok(); + stdout().write_all(b"\r\n").ok(); + } + _ = time::sleep(time::Duration::from_millis(100)) => { + if scheduler.running.read().get(id).is_none() { + stdout().write_all(b"Task finished, press `q` to quit\r\n").ok(); + break; + } + }, + Ok(_) = stdin.read(&mut quit) => { + if quit[0] == b'q' { + break; + } + } + } + } + + if let Some(task) = scheduler.running.write().get_mut(id) { + task.logger = None; + } + while quit[0] != b'q' { + stdin.read(&mut quit).await.ok(); + } + + disable_raw_mode().ok(); + emit!(Stop(false)).await; + }); + false } pub fn cancel(&mut self) -> bool { - let id = self.scheduler.running.read().values().skip(self.cursor).next().map(|t| t.id); + let id = self.scheduler.running.read().get_id(self.cursor); if !id.map(|id| self.scheduler.cancel(id)).unwrap_or(false) { return false; } diff --git a/core/src/tasks/file.rs b/core/src/tasks/workers/file.rs similarity index 85% rename from core/src/tasks/file.rs rename to core/src/tasks/workers/file.rs index 1c5e6ce3..d3aa5aa0 100644 --- a/core/src/tasks/file.rs +++ b/core/src/tasks/workers/file.rs @@ -4,11 +4,11 @@ use anyhow::Result; use futures::{future::BoxFuture, FutureExt}; use shared::{calculate_size, copy_with_progress}; use tokio::{fs, io::{self, ErrorKind::{AlreadyExists, NotFound}}, sync::mpsc}; -use tracing::{info, trace}; +use tracing::trace; -use super::TaskOp; +use crate::tasks::TaskOp; -pub(super) struct File { +pub(crate) struct File { rx: async_channel::Receiver, tx: async_channel::Sender, @@ -16,7 +16,7 @@ pub(super) struct File { } #[derive(Debug)] -pub(super) enum FileOp { +pub(crate) enum FileOp { Paste(FileOpPaste), Link(FileOpLink), Delete(FileOpDelete), @@ -24,7 +24,7 @@ pub(super) enum FileOp { } #[derive(Clone, Debug)] -pub(super) struct FileOpPaste { +pub(crate) struct FileOpPaste { pub id: usize, pub from: PathBuf, pub to: PathBuf, @@ -34,7 +34,7 @@ pub(super) struct FileOpPaste { } #[derive(Clone, Debug)] -pub(super) struct FileOpLink { +pub(crate) struct FileOpLink { pub id: usize, pub from: PathBuf, pub to: PathBuf, @@ -43,27 +43,27 @@ pub(super) struct FileOpLink { } #[derive(Clone, Debug)] -pub(super) struct FileOpDelete { +pub(crate) struct FileOpDelete { pub id: usize, pub target: PathBuf, pub length: u64, } #[derive(Clone, Debug)] -pub(super) struct FileOpTrash { +pub(crate) struct FileOpTrash { pub id: usize, pub target: PathBuf, pub length: u64, } impl File { - pub(super) fn new(sch: mpsc::UnboundedSender) -> Self { + pub(crate) fn new(sch: mpsc::UnboundedSender) -> Self { let (tx, rx) = async_channel::unbounded(); Self { tx, rx, sch } } #[inline] - pub(super) async fn recv(&self) -> Result<(usize, FileOp)> { + pub(crate) 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)), @@ -72,8 +72,8 @@ impl File { }) } - pub(super) async fn work(&self, task: &mut FileOp) -> Result<()> { - match task { + pub(crate) async fn work(&self, op: &mut FileOp) -> Result<()> { + match op { FileOp::Paste(task) => { match fs::remove_file(&task.to).await { Err(e) if e.kind() != NotFound => Err(e)?, @@ -90,7 +90,7 @@ impl File { break; } Ok(n) => { - trace!("Paste task advanced {}: {:?}", n, task); + self.log(task.id, format!("Paste task advanced {}: {:?}", n, task))?; self.sch.send(TaskOp::Adv(task.id, 0, n))? } Err(e) if e.kind() == NotFound => { @@ -100,7 +100,7 @@ impl File { // Operation not permitted (os error 1) // Attribute not found (os error 93) Err(e) if task.retry < 3 && matches!(e.raw_os_error(), Some(1) | Some(93)) => { - trace!("Paste task retry: {:?}", task); + self.log(task.id, format!("Paste task retry: {:?}", task))?; task.retry += 1; return Ok(self.tx.send(FileOp::Paste(task.clone())).await?); } @@ -113,7 +113,7 @@ impl File { let src = match fs::read_link(&task.from).await { Ok(src) => src, Err(e) if e.kind() == NotFound => { - trace!("Link task partially done: {:?}", task); + self.log(task.id, format!("Link task partially done: {:?}", task))?; return Ok(self.sch.send(TaskOp::Adv(task.id, 1, task.length))?); } Err(e) => Err(e)?, @@ -132,7 +132,7 @@ impl File { FileOp::Delete(task) => { if let Err(e) = fs::remove_file(&task.target).await { if e.kind() != NotFound && fs::symlink_metadata(&task.target).await.is_ok() { - info!("Delete task failed: {:?}, {}", task, e); + self.log(task.id, format!("Delete task failed: {:?}, {}", task, e))?; Err(e)? } } @@ -156,9 +156,13 @@ impl File { Ok(()) } + #[inline] + fn log(&self, id: usize, line: String) -> Result<()> { Ok(self.sch.send(TaskOp::Log(id, line))?) } + + #[inline] fn done(&self, id: usize) -> Result<()> { Ok(self.sch.send(TaskOp::Done(id))?) } - pub(super) async fn paste(&self, mut task: FileOpPaste) -> Result<()> { + pub(crate) async fn paste(&self, mut task: FileOpPaste) -> Result<()> { if task.cut { match fs::rename(&task.from, &task.to).await { Ok(_) => return self.done(task.id), @@ -188,7 +192,7 @@ impl File { let dest = root.join(src.components().skip(skip).collect::()); match fs::create_dir(&dest).await { Err(e) if e.kind() != AlreadyExists => { - info!("Create dir failed: {:?}, {}", dest, e); + self.log(task.id, format!("Create dir failed: {:?}, {}", dest, e))?; continue; } _ => {} @@ -197,7 +201,7 @@ impl File { let mut it = match fs::read_dir(&src).await { Ok(it) => it, Err(e) => { - info!("Read dir failed: {:?}, {}", src, e); + self.log(task.id, format!("Read dir failed: {:?}, {}", src, e))?; continue; } }; @@ -231,7 +235,7 @@ impl File { self.done(task.id) } - pub(super) async fn delete(&self, mut task: FileOpDelete) -> Result<()> { + pub(crate) async fn delete(&self, mut task: FileOpDelete) -> Result<()> { let meta = fs::symlink_metadata(&task.target).await?; if !meta.is_dir() { let id = task.id; @@ -268,7 +272,7 @@ impl File { self.done(task.id) } - pub(super) async fn trash(&self, mut task: FileOpTrash) -> Result<()> { + pub(crate) async fn trash(&self, mut task: FileOpTrash) -> Result<()> { let id = task.id; task.length = calculate_size(&task.target).await; @@ -286,7 +290,7 @@ impl File { if meta.is_ok() { meta } else { fs::symlink_metadata(path).await } } - pub(super) fn remove_empty_dirs(dir: &Path) -> BoxFuture<()> { + pub(crate) fn remove_empty_dirs(dir: &Path) -> BoxFuture<()> { trace!("Remove empty dirs: {:?}", dir); async move { let mut it = match fs::read_dir(dir).await { diff --git a/core/src/tasks/workers/mod.rs b/core/src/tasks/workers/mod.rs new file mode 100644 index 00000000..3c76a534 --- /dev/null +++ b/core/src/tasks/workers/mod.rs @@ -0,0 +1,7 @@ +mod file; +mod precache; +mod process; + +pub(super) use file::*; +pub(super) use precache::*; +pub(super) use process::*; diff --git a/core/src/tasks/precache.rs b/core/src/tasks/workers/precache.rs similarity index 78% rename from core/src/tasks/precache.rs rename to core/src/tasks/workers/precache.rs index 60340779..057666bb 100644 --- a/core/src/tasks/precache.rs +++ b/core/src/tasks/workers/precache.rs @@ -6,64 +6,63 @@ use parking_lot::Mutex; use shared::{calculate_size, Throttle}; use tokio::{fs, sync::mpsc}; -use super::TaskOp; -use crate::{emit, external, files::{File, FilesOp}}; +use crate::{emit, external, files::{File, FilesOp}, tasks::TaskOp}; -pub struct Precache { +pub(crate) struct Precache { rx: async_channel::Receiver, tx: async_channel::Sender, sch: mpsc::UnboundedSender, - pub(super) size_handing: Mutex>, + pub(crate) size_handing: Mutex>, } #[derive(Debug)] -pub(super) enum PrecacheOp { +pub(crate) enum PrecacheOp { Image(PrecacheOpImage), Video(PrecacheOpVideo), } #[derive(Debug)] -pub(super) struct PrecacheOpSize { +pub(crate) struct PrecacheOpSize { pub id: usize, pub target: PathBuf, pub throttle: Arc>, } #[derive(Debug)] -pub(super) struct PrecacheOpMime { +pub(crate) struct PrecacheOpMime { pub id: usize, pub targets: Vec, } #[derive(Debug)] -pub(super) struct PrecacheOpImage { +pub(crate) struct PrecacheOpImage { pub id: usize, pub target: PathBuf, } #[derive(Debug)] -pub(super) struct PrecacheOpVideo { +pub(crate) struct PrecacheOpVideo { pub id: usize, pub target: PathBuf, } impl Precache { - pub(super) fn new(sch: mpsc::UnboundedSender) -> Self { + pub(crate) fn new(sch: mpsc::UnboundedSender) -> Self { let (tx, rx) = async_channel::unbounded(); Self { tx, rx, sch, size_handing: Default::default() } } #[inline] - pub(super) async fn recv(&self) -> Result<(usize, PrecacheOp)> { + pub(crate) async fn recv(&self) -> Result<(usize, PrecacheOp)> { Ok(match self.rx.recv().await? { PrecacheOp::Image(t) => (t.id, PrecacheOp::Image(t)), PrecacheOp::Video(t) => (t.id, PrecacheOp::Video(t)), }) } - pub(super) async fn work(&self, task: &mut PrecacheOp) -> Result<()> { + pub(crate) async fn work(&self, task: &mut PrecacheOp) -> Result<()> { match task { PrecacheOp::Image(task) => { Image::precache(&task.target).await.ok(); @@ -85,7 +84,7 @@ impl Precache { #[inline] fn done(&self, id: usize) -> Result<()> { Ok(self.sch.send(TaskOp::Done(id))?) } - pub(super) async fn mime(&self, task: PrecacheOpMime) -> Result<()> { + pub(crate) async fn mime(&self, task: PrecacheOpMime) -> Result<()> { self.sch.send(TaskOp::New(task.id, 0))?; if let Ok(mimes) = external::file(&task.targets).await { emit!(Mimetype(mimes)); @@ -95,7 +94,7 @@ impl Precache { self.done(task.id) } - pub(super) async fn size(&self, task: PrecacheOpSize) -> Result<()> { + pub(crate) async fn size(&self, task: PrecacheOpSize) -> Result<()> { self.sch.send(TaskOp::New(task.id, 0))?; let length = Some(calculate_size(&task.target).await); @@ -118,7 +117,7 @@ impl Precache { self.done(task.id) } - pub(super) fn image(&self, id: usize, targets: Vec) -> Result<()> { + pub(crate) fn image(&self, id: usize, targets: Vec) -> Result<()> { for target in targets { self.sch.send(TaskOp::New(id, 0))?; self.tx.send_blocking(PrecacheOp::Image(PrecacheOpImage { id, target }))?; @@ -126,7 +125,7 @@ impl Precache { self.done(id) } - pub(super) fn video(&self, id: usize, targets: Vec) -> Result<()> { + pub(crate) fn video(&self, id: usize, targets: Vec) -> Result<()> { for target in targets { self.sch.send(TaskOp::New(id, 0))?; self.tx.send_blocking(PrecacheOp::Video(PrecacheOpVideo { id, target }))?; diff --git a/core/src/tasks/workers/process.rs b/core/src/tasks/workers/process.rs new file mode 100644 index 00000000..cde99ba5 --- /dev/null +++ b/core/src/tasks/workers/process.rs @@ -0,0 +1,84 @@ +use std::{ffi::OsString, process::Stdio}; + +use anyhow::Result; +use tokio::{io::{AsyncBufReadExt, BufReader}, process::Command, select, sync::{mpsc, oneshot}}; +use tracing::trace; + +use crate::{emit, tasks::TaskOp, BLOCKER}; + +pub(crate) struct Process { + sch: mpsc::UnboundedSender, +} + +#[derive(Debug)] +pub(crate) struct ProcessOpOpen { + pub id: usize, + pub cmd: String, + pub args: Vec, + pub block: bool, + pub cancel: oneshot::Sender<()>, +} + +impl Process { + pub(crate) fn new(sch: mpsc::UnboundedSender) -> Self { Self { sch } } + + #[inline] + fn log(&self, id: usize, line: String) -> Result<()> { Ok(self.sch.send(TaskOp::Log(id, line))?) } + + #[inline] + fn done(&self, id: usize) -> Result<()> { Ok(self.sch.send(TaskOp::Done(id))?) } + + pub(crate) async fn open(&self, mut task: ProcessOpOpen) -> Result<()> { + if task.block { + let _guard = BLOCKER.acquire().await.unwrap(); + emit!(Stop(true)).await; + + match Command::new(&task.cmd).args(&task.args).kill_on_drop(true).spawn() { + Ok(mut child) => { + child.wait().await.ok(); + } + Err(e) => { + trace!("Failed to spawn {}: {}", task.cmd, e); + } + } + + emit!(Stop(false)).await; + return Ok(()); + } + + self.sch.send(TaskOp::New(task.id, 0))?; + let mut child = Command::new(&task.cmd) + .args(&task.args) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true) + .spawn()?; + + let mut stdout = BufReader::new(child.stdout.take().unwrap()).lines(); + let mut stderr = BufReader::new(child.stderr.take().unwrap()).lines(); + loop { + select! { + _ = task.cancel.closed() => break, + Ok(Some(line)) = stdout.next_line() => { + self.log(task.id, line)?; + } + Ok(Some(line)) = stderr.next_line() => { + self.log(task.id, line)?; + } + Ok(status) = child.wait() => { + self.log(task.id, match status.code() { + Some(code) => format!("Exited with status code: {code}"), + None => "Process terminated by signal".to_string(), + })?; + if !status.success() { + return Ok(()); + } + break; + } + } + } + + self.sch.send(TaskOp::Adv(task.id, 1, 0))?; + self.done(task.id) + } +} diff --git a/cspell.json b/cspell.json index b96f1a26..48fee7e4 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"],"version":"0.2"} +{"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"],"version":"0.2","language":"en","flagWords":[]}