From d3f0c9c059a1f2c94208e8862729a19395511cf2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=89=E5=92=B2=E9=9B=85=20=C2=B7=20Misaki=20Masa?= Date: Thu, 7 Sep 2023 00:12:59 +0800 Subject: [PATCH] perf: load large folders in chunks (#117) --- Cargo.lock | 60 +++++++++---------- app/src/app.rs | 7 ++- core/src/event.rs | 4 +- core/src/external/fd.rs | 16 ++--- core/src/external/rg.rs | 16 ++--- core/src/external/zoxide.rs | 3 +- core/src/files/files.rs | 91 +++++++++++++++++------------ core/src/files/op.rs | 20 +++++-- core/src/input/input.rs | 2 +- core/src/manager/folder.rs | 3 +- core/src/manager/manager.rs | 14 ++--- core/src/manager/preview/preview.rs | 53 +++++++++-------- core/src/manager/tab.rs | 35 ++++++----- core/src/manager/watcher.rs | 69 +++++++++++----------- core/src/tasks/tasks.rs | 4 +- core/src/tasks/workers/file.rs | 6 +- core/src/which/which.rs | 2 +- shared/src/lib.rs | 2 - shared/src/stream.rs | 69 ---------------------- shared/src/url.rs | 56 ++++++++++++++---- 20 files changed, 270 insertions(+), 262 deletions(-) delete mode 100644 shared/src/stream.rs diff --git a/Cargo.lock b/Cargo.lock index 5b7a371d..b6235c6c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -225,9 +225,9 @@ checksum = "a3e2c3daef883ecc1b5d58c15adae93470a91d425f3532ba1695849656af3fc1" [[package]] name = "bytemuck" -version = "1.13.1" +version = "1.14.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "17febce684fd15d89027105661fec94afb475cb995fbc59d2865198446ba2eea" +checksum = "374d28ec25809ee0e23827c2ab573d729e293f281dfe393500e7ad618baa61c6" [[package]] name = "byteorder" @@ -264,9 +264,9 @@ checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd" [[package]] name = "chrono" -version = "0.4.28" +version = "0.4.29" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "95ed24df0632f708f5f6d8082675bef2596f7084dee3dd55f632290bf35bfe0f" +checksum = "d87d9d13be47a5b7c3907137f1290b0459a7f80efb26be8c52afb11963bccb02" dependencies = [ "android-tzdata", "iana-time-zone", @@ -305,7 +305,7 @@ dependencies = [ "heck", "proc-macro2", "quote", - "syn 2.0.29", + "syn 2.0.31", ] [[package]] @@ -700,7 +700,7 @@ checksum = "89ca545a94061b6365f2c7355b4b32bd20df3ff95f02da9329b34ccc3bd6ee72" dependencies = [ "proc-macro2", "quote", - "syn 2.0.29", + "syn 2.0.31", ] [[package]] @@ -1037,9 +1037,9 @@ dependencies = [ [[package]] name = "memchr" -version = "2.6.2" +version = "2.6.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5486aed0026218e61b8a01d5fbd5a0a134649abb71a0e53b7bc088529dced86e" +checksum = "8f232d6ef707e1956a43342693d2a31e72989554d58299d7a88738cc95b0d35c" [[package]] name = "memoffset" @@ -1182,9 +1182,9 @@ dependencies = [ [[package]] name = "object" -version = "0.32.0" +version = "0.32.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "77ac5bbd07aea88c60a577a1ce218075ffd59208b2d7ca97adf9bfc5aeb21ebe" +checksum = "9cf5f9dd3933bd50a9e1f149ec995f39ae2c496d31fd772c1fd45ebc27e902b0" dependencies = [ "memchr", ] @@ -1281,7 +1281,7 @@ checksum = "4359fd9c9171ec6e8c62926d6faaf553a8dc3f64e1507e76da7911b4f6a04405" dependencies = [ "proc-macro2", "quote", - "syn 2.0.29", + "syn 2.0.31", ] [[package]] @@ -1459,9 +1459,9 @@ dependencies = [ [[package]] name = "regex" -version = "1.9.4" +version = "1.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "12de2eff854e5fa4b1295edd650e227e9d8fb0c9e90b12e7f36d6a6811791a29" +checksum = "697061221ea1b4a94a624f67d0ae2bfe4e22b8a17b6a192afb11046542cc8c47" dependencies = [ "aho-corasick", "memchr", @@ -1471,9 +1471,9 @@ dependencies = [ [[package]] name = "regex-automata" -version = "0.3.7" +version = "0.3.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "49530408a136e16e5b486e883fbb6ba058e8e4e8ae6621a77b048b314336e629" +checksum = "c2f401f4955220693b56f8ec66ee9c78abffd8d1c4f23dc41a23839eb88f0795" dependencies = [ "aho-corasick", "memchr", @@ -1542,7 +1542,7 @@ checksum = "4eca7ac642d82aa35b60049a6eccb4be6be75e599bd2e9adb5f875a737654af2" dependencies = [ "proc-macro2", "quote", - "syn 2.0.29", + "syn 2.0.31", ] [[package]] @@ -1706,7 +1706,7 @@ dependencies = [ "proc-macro2", "quote", "rustversion", - "syn 2.0.29", + "syn 2.0.31", ] [[package]] @@ -1722,9 +1722,9 @@ dependencies = [ [[package]] name = "syn" -version = "2.0.29" +version = "2.0.31" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c324c494eba9d92503e6f1ef2e6df781e78f6a7705a0202d9801b198807d518a" +checksum = "718fa2415bcb8d8bd775917a1bf12a7931b6dfa890753378538118181e0cb398" dependencies = [ "proc-macro2", "quote", @@ -1754,22 +1754,22 @@ dependencies = [ [[package]] name = "thiserror" -version = "1.0.47" +version = "1.0.48" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "97a802ec30afc17eee47b2855fc72e0c4cd62be9b4efe6591edde0ec5bd68d8f" +checksum = "9d6d7a740b8a666a7e828dd00da9c0dc290dff53154ea77ac109281de90589b7" dependencies = [ "thiserror-impl", ] [[package]] name = "thiserror-impl" -version = "1.0.47" +version = "1.0.48" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6bb623b56e39ab7dcd4b1b98bb6c8f8d907ed255b18de254088016b27a8ee19b" +checksum = "49922ecae66cc8a249b77e68d1d0623c1b2c514f0060c27cdc68bd62a1219d35" dependencies = [ "proc-macro2", "quote", - "syn 2.0.29", + "syn 2.0.31", ] [[package]] @@ -1863,7 +1863,7 @@ checksum = "630bdcf245f78637c13ec01ffae6187cca34625e8c63150d424b59e55af2675e" dependencies = [ "proc-macro2", "quote", - "syn 2.0.29", + "syn 2.0.31", ] [[package]] @@ -1943,7 +1943,7 @@ checksum = "5f4f31f56159e98206da9efd823404b79b6ef3143b4a7ab76e67b1751b25a4ab" dependencies = [ "proc-macro2", "quote", - "syn 2.0.29", + "syn 2.0.31", ] [[package]] @@ -2109,9 +2109,9 @@ checksum = "49874b5167b65d7193b8aba1567f5c7d93d001cafc34600cee003eda787e483f" [[package]] name = "walkdir" -version = "2.3.3" +version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "36df944cda56c7d8d8b7496af378e6b16de9284591917d307c9b4d313c44e698" +checksum = "d71d857dc86794ca4c280d616f7da00d2dbfd8cd788846559a6813e6aa4b54ee" dependencies = [ "same-file", "winapi-util", @@ -2144,7 +2144,7 @@ dependencies = [ "once_cell", "proc-macro2", "quote", - "syn 2.0.29", + "syn 2.0.31", "wasm-bindgen-shared", ] @@ -2166,7 +2166,7 @@ checksum = "54681b18a46765f095758388f2d0cf16eb8d4169b639ab575a8f5693af210c7b" dependencies = [ "proc-macro2", "quote", - "syn 2.0.29", + "syn 2.0.31", "wasm-bindgen-backend", "wasm-bindgen-shared", ] diff --git a/app/src/app.rs b/app/src/app.rs index ab83a0ab..54328bf0 100644 --- a/app/src/app.rs +++ b/app/src/app.rs @@ -128,16 +128,17 @@ impl App { manager.refresh(); } Event::Files(op) => { - let read = matches!(op, FilesOp::Read(..)); + let calc = matches!(op, FilesOp::Full(..) | FilesOp::Part(..)); let b = match op { - FilesOp::Read(..) => manager.update_read(op), + FilesOp::Full(..) => manager.update_read(op), + FilesOp::Part(..) => manager.update_read(op), FilesOp::Size(..) => manager.update_read(op), FilesOp::IOErr(..) => manager.update_ioerr(op), }; if b { emit!(Render); } - if read { + if calc { tasks.precache_size(&manager.current().files); } } diff --git a/core/src/event.rs b/core/src/event.rs index 97039857..762d6e46 100644 --- a/core/src/event.rs +++ b/core/src/event.rs @@ -71,8 +71,8 @@ macro_rules! emit { $crate::Event::Ctrl($exec, $layer).emit(); }; - (Cd($op:expr)) => { - $crate::Event::Cd($op).emit(); + (Cd($url:expr)) => { + $crate::Event::Cd($url).emit(); }; (Files($op:expr)) => { $crate::Event::Files($op).emit(); diff --git a/core/src/external/fd.rs b/core/src/external/fd.rs index ba302f69..d914b19b 100644 --- a/core/src/external/fd.rs +++ b/core/src/external/fd.rs @@ -1,9 +1,10 @@ -use std::{process::Stdio, time::Duration}; +use std::process::Stdio; use anyhow::Result; -use shared::{StreamBuf, Url}; -use tokio::{io::{AsyncBufReadExt, BufReader}, process::Command, sync::mpsc}; -use tokio_stream::wrappers::UnboundedReceiverStream; +use shared::Url; +use tokio::{io::{AsyncBufReadExt, BufReader}, process::Command, sync::mpsc::{self, UnboundedReceiver}}; + +use crate::files::File; pub struct FdOpt { pub cwd: Url, @@ -12,7 +13,7 @@ pub struct FdOpt { pub subject: String, } -pub fn fd(opt: FdOpt) -> Result>> { +pub fn fd(opt: FdOpt) -> Result> { let mut child = Command::new("fd") .arg("--base-directory") .arg(&opt.cwd) @@ -28,11 +29,12 @@ pub fn fd(opt: FdOpt) -> Result>> { let mut it = BufReader::new(child.stdout.take().unwrap()).lines(); let (tx, rx) = mpsc::unbounded_channel(); - let rx = StreamBuf::new(UnboundedReceiverStream::new(rx), Duration::from_millis(300)); tokio::spawn(async move { while let Ok(Some(line)) = it.next_line().await { - tx.send(opt.cwd.join(line)).ok(); + if let Ok(file) = File::from(opt.cwd.join(line)).await { + tx.send(file).ok(); + } } child.wait().await.ok(); }); diff --git a/core/src/external/rg.rs b/core/src/external/rg.rs index 9e3a5fb1..c53ac06d 100644 --- a/core/src/external/rg.rs +++ b/core/src/external/rg.rs @@ -1,9 +1,10 @@ -use std::{process::Stdio, time::Duration}; +use std::process::Stdio; use anyhow::Result; -use shared::{StreamBuf, Url}; -use tokio::{io::{AsyncBufReadExt, BufReader}, process::Command, sync::mpsc}; -use tokio_stream::wrappers::UnboundedReceiverStream; +use shared::Url; +use tokio::{io::{AsyncBufReadExt, BufReader}, process::Command, sync::mpsc::{self, UnboundedReceiver}}; + +use crate::files::File; pub struct RgOpt { pub cwd: Url, @@ -11,7 +12,7 @@ pub struct RgOpt { pub subject: String, } -pub fn rg(opt: RgOpt) -> Result>> { +pub fn rg(opt: RgOpt) -> Result> { let mut child = Command::new("rg") .current_dir(&opt.cwd) .args(["--color=never", "--files-with-matches", "--smart-case"]) @@ -26,11 +27,12 @@ pub fn rg(opt: RgOpt) -> Result>> { let mut it = BufReader::new(child.stdout.take().unwrap()).lines(); let (tx, rx) = mpsc::unbounded_channel(); - let rx = StreamBuf::new(UnboundedReceiverStream::new(rx), Duration::from_millis(300)); tokio::spawn(async move { while let Ok(Some(line)) = it.next_line().await { - tx.send(opt.cwd.join(line)).ok(); + if let Ok(file) = File::from(opt.cwd.join(line)).await { + tx.send(file).ok(); + } } child.wait().await.ok(); }); diff --git a/core/src/external/zoxide.rs b/core/src/external/zoxide.rs index ae46e07d..d947f1dd 100644 --- a/core/src/external/zoxide.rs +++ b/core/src/external/zoxide.rs @@ -17,12 +17,11 @@ pub fn zoxide(opt: ZoxideOpt) -> Result>> { .spawn()?; let (tx, rx) = oneshot::channel(); - let cwd = opt.cwd.clone(); tokio::spawn(async move { if let Ok(output) = child.wait_with_output().await { let selected = String::from_utf8_lossy(&output.stdout).trim().to_string(); if !selected.is_empty() { - tx.send(Ok(Url::new(selected, &cwd))).ok(); + tx.send(Ok(Url::from(selected))).ok(); return; } } diff --git a/core/src/files/files.rs b/core/src/files/files.rs index 831c3ec3..2c69bba7 100644 --- a/core/src/files/files.rs +++ b/core/src/files/files.rs @@ -1,15 +1,16 @@ -use std::{collections::{BTreeMap, BTreeSet}, mem, ops::Deref}; +use std::{collections::{BTreeMap, BTreeSet}, mem, ops::Deref, sync::atomic::Ordering}; use anyhow::Result; use config::manager::SortBy; use shared::Url; -use tokio::fs; +use tokio::{fs, select, sync::mpsc::{self, UnboundedReceiver}}; -use super::{File, FilesSorter}; +use super::{File, FilesSorter, FILES_VERSION}; pub struct Files { - items: Vec, - hidden: Vec, + items: Vec, + hidden: Vec, + version: u64, sizes: BTreeMap, selected: BTreeSet, @@ -21,8 +22,9 @@ pub struct Files { impl Default for Files { fn default() -> Self { Self { - items: Default::default(), - hidden: Default::default(), + items: Default::default(), + hidden: Default::default(), + version: Default::default(), sizes: Default::default(), selected: Default::default(), @@ -40,25 +42,21 @@ impl Deref for Files { } impl Files { - pub async fn read(urls: Vec) -> Vec { - let mut items = Vec::with_capacity(urls.len()); - for url in urls { - if let Ok(file) = File::from(url).await { - items.push(file); - } - } - items - } - - pub async fn read_dir(url: &Url) -> Result> { + pub async fn from_dir(url: &Url) -> Result> { let mut it = fs::read_dir(url).await?; - let mut items = Vec::new(); - while let Ok(Some(item)) = it.next_entry().await { - if let Ok(meta) = item.metadata().await { - items.push(File::from_meta(Url::new(item.path(), url), meta).await); + let (tx, rx) = mpsc::unbounded_channel(); + + tokio::spawn(async move { + while let Ok(Some(item)) = it.next_entry().await { + select! { + _ = tx.closed() => break, + Ok(meta) = item.metadata() => { + tx.send(File::from_meta(Url::from(item.path()), meta).await).ok(); + } + } } - } - Ok(items) + }); + Ok(rx) } } @@ -83,10 +81,23 @@ impl Files { pub fn select_all(&mut self, state: Option) -> bool { match state { Some(true) => { + let b = if self.selected.len() < self.items.len() { + true + } else { + self.items.iter().any(|f| !self.selected.contains(&f.url)) + }; + self.selected = self.iter().map(|f| f.url_owned()).collect(); + b } Some(false) => { + if self.selected.is_empty() { + return false; + } + + let b = self.items.iter().any(|f| self.selected.contains(&f.url)); self.selected.clear(); + b } None => { for item in &self.items { @@ -96,9 +107,9 @@ impl Files { self.selected.insert(item.url_owned()); } } + !self.items.is_empty() } } - !self.items.is_empty() } pub fn select_index(&mut self, indices: &BTreeSet, state: Option) -> bool { @@ -111,26 +122,22 @@ impl Files { applied } - pub fn update_read(&mut self, mut items: Vec) -> bool { + pub fn update_full(&mut self, mut items: Vec) -> bool { if !self.show_hidden { (self.hidden, items) = items.into_iter().partition(|f| f.is_hidden); } self.sorter.sort(&mut items); self.items = items; + self.version = FILES_VERSION.fetch_add(1, Ordering::Relaxed); true } - pub fn update_size(&mut self, items: BTreeMap) -> bool { - self.sizes.extend(items); - if self.sorter.by == SortBy::Size { - self.sorter.sort(&mut self.items); - } - true - } - - // TODO: remove this - pub fn update_search(&mut self, items: Vec) -> bool { + pub fn update_part(&mut self, version: u64, items: Vec) -> bool { if !items.is_empty() { + if version != self.version { + return false; + } + if self.show_hidden { self.items.extend(items); } else { @@ -143,14 +150,22 @@ impl Files { return true; } + self.version = version; if !self.items.is_empty() { self.items.clear(); self.hidden.clear(); return true; } - false } + + pub fn update_size(&mut self, items: BTreeMap) -> bool { + self.sizes.extend(items); + if self.sorter.by == SortBy::Size { + self.sorter.sort(&mut self.items); + } + true + } } impl Files { @@ -166,7 +181,7 @@ impl Files { } #[inline] - pub fn position(&self, url: &Url) -> Option { self.iter().position(|f| f.url == *url) } + pub fn position(&self, url: &Url) -> Option { self.iter().position(|f| &f.url == url) } #[inline] pub fn duplicate(&self, idx: usize) -> Option { self.items.get(idx).cloned() } diff --git a/core/src/files/op.rs b/core/src/files/op.rs index 4d3723c0..346d880a 100644 --- a/core/src/files/op.rs +++ b/core/src/files/op.rs @@ -1,27 +1,35 @@ -use std::collections::BTreeMap; +use std::{collections::BTreeMap, sync::atomic::{AtomicU64, Ordering}}; use shared::Url; use super::File; +use crate::emit; + +pub(super) static FILES_VERSION: AtomicU64 = AtomicU64::new(0); #[derive(Debug)] pub enum FilesOp { - Read(Url, Vec), + Full(Url, Vec), + Part(Url, u64, Vec), Size(Url, BTreeMap), IOErr(Url), } impl FilesOp { #[inline] - pub fn url(&self) -> Url { + pub fn url(&self) -> &Url { match self { - Self::Read(url, _) => url, + Self::Full(url, _) => url, + Self::Part(url, ..) => url, Self::Size(url, _) => url, Self::IOErr(url) => url, } - .clone() } #[inline] - pub fn clear(url: &Url) -> Self { Self::Read(url.clone(), Vec::new()) } + pub fn prepare(url: &Url) -> u64 { + let version = FILES_VERSION.fetch_add(1, Ordering::Relaxed); + emit!(Files(Self::Part(url.clone(), version, Vec::new()))); + version + } } diff --git a/core/src/input/input.rs b/core/src/input/input.rs index 31752fb7..ca71e08d 100644 --- a/core/src/input/input.rs +++ b/core/src/input/input.rs @@ -308,7 +308,7 @@ impl Input { }; snap.cursor = snap.count().saturating_sub(snap.mode.delta()).min(snap.cursor); - if *snap == old { + if snap == &old { return false; } if !matches!(old.op, InputOp::None | InputOp::Select(_)) { diff --git a/core/src/manager/folder.rs b/core/src/manager/folder.rs index d9bb6166..3f765477 100644 --- a/core/src/manager/folder.rs +++ b/core/src/manager/folder.rs @@ -27,7 +27,8 @@ impl From<&Url> for Folder { impl Folder { pub fn update(&mut self, op: FilesOp) -> bool { let b = match op { - FilesOp::Read(_, items) => self.files.update_read(items), + FilesOp::Full(_, items) => self.files.update_full(items), + FilesOp::Part(_, version, items) => self.files.update_part(version, items), FilesOp::Size(_, items) => self.files.update_size(items), _ => unreachable!(), }; diff --git a/core/src/manager/manager.rs b/core/src/manager/manager.rs index 779024d5..a4730b9a 100644 --- a/core/src/manager/manager.rs +++ b/core/src/manager/manager.rs @@ -189,7 +189,7 @@ impl Manager { fs::File::create(path).await?; } - if let Ok(file) = File::from(Url::new(hovered, &cwd)).await { + if let Ok(file) = File::from(Url::from(hovered)).await { emit!(Hover(file)); emit!(Refresh); } @@ -326,18 +326,18 @@ impl Manager { } pub fn update_read(&mut self, op: FilesOp) -> bool { - let url = op.url(); + let url = op.url().clone(); let cwd = self.cwd().to_owned(); let hovered = self.hovered().map(|h| h.url_owned()); - let mut b = if cwd == url && !cwd.is_search() { + let mut b = if cwd == url { self.current_mut().update(op) } else if matches!(self.parent(), Some(p) if p.cwd == url) { self.active_mut().parent.as_mut().unwrap().update(op) } else { self.active_mut().history.entry(url.clone()).or_insert_with(|| Folder::from(&url)).update(op); - matches!(self.hovered(), Some(h) if *h.url() == url) + matches!(self.hovered(), Some(h) if h.url() == &url) }; b |= self.active_mut().parent.as_mut().map_or(false, |p| p.hover(&cwd)); @@ -351,11 +351,11 @@ impl Manager { pub fn update_ioerr(&mut self, op: FilesOp) -> bool { let url = op.url(); - let op = FilesOp::clear(&url); + let op = FilesOp::Full(url.clone(), Vec::new()); - if url == *self.cwd() { + if url == self.cwd() { self.current_mut().update(op); - } else if matches!(self.parent(), Some(p) if p.cwd == url) { + } else if matches!(self.parent(), Some(p) if &p.cwd == url) { self.active_mut().parent.as_mut().unwrap().update(op); } else { return false; diff --git a/core/src/manager/preview/preview.rs b/core/src/manager/preview/preview.rs index 04cdb6ec..d7353f78 100644 --- a/core/src/manager/preview/preview.rs +++ b/core/src/manager/preview/preview.rs @@ -1,9 +1,10 @@ -use std::sync::atomic::Ordering; +use std::{sync::atomic::Ordering, time::Duration}; use adaptor::Adaptor; use config::MANAGER; use shared::{MimeKind, PeekError, Url, MIME_DIR}; -use tokio::task::JoinHandle; +use tokio::{pin, task::JoinHandle}; +use tokio_stream::{wrappers::UnboundedReceiverStream, StreamExt}; use super::{Provider, INCR}; use crate::{emit, files::{Files, FilesOp}}; @@ -70,32 +71,36 @@ impl Preview { } self.reset(|_| true); - if files.is_some() || sequent { - emit!(Preview(PreviewLock { - url: url.clone(), - mime: MIME_DIR.to_owned(), - skip: self.skip, - data: PreviewData::Folder, - })); - } + emit!(Preview(PreviewLock { + url: url.clone(), + mime: MIME_DIR.to_owned(), + skip: self.skip, + data: PreviewData::Folder, + })); if sequent { return; } - let (url, skip) = (url.clone(), self.skip); + let url = url.clone(); self.handle = Some(tokio::spawn(async move { - emit!(Files(match Files::read_dir(&url).await { - Ok(items) => FilesOp::Read(url.clone(), items), - Err(_) => FilesOp::IOErr(url.clone()), - })); + let Ok(rx) = Files::from_dir(&url).await else { + emit!(Files(FilesOp::IOErr(url))); + return; + }; - emit!(Preview(PreviewLock { - url, - mime: MIME_DIR.to_owned(), - skip, - data: PreviewData::Folder, - })); + if files.is_some() { + emit!(Files(FilesOp::Full(url, UnboundedReceiverStream::new(rx).collect().await))); + return; + } + + let rx = UnboundedReceiverStream::new(rx).chunks_timeout(10000, Duration::from_millis(500)); + pin!(rx); + + let version = FilesOp::prepare(&url); + while let Some(chunk) = rx.next().await { + emit!(Files(FilesOp::Part(url.clone(), version, chunk))); + } })); } @@ -164,7 +169,7 @@ impl Preview { #[inline] pub fn same(&self, url: &Url, mime: &str) -> bool { if let Some(ref lock) = self.lock { - return lock.url == *url && lock.mime == mime && lock.skip == self.skip; + return &lock.url == url && lock.mime == mime && lock.skip == self.skip; } false } @@ -172,7 +177,7 @@ impl Preview { #[inline] pub fn same_mime(&self, url: &Url, mime: &str) -> bool { if let Some(ref lock) = self.lock { - return lock.url == *url && lock.mime == mime; + return &lock.url == url && lock.mime == mime; } false } @@ -180,7 +185,7 @@ impl Preview { #[inline] pub fn same_path(&self, url: &Url) -> bool { if let Some(ref lock) = self.lock { - return lock.url == *url; + return &lock.url == url; } false } diff --git a/core/src/manager/tab.rs b/core/src/manager/tab.rs index 0604c97b..9bb6a96d 100644 --- a/core/src/manager/tab.rs +++ b/core/src/manager/tab.rs @@ -1,13 +1,13 @@ -use std::{borrow::Cow, collections::{BTreeMap, BTreeSet}, ffi::{OsStr, OsString}, mem}; +use std::{borrow::Cow, collections::{BTreeMap, BTreeSet}, ffi::{OsStr, OsString}, mem, time::Duration}; use anyhow::{Error, Result}; use config::open::Opener; -use futures::StreamExt; use shared::{Defer, Url}; -use tokio::task::JoinHandle; +use tokio::{pin, task::JoinHandle}; +use tokio_stream::{wrappers::UnboundedReceiverStream, StreamExt}; use super::{Folder, Mode, Preview, PreviewLock}; -use crate::{emit, external::{self, FzfOpt, ZoxideOpt}, files::{File, Files, FilesOp, FilesSorter}, input::InputOpt, Event, BLOCKER}; +use crate::{emit, external::{self, FzfOpt, ZoxideOpt}, files::{File, FilesOp, FilesSorter}, input::InputOpt, Event, BLOCKER}; pub struct Tab { pub(super) mode: Mode, @@ -106,7 +106,7 @@ impl Tab { let rep = self.history_new(&target); let rep = mem::replace(&mut self.current, rep); - if !rep.cwd.is_search() { + if rep.cwd.is_regular() { self.history.insert(rep.cwd.clone(), rep); } @@ -127,7 +127,7 @@ impl Tab { emit!(Input(InputOpt::top("Change directory:").with_value(target.to_string_lossy()))); if let Ok(s) = result.await { - emit!(Cd(Url::new(s, &target))); + emit!(Cd(Url::from(s))); } }); false @@ -143,7 +143,7 @@ impl Tab { let rep = self.history_new(hovered.url()); let rep = mem::replace(&mut self.current, rep); - if !rep.cwd.is_search() { + if rep.cwd.is_regular() { self.history.insert(rep.cwd.clone(), rep); } @@ -178,7 +178,7 @@ impl Tab { let rep = self.history_new(¤t); let rep = mem::replace(&mut self.current, rep); - if !rep.cwd.is_search() { + if rep.cwd.is_regular() { self.history.insert(rep.cwd.clone(), rep); } @@ -237,21 +237,29 @@ impl Tab { handle.abort(); } - let cwd = self.current.cwd.clone(); + let cwd = self.current.cwd.to_search(); let hidden = self.show_hidden; self.search = Some(tokio::spawn(async move { let subject = emit!(Input(InputOpt::top("Search:"))).await?; - let mut rx = if grep { + let rx = if grep { external::rg(external::RgOpt { cwd: cwd.clone(), hidden, subject }) } else { external::fd(external::FdOpt { cwd: cwd.clone(), hidden, glob: false, subject }) }?; - emit!(Files(FilesOp::clear(&cwd))); + let rx = UnboundedReceiverStream::new(rx).chunks_timeout(1000, Duration::from_millis(300)); + pin!(rx); + + let version = FilesOp::prepare(&cwd); + let mut first = true; while let Some(chunk) = rx.next().await { - emit!(Files(FilesOp::Read(cwd.clone(), Files::read(chunk).await))); + if first { + emit!(Cd(cwd.clone())); + first = false; + } + emit!(Files(FilesOp::Part(cwd.clone(), version, chunk))); } Ok(()) })); @@ -265,8 +273,7 @@ impl Tab { if self.current.cwd.is_search() { self.preview_reset_image(); - let cwd = self.current.cwd.clone(); - let rep = self.history_new(&cwd); + let rep = self.history_new(&self.current.cwd.to_regular()); drop(mem::replace(&mut self.current, rep)); emit!(Refresh); } diff --git a/core/src/manager/watcher.rs b/core/src/manager/watcher.rs index 2c19c820..d501de04 100644 --- a/core/src/manager/watcher.rs +++ b/core/src/manager/watcher.rs @@ -1,14 +1,13 @@ use std::{collections::BTreeSet, sync::Arc, time::Duration}; -use futures::StreamExt; use indexmap::IndexMap; use notify::{event::{MetadataKind, ModifyKind}, EventKind, RecommendedWatcher, RecursiveMode, Watcher as _Watcher}; use parking_lot::RwLock; -use shared::{StreamBuf, Url}; -use tokio::{fs, sync::mpsc}; -use tokio_stream::wrappers::UnboundedReceiverStream; +use shared::Url; +use tokio::{fs, pin, sync::mpsc::{self, UnboundedReceiver}}; +use tokio_stream::{wrappers::UnboundedReceiverStream, StreamExt}; -use crate::{emit, external, files::{Files, FilesOp}}; +use crate::{emit, external, files::{File, Files, FilesOp}}; pub struct Watcher { watcher: RecommendedWatcher, @@ -18,8 +17,6 @@ pub struct Watcher { impl Watcher { pub(super) fn start() -> Self { let (tx, rx) = mpsc::unbounded_channel(); - let rx = StreamBuf::new(UnboundedReceiverStream::new(rx), Duration::from_millis(300)); - let watcher = RecommendedWatcher::new( { let tx = tx.clone(); @@ -70,6 +67,7 @@ impl Watcher { } pub(super) fn watch(&mut self, mut watched: BTreeSet<&Url>) { + watched.retain(|&u| u.is_regular()); let (to_unwatch, to_watch): (BTreeSet<_>, BTreeSet<_>) = { let guard = self.watched.read(); let keys = guard.keys().collect::>(); @@ -122,8 +120,12 @@ impl Watcher { } pub(super) fn trigger_dirs(&self, dirs: &[&Url]) { + let dirs: Vec<_> = dirs.iter().filter(|&u| u.is_regular()).map(|&u| u.clone()).collect(); + if dirs.is_empty() { + return; + } + let watched = self.watched.clone(); - let dirs: Vec<_> = dirs.iter().map(|&u| u.clone()).collect(); tokio::spawn(async move { for dir in dirs { Self::dir_changed(&dir, watched.clone()).await; @@ -131,10 +133,10 @@ impl Watcher { }); } - async fn changed( - mut rx: StreamBuf>, - watched: Arc>>>, - ) { + async fn changed(rx: UnboundedReceiver, watched: Arc>>>) { + let rx = UnboundedReceiverStream::new(rx).chunks_timeout(100, Duration::from_millis(200)); + pin!(rx); + while let Some(paths) = rx.next().await { let (mut files, mut dirs): (Vec<_>, Vec<_>) = Default::default(); for path in paths.into_iter().collect::>() { @@ -163,34 +165,35 @@ impl Watcher { } async fn dir_changed(url: &Url, watched: Arc>>>) { - let linked = watched + let linked: Vec<_> = watched .read() .iter() .map_while(|(k, v)| v.as_ref().and_then(|v| url.strip_prefix(v)).map(|v| k.join(v))) - .collect::>(); + .collect(); - let result = Files::read_dir(url).await; - if linked.is_empty() { - emit!(Files(match result { - Ok(items) => FilesOp::Read(url.clone(), items), - Err(_) => FilesOp::IOErr(url.clone()), - })); + let Ok(rx) = Files::from_dir(url).await else { + emit!(Files(FilesOp::IOErr(url.clone()))); + for ori in linked { + emit!(Files(FilesOp::IOErr(ori))); + } return; - } + }; + let linked_files = |files: &[File], ori: &Url| -> Vec { + let mut new = Vec::with_capacity(files.len()); + for file in files { + let mut file = file.clone(); + file.set_url(ori.join(file.url().strip_prefix(url).unwrap())); + new.push(file); + } + new + }; + + let files: Vec<_> = UnboundedReceiverStream::new(rx).collect().await; for ori in linked { - emit!(Files(match &result { - Ok(items) => { - let mut files = Vec::with_capacity(items.len()); - for item in items { - let mut file = item.clone(); - file.set_url(ori.join(item.url().strip_prefix(url).unwrap())); - files.push(file); - } - FilesOp::Read(ori, files) - } - Err(_) => FilesOp::IOErr(ori), - })); + let files = linked_files(&files, &ori); + emit!(Files(FilesOp::Full(ori, files))); } + emit!(Files(FilesOp::Full(url.clone(), files))); } } diff --git a/core/src/tasks/tasks.rs b/core/src/tasks/tasks.rs index e7f0e27d..8614087a 100644 --- a/core/src/tasks/tasks.rs +++ b/core/src/tasks/tasks.rs @@ -157,7 +157,7 @@ impl Tasks { pub fn file_cut(&self, src: &HashSet, dest: Url, force: bool) -> bool { for p in src { let to = dest.join(p.file_name().unwrap()); - if force && *p == to { + if force && p == &to { trace!("file_cut: same file, skipping {:?}", to); } else { self.scheduler.file_cut(p.clone(), to, force); @@ -169,7 +169,7 @@ impl Tasks { pub fn file_copy(&self, src: &HashSet, dest: Url, force: bool, follow: bool) -> bool { for p in src { let to = dest.join(p.file_name().unwrap()); - if force && *p == to { + if force && p == &to { trace!("file_copy: same file, skipping {:?}", to); } else { self.scheduler.file_copy(p.clone(), to, force, follow); diff --git a/core/src/tasks/workers/file.rs b/core/src/tasks/workers/file.rs index 4f52eacb..c55258a6 100644 --- a/core/src/tasks/workers/file.rs +++ b/core/src/tasks/workers/file.rs @@ -256,7 +256,7 @@ impl File { let mut dirs = VecDeque::from([task.target]); while let Some(target) = dirs.pop_front() { - let mut it = match fs::read_dir(&target).await { + let mut it = match fs::read_dir(target).await { Ok(it) => it, Err(_) => continue, }; @@ -268,11 +268,11 @@ impl File { }; if meta.is_dir() { - dirs.push_front(Url::new(entry.path(), &target)); + dirs.push_front(Url::from(entry.path())); continue; } - task.target = Url::new(entry.path(), &target); + task.target = Url::from(entry.path()); task.length = meta.len(); self.sch.send(TaskOp::New(task.id, meta.len()))?; self.tx.send(FileOp::Delete(task.clone())).await?; diff --git a/core/src/which/which.rs b/core/src/which/which.rs index a18f7e6d..21dd4a3a 100644 --- a/core/src/which/which.rs +++ b/core/src/which/which.rs @@ -23,7 +23,7 @@ impl Which { self.layer = layer; self.times = 1; self.cands = - KEYMAP.get(layer).iter().filter(|s| s.on.len() > 1 && s.on[0] == *key).cloned().collect(); + KEYMAP.get(layer).iter().filter(|s| s.on.len() > 1 && &s.on[0] == key).cloned().collect(); self.switch(true); true } diff --git a/shared/src/lib.rs b/shared/src/lib.rs index db1e9db0..3b12b26f 100644 --- a/shared/src/lib.rs +++ b/shared/src/lib.rs @@ -7,7 +7,6 @@ mod fns; mod fs; mod mime; mod ro_cell; -mod stream; mod term; mod throttle; mod time; @@ -20,7 +19,6 @@ pub use fns::*; pub use fs::*; pub use mime::*; pub use ro_cell::*; -pub use stream::*; pub use term::*; pub use throttle::*; pub use time::*; diff --git a/shared/src/stream.rs b/shared/src/stream.rs deleted file mode 100644 index 1c72dd9b..00000000 --- a/shared/src/stream.rs +++ /dev/null @@ -1,69 +0,0 @@ -use std::{mem, pin::Pin, task::{Context, Poll}, time::Duration}; - -use futures::{FutureExt, Stream, StreamExt}; -use tokio::time::{sleep, Instant, Sleep}; - -pub struct StreamBuf -where - S: Stream, -{ - stream: S, - interval: Duration, - - sleep: Sleep, - pending: Vec, -} - -impl Unpin for StreamBuf where S: Stream {} - -impl StreamBuf -where - S: Stream + Unpin, -{ - pub fn new(stream: S, interval: Duration) -> StreamBuf { - Self { stream, interval, sleep: sleep(Duration::ZERO), pending: Default::default() } - } - - pub fn flush(&mut self) { - let sleep = unsafe { Pin::new_unchecked(&mut self.sleep) }; - sleep.reset(Instant::now()); - } -} - -impl Stream for StreamBuf -where - S: Stream + Unpin, -{ - type Item = Vec; - - fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - let (mut stream, interval, mut sleep, pending) = unsafe { - let me = self.get_unchecked_mut(); - (Pin::new(&mut me.stream), me.interval, Pin::new_unchecked(&mut me.sleep), &mut me.pending) - }; - - while let Poll::Ready(next) = stream.poll_next_unpin(cx) { - match next { - Some(next) => pending.push(next), - None => { - if pending.is_empty() { - return Poll::Ready(None); - } - break; - } - } - } - - if pending.is_empty() { - return Poll::Pending; - } - - match sleep.poll_unpin(cx) { - Poll::Ready(_) => { - sleep.reset(Instant::now() + interval); - Poll::Ready(Some(mem::take(pending))) - } - Poll::Pending => Poll::Pending, - } - } -} diff --git a/shared/src/url.rs b/shared/src/url.rs index 4cff2a51..39c317f5 100644 --- a/shared/src/url.rs +++ b/shared/src/url.rs @@ -9,7 +9,7 @@ pub struct Url { #[derive(Clone, Copy, Debug, Default, Eq, Hash, Ord, PartialEq, PartialOrd)] pub enum UrlScheme { #[default] - None, + Regular, Search, Archive, } @@ -62,14 +62,26 @@ impl AsRef for Url { impl Url { #[inline] - pub fn new(url: impl Into, ctx: &Url) -> Self { - let mut url: Self = url.into(); - url.scheme = ctx.scheme; - url + pub fn join(&self, path: impl AsRef) -> Self { + let url = Self::from(self.path.join(path)); + match self.scheme { + UrlScheme::Regular => url, + UrlScheme::Search => url, + UrlScheme::Archive => url.into_archive(), + } } #[inline] - pub fn join(&self, path: impl AsRef) -> Self { Self::new(self.path.join(path), self) } + pub fn parent_url(&self) -> Option { + self.path.parent().map(|p| { + let url = Self::from(p); + match self.scheme { + UrlScheme::Regular => url, + UrlScheme::Search => url, + UrlScheme::Archive => url, + } + }) + } #[inline] pub fn strip_prefix(&self, base: impl AsRef) -> Option<&Path> { @@ -83,18 +95,42 @@ impl Url { impl Url { // --- Scheme #[inline] - pub fn is_none(&self) -> bool { self.scheme == UrlScheme::None } + pub fn is_regular(&self) -> bool { self.scheme == UrlScheme::Regular } + + #[inline] + pub fn to_regular(&self) -> Self { self.clone().into_regular() } + + #[inline] + pub fn into_regular(mut self) -> Self { + self.scheme = UrlScheme::Regular; + self + } #[inline] pub fn is_search(&self) -> bool { self.scheme == UrlScheme::Search } + #[inline] + pub fn to_search(&self) -> Self { self.clone().into_search() } + + #[inline] + pub fn into_search(mut self) -> Self { + self.scheme = UrlScheme::Search; + self + } + #[inline] pub fn is_archive(&self) -> bool { self.scheme == UrlScheme::Archive } + #[inline] + pub fn to_archive(&self) -> Self { self.clone().into_archive() } + + #[inline] + pub fn into_archive(mut self) -> Self { + self.scheme = UrlScheme::Archive; + self + } + // --- Path #[inline] pub fn set_path(&mut self, path: PathBuf) { self.path = path; } - - #[inline] - pub fn parent_url(&self) -> Option { self.path.parent().map(|p| Self::new(p, self)) } }