Acquire importer lock before adding transaction

Author: majecty

core/src/client/client.rs

@@ -251,7 +251,8 @@ impl Client {
             transactions.iter().filter_map(|bytes| UntrustedRlp::new(bytes).as_val().ok()).collect();
         let hashes: Vec<_> = transactions.iter().map(UnverifiedTransaction::hash).collect();
         self.transactions_received(&hashes, peer_id);
-        let results = self.importer.miner.import_external_transactions(self, transactions);
+        let _lock = self.importer.import_lock.lock();
+        let results = self.importer.miner.import_external_transactions(self, transactions, &_lock);
         results.len()
     }
 
@@ -728,7 +729,8 @@ impl BlockChainClient for Client {
 
     /// Import own transaction
     fn queue_own_transaction(&self, transaction: SignedTransaction) -> Result<(), Error> {
-        self.importer.miner.import_own_transaction(self, transaction)?;
+        let _lock = self.importer.import_lock.lock();
+        self.importer.miner.import_own_transaction(self, transaction, &_lock)?;
         Ok(())
     }
 

core/src/client/test_client.rs

@@ -306,7 +306,8 @@ impl TestBlockChainClient {
         let sender_address = public_to_address(&signed.signer_public());
         self.set_balance(sender_address, 10_000_000_000_000_000_000);
         let hash = signed.hash();
-        let res = self.miner.import_external_transactions(self, vec![signed.into()]);
+        let _import_lock = unimplemented!();
+        let res = self.miner.import_external_transactions(self, vec![signed.into()], _import_lock);
         let res = res.into_iter().next().unwrap().expect("Successful import");
         assert_eq!(res, TransactionImportResult::Current);
         hash
@@ -533,15 +534,15 @@ impl BlockChainClient for TestBlockChainClient {
     }
 
     fn queue_own_transaction(&self, transaction: SignedTransaction) -> Result<(), GenericError> {
-        self.miner.import_own_transaction(self, transaction)?;
+        self.miner.import_own_transaction(self, transaction, unimplemented!())?;
         Ok(())
     }
 
     fn queue_transactions(&self, transactions: Vec<Bytes>, _peer_id: NodeId) {
         // import right here
         let transactions =
             transactions.into_iter().filter_map(|bytes| UntrustedRlp::new(&bytes).as_val().ok()).collect();
-        self.miner.import_external_transactions(self, transactions);
+        self.miner.import_external_transactions(self, transactions, unimplemented!());
     }
 
     fn ready_transactions(&self, range: Range<u64>) -> PendingSignedTransactions {

core/src/miner/mem_pool.rs

@@ -17,6 +17,7 @@
 use std::collections::{BTreeSet, HashMap, HashSet};
 use std::ops::Range;
 use std::sync::Arc;
+use parking_lot::MutexGuard;
 
 use ckey::{public_to_address, Public};
 use ctypes::errors::{HistoryError, RuntimeError, SyntaxError};
@@ -240,6 +241,7 @@ impl MemPool {
         inserted_block_number: PoolingInstant,
         inserted_timestamp: u64,
         fetch_account: &F,
+        _importer_lock: &MutexGuard<()>,
     ) -> Vec<Result<TransactionImportResult, Error>>
     where
         F: Fn(&Public) -> AccountDetails, {
@@ -940,9 +942,7 @@ impl MemPool {
     }
 
     pub fn count_current_future_transactions(&self, range: Range<u64>) -> (usize, usize) {
-        let current = self.current
-            .queue
-            .len();
+        let current = self.current.queue.len();
         let future = self.future.queue.len();
         (current, future)
     }

core/src/miner/miner.rs

@@ -29,7 +29,7 @@ use ctypes::transaction::{Action, IncompleteTransaction, Timelock};
 use ctypes::{BlockHash, BlockNumber, Header, TxHash};
 use cvm::ChainTimeInfo;
 use kvdb::KeyValueDB;
-use parking_lot::{Mutex, RwLock};
+use parking_lot::{Mutex, MutexGuard, RwLock};
 use primitives::{Bytes, H256};
 
 use super::mem_pool::{Error as MemPoolError, MemPool};
@@ -261,6 +261,7 @@ impl Miner {
         transactions: Vec<UnverifiedTransaction>,
         default_origin: TxOrigin,
         mem_pool: &mut MemPool,
+        _importer_lock: &MutexGuard<()>,
     ) -> Vec<Result<TransactionImportResult, Error>> {
         let best_header = client.best_block_header().decode();
         let fake_header = best_header.generate_child();
@@ -351,7 +352,8 @@ impl Miner {
             }
         };
 
-        let insertion_results = mem_pool.add(to_insert, current_block_number, current_timestamp, &fetch_account);
+        let insertion_results =
+            mem_pool.add(to_insert, current_block_number, current_timestamp, &fetch_account, _importer_lock);
 
         debug_assert_eq!(insertion_results.len(), intermediate_results.iter().filter(|r| r.is_ok()).count());
         let mut insertion_results_index = 0;
@@ -784,35 +786,6 @@ impl MinerService for Miner {
         C: AccountData + BlockChainTrait + BlockProducer + EngineInfo + ImportBlock, {
         ctrace!(MINER, "chain_new_blocks");
 
-        // Then import all transactions...
-        {
-            let mut mem_pool = self.mem_pool.write();
-            for hash in retracted {
-                let block = chain.block(&(*hash).into()).expect(
-                    "Client is sending message after commit to db and inserting to chain; the block is available; qed",
-                );
-                let transactions = block.transactions();
-                let _ = self.add_transactions_to_pool(chain, transactions, TxOrigin::RetractedBlock, &mut mem_pool);
-            }
-        }
-
-        // ...and at the end remove the old ones
-        {
-            let fetch_account = |p: &Public| {
-                let address = public_to_address(p);
-                let a = chain.latest_regular_key_owner(&address).unwrap_or(address);
-
-                AccountDetails {
-                    seq: chain.latest_seq(&a),
-                    balance: chain.latest_balance(&a),
-                }
-            };
-            let current_block_number = chain.chain_info().best_block_number;
-            let current_timestamp = chain.chain_info().best_block_timestamp;
-            let mut mem_pool = self.mem_pool.write();
-            mem_pool.remove_old(&fetch_account, current_block_number, current_timestamp);
-        }
-
         if !self.options.no_reseal_timer {
             chain.set_min_timer();
         }
@@ -983,11 +956,12 @@ impl MinerService for Miner {
         &self,
         client: &C,
         transactions: Vec<UnverifiedTransaction>,
+        _importer_lock: &MutexGuard<()>,
     ) -> Vec<Result<TransactionImportResult, Error>> {
         ctrace!(EXTERNAL_PARCEL, "Importing external transactions");
         let results = {
             let mut mem_pool = self.mem_pool.write();
-            self.add_transactions_to_pool(client, transactions, TxOrigin::External, &mut mem_pool)
+            self.add_transactions_to_pool(client, transactions, TxOrigin::External, &mut mem_pool, _importer_lock)
         };
 
         if !results.is_empty()
@@ -1008,6 +982,7 @@ impl MinerService for Miner {
         &self,
         chain: &C,
         tx: SignedTransaction,
+        _importer_lock: &MutexGuard<()>,
     ) -> Result<TransactionImportResult, Error> {
         ctrace!(OWN_PARCEL, "Importing transaction: {:?}", tx);
 
@@ -1016,7 +991,7 @@ impl MinerService for Miner {
             let mut mem_pool = self.mem_pool.write();
             // We need to re-validate transactions
             let import = self
-                .add_transactions_to_pool(chain, vec![tx.into()], TxOrigin::Local, &mut mem_pool)
+                .add_transactions_to_pool(chain, vec![tx.into()], TxOrigin::Local, &mut mem_pool, _importer_lock)
                 .pop()
                 .expect("one result returned per added transaction; one added => one result; qed");
 
@@ -1057,6 +1032,7 @@ impl MinerService for Miner {
         platform_address: PlatformAddress,
         passphrase: Option<Password>,
         seq: Option<u64>,
+        _importer_lock: &MutexGuard<()>,
     ) -> Result<(TxHash, u64), Error> {
         let address = platform_address.try_into_address()?;
         let seq = match seq {
@@ -1089,7 +1065,7 @@ impl MinerService for Miner {
         let unverified = UnverifiedTransaction::new(tx, sig);
         let signed = SignedTransaction::try_new(unverified)?;
         let hash = signed.hash();
-        self.import_own_transaction(client, signed)?;
+        self.import_own_transaction(client, signed, _importer_lock)?;
 
         Ok((hash, seq))
     }
@@ -1225,7 +1201,7 @@ pub mod test {
         );
 
         let transactions = vec![transaction1.clone(), transaction2, transaction1];
-        miner.add_transactions_to_pool(client.as_ref(), transactions, TxOrigin::Local, &mut mem_pool);
+        // miner.add_transactions_to_pool(client.as_ref(), transactions, TxOrigin::Local, &mut mem_pool);
     }
 
     fn generate_test_client(db: Arc<dyn KeyValueDB>, miner: Arc<Miner>, scheme: &Scheme) -> Result<Arc<Client>, Error> {

core/src/miner/mod.rs

@@ -25,6 +25,7 @@ mod work_notify;
 
 use std::ops::Range;
 
+use parking_lot::MutexGuard;
 use ckey::{Address, Password, PlatformAddress};
 use cstate::{FindActionHandler, TopStateView};
 use ctypes::transaction::IncompleteTransaction;
@@ -123,13 +124,15 @@ pub trait MinerService: Send + Sync {
         &self,
         client: &C,
         transactions: Vec<UnverifiedTransaction>,
+        _importer_lock: &MutexGuard<()>,
     ) -> Vec<Result<TransactionImportResult, Error>>;
 
     /// Imports own (node owner) transaction to mem pool.
     fn import_own_transaction<C: MiningBlockChainClient + EngineInfo + TermInfo + TermInfo>(
         &self,
         chain: &C,
         tx: SignedTransaction,
+        _importer_lock: &MutexGuard<()>,
     ) -> Result<TransactionImportResult, Error>;
 
     /// Imports incomplete (node owner) transaction to mem pool.
@@ -141,6 +144,7 @@ pub trait MinerService: Send + Sync {
         platform_address: PlatformAddress,
         passphrase: Option<Password>,
         seq: Option<u64>,
+        _importer_lock: &MutexGuard<()>,
     ) -> Result<(TxHash, u64), Error>;
 
     /// Get a list of all pending transactions in the mem pool.

rpc/src/v1/impls/account.rs

@@ -109,6 +109,7 @@ where
                 platform_address,
                 passphrase,
                 seq,
+                unimplemented!()
             )
             .map_err(errors::core)?;