diff --git a/yazi-fs/src/path/relative.rs b/yazi-fs/src/path/relative.rs index 30a2b8e4..7b0808e2 100644 --- a/yazi-fs/src/path/relative.rs +++ b/yazi-fs/src/path/relative.rs @@ -21,7 +21,7 @@ fn path_relative_to_impl<'a>(from: PathCow<'_>, to: PathCow<'a>) -> Result(from: PathCow<'_>, to: PathCow<'a>) -> Result Provider for Local<'a> { } #[inline] - async fn symlink(&self, original: P, _is_dir: F) -> io::Result<()> + async fn symlink(&self, original: S, _is_dir: F) -> io::Result<()> where - P: AsPath, + S: AsStrand, F: AsyncFnOnce() -> io::Result, { #[cfg(unix)] { - let original = original.as_path().as_os()?; + let original = original.as_strand().as_os()?; tokio::fs::symlink(original, self.path).await } #[cfg(windows)] @@ -139,11 +139,11 @@ impl<'a> Provider for Local<'a> { } #[inline] - async fn symlink_dir

(&self, original: P) -> io::Result<()> + async fn symlink_dir(&self, original: S) -> io::Result<()> where - P: AsPath, + S: AsStrand, { - let original = original.as_path().as_os()?; + let original = original.as_strand().as_os()?; #[cfg(unix)] { @@ -156,11 +156,11 @@ impl<'a> Provider for Local<'a> { } #[inline] - async fn symlink_file

(&self, original: P) -> io::Result<()> + async fn symlink_file(&self, original: S) -> io::Result<()> where - P: AsPath, + S: AsStrand, { - let original = original.as_path().as_os()?; + let original = original.as_strand().as_os()?; #[cfg(unix)] { diff --git a/yazi-fs/src/provider/traits.rs b/yazi-fs/src/provider/traits.rs index cc7c43f4..cfe2db55 100644 --- a/yazi-fs/src/provider/traits.rs +++ b/yazi-fs/src/provider/traits.rs @@ -2,7 +2,7 @@ use std::io; use tokio::{io::{AsyncRead, AsyncSeek, AsyncWrite, AsyncWriteExt}, sync::mpsc}; use yazi_macro::ok_or_not_found; -use yazi_shared::{path::{AsPath, PathBufDyn}, strand::StrandCow, url::{AsUrl, Url, UrlBuf}}; +use yazi_shared::{path::{AsPath, PathBufDyn}, strand::{AsStrand, StrandCow}, url::{AsUrl, Url, UrlBuf}}; use crate::{cha::{Cha, ChaType}, provider::Attrs}; @@ -155,21 +155,21 @@ pub trait Provider: Sized { where P: AsPath; - fn symlink(&self, original: P, _is_dir: F) -> impl Future> + fn symlink(&self, original: S, _is_dir: F) -> impl Future> where - P: AsPath, + S: AsStrand, F: AsyncFnOnce() -> io::Result; - fn symlink_dir

(&self, original: P) -> impl Future> + fn symlink_dir(&self, original: S) -> impl Future> where - P: AsPath, + S: AsStrand, { self.symlink(original, async || Ok(true)) } - fn symlink_file

(&self, original: P) -> impl Future> + fn symlink_file(&self, original: S) -> impl Future> where - P: AsPath, + S: AsStrand, { self.symlink(original, async || Ok(false)) } diff --git a/yazi-scheduler/src/file/file.rs b/yazi-scheduler/src/file/file.rs index da141ab2..496e54c8 100644 --- a/yazi-scheduler/src/file/file.rs +++ b/yazi-scheduler/src/file/file.rs @@ -1,4 +1,4 @@ -use std::{collections::VecDeque, hash::{BuildHasher, Hash, Hasher}, time::Duration}; +use std::{collections::VecDeque, hash::{BuildHasher, Hash, Hasher}}; use anyhow::{Context, Result, anyhow}; use tokio::{io::{self, ErrorKind::{AlreadyExists, NotFound}}, sync::mpsc}; @@ -26,7 +26,10 @@ impl File { } pub(crate) async fn paste(&self, mut task: FileInPaste) -> Result<(), FileOutPaste> { - if task.cut && ok_or_not_found(provider::rename(&task.from, &task.to).await).is_ok() { + if task.cut + && !task.follow + && ok_or_not_found(provider::rename(&task.from, &task.to).await).is_ok() + { return Ok(self.ops.out(task.id, FileOutPaste::Succ)); } @@ -108,8 +111,17 @@ impl File { } Ok(n) => self.ops.out(task.id, FileOutPasteDo::Adv(n)), Err(e) if e.kind() == NotFound => { - warn!("Paste task partially done: {task:?}"); - break; + let Ok(cha) = Self::cha(&task.from, task.follow, None).await else { + warn!("Paste task partially done: {task:?}"); + break; + }; + + if cha.is_orphan() || (cha.is_link() && !task.follow) { + task.cha = Some(cha); + return Ok(self.queue(task.into_link(), NORMAL)); + } + + Err(e)? } // Operation not permitted (os error 1) // Attribute not found (os error 93) @@ -132,10 +144,6 @@ impl File { } pub(crate) async fn link_do(&self, task: FileInLink) -> Result<(), FileOutLink> { - if !task.from.scheme().covariant(task.to.scheme()) { - Err(anyhow!("Source and target must be on the same filesystem: {task:?}"))? - } - let mut src: PathCow = if task.resolve { ok_or_not_found!( provider::read_link(&task.from).await, diff --git a/yazi-scheduler/src/scheduler.rs b/yazi-scheduler/src/scheduler.rs index 2c72a50b..f14547ec 100644 --- a/yazi-scheduler/src/scheduler.rs +++ b/yazi-scheduler/src/scheduler.rs @@ -79,13 +79,7 @@ impl Scheduler { let mut ongoing = self.ongoing.lock(); let id = ongoing.add::(format!("Cut {} to {}", from.display(), to.display())); - let Ok(prefixed) = to.try_starts_with(&from) else { - return self - .ops - .out(id, FileOutPaste::Fail("Path being cut has a different encoding".to_owned())); - }; - - if prefixed && !to.covariant(&from) { + if to.try_starts_with(&from).unwrap_or(false) && !to.covariant(&from) { return self.ops.out(id, FileOutPaste::Fail("Cannot cut directory into itself".to_owned())); } @@ -103,11 +97,12 @@ impl Scheduler { }); let file = self.file.clone(); + let follow = !from.scheme().covariant(to.scheme()); self.send_micro(id, LOW, async move { if !force { to = unique_name(to, must_be_dir(&from)).await?; } - file.paste(FileInPaste { id, from, to, cha: None, cut: true, follow: false, retry: 0 }).await + file.paste(FileInPaste { id, from, to, cha: None, cut: true, follow, retry: 0 }).await }); } @@ -118,17 +113,12 @@ impl Scheduler { to.display() )); - let Ok(prefixed) = to.try_starts_with(&from) else { - return self - .ops - .out(id, FileOutPaste::Fail("Path being copied has a different encoding".to_owned())); - }; - - if prefixed && !to.covariant(&from) { + if to.try_starts_with(&from).unwrap_or(false) && !to.covariant(&from) { return self.ops.out(id, FileOutPaste::Fail("Cannot copy directory into itself".to_owned())); } let file = self.file.clone(); + let follow = follow || !from.scheme().covariant(to.scheme()); self.send_micro(id, LOW, async move { if !force { to = unique_name(to, must_be_dir(&from)).await?; @@ -160,14 +150,13 @@ impl Scheduler { to.display() )); - let Ok(prefixed) = to.try_starts_with(&from) else { - return self.ops.out( - id, - FileOutHardlink::Fail("Path being hardlinked has a different encoding".to_owned()), - ); - }; + if !from.scheme().covariant(to.scheme()) { + return self + .ops + .out(id, FileOutHardlink::Fail("Cannot hardlink cross filesystem".to_owned())); + } - if prefixed && !to.covariant(&from) { + if to.try_starts_with(&from).unwrap_or(false) && !to.covariant(&from) { return self .ops .out(id, FileOutHardlink::Fail("Cannot hardlink directory into itself".to_owned())); diff --git a/yazi-sftp/src/fs/read_dir.rs b/yazi-sftp/src/fs/read_dir.rs index 22e0a354..ddc2f0d3 100644 --- a/yazi-sftp/src/fs/read_dir.rs +++ b/yazi-sftp/src/fs/read_dir.rs @@ -1,6 +1,6 @@ -use std::{mem, sync::Arc}; +use std::{mem, sync::Arc, time::Duration}; -use crate::{Error, Session, SftpPath, fs::DirEntry, requests, responses}; +use crate::{Error, Operator, Session, SftpPath, fs::DirEntry, requests, responses}; pub struct ReadDir { session: Arc, @@ -12,6 +12,10 @@ pub struct ReadDir { done: bool, } +impl Drop for ReadDir { + fn drop(&mut self) { Operator::from(&self.session).close(&self.handle).ok(); } +} + impl ReadDir { pub(crate) fn new(session: &Arc, dir: SftpPath, handle: String) -> Self { Self { @@ -49,7 +53,12 @@ impl ReadDir { return Ok(()); } - self.name = match self.session.send(requests::ReadDir::new(&self.handle)).await { + let result = self + .session + .send_with_timeout(requests::ReadDir::new(&self.handle), Duration::from_mins(5)) + .await; + + self.name = match result { Ok(resp) => resp, Err(Error::Status(status)) if status.is_eof() => { self.done = true; diff --git a/yazi-sftp/src/session.rs b/yazi-sftp/src/session.rs index 0857bc2c..1ad674b5 100644 --- a/yazi-sftp/src/session.rs +++ b/yazi-sftp/src/session.rs @@ -3,7 +3,7 @@ use std::{any::TypeId, collections::HashMap, io::{self, ErrorKind}, sync::Arc, t use parking_lot::Mutex; use russh::{ChannelStream, client::Msg}; use serde::Serialize; -use tokio::{io::{AsyncReadExt, AsyncWriteExt, ReadHalf, WriteHalf}, select, sync::{mpsc, oneshot}, time::timeout}; +use tokio::{io::{AsyncReadExt, AsyncWriteExt, ReadHalf, WriteHalf}, select, sync::{mpsc, oneshot}}; use crate::{Error, Id, Packet, Receiver, responses}; @@ -93,12 +93,7 @@ impl Session { I: Into> + Serialize, O: TryFrom, Error = Error> + 'static, { - match timeout(Duration::from_secs(30), self.send_sync(input)?).await?? { - Packet::Status(status) if TypeId::of::() != TypeId::of::() => { - Err(Error::Status(status)) - } - response => response.try_into(), - } + self.send_with_timeout(input, Duration::from_secs(45)).await } pub fn send_sync<'a, I>(self: &Arc, input: I) -> Result @@ -118,5 +113,22 @@ impl Session { Ok(Receiver::new(self, id, rx)) } + pub async fn send_with_timeout<'a, I, O>( + self: &Arc, + input: I, + timeout: Duration, + ) -> Result + where + I: Into> + Serialize, + O: TryFrom, Error = Error> + 'static, + { + match tokio::time::timeout(timeout, self.send_sync(input)?).await?? { + Packet::Status(status) if TypeId::of::() != TypeId::of::() => { + Err(Error::Status(status)) + } + response => response.try_into(), + } + } + pub fn is_closed(self: &Arc) -> bool { self.tx.is_closed() } } diff --git a/yazi-vfs/src/provider/copier.rs b/yazi-vfs/src/provider/copier.rs index c07b0946..211a1bf1 100644 --- a/yazi-vfs/src/provider/copier.rs +++ b/yazi-vfs/src/provider/copier.rs @@ -7,12 +7,15 @@ use yazi_shared::url::{Url, UrlBuf}; use crate::provider::{self, Gate}; +const BUF_SIZE: usize = 512 * 1024; +const PER_CHUNK: u64 = 8 * 1024 * 1024; + pub(super) async fn copy_impl(from: Url<'_>, to: Url<'_>, attrs: Attrs) -> io::Result { let src = provider::open(from).await?; let dist = provider::create(to).await?; - let mut reader = BufReader::with_capacity(524288, src); - let mut writer = BufWriter::with_capacity(524288, dist); + let mut reader = BufReader::with_capacity(BUF_SIZE, src); + let mut writer = BufWriter::with_capacity(BUF_SIZE, dist); let written = tokio::io::copy(&mut reader, &mut writer).await?; writer.flush().await?; @@ -33,32 +36,39 @@ pub(super) fn copy_with_progress_impl( let (acc_, prog_tx_) = (acc.clone(), prog_tx.clone()); tokio::spawn(async move { - let (cha, mut src) = { - let f = provider::open(&*from).await?; - (f.metadata().await?, Some(f)) + let init = async { + let src = provider::open(&*from).await?; + let cha = src.metadata().await?; + + let dist = provider::create(&*to).await?; + dist.set_len(cha.len).await?; + Ok((cha, Some(src), Some(dist))) }; - let mut dist = { - let f = provider::create(&*to).await?; - f.set_len(cha.len).await?; - Some(f) + let (cha, mut src, mut dist) = match init.await { + Ok(r) => r, + Err(e) => { + prog_tx_.send(Err(e)).await.ok(); + done_tx.send(()).ok(); + return; + } }; - let chunks = (cha.len + 5242880 - 1) / 5242880; + let chunks = (cha.len + PER_CHUNK - 1) / PER_CHUNK; let mut result = futures::stream::iter(0..chunks) .map(|i| { let acc_ = acc_.clone(); let (from, to) = (from.clone(), to.clone()); let (src, dist) = (src.take(), dist.take()); async move { - let offset = i * 5242880; - let take = cha.len.saturating_sub(offset).min(5242880); + let offset = i * PER_CHUNK; + let take = cha.len.saturating_sub(offset).min(PER_CHUNK); - let mut src = BufReader::with_capacity(524288, match src { + let mut src = BufReader::with_capacity(BUF_SIZE, match src { Some(f) => f, None => provider::open(&*from).await?, }); - let mut dist = BufWriter::with_capacity(524288, match dist { + let mut dist = BufWriter::with_capacity(BUF_SIZE, match dist { Some(f) => f, None => Gate::default().write(true).open(&*to).await?, }); @@ -114,7 +124,6 @@ pub(super) fn copy_with_progress_impl( } done_tx.send(()).ok(); - Ok::<_, io::Error>(()) }); tokio::spawn(async move { diff --git a/yazi-vfs/src/provider/provider.rs b/yazi-vfs/src/provider/provider.rs index 4a93a644..218f9ca3 100644 --- a/yazi-vfs/src/provider/provider.rs +++ b/yazi-vfs/src/provider/provider.rs @@ -2,7 +2,7 @@ use std::io; use tokio::sync::mpsc; use yazi_fs::{cha::Cha, provider::{Attrs, Provider, local::Local}}; -use yazi_shared::{path::{AsPath, PathBufDyn}, url::{AsUrl, UrlBuf, UrlCow}}; +use yazi_shared::{path::PathBufDyn, strand::AsStrand, url::{AsUrl, UrlBuf, UrlCow}}; use super::{Providers, ReadDir, RwFile}; @@ -201,27 +201,27 @@ where } } -pub async fn symlink(link: U, original: P, is_dir: F) -> io::Result<()> +pub async fn symlink(link: U, original: S, is_dir: F) -> io::Result<()> where U: AsUrl, - P: AsPath, + S: AsStrand, F: AsyncFnOnce() -> io::Result, { Providers::new(link.as_url()).await?.symlink(original, is_dir).await } -pub async fn symlink_dir(link: U, original: P) -> io::Result<()> +pub async fn symlink_dir(link: U, original: S) -> io::Result<()> where U: AsUrl, - P: AsPath, + S: AsStrand, { Providers::new(link.as_url()).await?.symlink_dir(original).await } -pub async fn symlink_file(link: U, original: P) -> io::Result<()> +pub async fn symlink_file(link: U, original: S) -> io::Result<()> where U: AsUrl, - P: AsPath, + S: AsStrand, { Providers::new(link.as_url()).await?.symlink_file(original).await } diff --git a/yazi-vfs/src/provider/providers.rs b/yazi-vfs/src/provider/providers.rs index 59e4b00e..e66b8356 100644 --- a/yazi-vfs/src/provider/providers.rs +++ b/yazi-vfs/src/provider/providers.rs @@ -2,7 +2,7 @@ use std::io; use tokio::sync::mpsc; use yazi_fs::{cha::Cha, provider::{Attrs, Provider}}; -use yazi_shared::{path::{AsPath, PathBufDyn}, url::{Url, UrlBuf, UrlCow}}; +use yazi_shared::{path::{AsPath, PathBufDyn}, strand::AsStrand, url::{Url, UrlBuf, UrlCow}}; #[derive(Clone)] pub(super) enum Providers<'a> { @@ -161,9 +161,9 @@ impl<'a> Provider for Providers<'a> { } } - async fn symlink(&self, original: P, is_dir: F) -> io::Result<()> + async fn symlink(&self, original: S, is_dir: F) -> io::Result<()> where - P: AsPath, + S: AsStrand, F: AsyncFnOnce() -> io::Result, { match self { @@ -172,9 +172,9 @@ impl<'a> Provider for Providers<'a> { } } - async fn symlink_dir

(&self, original: P) -> io::Result<()> + async fn symlink_dir(&self, original: S) -> io::Result<()> where - P: AsPath, + S: AsStrand, { match self { Self::Local(p) => p.symlink_dir(original).await, @@ -182,9 +182,9 @@ impl<'a> Provider for Providers<'a> { } } - async fn symlink_file

(&self, original: P) -> io::Result<()> + async fn symlink_file(&self, original: S) -> io::Result<()> where - P: AsPath, + S: AsStrand, { match self { Self::Local(p) => p.symlink_file(original).await, diff --git a/yazi-vfs/src/provider/sftp/sftp.rs b/yazi-vfs/src/provider/sftp/sftp.rs index 6fdc063a..00e1797d 100644 --- a/yazi-vfs/src/provider/sftp/sftp.rs +++ b/yazi-vfs/src/provider/sftp/sftp.rs @@ -4,7 +4,7 @@ use tokio::{io::{AsyncWriteExt, BufReader, BufWriter}, sync::mpsc::Receiver}; use yazi_config::vfs::{ServiceSftp, Vfs}; use yazi_fs::{CWD, provider::{DirReader, FileHolder, Provider}}; use yazi_sftp::fs::{Attrs, Flags}; -use yazi_shared::{loc::LocBuf, path::{AsPath, PathBufDyn}, pool::InternStr, scheme::SchemeKind, url::{Url, UrlBuf, UrlCow, UrlLike}}; +use yazi_shared::{loc::LocBuf, path::{AsPath, PathBufDyn}, pool::InternStr, scheme::SchemeKind, strand::AsStrand, url::{Url, UrlBuf, UrlCow, UrlLike}}; use super::Cha; use crate::provider::sftp::Conn; @@ -175,14 +175,14 @@ impl<'a> Provider for Sftp<'a> { Ok(()) } - async fn symlink(&self, original: P, _is_dir: F) -> io::Result<()> + async fn symlink(&self, original: S, _is_dir: F) -> io::Result<()> where - P: AsPath, + S: AsStrand, F: AsyncFnOnce() -> io::Result, { - let original = original.as_path().as_unix()?; + let original = original.as_strand().encoded_bytes(); - Ok(self.op().await?.symlink(&original, self.path).await?) + Ok(self.op().await?.symlink(original, self.path).await?) } async fn symlink_metadata(&self) -> io::Result {