From 323b07642c27708e1996456f3ef55a898ea16408 Mon Sep 17 00:00:00 2001 From: sxyazi Date: Thu, 3 Aug 2023 07:23:07 +0800 Subject: [PATCH] .. --- 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 2703bddd..4756fd01 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) + } +}