perf: load large folders in chunks (#117)

This commit is contained in:
三咲雅 · Misaki Masa 2023-09-07 00:12:59 +08:00 committed by GitHub
parent eee68a99b4
commit d3f0c9c059
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
20 changed files with 270 additions and 262 deletions

60
Cargo.lock generated
View file

@ -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",
]

View file

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

View file

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

View file

@ -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<StreamBuf<UnboundedReceiverStream<Url>>> {
pub fn fd(opt: FdOpt) -> Result<UnboundedReceiver<File>> {
let mut child = Command::new("fd")
.arg("--base-directory")
.arg(&opt.cwd)
@ -28,11 +29,12 @@ pub fn fd(opt: FdOpt) -> Result<StreamBuf<UnboundedReceiverStream<Url>>> {
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();
});

View file

@ -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<StreamBuf<UnboundedReceiverStream<Url>>> {
pub fn rg(opt: RgOpt) -> Result<UnboundedReceiver<File>> {
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<StreamBuf<UnboundedReceiverStream<Url>>> {
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();
});

View file

@ -17,12 +17,11 @@ pub fn zoxide(opt: ZoxideOpt) -> Result<Receiver<Result<Url>>> {
.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;
}
}

View file

@ -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<File>,
hidden: Vec<File>,
items: Vec<File>,
hidden: Vec<File>,
version: u64,
sizes: BTreeMap<Url, u64>,
selected: BTreeSet<Url>,
@ -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<Url>) -> Vec<File> {
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<Vec<File>> {
pub async fn from_dir(url: &Url) -> Result<UnboundedReceiver<File>> {
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>) -> 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<usize>, state: Option<bool>) -> bool {
@ -111,26 +122,22 @@ impl Files {
applied
}
pub fn update_read(&mut self, mut items: Vec<File>) -> bool {
pub fn update_full(&mut self, mut items: Vec<File>) -> 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<Url, u64>) -> 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<File>) -> bool {
pub fn update_part(&mut self, version: u64, items: Vec<File>) -> 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<Url, u64>) -> 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<usize> { self.iter().position(|f| f.url == *url) }
pub fn position(&self, url: &Url) -> Option<usize> { self.iter().position(|f| &f.url == url) }
#[inline]
pub fn duplicate(&self, idx: usize) -> Option<File> { self.items.get(idx).cloned() }

View file

@ -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<File>),
Full(Url, Vec<File>),
Part(Url, u64, Vec<File>),
Size(Url, BTreeMap<Url, u64>),
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
}
}

View file

@ -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(_)) {

View file

@ -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!(),
};

View file

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

View file

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

View file

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

View file

@ -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::<BTreeSet<_>>();
@ -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<UnboundedReceiverStream<Url>>,
watched: Arc<RwLock<IndexMap<Url, Option<Url>>>>,
) {
async fn changed(rx: UnboundedReceiver<Url>, watched: Arc<RwLock<IndexMap<Url, Option<Url>>>>) {
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::<BTreeSet<_>>() {
@ -163,34 +165,35 @@ impl Watcher {
}
async fn dir_changed(url: &Url, watched: Arc<RwLock<IndexMap<Url, Option<Url>>>>) {
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::<Vec<_>>();
.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<File> {
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)));
}
}

View file

@ -157,7 +157,7 @@ impl Tasks {
pub fn file_cut(&self, src: &HashSet<Url>, 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<Url>, 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);

View file

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

View file

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

View file

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

View file

@ -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<S>
where
S: Stream,
{
stream: S,
interval: Duration,
sleep: Sleep,
pending: Vec<S::Item>,
}
impl<S> Unpin for StreamBuf<S> where S: Stream {}
impl<S> StreamBuf<S>
where
S: Stream + Unpin,
{
pub fn new(stream: S, interval: Duration) -> StreamBuf<S> {
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<S> Stream for StreamBuf<S>
where
S: Stream + Unpin,
{
type Item = Vec<S::Item>;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
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,
}
}
}

View file

@ -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<OsStr> for Url {
impl Url {
#[inline]
pub fn new(url: impl Into<Url>, ctx: &Url) -> Self {
let mut url: Self = url.into();
url.scheme = ctx.scheme;
url
pub fn join(&self, path: impl AsRef<Path>) -> 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<Path>) -> Self { Self::new(self.path.join(path), self) }
pub fn parent_url(&self) -> Option<Url> {
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<Path>) -> 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<Url> { self.path.parent().map(|p| Self::new(p, self)) }
}