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":[]}