X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=src%2Fmain.rs;h=ba326f77d65d76b965b433e6a652b4c4116f5e1f;hb=62770d56227d5458fd7502fedf065a80b6b16373;hp=b868570db764f9f702248adf3caeca973c4fe7be;hpb=5bd597ef34ea3546b95dfb83049dc9d34e03cd8e;p=dnsseed-rust

diff --git a/src/main.rs b/src/main.rs
index b868570..ba326f7 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -5,7 +5,7 @@ mod bgp_client;
 mod timeout_stream;
 mod datastore;
 
-use std::{cmp, env};
+use std::env;
 use std::collections::HashMap;
 use std::sync::{Arc, Mutex};
 use std::sync::atomic::{Ordering, AtomicBool};
@@ -37,6 +37,7 @@ static mut HEADER_MAP: Option<Box<Mutex<HashMap<sha256d::Hash, u64>>>> = None;
 static mut HEIGHT_MAP: Option<Box<Mutex<HashMap<u64, sha256d::Hash>>>> = None;
 static mut DATA_STORE: Option<Box<Store>> = None;
 static mut PRINTER: Option<Box<Printer>> = None;
+static mut TOR_PROXY: Option<SocketAddr> = None;
 pub static START_SHUTDOWN: AtomicBool = AtomicBool::new(false);
 static SCANNING: AtomicBool = AtomicBool::new(false);
 
@@ -77,7 +78,7 @@ pub fn scan_node(scan_time: Instant, node: SocketAddr, manual: bool) {
 	let peer = Delay::new(scan_time).then(move |_| {
 		printer.set_stat(Stat::NewConnection);
 		let timeout = store.get_u64(U64Setting::RunTimeout);
-		Peer::new(node.clone(), Duration::from_secs(timeout), printer)
+		Peer::new(node.clone(), unsafe { TOR_PROXY.as_ref().unwrap() }, Duration::from_secs(timeout), printer)
 	});
 	tokio::spawn(peer.and_then(move |(mut write, read)| {
 		TimeoutStream::new_timeout(read, scan_time + Duration::from_secs(store.get_u64(U64Setting::RunTimeout))).map_err(move |err| {
@@ -107,7 +108,7 @@ pub fn scan_node(scan_time: Instant, node: SocketAddr, manual: bool) {
 			}
 			state_lock.fail_reason = AddressState::TimeoutDuringRequest;
 			match msg {
-				NetworkMessage::Version(ver) => {
+				Some(NetworkMessage::Version(ver)) => {
 					if ver.start_height < 0 || ver.start_height as u64 > state_lock.request.0 + 1008*2 {
 						state_lock.fail_reason = AddressState::HighBlockCount;
 						return future::err(());
@@ -141,18 +142,18 @@ pub fn scan_node(scan_time: Instant, node: SocketAddr, manual: bool) {
 						return future::err(());
 					}
 				},
-				NetworkMessage::Verack => {
+				Some(NetworkMessage::Verack) => {
 					check_set_flag!(recvd_verack, "verack");
 					if let Err(_) = write.try_send(NetworkMessage::Ping(state_lock.pong_nonce)) {
 						return future::err(());
 					}
 				},
-				NetworkMessage::Ping(v) => {
+				Some(NetworkMessage::Ping(v)) => {
 					if let Err(_) = write.try_send(NetworkMessage::Pong(v)) {
 						return future::err(())
 					}
 				},
-				NetworkMessage::Pong(v) => {
+				Some(NetworkMessage::Pong(v)) => {
 					if v != state_lock.pong_nonce {
 						state_lock.fail_reason = AddressState::ProtocolViolation;
 						state_lock.msg = ("due to invalid pong nonce".to_string(), true);
@@ -163,7 +164,7 @@ pub fn scan_node(scan_time: Instant, node: SocketAddr, manual: bool) {
 						return future::err(());
 					}
 				},
-				NetworkMessage::Addr(addrs) => {
+				Some(NetworkMessage::Addr(addrs)) => {
 					if addrs.len() > 1000 {
 						state_lock.fail_reason = AddressState::ProtocolViolation;
 						state_lock.msg = (format!("due to oversized addr: {}", addrs.len()), true);
@@ -183,7 +184,7 @@ pub fn scan_node(scan_time: Instant, node: SocketAddr, manual: bool) {
 					}
 					unsafe { DATA_STORE.as_ref().unwrap() }.add_fresh_nodes(&addrs);
 				},
-				NetworkMessage::Block(block) => {
+				Some(NetworkMessage::Block(block)) => {
 					if block != state_lock.request.2 {
 						state_lock.fail_reason = AddressState::ProtocolViolation;
 						state_lock.msg = ("due to bad block".to_string(), true);
@@ -192,7 +193,7 @@ pub fn scan_node(scan_time: Instant, node: SocketAddr, manual: bool) {
 					check_set_flag!(recvd_block, "block");
 					return future::err(());
 				},
-				NetworkMessage::Inv(invs) => {
+				Some(NetworkMessage::Inv(invs)) => {
 					for inv in invs {
 						if inv.inv_type == InvType::Transaction {
 							state_lock.fail_reason = AddressState::EvilNode;
@@ -201,7 +202,7 @@ pub fn scan_node(scan_time: Instant, node: SocketAddr, manual: bool) {
 						}
 					}
 				},
-				NetworkMessage::Tx(_) => {
+				Some(NetworkMessage::Tx(_)) => {
 					state_lock.fail_reason = AddressState::EvilNode;
 					state_lock.msg = ("due to unrequested transaction".to_string(), true);
 					return future::err(());
@@ -278,17 +279,19 @@ fn scan_net() {
 		let printer = unsafe { PRINTER.as_ref().unwrap() };
 		let store = unsafe { DATA_STORE.as_ref().unwrap() };
 
+		let start_time = Instant::now();
 		let mut scan_nodes = store.get_next_scan_nodes();
 		printer.add_line(format!("Got {} addresses to scan", scan_nodes.len()), false);
-		let per_iter_time = Duration::from_millis(1000 / store.get_u64(U64Setting::ConnsPerSec));
-		let start_time = Instant::now();
-		let mut iter_time = start_time;
+		if !scan_nodes.is_empty() {
+			let per_iter_time = Duration::from_millis(datastore::SECS_PER_SCAN_RESULTS * 1000 / scan_nodes.len() as u64);
+			let mut iter_time = start_time;
 
-		for node in scan_nodes.drain(..) {
-			scan_node(iter_time, node, false);
-			iter_time += per_iter_time;
+			for node in scan_nodes.drain(..) {
+				scan_node(iter_time, node, false);
+				iter_time += per_iter_time;
+			}
 		}
-		Delay::new(cmp::max(iter_time, start_time + Duration::from_secs(1))).then(|_| {
+		Delay::new(start_time + Duration::from_secs(datastore::SECS_PER_SCAN_RESULTS)).then(move |_| {
 			if !START_SHUTDOWN.load(Ordering::Relaxed) {
 				scan_net();
 			}
@@ -299,7 +302,7 @@ fn scan_net() {
 
 fn make_trusted_conn(trusted_sockaddr: SocketAddr, bgp_client: Arc<BGPClient>) {
 	let printer = unsafe { PRINTER.as_ref().unwrap() };
-	let trusted_peer = Peer::new(trusted_sockaddr.clone(), Duration::from_secs(600), printer);
+	let trusted_peer = Peer::new(trusted_sockaddr.clone(), unsafe { TOR_PROXY.as_ref().unwrap() }, Duration::from_secs(600), printer);
 	let bgp_reload = Arc::clone(&bgp_client);
 	tokio::spawn(trusted_peer.and_then(move |(mut trusted_write, trusted_read)| {
 		printer.add_line("Connected to local peer".to_string(), false);
@@ -309,13 +312,13 @@ fn make_trusted_conn(trusted_sockaddr: SocketAddr, bgp_client: Arc<BGPClient>) {
 				return future::err(());
 			}
 			match msg {
-				NetworkMessage::Version(ver) => {
+				Some(NetworkMessage::Version(ver)) => {
 					if let Err(_) = trusted_write.try_send(NetworkMessage::Verack) {
 						return future::err(())
 					}
 					starting_height = ver.start_height;
 				},
-				NetworkMessage::Verack => {
+				Some(NetworkMessage::Verack) => {
 					if let Err(_) = trusted_write.try_send(NetworkMessage::SendHeaders) {
 						return future::err(());
 					}
@@ -330,10 +333,10 @@ fn make_trusted_conn(trusted_sockaddr: SocketAddr, bgp_client: Arc<BGPClient>) {
 						return future::err(());
 					}
 				},
-				NetworkMessage::Addr(addrs) => {
+				Some(NetworkMessage::Addr(addrs)) => {
 					unsafe { DATA_STORE.as_ref().unwrap() }.add_fresh_nodes(&addrs);
 				},
-				NetworkMessage::Headers(headers) => {
+				Some(NetworkMessage::Headers(headers)) => {
 					if headers.is_empty() {
 						return future::ok(());
 					}
@@ -375,7 +378,7 @@ fn make_trusted_conn(trusted_sockaddr: SocketAddr, bgp_client: Arc<BGPClient>) {
 						return future::err(())
 					}
 				},
-				NetworkMessage::Block(block) => {
+				Some(NetworkMessage::Block(block)) => {
 					let hash = block.header.bitcoin_hash();
 					let header_map = unsafe { HEADER_MAP.as_ref().unwrap() }.lock().unwrap();
 					let height = *header_map.get(&hash).expect("Got loose block from trusted peer we coulnd't have requested");
@@ -387,7 +390,7 @@ fn make_trusted_conn(trusted_sockaddr: SocketAddr, bgp_client: Arc<BGPClient>) {
 						}
 					}
 				},
-				NetworkMessage::Ping(v) => {
+				Some(NetworkMessage::Ping(v)) => {
 					if let Err(_) = trusted_write.try_send(NetworkMessage::Pong(v)) {
 						return future::err(())
 					}
@@ -408,8 +411,8 @@ fn make_trusted_conn(trusted_sockaddr: SocketAddr, bgp_client: Arc<BGPClient>) {
 }
 
 fn main() {
-	if env::args().len() != 4 {
-		println!("USAGE: dnsseed-rust datastore localPeerAddress bgp_peer");
+	if env::args().len() != 5 {
+		println!("USAGE: dnsseed-rust datastore localPeerAddress tor_proxy_addr bgp_peer");
 		return;
 	}
 
@@ -421,7 +424,7 @@ fn main() {
 	unsafe { REQUEST_BLOCK = Some(Box::new(Mutex::new(Arc::new((0, genesis_block(Network::Bitcoin).bitcoin_hash(), genesis_block(Network::Bitcoin)))))) };
 
 	let trt = tokio::runtime::Builder::new()
-		.blocking_threads(2).core_threads(num_cpus::get().max(1) * 3)
+		.blocking_threads(2).core_threads(num_cpus::get().max(1) * 2)
 		.build().unwrap();
 
 	let _ = trt.block_on_all(future::lazy(|| {
@@ -429,6 +432,10 @@ fn main() {
 		args.next();
 		let path = args.next().unwrap();
 		let trusted_sockaddr: SocketAddr = args.next().unwrap().parse().unwrap();
+
+		let tor_socks5_sockaddr: SocketAddr = args.next().unwrap().parse().unwrap();
+		unsafe { TOR_PROXY = Some(tor_socks5_sockaddr); }
+
 		let bgp_sockaddr: SocketAddr = args.next().unwrap().parse().unwrap();
 
 		Store::new(path).and_then(move |store| {