Fix parallel transactions race-condition

ethcore/engines/instant-seal/Cargo.toml

@@ -13,6 +13,7 @@ ethjson = { path = "../../../json" }
 ethereum-types = "0.6.0"
 keccak-hash = "0.2.0"
 machine = { path = "../../machine" }
+parking_lot = "0.8"
 trace = { path = "../../trace" }
 
 [dev-dependencies]

ethcore/engines/instant-seal/src/lib.rs

@@ -14,6 +14,8 @@
 // You should have received a copy of the GNU General Public License
 // along with Parity Ethereum.  If not, see <http://www.gnu.org/licenses/>.
 
+use std::collections::HashSet;
+
 use common_types::{
 	header::Header,
 	engines::{
@@ -22,13 +24,15 @@ use common_types::{
 		params::CommonParams,
 	},
 	errors::EthcoreError as Error,
+	BlockNumber,
 };
 use engine::Engine;
 use ethjson;
 use machine::{
 	ExecutedBlock,
 	Machine
 };
+use parking_lot::Mutex;
 
 
 /// `InstantSeal` params.
@@ -51,6 +55,7 @@ impl From<ethjson::spec::InstantSealParams> for InstantSealParams {
 pub struct InstantSeal {
 	params: InstantSealParams,
 	machine: Machine,
+	sealed_blocks: Mutex<HashSet<BlockNumber>>,
 }
 
 impl InstantSeal {
@@ -59,6 +64,7 @@ impl InstantSeal {
 		InstantSeal {
 			params,
 			machine,
+			sealed_blocks: Mutex::new(HashSet::new()),
 		}
 	}
 }
@@ -74,7 +80,14 @@ impl Engine for InstantSeal {
 		if block.transactions.is_empty() {
 			Seal::None
 		} else {
-			Seal::Regular(Vec::new())
+			let mut sealed_blocks = self.sealed_blocks.lock();
+			let block_number = block.header.number();
+			if !sealed_blocks.contains(&block_number) {
+				sealed_blocks.insert(block_number);
+				Seal::Regular(Vec::new())
+			} else {
+				Seal::None
+			}
 		}
 	}
 

ethcore/src/client/client.rs

@@ -1233,6 +1233,11 @@ impl Client {
 			_   => self.block_header(id).and_then(|h| h.decode().ok())
 		}
 	}
+
+	/// Maybe call `update_sealing` method on the Miner
+	pub fn maybe_update_sealing(&self) {
+		self.importer.miner.maybe_update_sealing(self);
+	}
 }
 
 impl DatabaseRestore for Client {

ethcore/src/miner/miner.rs

@@ -859,6 +859,7 @@ impl Miner {
 			false
 		}
 	}
+
 	/// Prepare pending block, check whether sealing is needed, and then update sealing.
 	fn prepare_and_update_sealing<C: miner::BlockChainClient>(&self, chain: &C) {
 		use miner::MinerService;
@@ -876,6 +877,13 @@ impl Miner {
 			SealingState::NotReady => { self.maybe_enable_sealing(); },
 		}
 	}
+
+	/// Call `update_sealing` if needed
+	pub fn maybe_update_sealing<C: miner::BlockChainClient>(&self, chain: &C) {
+		if self.transaction_queue.has_local_pending_transactions() {
+			self.prepare_and_update_sealing(chain);
+		}
+	}
 }
 
 impl miner::MinerService for Miner {
@@ -1242,7 +1250,7 @@ impl miner::MinerService for Miner {
 	/// Update sealing if required.
 	/// Prepare the block and work if the Engine does not seal internally.
 	fn update_sealing<C>(&self, chain: &C) where
-		C: BlockChain + CallContract + BlockProducer + SealedBlockImporter + Nonce + Sync,
+		C: miner::BlockChainClient,
 	{
 		trace!(target: "miner", "update_sealing");
 
@@ -1429,15 +1437,22 @@ impl miner::MinerService for Miner {
 						service_transaction_checker.as_ref(),
 					);
 					queue.cull(client);
+					if is_internal_import {
+						chain.maybe_update_sealing();
+					}
 				};
 
 				if let Err(e) = channel.send(ClientIoMessage::<Client>::execute(cull)) {
 					warn!(target: "miner", "Error queueing cull: {:?}", e);
 				}
 			} else {
 				self.transaction_queue.cull(client);
+				if is_internal_import {
+					self.maybe_update_sealing(chain);
+				}
 			}
 		}
+
 		if let Some(ref service_transaction_checker) = self.service_transaction_checker {
 			match service_transaction_checker.refresh_cache(chain) {
 				Ok(true) => {

ethcore/src/miner/mod.rs

@@ -84,7 +84,7 @@ pub trait MinerService : Send + Sync {
 
 	/// Update current pending block
 	fn update_sealing<C>(&self, chain: &C)
-		where C: BlockChain + CallContract + BlockProducer + SealedBlockImporter + Nonce + Sync;
+		where C: BlockChainClient;
 
 	// Notifications
 

miner/src/pool/queue.rs

@@ -458,6 +458,7 @@ impl TransactionQueue {
 			let state_readiness = ready::State::new(client.clone(), stale_id, nonce_cap);
 			removed += self.pool.write().cull(Some(chunk), state_readiness);
 		}
+		self.cached_pending.write().clear();
 		debug!(target: "txqueue", "Removed {} stalled transactions. {}", removed, self.status());
 	}