[2.x.x] Use blocking IO in P2P to reduce CPU load (#2855)

* Use blocking IO in P2P to reduce CPU load

chain/src/chain.rs

@@ -646,8 +646,7 @@ impl Chain {
 	/// TODO - Write this data to disk and validate the rebuilt kernel MMR.
 	pub fn kernel_data_write(&self, reader: &mut Read) -> Result<(), Error> {
 		let mut count = 0;
-		let mut stream =
-			StreamingReader::new(reader, ProtocolVersion::local(), Duration::from_secs(1));
+		let mut stream = StreamingReader::new(reader, ProtocolVersion::local());
 		while let Ok(_kernel) = TxKernelEntry::read(&mut stream) {
 			count += 1;
 		}

core/src/core/block.rs

@@ -180,9 +180,9 @@ impl Default for HeaderVersion {
 
 // self-conscious increment function courtesy of Jasper
 impl HeaderVersion {
-  fn next(&self) -> Self {
-    Self(self.0+1)
-  }
+	fn next(&self) -> Self {
+		Self(self.0 + 1)
+	}
 }
 
 impl HeaderVersion {

core/src/global.rs

@@ -17,13 +17,15 @@
 //! should be used sparingly.
 
 use crate::consensus::{
-	HeaderInfo, valid_header_version, graph_weight, BASE_EDGE_BITS, BLOCK_TIME_SEC,
+	graph_weight, valid_header_version, HeaderInfo, BASE_EDGE_BITS, BLOCK_TIME_SEC,
 	COINBASE_MATURITY, CUT_THROUGH_HORIZON, DAY_HEIGHT, DEFAULT_MIN_EDGE_BITS,
 	DIFFICULTY_ADJUST_WINDOW, INITIAL_DIFFICULTY, MAX_BLOCK_WEIGHT, PROOFSIZE,
 	SECOND_POW_EDGE_BITS, STATE_SYNC_THRESHOLD,
 };
 use crate::core::block::HeaderVersion;
-use crate::pow::{self, new_cuckatoo_ctx, new_cuckaroo_ctx, new_cuckarood_ctx, EdgeType, PoWContext};
+use crate::pow::{
+	self, new_cuckaroo_ctx, new_cuckarood_ctx, new_cuckatoo_ctx, EdgeType, PoWContext,
+};
 /// An enum collecting sets of parameters used throughout the
 /// code wherever mining is needed. This should allow for
 /// different sets of parameters for different purposes,
@@ -164,14 +166,16 @@ where
 	match chain_type {
 		// Mainnet has Cuckaroo(d)29 for AR and Cuckatoo31+ for AF
 		ChainTypes::Mainnet if edge_bits > 29 => new_cuckatoo_ctx(edge_bits, proof_size, max_sols),
-		ChainTypes::Mainnet if valid_header_version(height, HeaderVersion::new(2))
-						=> new_cuckarood_ctx(edge_bits, proof_size),
+		ChainTypes::Mainnet if valid_header_version(height, HeaderVersion::new(2)) => {
+			new_cuckarood_ctx(edge_bits, proof_size)
+		}
 		ChainTypes::Mainnet => new_cuckaroo_ctx(edge_bits, proof_size),
 
 		// Same for Floonet
 		ChainTypes::Floonet if edge_bits > 29 => new_cuckatoo_ctx(edge_bits, proof_size, max_sols),
-		ChainTypes::Floonet if valid_header_version(height, HeaderVersion::new(2))
-						=> new_cuckarood_ctx(edge_bits, proof_size),
+		ChainTypes::Floonet if valid_header_version(height, HeaderVersion::new(2)) => {
+			new_cuckarood_ctx(edge_bits, proof_size)
+		}
 		ChainTypes::Floonet => new_cuckaroo_ctx(edge_bits, proof_size),
 
 		// Everything else is Cuckatoo only

core/src/pow.rs

@@ -33,9 +33,9 @@ use num;
 
 #[macro_use]
 mod common;
-pub mod cuckatoo;
 pub mod cuckaroo;
 pub mod cuckarood;
+pub mod cuckatoo;
 mod error;
 #[allow(dead_code)]
 pub mod lean;
@@ -49,9 +49,9 @@ use chrono::prelude::{DateTime, NaiveDateTime, Utc};
 
 pub use self::common::EdgeType;
 pub use self::types::*;
-pub use crate::pow::cuckatoo::{new_cuckatoo_ctx, CuckatooContext};
 pub use crate::pow::cuckaroo::{new_cuckaroo_ctx, CuckarooContext};
 pub use crate::pow::cuckarood::{new_cuckarood_ctx, CuckaroodContext};
+pub use crate::pow::cuckatoo::{new_cuckatoo_ctx, CuckatooContext};
 pub use crate::pow::error::Error;
 
 const MAX_SOLS: u32 = 10;

core/src/pow/cuckarood.rs

@@ -22,11 +22,11 @@
 //! a rotation by 25, halves the number of graph nodes in each partition,
 //! and requires cycles to alternate between even- and odd-indexed edges.
 
+use crate::global;
 use crate::pow::common::{CuckooParams, EdgeType};
 use crate::pow::error::{Error, ErrorKind};
 use crate::pow::siphash::siphash_block;
 use crate::pow::{PoWContext, Proof};
-use crate::global;
 
 /// Instantiate a new CuckaroodContext as a PowContext. Note that this can't
 /// be moved in the PoWContext trait as this particular trait needs to be
@@ -69,8 +69,7 @@ where
 
 	fn verify(&self, proof: &Proof) -> Result<(), Error> {
 		if proof.proof_size() != global::proofsize() {
-			return Err(ErrorKind::Verification(
-				"wrong cycle length".to_owned(),))?;
+			return Err(ErrorKind::Verification("wrong cycle length".to_owned()))?;
 		}
 		let nonces = &proof.nonces;
 		let mut uvs = vec![0u64; 2 * proof.proof_size()];
@@ -92,10 +91,10 @@ where
 			}
 			let edge = to_edge!(T, siphash_block(&self.params.siphash_keys, nonces[n], 25));
 			let idx = 4 * ndir[dir] + 2 * dir;
-			uvs[idx  ] = to_u64!( edge        & nodemask);
-			uvs[idx+1] = to_u64!((edge >> 32) & nodemask);
-			xor0 ^= uvs[idx  ];
-			xor1 ^= uvs[idx+1];
+			uvs[idx] = to_u64!(edge & nodemask);
+			uvs[idx + 1] = to_u64!((edge >> 32) & nodemask);
+			xor0 ^= uvs[idx];
+			xor1 ^= uvs[idx + 1];
 			ndir[dir] += 1;
 		}
 		if xor0 | xor1 != 0 {
@@ -110,7 +109,8 @@ where
 			// follow cycle
 			j = i;
 			for k in (((i % 4) ^ 2)..(2 * self.params.proof_size)).step_by(4) {
-				if uvs[k] == uvs[i] { // find reverse edge endpoint identical to one at i
+				if uvs[k] == uvs[i] {
+					// find reverse edge endpoint identical to one at i
 					if j != i {
 						return Err(ErrorKind::Verification("branch in cycle".to_owned()))?;
 					}
@@ -173,11 +173,15 @@ mod test {
 	fn cuckarood19_29_vectors() {
 		let mut ctx19 = new_impl::<u64>(19, 42);
 		ctx19.params.siphash_keys = V1_19_HASH.clone();
-		assert!(ctx19.verify(&Proof::new(V1_19_SOL.to_vec().clone())).is_ok());
+		assert!(ctx19
+			.verify(&Proof::new(V1_19_SOL.to_vec().clone()))
+			.is_ok());
 		assert!(ctx19.verify(&Proof::zero(42)).is_err());
 		let mut ctx29 = new_impl::<u64>(29, 42);
 		ctx29.params.siphash_keys = V2_29_HASH.clone();
-		assert!(ctx29.verify(&Proof::new(V2_29_SOL.to_vec().clone())).is_ok());
+		assert!(ctx29
+			.verify(&Proof::new(V2_29_SOL.to_vec().clone()))
+			.is_ok());
 		assert!(ctx29.verify(&Proof::zero(42)).is_err());
 	}
 

core/src/ser.rs

@@ -22,7 +22,6 @@
 use crate::core::hash::{DefaultHashable, Hash, Hashed};
 use crate::global::PROTOCOL_VERSION;
 use crate::keychain::{BlindingFactor, Identifier, IDENTIFIER_SIZE};
-use crate::util::read_write::read_exact;
 use crate::util::secp::constants::{
 	AGG_SIGNATURE_SIZE, COMPRESSED_PUBLIC_KEY_SIZE, MAX_PROOF_SIZE, PEDERSEN_COMMITMENT_SIZE,
 	SECRET_KEY_SIZE,
@@ -35,7 +34,6 @@ use byteorder::{BigEndian, ByteOrder, ReadBytesExt};
 use std::fmt::{self, Debug};
 use std::io::{self, Read, Write};
 use std::marker;
-use std::time::Duration;
 use std::{cmp, error};
 
 /// Possible errors deriving from serializing or deserializing.
@@ -459,22 +457,16 @@ pub struct StreamingReader<'a> {
 	total_bytes_read: u64,
 	version: ProtocolVersion,
 	stream: &'a mut dyn Read,
-	timeout: Duration,
 }
 
 impl<'a> StreamingReader<'a> {
 	/// Create a new streaming reader with the provided underlying stream.
 	/// Also takes a duration to be used for each individual read_exact call.
-	pub fn new(
-		stream: &'a mut dyn Read,
-		version: ProtocolVersion,
-		timeout: Duration,
-	) -> StreamingReader<'a> {
+	pub fn new(stream: &'a mut dyn Read, version: ProtocolVersion) -> StreamingReader<'a> {
 		StreamingReader {
 			total_bytes_read: 0,
 			version,
 			stream,
-			timeout,
 		}
 	}
 
@@ -521,7 +513,7 @@ impl<'a> Reader for StreamingReader<'a> {
 	/// Read a fixed number of bytes.
 	fn read_fixed_bytes(&mut self, len: usize) -> Result<Vec<u8>, Error> {
 		let mut buf = vec![0u8; len];
-		read_exact(&mut self.stream, &mut buf, self.timeout, true)?;
+		self.stream.read_exact(&mut buf)?;
 		self.total_bytes_read += len as u64;
 		Ok(buf)
 	}