From 1a079bf05c17b4e3969b621747ed60f1a4d2b4a9 Mon Sep 17 00:00:00 2001 From: sxyazi Date: Fri, 5 Apr 2024 01:54:37 +0800 Subject: [PATCH] perf: re-implement file watcher in an async way --- yazi-core/src/lib.rs | 7 +- .../src/manager/commands/update_files.rs | 4 +- .../src/manager/commands/update_mimetype.rs | 4 +- yazi-core/src/manager/watcher.rs | 145 ++++++++++-------- 4 files changed, 89 insertions(+), 71 deletions(-) diff --git a/yazi-core/src/lib.rs b/yazi-core/src/lib.rs index e7ff1a7e..43ab4f08 100644 --- a/yazi-core/src/lib.rs +++ b/yazi-core/src/lib.rs @@ -22,4 +22,9 @@ pub mod which; pub use clipboard::*; pub use step::*; -pub fn init() { CLIPBOARD.with(Default::default); } +pub fn init() { + CLIPBOARD.with(Default::default); + + manager::WATCHED.with(Default::default); + manager::LINKED.with(Default::default); +} diff --git a/yazi-core/src/manager/commands/update_files.rs b/yazi-core/src/manager/commands/update_files.rs index a7fc1adb..690d3e74 100644 --- a/yazi-core/src/manager/commands/update_files.rs +++ b/yazi-core/src/manager/commands/update_files.rs @@ -3,7 +3,7 @@ use std::borrow::Cow; use yazi_proxy::ManagerProxy; use yazi_shared::{event::Cmd, fs::FilesOp, render}; -use crate::{folder::Folder, manager::Manager, tab::Tab, tasks::Tasks}; +use crate::{folder::Folder, manager::{Manager, LINKED}, tab::Tab, tasks::Tasks}; pub struct Opt { op: FilesOp, @@ -22,7 +22,7 @@ impl Manager { }; let mut ops = vec![opt.op]; - for u in self.watcher.linked.read().from_dir(ops[0].url()) { + for u in LINKED.read().from_dir(ops[0].url()) { ops.push(ops[0].chroot(u)); } diff --git a/yazi-core/src/manager/commands/update_mimetype.rs b/yazi-core/src/manager/commands/update_mimetype.rs index e069fcf5..28f3f523 100644 --- a/yazi-core/src/manager/commands/update_mimetype.rs +++ b/yazi-core/src/manager/commands/update_mimetype.rs @@ -3,7 +3,7 @@ use std::collections::HashMap; use yazi_dds::ValueSendable; use yazi_shared::{event::Cmd, fs::Url, render}; -use crate::{manager::Manager, tasks::Tasks}; +use crate::{manager::{Manager, LINKED}, tasks::Tasks}; pub struct Opt { data: ValueSendable, @@ -23,7 +23,7 @@ impl Manager { return; }; - let linked = self.watcher.linked.read(); + let linked = LINKED.read(); let updates = opt .data .into_table_string() diff --git a/yazi-core/src/manager/watcher.rs b/yazi-core/src/manager/watcher.rs index d2f8837f..2b8e8708 100644 --- a/yazi-core/src/manager/watcher.rs +++ b/yazi-core/src/manager/watcher.rs @@ -1,68 +1,67 @@ -use std::{collections::{HashMap, HashSet}, sync::Arc, time::{Duration, SystemTime}}; +use std::{collections::{HashMap, HashSet}, time::{Duration, SystemTime}}; use anyhow::Result; use notify::{event::{MetadataKind, ModifyKind}, EventKind, RecommendedWatcher, RecursiveMode, Watcher as _Watcher}; use parking_lot::RwLock; -use tokio::{fs, pin, sync::mpsc::{self, UnboundedReceiver}}; +use tokio::{fs, pin, sync::{mpsc::{self, UnboundedReceiver}, watch}}; use tokio_stream::{wrappers::UnboundedReceiverStream, StreamExt}; use tracing::error; use yazi_plugin::isolate; use yazi_proxy::WATCHER; -use yazi_shared::fs::{File, FilesOp, Url}; +use yazi_shared::{fs::{File, FilesOp, Url}, RoCell}; use super::Linked; use crate::folder::{Files, Folder}; +pub(crate) static WATCHED: RoCell>> = RoCell::new(); +pub static LINKED: RoCell> = RoCell::new(); + pub struct Watcher { - watcher: RecommendedWatcher, - watched: Arc>>, - pub linked: Arc>, + tx: watch::Sender<(HashSet, HashSet)>, } impl Watcher { pub(super) fn serve() -> Self { - let (tx, rx) = mpsc::unbounded_channel(); - let watcher = RecommendedWatcher::new( - { - let tx = tx.clone(); - move |res: Result| { - let Ok(event) = res else { return }; + let (in_tx, in_rx) = watch::channel(Default::default()); + let (out_tx, out_rx) = mpsc::unbounded_channel(); - match event.kind { - EventKind::Create(_) => {} - EventKind::Modify(kind) => match kind { - ModifyKind::Data(_) => {} - ModifyKind::Metadata(md) => match md { - MetadataKind::WriteTime => {} - MetadataKind::Permissions => {} - MetadataKind::Ownership => {} - _ => return, - }, - ModifyKind::Name(_) => {} + let watcher = RecommendedWatcher::new( + move |res: Result| { + let Ok(event) = res else { return }; + + match event.kind { + EventKind::Create(_) => {} + EventKind::Modify(kind) => match kind { + ModifyKind::Data(_) => {} + ModifyKind::Metadata(md) => match md { + MetadataKind::WriteTime => {} + MetadataKind::Permissions => {} + MetadataKind::Ownership => {} _ => return, }, - EventKind::Remove(_) => {} + ModifyKind::Name(_) => {} _ => return, - } + }, + EventKind::Remove(_) => {} + _ => return, + } - for path in event.paths { - tx.send(Url::from(path)).ok(); - } + for path in event.paths { + out_tx.send(Url::from(path)).ok(); } }, Default::default(), ); - let instance = - Self { watcher: watcher.unwrap(), watched: Default::default(), linked: Default::default() }; - tokio::spawn(Self::on_changed(rx)); - instance + tokio::spawn(Self::on_in(in_rx, watcher.unwrap())); + tokio::spawn(Self::on_out(out_rx)); + Self { tx: in_tx } } pub(super) fn watch(&mut self, mut new: HashSet<&Url>) { new.retain(|&u| u.is_regular()); let (to_unwatch, to_watch): (HashSet<_>, HashSet<_>) = { - let guard = self.watched.read(); + let guard = WATCHED.read(); let old: HashSet<_> = guard.iter().collect(); ( old.difference(&new).map(|&x| x.clone()).collect(), @@ -70,17 +69,7 @@ impl Watcher { ) }; - for u in to_unwatch { - self.watcher.unwatch(&u).ok(); - } - for u in to_watch { - if self.watcher.watch(&u, RecursiveMode::NonRecursive).is_err() { - new.remove(&u); - } - } - - *self.watched.write() = new.into_iter().cloned().collect(); - self.sync_linked(); + self.tx.send((to_unwatch, to_watch)).ok(); } pub(super) fn trigger_dirs(&self, folders: &[&Folder]) { @@ -114,33 +103,33 @@ impl Watcher { }); } - fn sync_linked(&self) { - let mut new = self.watched.read().clone(); - self.linked.write().retain(|k, _| new.remove(k)); - - let watched = self.watched.clone(); - let linked = self.linked.clone(); - macro_rules! go { - ($todo:expr) => { - for from in $todo { - match fs::canonicalize(&from).await { - Ok(to) if to != *from && watched.read().contains(&from) => { - linked.write().insert(from, Url::from(to)); - } - _ => {} + async fn on_in( + mut rx: watch::Receiver<(HashSet, HashSet)>, + mut watcher: RecommendedWatcher, + ) { + loop { + { + let (ref to_unwatch, ref to_watch) = *rx.borrow_and_update(); + for u in to_unwatch { + if watcher.unwatch(u).is_ok() { + WATCHED.write().remove(u); } } - }; - } + for u in to_watch { + if watcher.watch(u, RecursiveMode::NonRecursive).is_ok() { + WATCHED.write().insert(u.clone()); + } + } + } - tokio::spawn(async move { - let old: Vec<_> = linked.read().keys().cloned().collect(); - go!(new); - go!(old); - }); + Self::sync_linked(); + if rx.changed().await.is_err() { + break; + } + } } - async fn on_changed(rx: UnboundedReceiver) { + async fn on_out(rx: UnboundedReceiver) { // TODO: revert this once a new notification is implemented let rx = UnboundedReceiverStream::new(rx).chunks_timeout(1000, Duration::from_millis(50)); pin!(rx); @@ -171,4 +160,28 @@ impl Watcher { } } } + + fn sync_linked() { + let mut new = WATCHED.read().clone(); + LINKED.write().retain(|k, _| new.remove(k)); + + macro_rules! go { + ($todo:expr) => { + for from in $todo { + match fs::canonicalize(&from).await { + Ok(to) if to != *from && WATCHED.read().contains(&from) => { + LINKED.write().insert(from, Url::from(to)); + } + _ => {} + } + } + }; + } + + tokio::spawn(async move { + let old: Vec<_> = LINKED.read().keys().cloned().collect(); + go!(new); + go!(old); + }); + } }