From 4adfa4a0613ddd622b07f2be3a5d17a01e1f5aba Mon Sep 17 00:00:00 2001 From: sxyazi Date: Thu, 3 Aug 2023 18:33:05 +0800 Subject: [PATCH] .. --- Cargo.lock | 12 +-- app/src/app.rs | 1 + app/src/executor.rs | 2 + config/preset/keymap.toml | 1 + core/Cargo.toml | 2 +- core/src/tasks/mod.rs | 2 + core/src/tasks/running.rs | 12 ++- core/src/tasks/scheduler.rs | 10 ++- core/src/tasks/task.rs | 81 ++++++++++++++++++++ core/src/tasks/tasks.rs | 123 +++++++++++++++++------------- core/src/tasks/workers/process.rs | 8 +- 11 files changed, 186 insertions(+), 68 deletions(-) create mode 100644 core/src/tasks/task.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/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/tasks/mod.rs b/core/src/tasks/mod.rs index b991e0f5..b79f612a 100644 --- a/core/src/tasks/mod.rs +++ b/core/src/tasks/mod.rs @@ -1,10 +1,12 @@ mod running; mod scheduler; +mod task; mod tasks; mod workers; use running::*; use scheduler::*; +use task::*; pub use tasks::*; pub const TASKS_PADDING: u16 = 2; diff --git a/core/src/tasks/running.rs b/core/src/tasks/running.rs index c6bd010e..5b068bfd 100644 --- a/core/src/tasks/running.rs +++ b/core/src/tasks/running.rs @@ -21,7 +21,15 @@ impl Running { } #[inline] - pub(super) fn get(&mut self, id: usize) -> Option<&mut Task> { self.all.get_mut(&id) } + 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() } @@ -40,7 +48,7 @@ impl Running { id: usize, stage: TaskStage, ) -> Option> { - if let Some(task) = self.get(id) { + if let Some(task) = self.get_mut(id) { if stage > task.stage { task.stage = stage; } diff --git a/core/src/tasks/scheduler.rs b/core/src/tasks/scheduler.rs index 1f700f8f..54d2fc95 100644 --- a/core/src/tasks/scheduler.rs +++ b/core/src/tasks/scheduler.rs @@ -105,20 +105,24 @@ impl Scheduler { 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(id) { + 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; } 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 7dbc52f7..0fd10e6c 100644 --- a/core/src/tasks/tasks.rs +++ b/core/src/tasks/tasks.rs @@ -1,60 +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 logs: 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(), - logs: Default::default(), - found: 0, - processed: 0, - todo: 0, - done: 0, - } - } -} - -#[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, -} +use super::{task::TaskSummary, Scheduler, TASKS_PADDING, TASKS_PERCENT}; +use crate::{emit, files::{File, Files}, input::InputOpt, Position, BLOCKER}; pub struct Tasks { scheduler: Arc, @@ -100,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/workers/process.rs b/core/src/tasks/workers/process.rs index 93163886..f96e315a 100644 --- a/core/src/tasks/workers/process.rs +++ b/core/src/tasks/workers/process.rs @@ -66,7 +66,13 @@ impl Process { self.log(task.id, line)?; } Ok(status) = child.wait() => { - self.log(task.id, format!("Exited with {:?}", status))?; + 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; } }