Ability to configure ping timeout interval

Cargo.lock

@@ -428,7 +428,7 @@ dependencies = [
 
 [[package]]
 name = "netapp"
-version = "0.5.0"
+version = "0.5.2"
 dependencies = [
  "arc-swap",
  "async-trait",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "netapp"
-version = "0.5.0"
+version = "0.5.2"
 authors = ["Alex Auvolat <alex@adnab.me>"]
 edition = "2018"
 license-file = "LICENSE"

src/peering/fullmesh.rs

@@ -25,9 +25,10 @@ const CONN_RETRY_INTERVAL: Duration = Duration::from_secs(30);
 const CONN_MAX_RETRIES: usize = 10;
 const PING_INTERVAL: Duration = Duration::from_secs(15);
 const LOOP_DELAY: Duration = Duration::from_secs(1);
-const PING_TIMEOUT: Duration = Duration::from_secs(10);
 const FAILED_PING_THRESHOLD: usize = 4;
 
+const DEFAULT_PING_TIMEOUT_MILLIS: u64 = 10_000;
+
 // -- Protocol messages --
 
 #[derive(Serialize, Deserialize)]
@@ -184,6 +185,8 @@ pub struct FullMeshPeeringStrategy {
 	next_ping_id: AtomicU64,
 	ping_endpoint: Arc<Endpoint<PingMessage, Self>>,
 	peer_list_endpoint: Arc<Endpoint<PeerListMessage, Self>>,
+
+	ping_timeout_millis: AtomicU64,
 }
 
 impl FullMeshPeeringStrategy {
@@ -220,6 +223,7 @@ impl FullMeshPeeringStrategy {
 			next_ping_id: AtomicU64::new(42),
 			ping_endpoint: netapp.endpoint("__netapp/peering/fullmesh.rs/Ping".into()),
 			peer_list_endpoint: netapp.endpoint("__netapp/peering/fullmesh.rs/PeerList".into()),
+			ping_timeout_millis: DEFAULT_PING_TIMEOUT_MILLIS.into(),
 		});
 
 		strat.update_public_peer_list(&strat.known_hosts.read().unwrap());
@@ -331,6 +335,12 @@ impl FullMeshPeeringStrategy {
 		self.public_peer_list.load_full()
 	}
 
+	/// Set the timeout for ping messages, in milliseconds
+	pub fn set_ping_timeout_millis(&self, timeout: u64) {
+		self.ping_timeout_millis
+			.store(timeout, atomic::Ordering::Relaxed);
+	}
+
 	// -- internal stuff --
 
 	fn update_public_peer_list(&self, known_hosts: &KnownHosts) {
@@ -372,6 +382,8 @@ impl FullMeshPeeringStrategy {
 		let peer_list_hash = self.known_hosts.read().unwrap().hash;
 		let ping_id = self.next_ping_id.fetch_add(1u64, atomic::Ordering::Relaxed);
 		let ping_time = Instant::now();
+		let ping_timeout =
+			Duration::from_millis(self.ping_timeout_millis.load(atomic::Ordering::Relaxed));
 		let ping_msg = PingMessage {
 			id: ping_id,
 			peer_list_hash,
@@ -385,7 +397,7 @@ impl FullMeshPeeringStrategy {
 		);
 		let ping_response = select! {
 			r = self.ping_endpoint.call(&id, ping_msg, PRIO_HIGH) => r,
-			_ = tokio::time::sleep(PING_TIMEOUT) => Err(Error::Message("Ping timeout".into())),
+			_ = tokio::time::sleep(ping_timeout) => Err(Error::Message("Ping timeout".into())),
 		};
 
 		match ping_response {

src/util.rs

@@ -1,5 +1,4 @@
 use std::net::SocketAddr;
-use std::net::ToSocketAddrs;
 
 use log::info;
 use serde::Serialize;
@@ -71,6 +70,8 @@ pub fn parse_peer_addr(peer: &str) -> Option<(NodeID, SocketAddr)> {
 /// Parse and resolve a peer's address including public key, written in the format:
 /// `<public key hex>@<ip or hostname>:<port>`
 pub fn parse_and_resolve_peer_addr(peer: &str) -> Option<(NodeID, Vec<SocketAddr>)> {
+	use std::net::ToSocketAddrs;
+
 	let delim = peer.find('@')?;
 	let (key, host) = peer.split_at(delim);
 	let pubkey = NodeID::from_slice(&hex::decode(&key).ok()?)?;
@@ -80,3 +81,18 @@ pub fn parse_and_resolve_peer_addr(peer: &str) -> Option<(NodeID, Vec<SocketAddr
 	}
 	Some((pubkey, hosts))
 }
+
+/// async version of parse_and_resolve_peer_addr
+pub async fn parse_and_resolve_peer_addr_async(peer: &str) -> Option<(NodeID, Vec<SocketAddr>)> {
+	let delim = peer.find('@')?;
+	let (key, host) = peer.split_at(delim);
+	let pubkey = NodeID::from_slice(&hex::decode(&key).ok()?)?;
+	let hosts = tokio::net::lookup_host(&host[1..])
+		.await
+		.ok()?
+		.collect::<Vec<_>>();
+	if hosts.is_empty() {
+		return None;
+	}
+	Some((pubkey, hosts))
+}