perf: re-implement file watcher in an async way

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

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

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

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<RwLock<HashSet<Url>>> = RoCell::new();
+pub static LINKED: RoCell<RwLock<Linked>> = RoCell::new();
+
 pub struct Watcher {
-	watcher:    RecommendedWatcher,
-	watched:    Arc<RwLock<HashSet<Url>>>,
-	pub linked: Arc<RwLock<Linked>>,
+	tx: watch::Sender<(HashSet<Url>, HashSet<Url>)>,
 }
 
 impl Watcher {
 	pub(super) fn serve() -> Self {
-		let (tx, rx) = mpsc::unbounded_channel();
-		let watcher = RecommendedWatcher::new(
-			{
-				let tx = tx.clone();
-				move |res: Result<notify::Event, notify::Error>| {
-					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<notify::Event, notify::Error>| {
+				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<Url>, HashSet<Url>)>,
+		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<Url>) {
+	async fn on_out(rx: UnboundedReceiver<Url>) {
 		// 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);
+		});
+	}
 }