From 32f6acd5f7c8f628b3e40a442e8cf3fdfe76a2e1 Mon Sep 17 00:00:00 2001 From: sxyazi Date: Thu, 3 Aug 2023 07:23:07 +0800 Subject: [PATCH 1/2] .. --- README.md | 2 +- core/src/manager/tab.rs | 8 +- core/src/tasks/mod.rs | 11 +-- core/src/tasks/process.rs | 92 ------------------ core/src/tasks/running.rs | 65 +++++++++++++ core/src/tasks/scheduler.rs | 113 +++++------------------ core/src/tasks/{ => workers}/file.rs | 29 +++--- core/src/tasks/workers/mod.rs | 7 ++ core/src/tasks/{ => workers}/precache.rs | 31 +++---- core/src/tasks/workers/process.rs | 62 +++++++++++++ 10 files changed, 196 insertions(+), 224 deletions(-) delete mode 100644 core/src/tasks/process.rs create mode 100644 core/src/tasks/running.rs rename core/src/tasks/{ => workers}/file.rs (92%) 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/README.md b/README.md index 755dcba7..67547d0b 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/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..b991e0f5 100644 --- a/core/src/tasks/mod.rs +++ b/core/src/tasks/mod.rs @@ -1,13 +1,10 @@ -mod file; -mod precache; -mod process; +mod running; mod scheduler; mod tasks; +mod workers; -use file::*; -pub use precache::*; -use process::*; -pub use scheduler::*; +use running::*; +use scheduler::*; 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..c6bd010e --- /dev/null +++ b/core/src/tasks/running.rs @@ -0,0 +1,65 @@ +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(&mut self, id: usize) -> Option<&mut Task> { self.all.get_mut(&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(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..32abdd16 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 task)) = file.recv() => { + if !running.read().exists(id) { + trace!("Skipping task {:?} as it was removed", task); + 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 task).await { + info!("Failed to work on task {:?}: {}", task, e); + } else { + trace!("Finished task {:?}", task); } - 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 task)) = precache.recv() => { + if !running.read().exists(id) { + trace!("Skipping task {:?} as it was removed", task); + 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 task).await { + info!("Failed to work on task {:?}: {}", task, e); + } else { + trace!("Finished task {:?}", task); } + } } } }); diff --git a/core/src/tasks/file.rs b/core/src/tasks/workers/file.rs similarity index 92% rename from core/src/tasks/file.rs rename to core/src/tasks/workers/file.rs index 1c5e6ce3..3c8a9017 100644 --- a/core/src/tasks/file.rs +++ b/core/src/tasks/workers/file.rs @@ -6,9 +6,9 @@ use shared::{calculate_size, copy_with_progress}; use tokio::{fs, io::{self, ErrorKind::{AlreadyExists, NotFound}}, sync::mpsc}; use tracing::{info, 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,7 +72,7 @@ impl File { }) } - pub(super) async fn work(&self, task: &mut FileOp) -> Result<()> { + pub(crate) async fn work(&self, task: &mut FileOp) -> Result<()> { match task { FileOp::Paste(task) => { match fs::remove_file(&task.to).await { @@ -156,9 +156,10 @@ impl File { Ok(()) } + #[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), @@ -231,7 +232,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 +269,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 +287,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..a976da2f --- /dev/null +++ b/core/src/tasks/workers/process.rs @@ -0,0 +1,62 @@ +use std::{ffi::OsString, process::Stdio}; + +use anyhow::Result; +use tokio::{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 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(()); + } + + let status = Command::new(&task.cmd) + .args(&task.args) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .kill_on_drop(true) + .status(); + + self.sch.send(TaskOp::New(task.id, 0))?; + select! { + _ = task.cancel.closed() => {}, + Ok(status) = status => { + trace!("{} exited with {:?}", task.cmd, status); + } + } + self.done(task.id) + } +} From ff1c26e54fbf129e6d3518360e516b7e5316796b Mon Sep 17 00:00:00 2001 From: sxyazi Date: Thu, 3 Aug 2023 08:14:50 +0800 Subject: [PATCH 2/2] .. --- core/src/tasks/scheduler.rs | 33 +++++++++++++++------------ core/src/tasks/tasks.rs | 14 +++++++++++- core/src/tasks/workers/file.rs | 21 +++++++++-------- core/src/tasks/workers/process.rs | 38 +++++++++++++++++++++---------- cspell.json | 2 +- 5 files changed, 71 insertions(+), 37 deletions(-) diff --git a/core/src/tasks/scheduler.rs b/core/src/tasks/scheduler.rs index 32abdd16..1f700f8f 100644 --- a/core/src/tasks/scheduler.rs +++ b/core/src/tasks/scheduler.rs @@ -70,26 +70,26 @@ impl Scheduler { Ok(fut) = rx.recv() => { fut.await; } - Ok((id, mut task)) = file.recv() => { + Ok((id, mut op)) = file.recv() => { if !running.read().exists(id) { - trace!("Skipping task {:?} as it was removed", task); + trace!("Skipping task {:?} as it was removed", op); continue; } - if let Err(e) = file.work(&mut task).await { - info!("Failed to work on task {:?}: {}", task, e); + if let Err(e) = file.work(&mut op).await { + info!("Failed to work on task {:?}: {}", op, e); } else { - trace!("Finished task {:?}", task); + trace!("Finished task {:?}", op); } } - Ok((id, mut task)) = precache.recv() => { + Ok((id, mut op)) = precache.recv() => { if !running.read().exists(id) { - trace!("Skipping task {:?} as it was removed", task); + trace!("Skipping task {:?} as it was removed", op); continue; } - if let Err(e) = precache.work(&mut task).await { - info!("Failed to work on task {:?}: {}", task, e); + if let Err(e) = precache.work(&mut op).await { + info!("Failed to work on task {:?}: {}", op, e); } else { - trace!("Finished task {:?}", task); + trace!("Finished task {:?}", op); } } } @@ -102,14 +102,20 @@ 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) { task.found += 1; task.todo += size; } } + TaskOp::Log(id, line) => { + if let Some(task) = running.write().get(id) { + task.logs.push_str(&line); + task.logs.push('\n'); + } + } TaskOp::Adv(id, processed, size) => { let mut running = running.write(); if let Some(task) = running.get(id) { @@ -303,7 +309,7 @@ impl Scheduler { }) }); - let _ = self.todo.send_blocking({ + tokio::spawn({ let process = self.process.clone(); let opener = opener.clone(); async move { @@ -312,7 +318,6 @@ impl Scheduler { .await .ok(); } - .boxed() }); } diff --git a/core/src/tasks/tasks.rs b/core/src/tasks/tasks.rs index 23ef5817..7dbc52f7 100644 --- a/core/src/tasks/tasks.rs +++ b/core/src/tasks/tasks.rs @@ -12,6 +12,7 @@ pub struct Task { pub id: usize, pub name: String, pub stage: TaskStage, + pub logs: String, pub found: u32, pub processed: u32, @@ -22,7 +23,16 @@ pub struct Task { impl Task { pub fn new(id: usize, name: String) -> Self { - Self { id, name, stage: Default::default(), found: 0, processed: 0, todo: 0, done: 0 } + Self { + id, + name, + stage: Default::default(), + logs: Default::default(), + found: 0, + processed: 0, + todo: 0, + done: 0, + } } } @@ -30,6 +40,8 @@ impl Task { 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 diff --git a/core/src/tasks/workers/file.rs b/core/src/tasks/workers/file.rs index 3c8a9017..d3aa5aa0 100644 --- a/core/src/tasks/workers/file.rs +++ b/core/src/tasks/workers/file.rs @@ -4,7 +4,7 @@ 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 crate::tasks::TaskOp; @@ -72,8 +72,8 @@ impl File { }) } - pub(crate) 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,6 +156,9 @@ 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))?) } @@ -189,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; } _ => {} @@ -198,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; } }; diff --git a/core/src/tasks/workers/process.rs b/core/src/tasks/workers/process.rs index a976da2f..93163886 100644 --- a/core/src/tasks/workers/process.rs +++ b/core/src/tasks/workers/process.rs @@ -1,7 +1,7 @@ use std::{ffi::OsString, process::Stdio}; use anyhow::Result; -use tokio::{process::Command, select, sync::{mpsc, oneshot}}; +use tokio::{io::{AsyncBufReadExt, BufReader}, process::Command, select, sync::{mpsc, oneshot}}; use tracing::trace; use crate::{emit, tasks::TaskOp, BLOCKER}; @@ -22,6 +22,9 @@ pub(crate) struct ProcessOpOpen { 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))?) } @@ -43,18 +46,29 @@ impl Process { return Ok(()); } - let status = Command::new(&task.cmd) - .args(&task.args) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .kill_on_drop(true) - .status(); - self.sch.send(TaskOp::New(task.id, 0))?; - select! { - _ = task.cancel.closed() => {}, - Ok(status) = status => { - trace!("{} exited with {:?}", task.cmd, status); + 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, format!("Exited with {:?}", status))?; + break; + } } } 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":[]}