From 0233a0e952efb0a6223cc9b4715872ecc9765f6e Mon Sep 17 00:00:00 2001 From: sxyazi Date: Tue, 12 Mar 2024 20:52:09 +0800 Subject: [PATCH] .. --- Cargo.lock | 1 + yazi-core/src/manager/commands/bulk_rename.rs | 2 +- yazi-core/src/manager/commands/open.rs | 2 +- yazi-core/src/tasks/commands/open_with.rs | 5 +- yazi-core/src/tasks/commands/process_exec.rs | 3 +- yazi-core/src/tasks/process.rs | 29 ++++--- yazi-core/src/tasks/tasks.rs | 4 +- yazi-fm/src/executor.rs | 1 + yazi-proxy/src/options/process.rs | 8 +- yazi-scheduler/Cargo.toml | 1 + yazi-scheduler/src/process/op.rs | 46 ++++++++--- yazi-scheduler/src/process/process.rs | 80 +++++++++---------- yazi-scheduler/src/process/shell.rs | 5 -- yazi-scheduler/src/scheduler.rs | 55 +++++++------ 14 files changed, 139 insertions(+), 103 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 1f8328ef..ec370fff 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2863,6 +2863,7 @@ dependencies = [ "regex", "tokio", "tokio-stream", + "tokio-util", "tracing", "trash", "yazi-adaptor", diff --git a/yazi-core/src/manager/commands/bulk_rename.rs b/yazi-core/src/manager/commands/bulk_rename.rs index fcea9fb7..eb7778af 100644 --- a/yazi-core/src/manager/commands/bulk_rename.rs +++ b/yazi-core/src/manager/commands/bulk_rename.rs @@ -1,4 +1,4 @@ -use std::{collections::HashMap, ffi::{OsStr, OsString}, io::{stdout, BufWriter, Write}, path::PathBuf}; +use std::{collections::HashMap, ffi::OsStr, io::{stdout, BufWriter, Write}, path::PathBuf}; use anyhow::{anyhow, Result}; use tokio::{fs::{self, OpenOptions}, io::{stdin, AsyncReadExt, AsyncWriteExt}}; diff --git a/yazi-core/src/manager/commands/open.rs b/yazi-core/src/manager/commands/open.rs index 8c1eaca5..24852522 100644 --- a/yazi-core/src/manager/commands/open.rs +++ b/yazi-core/src/manager/commands/open.rs @@ -84,7 +84,7 @@ impl Manager { if targets.is_empty() { return; } else if !opt.interactive { - return tasks.process_from_files(&opt.hovered, &targets); + return tasks.process_from_files(opt.hovered, targets); } let openers: Vec<_> = OPEN.common_openers(&targets).into_iter().cloned().collect(); diff --git a/yazi-core/src/tasks/commands/open_with.rs b/yazi-core/src/tasks/commands/open_with.rs index aaff5b3f..4712740f 100644 --- a/yazi-core/src/tasks/commands/open_with.rs +++ b/yazi-core/src/tasks/commands/open_with.rs @@ -5,7 +5,10 @@ use crate::tasks::Tasks; impl Tasks { pub fn open_with(&mut self, opt: impl TryInto) { if let Ok(opt) = opt.try_into() { - self.process_from_opener(&opt.opener, &opt.targets); + self.process_from_opener( + opt.opener, + opt.targets.into_iter().map(|u| u.into_os_string()).collect(), + ); } } } diff --git a/yazi-core/src/tasks/commands/process_exec.rs b/yazi-core/src/tasks/commands/process_exec.rs index 94676e6a..e53ba8ce 100644 --- a/yazi-core/src/tasks/commands/process_exec.rs +++ b/yazi-core/src/tasks/commands/process_exec.rs @@ -5,8 +5,7 @@ use crate::tasks::Tasks; impl Tasks { pub fn process_exec(&mut self, opt: impl TryInto) { if let Ok(opt) = opt.try_into() { - // FIXME - // self.process_from_opener(&opt.opener, &opt.targets); + self.scheduler.process_open(opt.opener, opt.args, Some(opt.done)); } } } diff --git a/yazi-core/src/tasks/process.rs b/yazi-core/src/tasks/process.rs index f3b65eac..b8b98cf3 100644 --- a/yazi-core/src/tasks/process.rs +++ b/yazi-core/src/tasks/process.rs @@ -1,4 +1,4 @@ -use std::{collections::HashMap, ffi::OsStr}; +use std::{collections::HashMap, ffi::OsString, mem}; use yazi_config::{open::Opener, OPEN}; use yazi_shared::fs::Url; @@ -6,25 +6,36 @@ use yazi_shared::fs::Url; use super::Tasks; impl Tasks { - pub fn process_from_files(&self, hovered: &Url, targets: &[(Url, String)]) { + pub fn process_from_files(&self, hovered: Url, targets: Vec<(Url, String)>) { let mut openers = HashMap::new(); for (url, mime) in targets { - if let Some(opener) = OPEN.openers(url, mime).and_then(|o| o.first().copied()) { - openers.entry(opener).or_insert_with(|| vec![hovered]).push(url); + if let Some(opener) = OPEN.openers(&url, mime).and_then(|o| o.first().copied()) { + openers.entry(opener).or_insert_with(|| vec![hovered.clone()]).push(url); } } for (opener, args) in openers { - self.process_from_opener(opener, &args); + self.process_from_opener( + opener.clone(), + args.into_iter().map(|u| u.into_os_string()).collect(), + ); } } - pub fn process_from_opener(&self, opener: &Opener, args: &[impl AsRef]) { + pub fn process_from_opener(&self, opener: Opener, mut args: Vec) { if opener.spread { - self.scheduler.process_open(opener, args); + self.scheduler.process_open(opener, args, None); return; } - for target in args.iter().skip(1) { - self.scheduler.process_open(opener, &[&args[0], target]); + if args.is_empty() { + return; + } + if args.len() == 2 { + self.scheduler.process_open(opener, args, None); + return; + } + let hovered = mem::take(&mut args[0]); + for target in args.into_iter().skip(1) { + self.scheduler.process_open(opener.clone(), vec![hovered.clone(), target], None); } } } diff --git a/yazi-core/src/tasks/tasks.rs b/yazi-core/src/tasks/tasks.rs index f555a877..d84138cd 100644 --- a/yazi-core/src/tasks/tasks.rs +++ b/yazi-core/src/tasks/tasks.rs @@ -1,8 +1,8 @@ -use std::{collections::{HashMap, HashSet}, ffi::OsStr, mem, sync::Arc, time::Duration}; +use std::{collections::{HashMap, HashSet}, mem, sync::Arc, time::Duration}; use tokio::time::sleep; use tracing::debug; -use yazi_config::{manager::SortBy, open::Opener, plugin::{PluginRule, MAX_PRELOADERS}, OPEN, PLUGIN}; +use yazi_config::{manager::SortBy, plugin::{PluginRule, MAX_PRELOADERS}, PLUGIN}; use yazi_plugin::ValueSendable; use yazi_scheduler::{Scheduler, TaskSummary}; use yazi_shared::{emit, event::Cmd, fs::{File, Url}, term::Term, Layer, MIME_DIR}; diff --git a/yazi-fm/src/executor.rs b/yazi-fm/src/executor.rs index e3163577..97eb60f1 100644 --- a/yazi-fm/src/executor.rs +++ b/yazi-fm/src/executor.rs @@ -156,6 +156,7 @@ impl<'a> Executor<'a> { on!(inspect); on!(cancel); on!(open_with); + on!(process_exec); #[allow(clippy::single_match)] match cmd.name.as_str() { diff --git a/yazi-proxy/src/options/process.rs b/yazi-proxy/src/options/process.rs index 9dbf8e30..5f77763f 100644 --- a/yazi-proxy/src/options/process.rs +++ b/yazi-proxy/src/options/process.rs @@ -1,14 +1,14 @@ use std::ffi::OsString; +use tokio::sync::oneshot; +use yazi_config::open::Opener; use yazi_shared::event::Cmd; // --- Exec -#[derive(Default)] pub struct ProcessExecOpt { - pub cmd: OsString, + pub opener: Opener, pub args: Vec, - pub block: bool, - pub orphan: bool, + pub done: oneshot::Sender<()>, } impl TryFrom for ProcessExecOpt { diff --git a/yazi-scheduler/Cargo.toml b/yazi-scheduler/Cargo.toml index 51b252f3..ed4c29db 100644 --- a/yazi-scheduler/Cargo.toml +++ b/yazi-scheduler/Cargo.toml @@ -25,6 +25,7 @@ parking_lot = "^0" regex = "^1" tokio = { version = "^1", features = [ "parking_lot", "rt-multi-thread" ] } tokio-stream = "^0" +tokio-util = "^0" # Logging tracing = { version = "^0", features = [ "max_level_debug", "release_max_level_warn" ] } diff --git a/yazi-scheduler/src/process/op.rs b/yazi-scheduler/src/process/op.rs index 374f9571..1981522c 100644 --- a/yazi-scheduler/src/process/op.rs +++ b/yazi-scheduler/src/process/op.rs @@ -1,21 +1,45 @@ use std::ffi::OsString; -use tokio::sync::oneshot; +use tokio_util::sync::CancellationToken; use super::ShellOpt; #[derive(Debug)] -pub struct ProcessOpOpen { - pub id: usize, - pub cmd: OsString, - pub args: Vec, - pub block: bool, - pub orphan: bool, - pub cancel: oneshot::Sender<()>, +pub struct ProcessOpOrphan { + pub id: usize, + pub cmd: OsString, + pub args: Vec, } -impl From for ShellOpt { - fn from(op: ProcessOpOpen) -> Self { - Self { cmd: op.cmd, args: op.args, piped: false, orphan: op.orphan } +impl From for ShellOpt { + fn from(op: ProcessOpOrphan) -> Self { + Self { cmd: op.cmd, args: op.args, piped: false, orphan: true } + } +} + +#[derive(Debug)] +pub struct ProcessOpBlock { + pub id: usize, + pub cmd: OsString, + pub args: Vec, +} + +impl From for ShellOpt { + fn from(op: ProcessOpBlock) -> Self { + Self { cmd: op.cmd, args: op.args, piped: false, orphan: false } + } +} + +#[derive(Debug)] +pub struct ProcessOpBg { + pub id: usize, + pub cmd: OsString, + pub args: Vec, + pub ct: CancellationToken, +} + +impl From for ShellOpt { + fn from(op: ProcessOpBg) -> Self { + Self { cmd: op.cmd, args: op.args, piped: true, orphan: false } } } diff --git a/yazi-scheduler/src/process/process.rs b/yazi-scheduler/src/process/process.rs index 587fe497..10da2bad 100644 --- a/yazi-scheduler/src/process/process.rs +++ b/yazi-scheduler/src/process/process.rs @@ -3,7 +3,7 @@ use tokio::{io::{AsyncBufReadExt, BufReader}, select, sync::mpsc}; use yazi_proxy::{AppProxy, HIDER}; use yazi_shared::Defer; -use super::{ProcessOpOpen, ShellOpt}; +use super::{ProcessOpBg, ProcessOpBlock, ProcessOpOrphan, ShellOpt}; use crate::TaskProg; pub struct Process { @@ -13,15 +13,44 @@ pub struct Process { impl Process { pub fn new(prog: mpsc::UnboundedSender) -> Self { Self { prog } } - pub async fn open(&self, mut task: ProcessOpOpen) -> Result<()> { - if task.block { - return self.open_block(task).await; + pub async fn orphan(&self, task: ProcessOpOrphan) -> Result<()> { + let id = task.id; + match super::shell(task.into()) { + Ok(_) => self.succ(id)?, + Err(e) => { + self.prog.send(TaskProg::New(id, 0))?; + self.fail(id, format!("Failed to spawn process: {e}"))?; + } } - if task.orphan { - return self.open_orphan(task).await; + Ok(()) + } + + pub async fn block(&self, task: ProcessOpBlock) -> Result<()> { + let _permit = HIDER.acquire().await.unwrap(); + let _defer = Defer::new(AppProxy::resume); + AppProxy::stop().await; + + let (id, cmd) = (task.id, task.cmd.clone()); + let result = super::shell(task.into()); + if let Err(e) = result { + AppProxy::notify_warn(&cmd.to_string_lossy(), &format!("Failed to spawn process: {e}")); + return self.succ(id); } + let status = result.unwrap().wait().await?; + if !status.success() { + let content = match status.code() { + Some(code) => format!("Process exited with status code: {code}"), + None => "Process terminated by signal".to_string(), + }; + AppProxy::notify_warn(&cmd.to_string_lossy(), &content); + } + + self.succ(id) + } + + pub async fn bg(&self, task: ProcessOpBg) -> Result<()> { self.prog.send(TaskProg::New(task.id, 0))?; let mut child = super::shell(ShellOpt { cmd: task.cmd, args: task.args, piped: true, ..Default::default() })?; @@ -30,7 +59,7 @@ impl Process { let mut stderr = BufReader::new(child.stderr.take().unwrap()).lines(); loop { select! { - _ = task.cancel.closed() => { + _ = task.ct.cancelled() => { child.start_kill().ok(); break; } @@ -56,43 +85,6 @@ impl Process { self.prog.send(TaskProg::Adv(task.id, 1, 0))?; self.succ(task.id) } - - async fn open_block(&self, task: ProcessOpOpen) -> Result<()> { - let _permit = HIDER.acquire().await.unwrap(); - let _defer = Defer::new(AppProxy::resume); - AppProxy::stop().await; - - let (id, cmd) = (task.id, task.cmd.clone()); - let result = super::shell(task.into()); - if let Err(e) = result { - AppProxy::notify_warn(&cmd.to_string_lossy(), &format!("Failed to spawn process: {e}")); - return self.succ(id); - } - - let status = result.unwrap().wait().await?; - if !status.success() { - let content = match status.code() { - Some(code) => format!("Process exited with status code: {code}"), - None => "Process terminated by signal".to_string(), - }; - AppProxy::notify_warn(&cmd.to_string_lossy(), &content); - } - - self.succ(id) - } - - async fn open_orphan(&self, task: ProcessOpOpen) -> Result<()> { - let id = task.id; - match super::shell(task.into()) { - Ok(_) => self.succ(id)?, - Err(e) => { - self.prog.send(TaskProg::New(id, 0))?; - self.fail(id, format!("Failed to spawn process: {e}"))?; - } - } - - Ok(()) - } } impl Process { diff --git a/yazi-scheduler/src/process/shell.rs b/yazi-scheduler/src/process/shell.rs index cc7039b7..2be7f2fc 100644 --- a/yazi-scheduler/src/process/shell.rs +++ b/yazi-scheduler/src/process/shell.rs @@ -12,11 +12,6 @@ pub struct ShellOpt { } impl ShellOpt { - pub fn with_piped(mut self) -> Self { - self.piped = true; - self - } - #[inline] fn stdio(&self) -> Stdio { if self.orphan { diff --git a/yazi-scheduler/src/scheduler.rs b/yazi-scheduler/src/scheduler.rs index 2fdb3764..392f3cba 100644 --- a/yazi-scheduler/src/scheduler.rs +++ b/yazi-scheduler/src/scheduler.rs @@ -1,14 +1,15 @@ -use std::{ffi::OsStr, sync::Arc, time::Duration}; +use std::{ffi::OsString, sync::Arc, time::Duration}; use futures::{future::BoxFuture, FutureExt}; use parking_lot::Mutex; use tokio::{fs, select, sync::{mpsc::{self, UnboundedReceiver}, oneshot}}; +use tokio_util::sync::CancellationToken; use yazi_config::{open::Opener, plugin::PluginRule, TASKS}; use yazi_plugin::ValueSendable; use yazi_shared::{fs::{unique_path, Url}, Throttle}; use super::{Ongoing, TaskProg, TaskStage}; -use crate::{file::{File, FileOpDelete, FileOpLink, FileOpPaste, FileOpTrash}, plugin::{Plugin, PluginOpEntry}, preload::{Preload, PreloadOpRule, PreloadOpSize}, process::{Process, ProcessOpOpen}, TaskKind, TaskOp, HIGH, LOW, NORMAL}; +use crate::{file::{File, FileOpDelete, FileOpLink, FileOpPaste, FileOpTrash}, plugin::{Plugin, PluginOpEntry}, preload::{Preload, PreloadOpRule, PreloadOpSize}, process::{Process, ProcessOpBg, ProcessOpBlock, ProcessOpOrphan}, TaskKind, TaskOp, HIGH, LOW, NORMAL}; pub struct Scheduler { pub file: Arc, @@ -332,46 +333,54 @@ impl Scheduler { } } - pub fn process_open(&self, opener: &Opener, args: &[impl AsRef]) { + pub fn process_open( + &self, + opener: Opener, + args: Vec, + done: Option>, + ) { let name = { - let s = format!("Run `{}`", opener.run); - let args = args.iter().map(|a| a.as_ref().to_string_lossy()).collect::>().join(" "); - if args.is_empty() { s } else { format!("{s} with `{args}`") } + let args = args.iter().map(|a| a.to_string_lossy()).collect::>().join(" "); + if args.is_empty() { + format!("Run {:?}", opener.run) + } else { + format!("Run {:?} with `{args}`", opener.run) + } }; + let ct = CancellationToken::new(); let mut ongoing = self.ongoing.lock(); - let id = ongoing.add(TaskKind::User, name); - let (cancel_tx, mut cancel_rx) = oneshot::channel(); + let id = ongoing.add(TaskKind::User, name); ongoing.hooks.insert(id, { + let ct = ct.clone(); let ongoing = self.ongoing.clone(); Box::new(move |canceled: bool| { async move { - if canceled { - cancel_rx.close(); - } ongoing.lock().try_remove(id, TaskStage::Hooked); + if canceled { + ct.cancel(); + } + if let Some(tx) = done { + tx.send(()).ok(); + } } .boxed() }) }); - let args = args.iter().map(|a| a.as_ref().to_os_string()).collect::>(); + // FIXME: use micro instead tokio::spawn({ let process = self.process.clone(); let opener = opener.clone(); async move { - process - .open(ProcessOpOpen { - id, - cmd: opener.run.into(), - args, - block: opener.block, - orphan: opener.orphan, - cancel: cancel_tx, - }) - .await - .ok(); + if opener.orphan { + process.orphan(ProcessOpOrphan { id, cmd: opener.run.into(), args }).await.ok(); + } else if opener.block { + process.block(ProcessOpBlock { id, cmd: opener.run.into(), args }).await.ok(); + } else { + process.bg(ProcessOpBg { id, cmd: opener.run.into(), args, ct }).await.ok(); + } } }); }