UDP Discovery

ethcore/src/block_queue.rs

@@ -153,19 +153,19 @@ impl BlockQueue {
 	}
 
 	fn verify(verification: Arc<Mutex<Verification>>, engine: Arc<Box<Engine>>, wait: Arc<Condvar>, ready: Arc<QueueSignal>, deleting: Arc<AtomicBool>, empty: Arc<Condvar>) {
-		while !deleting.load(AtomicOrdering::Relaxed) {
+		while !deleting.load(AtomicOrdering::Acquire) {
 			{
 				let mut lock = verification.lock().unwrap();
 
 				if lock.unverified.is_empty() && lock.verifying.is_empty() {
 					empty.notify_all();
 				}
 
-				while lock.unverified.is_empty() && !deleting.load(AtomicOrdering::Relaxed) {
+				while lock.unverified.is_empty() && !deleting.load(AtomicOrdering::Acquire) {
 					lock = wait.wait(lock).unwrap();
 				}
 
-				if deleting.load(AtomicOrdering::Relaxed) {
+				if deleting.load(AtomicOrdering::Acquire) {
 					return;
 				}
 			}
@@ -347,7 +347,7 @@ impl MayPanic for BlockQueue {
 impl Drop for BlockQueue {
 	fn drop(&mut self) {
 		self.clear();
-		self.deleting.store(true, AtomicOrdering::Relaxed);
+		self.deleting.store(true, AtomicOrdering::Release);
 		self.more_to_verify.notify_all();
 		for t in self.verifiers.drain(..) {
 			t.join().unwrap();

ethcore/src/client.rs

@@ -419,11 +419,11 @@ impl BlockChainClient for Client {
 	}
 
 	fn state_data(&self, _hash: &H256) -> Option<Bytes> {
-		unimplemented!();
+		None
 	}
 
 	fn block_receipts(&self, _hash: &H256) -> Option<Bytes> {
-		unimplemented!();
+		None
 	}
 
 	fn import_block(&self, bytes: Bytes) -> ImportResult {

ethcore/src/service.rs

@@ -124,8 +124,6 @@ impl IoHandler<NetSyncMessage> for ClientIoHandler {
 	}
 }
 
-// TODO: rewrite into something that doesn't dependent on the testing environment having a particular port ready for use.
-/*
 #[cfg(test)]
 mod tests {
 	use super::*;
@@ -136,8 +134,7 @@ mod tests {
 	fn it_can_be_started() {
 		let spec = get_test_spec();
 		let temp_path = RandomTempPath::new();
-		let service = ClientService::start(spec, NetworkConfiguration::new(), &temp_path.as_path());
+		let service = ClientService::start(spec, NetworkConfiguration::new_with_port(40456), &temp_path.as_path());
 		assert!(service.is_ok());
 	}
 }
-*/

parity/main.rs

@@ -37,6 +37,7 @@ extern crate ethcore_rpc as rpc;
 
 use std::net::{SocketAddr};
 use std::env;
+use std::path::PathBuf;
 use rlog::{LogLevelFilter};
 use env_logger::LogBuilder;
 use ctrlc::CtrlC;
@@ -69,8 +70,10 @@ Options:
 
   --no-bootstrap           Don't bother trying to connect to any nodes initially.
   --listen-address URL     Specify the IP/port on which to listen for peers [default: 0.0.0.0:30304].
-  --public-address URL     Specify the IP/port on which peers may connect [default: 0.0.0.0:30304].
+  --public-address URL     Specify the IP/port on which peers may connect.
   --address URL            Equivalent to --listen-address URL --public-address URL.
+  --peers NUM  	           Try to manintain that many peers [default: 25].
+  --no-discovery   	       Disable new peer discovery.
   --upnp                   Use UPnP to try to figure out the correct network settings.
   --node-key KEY           Specify node secret key as hex string.
 
@@ -95,8 +98,10 @@ struct Args {
 	flag_keys_path: String,
 	flag_no_bootstrap: bool,
 	flag_listen_address: String,
-	flag_public_address: String,
+	flag_public_address: Option<String>,
 	flag_address: Option<String>,
+	flag_peers: u32,
+	flag_no_discovery: bool,
 	flag_upnp: bool,
 	flag_node_key: Option<String>,
 	flag_cache_pref_size: usize,
@@ -165,7 +170,7 @@ impl Configuration {
 		self.args.flag_db_path.replace("$HOME", env::home_dir().unwrap().to_str().unwrap())
 	}
 
-	fn keys_path(&self) -> String {
+	fn _keys_path(&self) -> String {
 		self.args.flag_keys_path.replace("$HOME", env::home_dir().unwrap().to_str().unwrap())	
 	}
 
@@ -187,21 +192,23 @@ impl Configuration {
 		}
 	}
 
-	fn net_addresses(&self) -> (SocketAddr, SocketAddr) {
-		let listen_address;
-		let public_address;
+	fn net_addresses(&self) -> (Option<SocketAddr>, Option<SocketAddr>) {
+		let mut listen_address = None;
+		let mut public_address = None;
 
-		match self.args.flag_address {
-			None => {
-				listen_address = SocketAddr::from_str(self.args.flag_listen_address.as_ref()).expect("Invalid listen address given with --listen-address");
-				public_address = SocketAddr::from_str(self.args.flag_public_address.as_ref()).expect("Invalid public address given with --public-address");
-			}
-			Some(ref a) => {
-				public_address = SocketAddr::from_str(a.as_ref()).expect("Invalid listen/public address given with --address");
-				listen_address = public_address;
+		if let Some(ref a) = self.args.flag_address {
+			public_address = Some(SocketAddr::from_str(a.as_ref()).expect("Invalid listen/public address given with --address"));
+			listen_address = public_address;
+		}
+		if listen_address.is_none() {
+			listen_address = Some(SocketAddr::from_str(self.args.flag_listen_address.as_ref()).expect("Invalid listen address given with --listen-address"));
+		}
+		if let Some(ref a) = self.args.flag_public_address {
+			if public_address.is_some() {
+				panic!("Conflicting flags: --address and --public-address");
 			}
-		};
-
+			public_address = Some(SocketAddr::from_str(a.as_ref()).expect("Invalid listen address given with --public-address"));
+		}
 		(listen_address, public_address)
 	}
 
@@ -236,6 +243,11 @@ impl Configuration {
 		net_settings.listen_address = listen;
 		net_settings.public_address = public;
 		net_settings.use_secret = self.args.flag_node_key.as_ref().map(|s| Secret::from_str(&s).expect("Invalid key string"));
+		net_settings.discovery_enabled = !self.args.flag_no_discovery;
+		net_settings.ideal_peers = self.args.flag_peers;
+		let mut net_path = PathBuf::from(&self.path());
+		net_path.push("network");
+		net_settings.config_path = Some(net_path.to_str().unwrap().to_owned());
 
 		// Build client
 		let mut service = ClientService::start(spec, net_settings, &Path::new(&self.path())).unwrap();

sync/src/chain.rs

@@ -82,7 +82,7 @@ const RECEIPTS_PACKET: u8 = 0x10;
 
 const NETWORK_ID: U256 = ONE_U256; //TODO: get this from parent
 
-const CONNECTION_TIMEOUT_SEC: f64 = 30f64;
+const CONNECTION_TIMEOUT_SEC: f64 = 10f64;
 
 struct Header {
 	/// Header data
@@ -314,7 +314,7 @@ impl ChainSync {
 		}
 
 		self.peers.insert(peer_id.clone(), peer);
-		info!(target: "sync", "Connected {}:{}", peer_id, io.peer_info(peer_id));
+		debug!(target: "sync", "Connected {}:{}", peer_id, io.peer_info(peer_id));
 		self.sync_peer(io, peer_id, false);
 		Ok(())
 	}
@@ -545,7 +545,7 @@ impl ChainSync {
 	pub fn on_peer_aborting(&mut self, io: &mut SyncIo, peer: PeerId) {
 		trace!(target: "sync", "== Disconnecting {}", peer);
 		if self.peers.contains_key(&peer) {
-			info!(target: "sync", "Disconnected {}", peer);
+			debug!(target: "sync", "Disconnected {}", peer);
 			self.clear_peer_download(peer);
 			self.peers.remove(&peer);
 			self.continue_sync(io);
@@ -1179,7 +1179,7 @@ impl ChainSync {
 		for (peer_id, peer_number) in updated_peers {
 			let mut peer_best = self.peers.get(&peer_id).unwrap().latest_hash.clone();
 			if best_number - peer_number > MAX_PEERS_PROPAGATION as BlockNumber {
-				// If we think peer is too far behind just end one latest hash
+				// If we think peer is too far behind just send one latest hash
 				peer_best = last_parent.clone();
 			}
 			sent = sent + match ChainSync::create_new_hashes_rlp(io.chain(), &peer_best, &local_best) {

util/Cargo.toml

@@ -30,6 +30,7 @@ clippy = { version = "0.0.42", optional = true }
 json-tests = { path = "json-tests" }
 target_info = "0.1.0"
 igd = "0.4.2"
+libc = "0.2.7"
 
 [features]
 default = []

util/fdlimit/src/raise_fd_limit.rs

@@ -57,5 +57,28 @@ pub unsafe fn raise_fd_limit() {
     }
 }
 
-#[cfg(not(any(target_os = "macos", target_os = "ios")))]
+#[cfg(any(target_os = "linux"))]
+#[allow(non_camel_case_types)]
+pub unsafe fn raise_fd_limit() {
+    use libc;
+    use std::io;
+
+    // Fetch the current resource limits
+    let mut rlim = libc::rlimit{rlim_cur: 0, rlim_max: 0};
+    if libc::getrlimit(libc::RLIMIT_NOFILE, &mut rlim) != 0 {
+        let err = io::Error::last_os_error();
+        panic!("raise_fd_limit: error calling getrlimit: {}", err);
+    }
+
+    // Set soft limit to hard imit
+    rlim.rlim_cur = rlim.rlim_max;
+
+    // Set our newly-increased resource limit
+    if libc::setrlimit(libc::RLIMIT_NOFILE, &rlim) != 0 {
+        let err = io::Error::last_os_error();
+        panic!("raise_fd_limit: error calling setrlimit: {}", err);
+    }
+}
+
+#[cfg(not(any(target_os = "macos", target_os = "ios", target_os = "linux")))]
 pub unsafe fn raise_fd_limit() {}

util/src/bytes.rs

@@ -170,32 +170,8 @@ pub trait BytesConvertable {
 	fn to_bytes(&self) -> Bytes { self.as_slice().to_vec() }
 }
 
-impl<'a> BytesConvertable for &'a [u8] {
-	fn bytes(&self) -> &[u8] { self }
-}
-
-impl BytesConvertable for Vec<u8> {
-	fn bytes(&self) -> &[u8] { self }
-}
-
-impl BytesConvertable for String {
-	fn bytes(&self) -> &[u8] { &self.as_bytes() }
-}
-
-macro_rules! impl_bytes_convertable_for_array {
-	($zero: expr) => ();
-	($len: expr, $($idx: expr),*) => {
-		impl BytesConvertable for [u8; $len] {
-			fn bytes(&self) -> &[u8] { self }
-		}
-		impl_bytes_convertable_for_array! { $($idx),* }
-	}
-}
-
-// -1 at the end is not expanded
-impl_bytes_convertable_for_array! {
-		32, 31, 30, 29, 28, 27, 26, 25, 24, 23, 22, 21, 20, 19, 18, 17, 16,
-		15, 14, 13, 12, 11, 10, 9, 8, 7, 6, 5, 4, 3, 2, 1, 0, -1
+impl<T> BytesConvertable for T where T: AsRef<[u8]> {
+	fn bytes(&self) -> &[u8] { self.as_ref() }
 }
 
 #[test]

util/src/hash.rs

@@ -77,12 +77,6 @@ macro_rules! impl_hash {
 		/// Unformatted binary data of fixed length.
 		pub struct $from (pub [u8; $size]);
 
-		impl BytesConvertable for $from {
-			fn bytes(&self) -> &[u8] {
-				&self.0
-			}
-		}
-
 		impl Deref for $from {
 			type Target = [u8];
 
@@ -92,6 +86,13 @@ macro_rules! impl_hash {
 			}
 		}
 
+		impl AsRef<[u8]> for $from {
+			#[inline]
+			fn as_ref(&self) -> &[u8] {
+				&self.0
+			}
+		}
+
 		impl DerefMut for $from {
 			#[inline]
 			fn deref_mut(&mut self) -> &mut [u8] {

util/src/io/service.rs

@@ -256,6 +256,11 @@ impl<Message> Handler for IoManager<Message> where Message: Send + Clone + Sync
 			IoMessage::DeregisterStream { handler_id, token } => {
 				let handler = self.handlers.get(handler_id).expect("Unknown handler id").clone();
 				handler.deregister_stream(token, event_loop);
+				// unregister a timer associated with the token (if any)
+				let timer_id = token + handler_id * TOKENS_PER_HANDLER;
+				if let Some(timer) = self.timers.write().unwrap().remove(&timer_id) {
+					event_loop.clear_timeout(timer.timeout);
+				}
 			},
 			IoMessage::UpdateStreamRegistration { handler_id, token } => {
 				let handler = self.handlers.get(handler_id).expect("Unknown handler id").clone();

util/src/io/worker.rs

@@ -44,6 +44,7 @@ pub struct Worker {
 	thread: Option<JoinHandle<()>>,
 	wait: Arc<Condvar>,
 	deleting: Arc<AtomicBool>,
+	wait_mutex: Arc<Mutex<()>>,
 }
 
 impl Worker {
@@ -61,6 +62,7 @@ impl Worker {
 			thread: None,
 			wait: wait.clone(),
 			deleting: deleting.clone(),
+			wait_mutex: wait_mutex.clone(),
 		};
 		worker.thread = Some(thread::Builder::new().name(format!("IO Worker #{}", index)).spawn(
 			move || {
@@ -77,13 +79,17 @@ impl Worker {
 						wait_mutex: Arc<Mutex<()>>,
 						deleting: Arc<AtomicBool>)
 						where Message: Send + Sync + Clone + 'static {
-		while !deleting.load(AtomicOrdering::Relaxed) {
+		loop {
 			{
 				let lock = wait_mutex.lock().unwrap();
-				let _ = wait.wait(lock).unwrap();
-				if deleting.load(AtomicOrdering::Relaxed) {
+				if deleting.load(AtomicOrdering::Acquire) {
 					return;
 				}
+				let _ = wait.wait(lock).unwrap();
+			}
+
+			if deleting.load(AtomicOrdering::Acquire) {
+				return;
 			}
 			while let chase_lev::Steal::Data(work) = stealer.steal() {
 				Worker::do_work(work, channel.clone());
@@ -114,7 +120,8 @@ impl Worker {
 
 impl Drop for Worker {
 	fn drop(&mut self) {
-		self.deleting.store(true, AtomicOrdering::Relaxed);
+		let _ = self.wait_mutex.lock();
+		self.deleting.store(true, AtomicOrdering::Release);
 		self.wait.notify_all();
 		let thread = mem::replace(&mut self.thread, None).unwrap();
 		thread.join().ok();

util/src/lib.rs

@@ -106,6 +106,7 @@ extern crate serde;
 #[macro_use]
 extern crate log as rlog;
 extern crate igd;
+extern crate libc;
 
 pub mod standard;
 #[macro_use]