Properly implement watches for Basalt

examples/basalt.rs

@@ -12,6 +12,8 @@ use structopt::StructOpt;
 use sodiumoxide::crypto::auth;
 use sodiumoxide::crypto::sign::ed25519;
 
+use tokio::sync::watch;
+
 use netapp::endpoint::*;
 use netapp::peering::basalt::*;
 use netapp::proto::*;
@@ -126,18 +128,18 @@ async fn main() {
 	let watch_cancel = netapp::util::watch_ctrl_c();
 
 	tokio::join!(
-		example.clone().sampling_loop(),
+		example.clone().sampling_loop(watch_cancel.clone()),
 		example
 			.netapp
 			.clone()
-			.listen(listen_addr, public_addr, watch_cancel),
+			.listen(listen_addr, public_addr, watch_cancel.clone()),
-		example.basalt.clone().run(),
+		example.basalt.clone().run(watch_cancel.clone()),
 	);
 }
 
 impl Example {
-	async fn sampling_loop(self: Arc<Self>) {
+	async fn sampling_loop(self: Arc<Self>, must_exit: watch::Receiver<bool>) {
-		loop {
+		while !*must_exit.borrow() {
 			tokio::time::sleep(Duration::from_secs(10)).await;
 
 			let peers = self.basalt.sample(10);

src/peering/basalt.rs

@@ -11,6 +11,8 @@ use serde::{Deserialize, Serialize};
 
 use sodiumoxide::crypto::hash;
 
+use tokio::sync::watch;
+
 use crate::endpoint::*;
 use crate::netapp::*;
 use crate::proto::*;
@@ -303,18 +305,19 @@ impl Basalt {
 			.collect::<Vec<_>>()
 	}
 
-	pub async fn run(self: Arc<Self>) {
+	pub async fn run(self: Arc<Self>, must_exit: watch::Receiver<bool>) {
 		for peer in self.bootstrap_peers.iter() {
 			tokio::spawn(self.clone().try_connect(*peer));
 		}
 
-		let pushpull_loop = self.clone().run_pushpull_loop();
+		tokio::join!(
-		let reset_loop = self.run_reset_loop();
+			self.clone().run_pushpull_loop(must_exit.clone()),
-		tokio::join!(pushpull_loop, reset_loop);
+			self.clone().run_reset_loop(must_exit.clone()),
+		);
 	}
 
-	async fn run_pushpull_loop(self: Arc<Self>) {
+	async fn run_pushpull_loop(self: Arc<Self>, must_exit: watch::Receiver<bool>) {
-		loop {
+		while !*must_exit.borrow() {
 			tokio::time::sleep(self.param.exchange_interval).await;
 
 			let peers = self.view.read().unwrap().sample(2);
@@ -360,8 +363,8 @@ impl Basalt {
 		}
 	}
 
-	async fn run_reset_loop(self: Arc<Self>) {
+	async fn run_reset_loop(self: Arc<Self>, must_exit: watch::Receiver<bool>) {
-		loop {
+		while !*must_exit.borrow() {
 			tokio::time::sleep(self.param.reset_interval).await;
 
 			{

src/peering/fullmesh.rs

@@ -178,12 +178,12 @@ impl FullMeshPeeringStrategy {
 			// 1. Read current state: get list of connected peers (ping them)
 			let (to_ping, to_retry) = {
 				let known_hosts = self.known_hosts.read().unwrap();
-				debug!("known_hosts: {} peers", known_hosts.list.len());
+				trace!("known_hosts: {} peers", known_hosts.list.len());
 
 				let mut to_ping = vec![];
 				let mut to_retry = vec![];
 				for (id, info) in known_hosts.list.iter() {
-					debug!("{}, {:?}", hex::encode(id), info);
+					trace!("{}, {:?}", hex::encode(id), info);
 					match info.state {
 						PeerConnState::Connected => {
 							let must_ping = match info.last_seen {