This commit is contained in:
sxyazi 2023-08-03 18:33:05 +08:00
parent 43445e2443
commit 4adfa4a061
No known key found for this signature in database
11 changed files with 186 additions and 68 deletions

12
Cargo.lock generated
View file

@ -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",

View file

@ -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 {

View file

@ -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,
}

View file

@ -110,6 +110,7 @@ keymap = [
{ on = [ "<Up>" ], exec = "arrow -1" },
{ on = [ "<Down>" ], exec = "arrow 1" },
{ on = [ "<Enter>" ], exec = "inspect" },
{ on = [ "x" ], exec = "cancel" },
]

View file

@ -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"

View file

@ -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;

View file

@ -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<usize> {
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<BoxFuture<'static, ()>> {
if let Some(task) = self.get(id) {
if let Some(task) = self.get_mut(id) {
if stage > task.stage {
task.stage = stage;
}

View file

@ -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;
}

81
core/src/tasks/task.rs Normal file
View file

@ -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<mpsc::UnboundedSender<String>>,
}
#[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<TaskSummary> 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,
}

View file

@ -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<Scheduler>,
@ -100,13 +53,73 @@ impl Tasks {
old != self.cursor
}
pub fn paginate(&self) -> Vec<Task> {
pub fn paginate(&self) -> Vec<TaskSummary> {
let running = self.scheduler.running.read();
running.values().take(Self::limit()).cloned().collect::<Vec<_>>()
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;
}

View file

@ -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;
}
}