mroot/yazi
1
0
Fork 0
mirror of https://github.com/sxyazi/yazi.git synced 2026-07-21 23:01:05 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

feat: show the output of running tasks in real-time (#17)

Browse source
This commit is contained in:
三咲雅 · Misaki Masa 2023-08-03 18:55:40 +08:00 • committed by GitHub
parent d5d443031f
commit 85f09e0103
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
18 changed files with 415 additions and 289 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

2
README.md
View file

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

1
app/src/app.rs
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 {

2
app/src/executor.rs
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,
}

1
config/preset/keymap.toml
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" },
]

2
core/Cargo.toml
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"

8
core/src/manager/tab.rs
View file

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

13
core/src/tasks/mod.rs
View file

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

92
core/src/tasks/process.rs
View file

@ -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<ProcessOp>,
tx: async_channel::Sender<ProcessOp>,
sch: mpsc::UnboundedSender<TaskOp>,
}
#[derive(Debug)]
pub(super) enum ProcessOp {
Open(ProcessOpOpen),
}
#[derive(Debug)]
pub(super) struct ProcessOpOpen {
pub id: usize,
pub cmd: String,
pub args: Vec<OsString>,
pub block: bool,
pub cancel: oneshot::Sender<()>,
}
impl Process {
pub(super) fn new(sch: mpsc::UnboundedSender<TaskOp>) -> 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)
}
}

73
core/src/tasks/running.rs Normal file
View file

@ -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<usize, Box<dyn (FnOnce(bool) -> BoxFuture<'static, ()>) + Send + Sync>>,
pub(super) all: BTreeMap<usize, Task>,
}
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<usize> {
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<Item = &Task> { 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<BoxFuture<'static, ()>> {
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
}
}

134
core/src/tasks/scheduler.rs
View file

@ -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<usize, Box<dyn (FnOnce(bool) -> BoxFuture<'static, ()>) + Send + Sync>>,
all: BTreeMap<usize, Task>,
}
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<Item = &Task> { self.all.values() }
#[inline]
fn is_empty(&self) -> bool { self.all.is_empty() }
fn try_remove(&mut self, id: usize, stage: TaskStage) -> Option<BoxFuture<'static, ()>> {
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<File>,
precache: Arc<Precache>,
@ -112,7 +57,6 @@ impl Scheduler {
fn schedule_macro(&self, rx: Receiver<BoxFuture<'static, ()>>) {
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()
});
}

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

111
core/src/tasks/tasks.rs
View file

@ -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<Scheduler>,
@ -88,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;
}

48
core/src/tasks/file.rs → core/src/tasks/workers/file.rs
View file

@ -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<FileOp>,
tx: async_channel::Sender<FileOp>,
@ -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<TaskOp>) -> Self {
pub(crate) fn new(sch: mpsc::UnboundedSender<TaskOp>) -> 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::<PathBuf>());
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 {

7
core/src/tasks/workers/mod.rs Normal file
View file

@ -0,0 +1,7 @@
mod file;
mod precache;
mod process;
pub(super) use file::*;
pub(super) use precache::*;
pub(super) use process::*;

31
core/src/tasks/precache.rs → core/src/tasks/workers/precache.rs
View file

@ -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<PrecacheOp>,
tx: async_channel::Sender<PrecacheOp>,
sch: mpsc::UnboundedSender<TaskOp>,
pub(super) size_handing: Mutex<BTreeSet<PathBuf>>,
pub(crate) size_handing: Mutex<BTreeSet<PathBuf>>,
}
#[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<Throttle<(PathBuf, File)>>,
}
#[derive(Debug)]
pub(super) struct PrecacheOpMime {
pub(crate) struct PrecacheOpMime {
pub id: usize,
pub targets: Vec<PathBuf>,
}
#[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<TaskOp>) -> Self {
pub(crate) fn new(sch: mpsc::UnboundedSender<TaskOp>) -> 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<PathBuf>) -> Result<()> {
pub(crate) fn image(&self, id: usize, targets: Vec<PathBuf>) -> 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<PathBuf>) -> Result<()> {
pub(crate) fn video(&self, id: usize, targets: Vec<PathBuf>) -> Result<()> {
for target in targets {
self.sch.send(TaskOp::New(id, 0))?;
self.tx.send_blocking(PrecacheOp::Video(PrecacheOpVideo { id, target }))?;

84
core/src/tasks/workers/process.rs Normal file
View file

@ -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<TaskOp>,
}
#[derive(Debug)]
pub(crate) struct ProcessOpOpen {
pub id: usize,
pub cmd: String,
pub args: Vec<OsString>,
pub block: bool,
pub cancel: oneshot::Sender<()>,
}
impl Process {
pub(crate) fn new(sch: mpsc::UnboundedSender<TaskOp>) -> 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)
}
}

2
cspell.json
View file

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