diff --git a/p2p/src/peers.rs b/p2p/src/peers.rs index 3448c6dd..0f10a84e 100644 --- a/p2p/src/peers.rs +++ b/p2p/src/peers.rs @@ -77,6 +77,7 @@ impl Peers { last_banned: 0, ban_reason: ReasonForBan::None, last_connected: Utc::now().timestamp(), + last_attempt: Utc::now().timestamp(), }; if !enough_outbound || !peer.info.is_outbound() { debug!("Adding newly connected peer {}.", peer_data.addr); @@ -101,6 +102,7 @@ impl Peers { last_banned: Utc::now().timestamp(), ban_reason, last_connected: Utc::now().timestamp(), + last_attempt: Utc::now().timestamp(), }; debug!("Banning peer {}.", addr); self.save_peer(&peer_data) @@ -723,6 +725,7 @@ impl NetAdapter for Peers { last_banned: 0, ban_reason: ReasonForBan::None, last_connected: 0, + last_attempt: 0, }; to_save.push(peer); } diff --git a/p2p/src/store.rs b/p2p/src/store.rs index 96909aab..208cd5d8 100644 --- a/p2p/src/store.rs +++ b/p2p/src/store.rs @@ -56,6 +56,8 @@ pub struct PeerData { pub ban_reason: ReasonForBan, /// Time when we last connected to this peer. pub last_connected: i64, + /// Time when last connection attempt happened to this peer. + pub last_attempt: i64, } impl Writeable for PeerData { @@ -68,7 +70,8 @@ impl Writeable for PeerData { [write_u8, self.flags as u8], [write_i64, self.last_banned], [write_i32, self.ban_reason as i32], - [write_i64, self.last_connected] + [write_i64, self.last_connected], + [write_i64, self.last_attempt] ); Ok(()) } @@ -82,12 +85,10 @@ impl Readable for PeerData { let (fl, lb, br) = ser_multiread!(reader, read_u8, read_i64, read_i32); let lc = reader.read_i64(); - // this only works because each PeerData is read in its own vector and this - // is the last data element - let last_connected = match lc { - Err(_) => Utc::now().timestamp(), - Ok(lc) => lc, - }; + let last_connected = lc.unwrap_or_else(|_| Utc::now().timestamp()); + + let la = reader.read_i64(); + let last_attempt = la.unwrap_or_else(|_| 0); let user_agent = String::from_utf8(ua).map_err(|_| ser::Error::CorruptedData)?; let capabilities = Capabilities::from_bits_truncate(capab); @@ -98,10 +99,11 @@ impl Readable for PeerData { addr, capabilities, user_agent, - flags: flags, + flags, last_banned: lb, ban_reason, last_connected, + last_attempt, }), None => Err(ser::Error::CorruptedData), } @@ -188,6 +190,7 @@ impl PeerStore { /// Convenience method to load a peer data, update its status and save it /// back. If new state is Banned its last banned time will be updated too. + /// If new state is Defunct last connection attempt will be updated too. pub fn update_state(&self, peer_addr: PeerAddr, new_state: State) -> Result<(), Error> { let batch = self.db.batch()?; @@ -198,6 +201,8 @@ impl PeerStore { peer.flags = new_state; if new_state == State::Banned { peer.last_banned = Utc::now().timestamp(); + } else if new_state == State::Defunct { + peer.last_attempt = Utc::now().timestamp(); } batch.put_ser(&peer_key(peer_addr)[..], &peer)?; diff --git a/servers/src/grin/seed.rs b/servers/src/grin/seed.rs index 245ce139..deb42ec4 100644 --- a/servers/src/grin/seed.rs +++ b/servers/src/grin/seed.rs @@ -161,8 +161,8 @@ fn monitor_peers(peers: Arc, config: p2p::P2PConfig, tx: mpsc::Sende banned_count += 1; } } - p2p::State::Healthy => healthy.push(x.addr), - p2p::State::Defunct => defuncts.push(x.addr), + p2p::State::Healthy => healthy.push(x), + p2p::State::Defunct => defuncts.push(x), p2p::State::Unknown => unknown.push(x.addr), } total_count += 1; @@ -233,27 +233,42 @@ fn monitor_peers(peers: Arc, config: p2p::P2PConfig, tx: mpsc::Sende // as many nodes in our db are not publicly accessible let mut new_peers = vec![]; let max_peer_attempts = 128; - // check random 64 disconnected healthy peers. + let max_attempt_delay = Duration::hours(1).num_seconds(); + // check maximum 64 random disconnected healthy peers no more often than 1 hour per peer. for hpa in healthy .iter() - .filter(|p| peers.get_connected_peer(**p).is_none()) + .filter(|p| { + peers.get_connected_peer(p.addr).is_none() + && Utc::now().timestamp() - p.last_attempt >= max_attempt_delay + }) .choose_multiple(&mut thread_rng(), max_peer_attempts / 2) { - new_peers.push(hpa); + new_peers.push(&hpa.addr); } - // check random 32 unknown peers received from peer list request. + // always check min 32 (max 96, if there are no healthy) random unknown peers received from peer list request. + let req_unk_count = cmp::max( + max_peer_attempts / 2 - new_peers.len() + max_peer_attempts / 4, + max_peer_attempts / 4, + ); for upa in unknown .iter() - .choose_multiple(&mut thread_rng(), max_peer_attempts / 4) + .choose_multiple(&mut thread_rng(), req_unk_count) { new_peers.push(upa); } - // always check 32 random defunct peers. + debug!( + "monitor_peers: check {} healthy, {} unknown, {} defuncts", + new_peers.len() - req_unk_count, + req_unk_count, + max_peer_attempts - new_peers.len() + ); + // check min 32 (max 128, if there are no healthy and unknown) random defunct peers no more often than 1 hour per peer. for dpa in defuncts .iter() - .choose_multiple(&mut thread_rng(), max_peer_attempts / 4) + .filter(|p| Utc::now().timestamp() - p.last_attempt >= max_attempt_delay) + .choose_multiple(&mut thread_rng(), max_peer_attempts - new_peers.len()) { - new_peers.push(dpa); + new_peers.push(&dpa.addr); } // Only queue up connection attempts for candidate peers where we @@ -358,6 +373,9 @@ fn listen_for_addrs( .name("peer_connect".to_string()) .spawn(move || match p2p_c.connect(addr) { Ok(p) => { + if peers_c.enough_outbound_peers() { + return; + } // If peer advertizes PEER_LIST then ask it for more peers that support PEER_LIST. // We want to build a local db of possible peers to connect to. // We do not necessarily care (at this point in time) what other capabilities these peers support.