This commit is contained in:
sxyazi 2023-08-03 07:23:07 +08:00
parent d5d443031f
commit 323b07642c
No known key found for this signature in database
10 changed files with 196 additions and 224 deletions

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

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 {

View file

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

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

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

@ -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<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(&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<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(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
}
}

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

View file

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

View file

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

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 }))?;

View file

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