Merge pull request #1497 from ethcore/clone-work

Optionally clone block behind work-package

ethcore/src/client/client.rs

@@ -362,6 +362,9 @@ impl Client {
 		let receipts = block.receipts().clone();
 		let traces = From::from(block.traces().clone().unwrap_or_else(Vec::new));
 
+		// CHECK! I *think* this is fine, even if the state_root is equal to another
+		// already-imported block of the same number.
+		// TODO: Prove it with a test.
 		block.drain().commit(number, hash, ancient).expect("State DB commit failed.");
 
 		// And update the chain after commit to prevent race conditions

ethcore/src/miner/miner.rs

@@ -65,6 +65,8 @@ pub struct MinerOptions {
 	pub pending_set: PendingSet,
 	/// How many historical work packages can we store before running out?
 	pub work_queue_size: usize,
+	/// Can we submit two different solutions for the same block and expect both to result in an import?
+	pub enable_resubmission: bool,
 }
 
 impl Default for MinerOptions {
@@ -79,6 +81,7 @@ impl Default for MinerOptions {
 			pending_set: PendingSet::AlwaysQueue,
 			reseal_min_period: Duration::from_secs(0),
 			work_queue_size: 20,
+			enable_resubmission: true,
 		}
 	}
 }
@@ -155,9 +158,10 @@ impl Miner {
 	fn prepare_sealing(&self, chain: &MiningBlockChainClient) {
 		trace!(target: "miner", "prepare_sealing: entering");
 
-		let (transactions, mut open_block) = {
+		let (transactions, mut open_block, last_work_hash) = {
 			let transactions = {self.transaction_queue.lock().unwrap().top_transactions()};
 			let mut sealing_work = self.sealing_work.lock().unwrap();
+			let last_work_hash = sealing_work.peek_last_ref().map(|pb| pb.block().fields().header.hash());
 			let best_hash = chain.best_block_header().sha3();
 /*
 			// check to see if last ClosedBlock in would_seals is actually same parent block.
@@ -184,7 +188,7 @@ impl Miner {
 					)
 				}
 			};
-			(transactions, open_block)
+			(transactions, open_block, last_work_hash)
 		};
 
 		let mut invalid_transactions = HashSet::new();
@@ -252,7 +256,8 @@ impl Miner {
 
 		let work = {
 			let mut sealing_work = self.sealing_work.lock().unwrap();
-			let work = if sealing_work.peek_last_ref().map_or(true, |pb| pb.block().fields().header.hash() != block.block().fields().header.hash()) {
+			trace!(target: "miner", "Checking whether we need to reseal: last={:?}, this={:?}", last_work_hash, block.block().fields().header.hash());
+			let work = if last_work_hash.map_or(true, |h| h != block.block().fields().header.hash()) {
 				trace!(target: "miner", "Pushing a new, refreshed or borrowed pending {}...", block.block().fields().header.hash());
 				let pow_hash = block.block().fields().header.hash();
 				let number = block.block().fields().header.number();
@@ -611,7 +616,7 @@ impl MinerService for Miner {
 	}
 
 	fn submit_seal(&self, chain: &MiningBlockChainClient, pow_hash: H256, seal: Vec<Bytes>) -> Result<(), Error> {
-		let result = if let Some(b) = self.sealing_work.lock().unwrap().take_used_if(|b| &b.hash() == &pow_hash) {
+		let result = if let Some(b) = self.sealing_work.lock().unwrap().get_used_if(if self.options.enable_resubmission { GetAction::Clone } else { GetAction::Take }, |b| &b.hash() == &pow_hash) {
 			b.lock().try_seal(self.engine(), seal).or_else(|_| {
 				warn!(target: "miner", "Mined solution rejected: Invalid.");
 				Err(Error::PowInvalid)

parity/cli.rs

@@ -169,6 +169,10 @@ Sealing/Mining Options:
                            more than 32 characters.
   --tx-queue-size LIMIT    Maximum amount of transactions in the queue (waiting
                            to be included in next block) [default: 1024].
+  --remove-solved          Move solved blocks from the work package queue
+                           instead of cloning them. This gives a slightly
+                           faster import speed, but means that extra solutions
+                           submitted for the same work package will go unused.
   --notify-work URLS       URLs to which work package notifications are pushed.
                            URLS should be a comma-delimited list of HTTP URLs.
 
@@ -313,6 +317,7 @@ pub struct Args {
 	pub flag_reseal_on_txs: String,
 	pub flag_reseal_min_period: u64,
 	pub flag_work_queue_size: usize,
+	pub flag_remove_solved: bool,
 	pub flag_tx_gas_limit: Option<String>,
 	pub flag_relay_set: String,
 	pub flag_author: Option<String>,

parity/configuration.rs

@@ -112,6 +112,7 @@ impl Configuration {
 			},
 			reseal_min_period: Duration::from_millis(self.args.flag_reseal_min_period),
 			work_queue_size: self.args.flag_work_queue_size,
+			enable_resubmission: !self.args.flag_remove_solved,
 		}
 	}
 

rpc/src/v1/tests/eth.rs

@@ -52,6 +52,7 @@ fn sync_provider() -> Arc<TestSyncProvider> {
 fn miner_service(spec: Spec, accounts: Arc<AccountProvider>) -> Arc<Miner> {
 	Miner::new(
 		MinerOptions {
+			new_work_notify: vec![],
 			force_sealing: true,
 			reseal_on_external_tx: true,
 			reseal_on_own_tx: true,
@@ -60,7 +61,7 @@ fn miner_service(spec: Spec, accounts: Arc<AccountProvider>) -> Arc<Miner> {
 			pending_set: PendingSet::SealingOrElseQueue,
 			reseal_min_period: Duration::from_secs(0),
 			work_queue_size: 50,
-			new_work_notify: vec![],
+			enable_resubmission: true,
 		},
 		spec,
 		Some(accounts)

util/src/using_queue.rs

@@ -27,6 +27,14 @@ pub struct UsingQueue<T> where T: Clone {
 	max_size: usize,
 }
 
+/// Take an item or just clone it?
+pub enum GetAction {
+	/// Remove the item, faster but you can't get it back.
+	Take,
+	/// Clone the item, slower but you can get it again.
+	Clone,
+}
+
 impl<T> UsingQueue<T> where T: Clone {
 	/// Create a new struct with a maximum size of `max_size`.
 	pub fn new(max_size: usize) -> UsingQueue<T> {
@@ -74,6 +82,20 @@ impl<T> UsingQueue<T> where T: Clone {
 		self.in_use.iter().position(|r| predicate(r)).map(|i| self.in_use.remove(i))
 	}
 
+	/// Returns `Some` item which is the first that `f` returns `true` with a reference to it
+	/// as a parameter or `None` if no such item exists in the queue.
+	pub fn clone_used_if<P>(&mut self, predicate: P) -> Option<T> where P: Fn(&T) -> bool {
+		self.in_use.iter().find(|r| predicate(r)).cloned()
+	}
+
+	/// Fork-function for `take_used_if` and `clone_used_if`.
+	pub fn get_used_if<P>(&mut self, action: GetAction, predicate: P) -> Option<T> where P: Fn(&T) -> bool {
+		match action {
+			GetAction::Take => self.take_used_if(predicate),
+			GetAction::Clone => self.clone_used_if(predicate),
+		}
+	}
+
 	/// Returns the most recently pushed block if `f` returns `true` with a reference to it as
 	/// a parameter, otherwise `None`.
 	/// Will not destroy a block if a reference to it has previously been returned by `use_last_ref`,
@@ -94,18 +116,66 @@ impl<T> UsingQueue<T> where T: Clone {
 }
 
 #[test]
-fn should_find_when_pushed() {
+fn should_not_find_when_pushed() {
 	let mut q = UsingQueue::new(2);
 	q.push(1);
 	assert!(q.take_used_if(|i| i == &1).is_none());
 }
 
+#[test]
+fn should_not_find_when_pushed_with_clone() {
+	let mut q = UsingQueue::new(2);
+	q.push(1);
+	assert!(q.clone_used_if(|i| i == &1).is_none());
+}
+
 #[test]
 fn should_find_when_pushed_and_used() {
 	let mut q = UsingQueue::new(2);
 	q.push(1);
 	q.use_last_ref();
-	assert!(q.take_used_if(|i| i == &1).is_some());
+	assert!(q.take_used_if(|i| i == &1).unwrap() == 1);
+}
+
+#[test]
+fn should_have_same_semantics_for_get_take_clone() {
+	let mut q = UsingQueue::new(2);
+	q.push(1);
+	assert!(q.get_used_if(GetAction::Clone, |i| i == &1).is_none());
+	assert!(q.get_used_if(GetAction::Take, |i| i == &1).is_none());
+	q.use_last_ref();
+	assert!(q.get_used_if(GetAction::Clone, |i| i == &1).unwrap() == 1);
+	assert!(q.get_used_if(GetAction::Clone, |i| i == &1).unwrap() == 1);
+	assert!(q.get_used_if(GetAction::Take, |i| i == &1).unwrap() == 1);
+	assert!(q.get_used_if(GetAction::Clone, |i| i == &1).is_none());
+	assert!(q.get_used_if(GetAction::Take, |i| i == &1).is_none());
+}
+
+#[test]
+fn should_find_when_pushed_and_used_with_clone() {
+	let mut q = UsingQueue::new(2);
+	q.push(1);
+	q.use_last_ref();
+	assert!(q.clone_used_if(|i| i == &1).unwrap() == 1);
+}
+
+#[test]
+fn should_not_find_again_when_pushed_and_taken() {
+	let mut q = UsingQueue::new(2);
+	q.push(1);
+	q.use_last_ref();
+	assert!(q.take_used_if(|i| i == &1).unwrap() == 1);
+	assert!(q.clone_used_if(|i| i == &1).is_none());
+}
+
+#[test]
+fn should_find_again_when_pushed_and_cloned() {
+	let mut q = UsingQueue::new(2);
+	q.push(1);
+	q.use_last_ref();
+	assert!(q.clone_used_if(|i| i == &1).unwrap() == 1);
+	assert!(q.clone_used_if(|i| i == &1).unwrap() == 1);
+	assert!(q.take_used_if(|i| i == &1).unwrap() == 1);
 }
 
 #[test]