X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=src%2Fdatastore.rs;h=1ae9dd95ad11c301426e2b697cd6af20ecafbdb0;hb=f44ddf483a9e7f8cc136f6ac64de3e135429fd59;hp=829fe0e1e7c7e9871b91af4da328636bc50a6e89;hpb=2590cec11b1c75ecd959267f07596219b40c2daa;p=dnsseed-rust

diff --git a/src/datastore.rs b/src/datastore.rs
index 829fe0e..1ae9dd9 100644
--- a/src/datastore.rs
+++ b/src/datastore.rs
@@ -19,6 +19,7 @@ use regex::Regex;
 use crate::bgp_client::BGPClient;
 
 pub const SECS_PER_SCAN_RESULTS: u64 = 15;
+const MAX_CONNS_PER_SEC_PER_STATUS: u64 = 30;
 
 #[derive(Clone, Copy, Hash, PartialEq, Eq)]
 pub enum AddressState {
@@ -124,6 +125,7 @@ struct Node {
 	last_good: Instant, // Ignored unless state is Good or WasGood
 	last_services: u64,
 	state: AddressState,
+	queued: bool,
 }
 
 /// Essentially SocketAddr but without a traffic class or scope
@@ -305,6 +307,7 @@ impl Store {
 					last_services,
 					last_update: Instant::now(),
 					last_good: Instant::now(),
+					queued: true,
 				};
 				if node.state == AddressState::Good {
 					for i in 0..64 {
@@ -362,6 +365,7 @@ impl Store {
 						last_services: 0,
 						last_update: cur_time,
 						last_good: cur_time,
+						queued: true,
 					});
 					nodes.state_next_scan[AddressState::Untested.to_num() as usize].push((cur_time, addr.into()));
 					res += 1;
@@ -393,6 +397,7 @@ impl Store {
 			last_services: 0,
 			last_update: now,
 			last_good: now,
+			queued: false,
 		});
 		let ret = state_ref.state;
 		if (state_ref.state == AddressState::Good || state_ref.state == AddressState::WasGood)
@@ -405,7 +410,10 @@ impl Store {
 				}
 			}
 			state_ref.last_services = 0;
-			nodes.state_next_scan[AddressState::WasGood.to_num() as usize].push((now, addr));
+			if !state_ref.queued {
+				nodes.state_next_scan[AddressState::WasGood.to_num() as usize].push((now, addr));
+				state_ref.queued = true;
+			}
 		} else {
 			state_ref.state = state;
 			if state == AddressState::Good {
@@ -419,7 +427,10 @@ impl Store {
 				state_ref.last_services = services;
 				state_ref.last_good = now;
 			}
-			nodes.state_next_scan[state.to_num() as usize].push((now, addr));
+			if !state_ref.queued {
+				nodes.state_next_scan[state.to_num() as usize].push((now, addr));
+				state_ref.queued = true;
+			}
 		}
 		state_ref.last_update = now;
 		ret
@@ -595,15 +606,18 @@ impl Store {
 		let cur_time = Instant::now();
 
 		{
-			let mut nodes = self.nodes.write().unwrap();
+			let mut nodes_lock = self.nodes.write().unwrap();
+			let nodes = nodes_lock.borrow_mut();
 			for (idx, state_nodes) in nodes.state_next_scan.iter_mut().enumerate() {
 				let rescan_interval = cmp::max(self.get_u64(U64Setting::RescanInterval(AddressState::from_num(idx as u8).unwrap())), 1);
 				let cmp_time = cur_time - Duration::from_secs(rescan_interval);
-				let split_point = cmp::min(SECS_PER_SCAN_RESULTS * state_nodes.len() as u64 / rescan_interval,
+				let split_point = cmp::min(cmp::min(SECS_PER_SCAN_RESULTS * state_nodes.len() as u64 / rescan_interval,
+							SECS_PER_SCAN_RESULTS * MAX_CONNS_PER_SEC_PER_STATUS),
 						state_nodes.binary_search_by(|a| a.0.cmp(&cmp_time)).unwrap_or_else(|idx| idx) as u64);
 				let mut new_nodes = state_nodes.split_off(split_point as usize);
 				mem::swap(&mut new_nodes, state_nodes);
 				for (_, node) in new_nodes.drain(..) {
+					nodes.nodes_to_state.get_mut(&node).unwrap().queued = false;
 					res.push((&node).into());
 				}
 			}