diff --git a/easytier/src/dns/node.rs b/easytier/src/dns/node.rs index e3f2e230..a1733740 100644 --- a/easytier/src/dns/node.rs +++ b/easytier/src/dns/node.rs @@ -166,7 +166,7 @@ impl DnsNode { last_heartbeat = Instant::now(); } - _ = self.mgr.dirty.notified() => {} + _ = self.mgr.dirty.wait() => {} event = subscriber.recv() => { match event { diff --git a/easytier/src/dns/peer_mgr.rs b/easytier/src/dns/peer_mgr.rs index 78fa1ff7..e81d7708 100644 --- a/easytier/src/dns/peer_mgr.rs +++ b/easytier/src/dns/peer_mgr.rs @@ -75,7 +75,6 @@ impl DnsPeerMgrInner { pub async fn refresh(&self, peer_id: PeerId) { if peer_id == self.peer_mgr.my_peer_id() { self.dirty.mark(); - self.dirty.notify_one(); return; } @@ -91,25 +90,23 @@ impl DnsPeerMgrInner { return; } - self.dirty.mark(); + let mut invalidate = route.dns.is_empty(); - let invalidate = route.dns.is_empty() - || match self.fetch(peer_id).await { - Ok(info) => { - self.peers.insert(peer_id, info).await; - false - } + if !invalidate { + match self.fetch(peer_id).await { + Ok(info) => self.peers.insert(peer_id, info).await, Err(error) => { tracing::warn!(%peer_id, ?error, "failed to fetch dns export config from peer"); - true + invalidate = true; } }; + } if invalidate { self.peers.invalidate(&peer_id).await; } - self.dirty.notify_one(); + self.dirty.mark(); } #[instrument(skip(self), level = "trace", ret)] diff --git a/easytier/src/dns/server.rs b/easytier/src/dns/server.rs index c9827ca2..a61b9483 100644 --- a/easytier/src/dns/server.rs +++ b/easytier/src/dns/server.rs @@ -225,7 +225,7 @@ impl DnsServer { let reload_catalog = async { loop { - dirty.catalog.notified().await; + dirty.catalog.wait().await; if dirty.catalog.reset() { self.catalog.replace(self.mgr.catalog()).await; } @@ -235,7 +235,7 @@ impl DnsServer { let reload_addresses = async { loop { - dirty.addresses.notified().await; + dirty.addresses.wait().await; if dirty.addresses.reset() { if let Err(e) = self.reload_addresses(self.mgr.iter_addresses()).await { tracing::error!("failed to reload addresses: {:?}", e); @@ -248,7 +248,7 @@ impl DnsServer { let reload_listeners = async { loop { - dirty.listeners.notified().await; + dirty.listeners.wait().await; if dirty.listeners.reset() { if let Err(e) = self .reload_listeners(self.mgr.iter_listeners(), &mut runtime) diff --git a/easytier/src/dns/utils/dirty.rs b/easytier/src/dns/utils/dirty.rs index 7fa690b1..804b95b8 100644 --- a/easytier/src/dns/utils/dirty.rs +++ b/easytier/src/dns/utils/dirty.rs @@ -1,39 +1,32 @@ -use derive_more::Deref; -use std::sync::atomic::{AtomicBool, Ordering}; -use tokio::sync::Notify; +use tokio::sync::watch; -#[derive(Debug, Deref)] +#[derive(Debug)] pub struct DirtyFlag { - dirty: AtomicBool, - #[deref] - notify: Notify, + tx: watch::Sender, + rx: watch::Receiver, } impl DirtyFlag { pub fn new(value: bool) -> Self { - let notify = Notify::new(); - - if value { - notify.notify_one(); - } - - Self { - dirty: AtomicBool::new(value), - notify, - } + let (tx, rx) = watch::channel(value); + Self { tx, rx } } pub fn mark(&self) { - self.dirty.store(true, Ordering::Release); - self.notify.notify_one(); + self.tx.send(true).ok(); } pub fn peek(&self) -> bool { - self.dirty.load(Ordering::Acquire) + *self.tx.borrow() } pub fn reset(&self) -> bool { - self.dirty.swap(false, Ordering::Acquire) + self.tx.send_replace(false) + } + + pub async fn wait(&self) { + let mut rx = self.rx.clone(); + let _ = rx.wait_for(|v| *v).await; } }