Remove unnecessary lock around request submission

There is no need to lock the executor to submit request to pool. The
submission channel is MPMC queue which handles synchronization
internally. There was a non-atomic counter of requests processed, this
was fixed in execution.

Author: maxkozlovsky

Cargo.lock

@@ -20,7 +20,6 @@ alloy-sol-types = { workspace = true }
 futures = { workspace = true }
 hex = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
-tokio = { workspace = true }
 tracing = { workspace = true }
 
 [build-dependencies]

monad-ethcall/Cargo.toml

@@ -32,7 +32,6 @@ use monad_chain_config::{
     MONAD_TESTNET2_CHAIN_ID, MONAD_TESTNET_CHAIN_ID,
 };
 use serde::{Deserialize, Serialize};
-use tokio::sync::Mutex;
 use tracing::{info, warn};
 
 #[allow(dead_code, non_camel_case_types, non_upper_case_globals)]
@@ -191,7 +190,7 @@ pub async fn eth_call(
     sender: Address,
     block_number: u64,
     block_id: Option<[u8; 32]>,
-    eth_call_executor: Arc<Mutex<EthCallExecutor>>,
+    eth_call_executor: Arc<EthCallExecutor>,
     state_override_set: &StateOverrideSet,
     tracer: MonadTracer,
     gas_specified: bool,
@@ -308,15 +307,11 @@ pub async fn eth_call(
     let (send, recv) = channel();
     let sender_ctx = Box::new(SenderContext { sender: send });
 
-    // hold lock on executor while submitting the task
-    let executor_lock = eth_call_executor.lock().await;
-    let eth_call_executor = executor_lock.eth_call_executor;
-
     unsafe {
         let sender_ctx_ptr = Box::into_raw(sender_ctx);
 
         bindings::monad_eth_call_executor_submit(
-            eth_call_executor,
+            eth_call_executor.eth_call_executor,
             chain_config,
             rlp_encoded_tx.as_ptr(),
             rlp_encoded_tx.len(),
@@ -335,9 +330,6 @@ pub async fn eth_call(
         )
     };
 
-    // lock is dropped after the task has been submitted
-    drop(executor_lock);
-
     let result = match recv.await {
         Ok(r) => r,
         Err(e) => {

monad-ethcall/src/lib.rs

@@ -618,7 +618,7 @@ impl CallParams {
 #[tracing::instrument(level = "debug")]
 async fn prepare_eth_call<T: Triedb + TriedbPath>(
     triedb_env: &T,
-    eth_call_executor: Arc<Mutex<EthCallExecutor>>,
+    eth_call_executor: Arc<EthCallExecutor>,
     chain_id: u64,
     eth_call_provider_gas_limit: u64,
     mut params: CallParams,
@@ -735,7 +735,7 @@ async fn prepare_eth_call<T: Triedb + TriedbPath>(
 #[rpc(method = "eth_call", ignore = "chain_id", ignore = "eth_call_executor")]
 pub async fn monad_eth_call<T: Triedb + TriedbPath>(
     triedb_env: &T,
-    eth_call_executor: Arc<Mutex<EthCallExecutor>>,
+    eth_call_executor: Arc<EthCallExecutor>,
     chain_id: u64,
     eth_call_provider_gas_limit: u64,
     params: MonadEthCallParams,
@@ -771,7 +771,7 @@ pub async fn monad_eth_call<T: Triedb + TriedbPath>(
 #[allow(non_snake_case)]
 pub async fn monad_debug_traceCall<T: Triedb + TriedbPath>(
     triedb_env: &T,
-    eth_call_executor: Arc<Mutex<EthCallExecutor>>,
+    eth_call_executor: Arc<EthCallExecutor>,
     chain_id: u64,
     eth_call_gas_limit: u64,
     params: MonadDebugTraceCallParams,

monad-rpc/src/handlers/eth/call.rs

@@ -26,7 +26,6 @@ use monad_rpc_docs::rpc;
 use monad_triedb_utils::triedb_env::{BlockKey, FinalizedBlockKey, ProposedBlockKey, Triedb};
 use monad_types::{BlockId, Hash, SeqNum};
 use serde::Deserialize;
-use tokio::sync::Mutex;
 use tracing::trace;
 
 use crate::{
@@ -43,7 +42,7 @@ trait EthCallProvider {
     async fn eth_call(
         &self,
         txn: TxEnvelope,
-        eth_call_executor: Option<Arc<Mutex<EthCallExecutor>>>,
+        eth_call_executor: Option<Arc<EthCallExecutor>>,
     ) -> CallResult;
 }
 
@@ -80,7 +79,7 @@ impl EthCallProvider for GasEstimator {
     async fn eth_call(
         &self,
         txn: TxEnvelope,
-        eth_call_executor: Option<Arc<Mutex<EthCallExecutor>>>,
+        eth_call_executor: Option<Arc<EthCallExecutor>>,
     ) -> CallResult {
         let (block_number, block_id) = match self.block_key {
             BlockKey::Finalized(FinalizedBlockKey(SeqNum(n))) => (n, None),
@@ -111,7 +110,7 @@ impl EthCallProvider for GasEstimator {
 
 async fn estimate_gas<T: EthCallProvider>(
     provider: &T,
-    eth_call_executor: Option<Arc<Mutex<EthCallExecutor>>>,
+    eth_call_executor: Option<Arc<EthCallExecutor>>,
     call_request: &mut CallRequest,
     original_tx_gas: U256,
     provider_gas_limit: u64,
@@ -231,7 +230,7 @@ pub struct MonadEthEstimateGasParams {
 /// Generates and returns an estimate of how much gas is necessary to allow the transaction to complete.
 pub async fn monad_eth_estimateGas<T: Triedb>(
     triedb_env: &T,
-    eth_call_executor: Arc<Mutex<EthCallExecutor>>,
+    eth_call_executor: Arc<EthCallExecutor>,
     chain_id: u64,
     provider_gas_limit: u64,
     params: MonadEthEstimateGasParams,
@@ -467,11 +466,7 @@ mod tests {
     }
 
     impl EthCallProvider for MockGasEstimator {
-        async fn eth_call(
-            &self,
-            txn: TxEnvelope,
-            _: Option<Arc<Mutex<EthCallExecutor>>>,
-        ) -> CallResult {
+        async fn eth_call(&self, txn: TxEnvelope, _: Option<Arc<EthCallExecutor>>) -> CallResult {
             if txn.gas_limit() >= self.gas_used + self.gas_refund {
                 CallResult::Success(SuccessCallResult {
                     gas_used: self.gas_used,

monad-rpc/src/handlers/eth/gas.rs

@@ -23,7 +23,7 @@ use actix_web::{
 use monad_archive::prelude::ArchiveReader;
 use monad_ethcall::EthCallExecutor;
 use monad_triedb_utils::triedb_env::TriedbEnv;
-use tokio::sync::{Mutex, Semaphore};
+use tokio::sync::Semaphore;
 use tracing_actix_web::RootSpanBuilder;
 
 use super::eth::call::EthCallStatsTracker;
@@ -36,7 +36,7 @@ use crate::{
 pub struct MonadRpcResources {
     pub txpool_bridge_client: EthTxPoolBridgeClient,
     pub triedb_reader: Option<TriedbEnv>,
-    pub eth_call_executor: Option<Arc<Mutex<EthCallExecutor>>>,
+    pub eth_call_executor: Option<Arc<EthCallExecutor>>,
     pub eth_call_executor_fibers: usize,
     pub eth_call_stats_tracker: Option<Arc<EthCallStatsTracker>>,
     pub archive_reader: Option<ArchiveReader>,
@@ -62,7 +62,7 @@ impl MonadRpcResources {
     pub fn new(
         txpool_bridge_client: EthTxPoolBridgeClient,
         triedb_reader: Option<TriedbEnv>,
-        eth_call_executor: Option<Arc<Mutex<EthCallExecutor>>>,
+        eth_call_executor: Option<Arc<EthCallExecutor>>,
         eth_call_executor_fibers: usize,
         archive_reader: Option<ArchiveReader>,
         chain_id: u64,

monad-rpc/src/handlers/resources.rs

@@ -270,12 +270,12 @@ async fn main() -> std::io::Result<()> {
     };
 
     let eth_call_executor = args.triedb_path.clone().as_deref().map(|path| {
-        Arc::new(tokio::sync::Mutex::new(EthCallExecutor::new(
+        Arc::new(EthCallExecutor::new(
             low_pool_config,
             high_pool_config,
             args.eth_call_executor_node_lru_max_mem,
             path,
-        )))
+        ))
     });
 
     let meter_provider: Option<opentelemetry_sdk::metrics::SdkMeterProvider> =

monad-rpc/src/main.rs

@@ -47,7 +47,7 @@ pub fn rlp_decode_account(account_rlp: Vec<u8>) -> Option<EthAccount> {
         return None;
     };
 
-    let mut is_delegated = false;
+    let is_delegated = false;
     let code_hash = if buf.is_empty() {
         None
     } else {