Merge pull request #457 from jjyr/reduce-mem-pool-lock

perf: reduce mem-pool lock

crates/config/src/config.rs

@@ -238,4 +238,6 @@ pub struct StoreConfig {
     pub options: HashMap<String, String>,
     #[serde(default)]
     pub options_file: Option<PathBuf>,
+    #[serde(default)]
+    pub cache_size: Option<usize>,
 }

crates/db/src/db.rs

@@ -27,9 +27,10 @@ impl RocksDB {
         let cf_names: Vec<_> = (0..columns).map(|c| c.to_string()).collect();
 
         let (mut opts, cf_descriptors) = if let Some(ref file) = config.options_file {
-            let mut full_opts = FullOptions::load_from_file(file, None, false).map_err(|err| {
-                internal_error(format!("failed to load the options file: {}", err))
-            })?;
+            let mut full_opts = FullOptions::load_from_file(file, config.cache_size, false)
+                .map_err(|err| {
+                    internal_error(format!("failed to load the options file: {}", err))
+                })?;
             let cf_names_str: Vec<&str> = cf_names.iter().map(|s| s.as_str()).collect();
             full_opts
                 .complete_column_families(&cf_names_str, false)
@@ -247,6 +248,7 @@ mod tests {
                 opts
             },
             options_file: None,
+            cache_size: None,
         };
         RocksDB::open(&config, 2); // no panic
     }
@@ -261,6 +263,7 @@ mod tests {
             path: tmp_dir.as_ref().to_path_buf(),
             options: HashMap::new(),
             options_file: None,
+            cache_size: None,
         };
         RocksDB::open(&config, 2); // no panic
     }
@@ -280,6 +283,7 @@ mod tests {
                 opts
             },
             options_file: None,
+            cache_size: None,
         };
         RocksDB::open(&config, 2); // panic
     }

crates/mem-pool/src/pool.rs

@@ -75,6 +75,24 @@ impl Default for OutputParam {
     }
 }
 
+pub fn fetch_state_db_with_mode<'a>(
+    db: &'a StoreTransaction,
+    mode: MemBlockDBMode,
+    block_number: u64,
+) -> Result<StateDBTransaction<'a>> {
+    // Repackage offset must be smaller than NewBlock, so that it has clean state.
+    let db_mode = match mode {
+        MemBlockDBMode::NewBlock => StateDBMode::Write(WriteContext::new(u32::MAX)),
+        MemBlockDBMode::Package => StateDBMode::Write(WriteContext::new(0)),
+    };
+    StateDBTransaction::from_checkpoint(
+        db,
+        CheckPoint::new(block_number, SubState::MemBlock),
+        db_mode,
+    )
+    .map_err(|err| anyhow!("err: {}", err))
+}
+
 /// MemPool
 pub struct MemPool {
     /// store
@@ -241,17 +259,7 @@ impl MemPool {
         db: &'a StoreTransaction,
         mode: MemBlockDBMode,
     ) -> Result<StateDBTransaction<'a>> {
-        // Repackage offset must be smaller than NewBlock, so that it has clean state.
-        let db_mode = match mode {
-            MemBlockDBMode::NewBlock => StateDBMode::Write(WriteContext::new(u32::MAX)),
-            MemBlockDBMode::Package => StateDBMode::Write(WriteContext::new(0)),
-        };
-        StateDBTransaction::from_checkpoint(
-            db,
-            CheckPoint::new(self.current_tip.1, SubState::MemBlock),
-            db_mode,
-        )
-        .map_err(|err| anyhow!("err: {}", err))
+        fetch_state_db_with_mode(db, mode, self.current_tip.1)
     }
 
     /// Push a layer2 tx into pool

crates/rpc-server/src/registry.rs

@@ -17,7 +17,7 @@ use gw_jsonrpc_types::{
     },
     test_mode::{ShouldProduceBlock, TestModePayload},
 };
-use gw_mem_pool::custodian::AvailableCustodians;
+use gw_mem_pool::{custodian::AvailableCustodians, pool::MemBlockDBMode};
 use gw_rpc_client::rpc_client::RPCClient;
 use gw_store::{
     chain_view::ChainView,
@@ -407,7 +407,6 @@ impl TryFrom<u8> for GetTxVerbose {
 async fn get_transaction(
     Params(param): Params<GetTxParams>,
     store: Data<Store>,
-    mem_pool: Data<MemPool>,
 ) -> Result<Option<L2TransactionWithStatus>, RpcError> {
     let (tx_hash, verbose) = match param {
         GetTxParams::Default((tx_hash,)) => (to_h256(tx_hash), GetTxVerbose::TxWithStatus),
@@ -439,20 +438,7 @@ async fn get_transaction(
             status = L2TransactionStatus::Committed;
         }
         None => {
-            tx_opt = match db.get_mem_pool_transaction(&tx_hash)? {
-                Some(tx) => Some(tx),
-                None => {
-                    // the tx maybe in the mem-pool but not finalized
-                    // so we try to sync with mem-pool, then fetch from db again
-                    if let Some(mem_pool) = mem_pool.as_ref() {
-                        // we only need to sync with mem-pool, wait for tx get finalized.
-                        mem_pool.lock().await;
-                        db.get_mem_pool_transaction(&tx_hash)?.map(Into::into)
-                    } else {
-                        None
-                    }
-                }
-            };
+            tx_opt = db.get_mem_pool_transaction(&tx_hash)?;
             status = L2TransactionStatus::Pending;
         }
     };
@@ -555,7 +541,6 @@ async fn get_tip_block_hash(store: Data<Store>) -> Result<JsonH256> {
 async fn get_transaction_receipt(
     Params((tx_hash,)): Params<(JsonH256,)>,
     store: Data<Store>,
-    mem_pool: Data<MemPool>,
 ) -> Result<Option<TxReceipt>> {
     let tx_hash = to_h256(tx_hash);
     let db = store.begin_transaction();
@@ -567,31 +552,18 @@ async fn get_transaction_receipt(
         return Ok(Some(receipt));
     }
     // search from mem pool
-    match db.get_mem_pool_transaction_receipt(&tx_hash)? {
-        Some(receipt) => Ok(Some(receipt.into())),
-        None => {
-            // the tx maybe in the mem-pool but not finalized
-            // so we try to sync with mem-pool, then fetch from db again
-            if let Some(mem_pool) = mem_pool.as_ref() {
-                // we only need to sync with mem-pool, wait for tx get finalized.
-                mem_pool.lock().await;
-                let receipt_opt = db
-                    .get_mem_pool_transaction_receipt(&tx_hash)?
-                    .map(Into::into);
-                Ok(receipt_opt)
-            } else {
-                Ok(None)
-            }
-        }
-    }
+    Ok(db
+        .get_mem_pool_transaction_receipt(&tx_hash)?
+        .map(Into::into))
 }
 
 async fn execute_l2transaction(
     Params((l2tx,)): Params<(JsonBytes,)>,
     mem_pool: Data<MemPool>,
+    generator: Data<Generator>,
     store: Data<Store>,
 ) -> Result<RunResult, RpcError> {
-    let mem_pool = match &*mem_pool {
+    let _mem_pool = match &*mem_pool {
         Some(mem_pool) => mem_pool,
         None => {
             return Err(mem_pool_is_disabled_err());
@@ -615,8 +587,26 @@ async fn execute_l2transaction(
         .build();
 
     let mut run_result = {
-        let mem_pool = mem_pool.lock().await;
-        mem_pool.unchecked_execute_transaction(&tx, &block_info)?
+        let tip_block_hash = store.get_tip_block_hash()?;
+        let db = store.begin_transaction();
+        let state_db =
+            gw_mem_pool::pool::fetch_state_db_with_mode(&db, MemBlockDBMode::NewBlock, number)?;
+        let state = state_db.state_tree()?;
+        let chain_view = ChainView::new(&db, tip_block_hash);
+        // verify tx signature
+        generator.check_transaction_signature(&state, &tx)?;
+        // tx basic verification
+        generator.verify_transaction(&state, &tx)?;
+        // execute tx
+        let raw_tx = tx.raw();
+        let run_result = generator.unchecked_execute_transaction(
+            &chain_view,
+            &state,
+            &block_info,
+            &raw_tx,
+            100000000,
+        )?;
+        run_result
     };
 
     if run_result.exit_code != 0 {

crates/tools/src/generate_config.rs

@@ -253,6 +253,7 @@ pub fn generate_node_config(args: GenerateNodeConfigArgs) -> Result<Config> {
         path: "".into(),
         options: HashMap::new(),
         options_file: None,
+        cache_size: None,
     };
     let genesis_committed_info = L2BlockCommittedInfo {
         block_hash,