Repository: Phala-Network/phala-blockchain (language: rust)

Issue:
Concurrent query issues in pRuntime
During my recent testing, I found two concurrency issues with pRuntime:

- [ ]  When a query is blocked inside the contract, the next query will get stuck (unable to enter the rocket).
    
    We didn't anticipate this problem before. Even though we have 8 rocket threads, only two concurrent requests can cause the blockage. We thought that Tokio's work-stealing mechanism would distribute the waiting tasks in the blocked thread to other threads, but in fact, it doesn't.
    
    This problem can be solved by using tokio::spawn_blocking().
    
- [ ]  When pRuntime is synchronizing blocks and executing transactions, queries cannot be processed because Phactory is locked.
    This is a known issue. When executing transactions, the global Phactory instance is locked. Almost all of the pRPC requests lock Phactory to retrieve the state, including queries. Most RPC release Phacktory quickly after retrieving the required state because it's read-only. However, executing transactions requires modifying the state, which locks it for a long duration, making it impossible to process queries(and any other requests) at the same time.
    To solve this problem, we may need to organize and isolate the states in Phactory, use some kind of RCU lock mechanism, or change to a fine-grained lock to minimize the locking time of transactions.

<details>
  <summary>Chinese</summary>
  <p>
我这两天测试发现pRuntime存在的两个并发问题：

- 当一个query阻塞在contract里面后，下一个query会卡住（无法进入rocket）
    
    这个问题之前没有预料到，即使我们开8个rocket线程，仅仅两个并发请求就会卡住第二个。一直认为tokio的work stealing机制会将阻塞线程中的堆积任务分派到其它线程去，事实上并不能。
    
    此问题可以使用tokio::spawn_blocking()解决。
    
- 当pruntime正在同步区块执行交易时，因为Phactory被加锁而无法处理query
    
    这个是已知问题。当执行交易时，会将全局Phactory实例锁住，会更新里面较多的状态。我们几乎所有请求都会锁Phactory取状态，query也会锁，只是取出需要的状态后很快释放，因为它是只读的。而执行交易需要修改状态，会加锁较长时间，导致无法同时处理query。
    
    要解决这个问题可能需要对Phactory中的状态进行梳理隔离，通过某种RCU锁机制，或改成细粒度锁，尽量缩短tx的加锁时间
  </p>
</details>


Discussion / hints:

Fix patch:
diff --git a/Cargo.lock b/Cargo.lock
index 42325127bb..6ce46b08dc 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -7526,6 +7526,7 @@ dependencies = [
  "hex",
  "hex-literal",
  "hex_fmt",
+ "im",
  "insta",
  "itertools",
  "log",
@@ -7584,6 +7585,7 @@ dependencies = [
  "base64 0.13.0",
  "derive_more",
  "frame-system",
+ "im",
  "insta",
  "log",
  "parity-scale-codec",
@@ -7657,6 +7659,7 @@ dependencies = [
  "derive_more",
  "environmental",
  "hex",
+ "im",
  "log",
  "parity-scale-codec",
  "phala-serde-more",
@@ -7664,6 +7667,7 @@ dependencies = [
  "serde",
  "sp-core",
  "spin 0.9.3",
+ "tracing",
 ]
 
 [[package]]
diff --git a/crates/phactory/Cargo.toml b/crates/phactory/Cargo.toml
index d4365aafd1..f8b14320b4 100644
--- a/crates/phactory/Cargo.toml
+++ b/crates/phactory/Cargo.toml
@@ -85,6 +85,7 @@ reqwest-env-proxy = { path = "../reqwest-env-proxy" }
 
 environmental = "1"
 once_cell = "1"
+im = "15"
 
 [dev-dependencies]
 insta = "1.7.2"
diff --git a/crates/phactory/api/Cargo.toml b/crates/phactory/api/Cargo.toml
index 99fdf5945a..4f91737ef8 100644
--- a/crates/phactory/api/Cargo.toml
+++ b/crates/phactory/api/Cargo.toml
@@ -34,6 +34,7 @@ reqwest = { version = "0.11", optional = true, default-features = false, feature
 
 primitive-types = { version = "0.12.1", optional = true, default-features = false }
 serde_json = "1.0.79"
+im = "15"
 
 [dev-dependencies]
 insta = "1.13.0"
diff --git a/crates/phactory/api/src/ecall_args.rs b/crates/phactory/api/src/ecall_args.rs
index 1bcb34d66e..76f9c05288 100644
--- a/crates/phactory/api/src/ecall_args.rs
+++ b/crates/phactory/api/src/ecall_args.rs
@@ -39,6 +39,9 @@ pub struct InitArgs {
 
     /// Only sync blocks into pruntime without dispatching messages.
     pub safe_mode_level: u8,
+
+    /// Disable the RCU policy to update the Phactory state.
+    pub no_rcu: bool,
 }
 
 pub use phala_git_revision::git_revision;
diff --git a/crates/phactory/api/src/storage_sync.rs b/crates/phactory/api/src/storage_sync.rs
index b5b5b268a5..318c10fda7 100644
--- a/crates/phactory/api/src/storage_sync.rs
+++ b/crates/phactory/api/src/storage_sync.rs
@@ -2,7 +2,7 @@ use super::blocks::{
     AuthoritySetChange, BlockHeaderWithChanges, HeaderToSync, RuntimeHasher, StorageProof,
 };
 
-use alloc::collections::VecDeque;
+use im::Vector as VecDeque;
 use alloc::string::String;
 use alloc::vec::Vec;
 use chain::Hash;
@@ -274,7 +274,7 @@ pub struct Counters {
     pub waiting_for_paraheaders: bool,
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Clone)]
 pub struct SolochainSynchronizer<Validator> {
     sync_state: BlockSyncState<Validator>,
     state_roots: VecDeque<Hash>,
@@ -346,7 +346,7 @@ impl<Validator: BlockValidator> StorageSynchronizer for SolochainSynchronizer<Va
     }
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Clone)]
 pub struct ParachainSynchronizer<Validator> {
     sync_state: BlockSyncState<Validator>,
     last_relaychain_state_root: Option<Hash>,
@@ -475,7 +475,7 @@ impl<Validator: BlockValidator> StorageSynchronizer for ParachainSynchronizer<Va
 
 // We create this new type to help serialize the original dyn StorageSynchronizer.
 // Because it it impossible to impl Serialize/Deserialize for dyn StorageSynchronizer.
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Clone)]
 pub enum Synchronizer<Validator> {
     Solo(SolochainSynchronizer<Validator>),
     Para(ParachainSynchronizer<Validator>),
diff --git a/crates/phactory/src/benchmark.rs b/crates/phactory/src/benchmark.rs
index 8db616b847..17d30dec07 100644
--- a/crates/phactory/src/benchmark.rs
+++ b/crates/phactory/src/benchmark.rs
@@ -106,6 +106,10 @@ pub fn paused() -> bool {
     !FLAGS.lock().unwrap().is_empty()
 }
 
+pub fn syncing() -> bool {
+    check_flag(Flags::SYNCING)
+}
+
 pub fn set_flag(flag: Flags, on: bool) {
     let mut guard = FLAGS.lock().unwrap();
     if on {
diff --git a/crates/phactory/src/bin_api_service.rs b/crates/phactory/src/bin_api_service.rs
index c89bbeb373..1775c362b8 100644
--- a/crates/phactory/src/bin_api_service.rs
+++ b/crates/phactory/src/bin_api_service.rs
@@ -50,7 +50,7 @@ impl<Platform: pal::Platform + Serialize + DeserializeOwned> Phactory<Platform>
         req_id: u64,
         input: blocks::DispatchBlockReq,
     ) -> Result<Value, Value> {
-        let resp = self.dispatch_block(req_id, input.blocks).map_err(display)?;
+        let resp = self.dispatch_blocks(req_id, input.blocks).map_err(display)?;
         Ok(json!({ "dispatched_to": resp.synced_to }))
     }
 
diff --git a/crates/phactory/src/contracts/pink.rs b/crates/phactory/src/contracts/pink.rs
index 63c6302f42..4e2f5cb6ba 100644
--- a/crates/phactory/src/contracts/pink.rs
+++ b/crates/phactory/src/contracts/pink.rs
@@ -509,7 +509,7 @@ impl Cluster {
     }
 
     pub(crate) async fn handle_query(
-        &mut self,
+        mut self,
         contract_id: &AccountId,
         origin: Option<&AccountId>,
         req: Query,
@@ -552,27 +552,34 @@ impl Cluster {
                     context.req_id,
                 );
                 let log_handler = context.log_handler.clone();
-                context::using(&mut ctx, move || {
-                    let mut runtime = self.runtime_mut(log_handler);
-                    if deposit > 0 {
-                        runtime.deposit(origin.clone(), deposit);
-                    }
-                    let args = TransactionArguments {
-                        origin,
-                        transfer,
-                        gas_limit: WEIGHT_REF_TIME_PER_SECOND * 10,
-                        gas_free: true,
-                        storage_deposit_limit: None,
-                    };
-                    let ink_result =
-                        runtime.contract_call(contract_id.clone(), input_data, mode, args);
-                    let effects = if mode.is_estimating() {
-                        None
-                    } else {
-                        runtime.effects.take().map(|e| e.into_query_only_effects())
-                    };
-                    Ok((Response::Payload(ink_result), effects))
+                let span = tracing::Span::current();
+                let contract_id = contract_id.clone();
+                tokio::task::spawn_blocking(move || {
+                    let _guard = span.enter();
+                    context::using(&mut ctx, move || {
+                        let mut runtime = self.runtime_mut(log_handler);
+                        if deposit > 0 {
+                            runtime.deposit(origin.clone(), deposit);
+                        }
+                        let args = TransactionArguments {
+                            origin,
+                            transfer,
+                            gas_limit: WEIGHT_REF_TIME_PER_SECOND * 10,
+                            gas_free: true,
+                            storage_deposit_limit: None,
+                        };
+                        let ink_result =
+                            runtime.contract_call(contract_id, input_data, mode, args);
+                        let effects = if mode.is_estimating() {
+                            None
+                        } else {
+                            runtime.effects.take().map(|e| e.into_query_only_effects())
+                        };
+                        Ok((Response::Payload(ink_result), effects))
+                    })
                 })
+                .await
+                .map_err(|_| QueryError::RuntimeError("Failed to spawn blocking task".into()))?
             }
             Query::SidevmQuery(payload) => {
                 let handle = context
diff --git a/crates/phactory/src/contracts/support.rs b/crates/phactory/src/contracts/support.rs
index e0f3b634ae..eab1f1e5a4 100644
--- a/crates/phactory/src/contracts/support.rs
+++ b/crates/phactory/src/contracts/support.rs
@@ -105,7 +105,7 @@ impl<'de> Deserialize<'de> for SidevmHandle {
     }
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Clone)]
 struct SidevmInfo {
     code: Vec<u8>,
     code_hash: H256,
@@ -119,7 +119,7 @@ pub(crate) enum SidevmCode {
     Code(Vec<u8>),
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Clone)]
 pub struct FatContract {
     send_mq: SignedMessageChannel,
     cmd_rcv_mq: SecretReceiver<RawData>,
diff --git a/crates/phactory/src/contracts/support/keeper.rs b/crates/phactory/src/contracts/support/keeper.rs
index 8cc717e4c8..d851dbf754 100644
--- a/crates/phactory/src/contracts/support/keeper.rs
+++ b/crates/phactory/src/contracts/support/keeper.rs
@@ -1,9 +1,9 @@
+use im::OrdMap as BTreeMap;
 use pink::types::AccountId;
 use serde::{Deserialize, Serialize};
 use sidevm::service::Spawner;
-use std::collections::BTreeMap;
 
-use crate::contracts::FatContract;
+use crate::{contracts::FatContract, im_helpers::ordmap_for_each_mut};
 
 type ContractMap = BTreeMap<AccountId, FatContract>;
 
@@ -14,6 +14,15 @@ pub struct ContractsKeeper {
     pub(crate) weight_changed: bool,
 }
 
+impl Clone for ContractsKeeper {
+    fn clone(&self) -> Self {
+        Self {
+            contracts: self.contracts.clone(),
+            weight_changed: self.weight_changed,
+        }
+    }
+}
+
 impl ContractsKeeper {
     pub fn insert(&mut self, contract: FatContract) {
         self.contracts.insert(contract.address().clone(), contract);
@@ -36,15 +45,17 @@ impl ContractsKeeper {
     }
 
     pub fn try_restart_sidevms(&mut self, spawner: &Spawner) {
-        for contract in self.contracts.values_mut() {
+        ordmap_for_each_mut(&mut self.contracts, |(_k, contract)| {
             if let Err(err) = contract.restart_sidevm_if_needed(spawner) {
                 error!("Failed to restart sidevm instance: {:?}", err);
             }
-        }
+        });
     }
 
     pub fn drain(&mut self) -> impl Iterator<Item = FatContract> {
-        std::mem::take(&mut self.contracts).into_values()
+        std::mem::take(&mut self.contracts)
+            .into_iter()
+            .map(|(_, v)| v)
     }
 
     pub fn iter(&self) -> impl Iterator<Item = (&AccountId, &FatContract)> {
diff --git a/crates/phactory/src/im_helpers.rs b/crates/phactory/src/im_helpers.rs
new file mode 100644
index 0000000000..9f125589cf
--- /dev/null
+++ b/crates/phactory/src/im_helpers.rs
@@ -0,0 +1,9 @@
+pub fn ordmap_for_each_mut<K: Ord + Clone, V: Clone>(
+    map: &mut im::OrdMap<K, V>,
+    mut f: impl FnMut((&K, &mut V)),
+) {
+    let snapshot = map.clone();
+    for key in snapshot.keys() {
+        f((key, map.get_mut(key).unwrap()));
+    }
+}
diff --git a/crates/phactory/src/lib.rs b/crates/phactory/src/lib.rs
index f529e1b752..df674f872d 100644
--- a/crates/phactory/src/lib.rs
+++ b/crates/phactory/src/lib.rs
@@ -17,7 +17,7 @@ use serde::{
 };
 
 use crate::light_validation::LightValidation;
-use std::{collections::BTreeMap, str::FromStr};
+use std::{collections::BTreeMap, str::FromStr, sync::Arc};
 use std::{fs::File, io::ErrorKind, path::PathBuf};
 use std::{io::Write, marker::PhantomData};
 use std::{path::Path, str};
@@ -64,6 +64,7 @@ pub mod benchmark;
 mod bin_api_service;
 mod contracts;
 mod cryptography;
+mod im_helpers;
 mod light_validation;
 mod prpc_service;
 mod secret_channel;
@@ -75,7 +76,7 @@ mod types;
 // runtime definition locally.
 type RuntimeHasher = <chain::Runtime as frame_system::Config>::Hashing;
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Clone)]
 struct RuntimeState {
     send_mq: MessageSendQueue,
 
@@ -213,12 +214,12 @@ enum RuntimeDataSeal {
     V1(PersistentRuntimeData),
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Clone)]
 #[serde(bound(deserialize = "Platform: Deserialize<'de>"))]
 pub struct Phactory<Platform> {
     platform: Platform,
     #[serde(skip)]
-    pub args: InitArgs,
+    pub args: Arc<InitArgs>,
     dev_mode: bool,
     attestation_provider: Option<AttestationProvider>,
     machine_id: Vec<u8>,
@@ -251,6 +252,16 @@ pub struct Phactory<Platform> {
     #[serde(skip)]
     trusted_sk: bool,
 
+    #[serde(skip)]
+    #[serde(default = "Instant::now")]
+    uptime: Instant,
+
+    #[serde(skip)]
+    pub(crate) rcu_dispatching: bool,
+
+    #[serde(skip)]
+    pub(crate) pending_effects: Vec<::pink::types::ExecSideEffects>,
+
     #[serde(skip)]
     #[serde(default = "Instant::now")]
     started_at: Instant,
@@ -283,6 +294,9 @@ impl<Platform: pal::Platform> Phactory<Platform> {
             netconfig: Default::default(),
             can_load_chain_state: false,
             trusted_sk: false,
+            uptime: Instant::now(),
+            rcu_dispatching: false,
+            pending_effects: Vec::new(),
             started_at: Instant::now(),
         }
     }
@@ -293,11 +307,11 @@ impl<Platform: pal::Platform> Phactory<Platform> {
         }
 
         self.can_load_chain_state = !system::gk_master_key_exists(&args.sealing_path);
-        self.args = args;
+        self.args = Arc::new(args);
     }
 
     pub fn set_args(&mut self, args: InitArgs) {
-        self.args = args;
+        self.args = Arc::new(args);
         if let Some(system) = &mut self.system {
             system.sealing_path = self.args.sealing_path.clone();
             system.storage_path = self.args.storage_path.clone();
diff --git a/crates/phactory/src/light_validation/mod.rs b/crates/phactory/src/light_validation/mod.rs
index 395ec83d7c..1694cded9e 100644
--- a/crates/phactory/src/light_validation/mod.rs
+++ b/crates/phactory/src/light_validation/mod.rs
@@ -39,7 +39,7 @@ mod justification;
 pub mod storage_proof;
 mod types;
 
-use std::collections::BTreeMap;
+use im::OrdMap as BTreeMap;
 use std::fmt;
 
 use anyhow::Result;
diff --git a/crates/phactory/src/prpc_service.rs b/crates/phactory/src/prpc_service.rs
index 511812d130..01e71c683e 100644
--- a/crates/phactory/src/prpc_service.rs
+++ b/crates/phactory/src/prpc_service.rs
@@ -28,8 +28,7 @@ use phala_types::{
     WorkerEndpointPayload, WorkerPublicKey, WorkerRegistrationInfoV2,
 };
 use sp_application_crypto::UncheckedFrom;
-use tokio::sync::oneshot::{channel, Sender};
-use tracing::{info, error};
+use tracing::{error, info};
 
 type RpcResult<T> = Result<T, RpcError>;
 
@@ -249,7 +248,7 @@ impl<Platform: pal::Platform + Serialize + DeserializeOwned> Phactory<Platform>
         })
     }
 
-    pub(crate) fn dispatch_block(
+    pub(crate) fn dispatch_blocks(
         &mut self,
         req_id: u64,
         mut blocks: Vec<blocks::BlockHeaderWithChanges>,
@@ -572,8 +571,9 @@ impl<Platform: pal::Platform + Serialize + DeserializeOwned> Phactory<Platform>
         &mut self,
         req_id: u64,
         request: pb::ContractQueryRequest,
-        effects_queue: Sender<ExecSideEffects>,
-    ) -> RpcResult<impl Future<Output = RpcResult<pb::ContractQueryResponse>>> {
+    ) -> RpcResult<
+        impl Future<Output = RpcResult<(pb::ContractQueryResponse, Option<ExecSideEffects>)>>,
+    > {
         if self.args.safe_mode_level > 0 {
             return Err(from_display("Query is unavailable in safe mode"));
         }
@@ -639,13 +639,6 @@ impl<Platform: pal::Platform + Serialize + DeserializeOwned> Phactory<Platform>
 
         Ok(async move {
             let (response, effects) = query_future.await?;
-
-            if let Some(effects) = effects {
-                effects_queue
-                    .send(effects)
-                    .map_err(|_| from_display("Failed to apply side effects"))?;
-            }
-
             let response = contract::ContractQueryResponse {
                 nonce: head.nonce,
                 result: contract::Data(response),
@@ -661,7 +654,7 @@ impl<Platform: pal::Platform + Serialize + DeserializeOwned> Phactory<Platform>
             )
             .map_err(from_debug)?;
 
-            Ok(pb::ContractQueryResponse::new(encrypted_resp))
+            Ok((pb::ContractQueryResponse::new(encrypted_resp), effects))
         })
     }
 
@@ -1041,15 +1034,20 @@ impl<T> Drop for LogOnDrop<T> {
 }
 
 impl<Platform: pal::Platform> RpcService<Platform> {
-    pub fn lock_phactory(&self) -> LogOnDrop<MutexGuard<'_, Phactory<Platform>>> {
+    pub fn lock_phactory(
+        &self,
+        allow_rcu: bool,
+    ) -> RpcResult<LogOnDrop<MutexGuard<'_, Phactory<Platform>>>> {
         debug!(target: "phactory::lock", "Locking phactory...");
         let guard = self.phactory.lock().unwrap();
         debug!(target: "phactory::lock", "Locked phactory");
-
-        LogOnDrop {
+        if !allow_rcu && guard.rcu_dispatching {
+            return Err(from_display("RCU in progress, please try the request again later"));
+        }
+        Ok(LogOnDrop {
             inner: guard,
             msg: "Unlocked phactory",
-        }
+        })
     }
 }
 
@@ -1080,7 +1078,7 @@ fn create_attestation_report_on<Platform: pal::Platform>(
 impl<Platform: pal::Platform + Serialize + DeserializeOwned> PhactoryApi for RpcService<Platform> {
     /// Get basic information about Phactory state.
     async fn get_info(&mut self, _request: ()) -> RpcResult<pb::PhactoryInfo> {
-        let info = self.lock_phactory().get_info();
+        let info = self.lock_phactory(true)?.get_info();
         info!("Got info: {:?}", info.debug_info());
         Ok(info)
     }
@@ -1089,7 +1087,7 @@ impl<Platform: pal::Platform + Serialize + DeserializeOwned> PhactoryApi for Rpc
     async fn sync_header(&mut self, request: pb::HeadersToSync) -> RpcResult<pb::SyncedTo> {
         let headers = request.decode_headers()?;
         let authority_set_change = request.decode_authority_set_change()?;
-        self.lock_phactory()
+        self.lock_phactory(false)?
             .sync_header(headers, authority_set_change)
     }
 
@@ -1099,7 +1097,7 @@ impl<Platform: pal::Platform + Serialize + DeserializeOwned> PhactoryApi for Rpc
         request: pb::ParaHeadersToSync,
     ) -> RpcResult<pb::SyncedTo> {
         let headers = request.decode_headers()?;
-        self.lock_phactory()
+        self.lock_phactory(false)?
             .sync_para_header(headers, request.proof)
     }
 
@@ -1107,7 +1105,7 @@ impl<Platform: pal::Platform + Serialize + DeserializeOwned> PhactoryApi for Rpc
         &mut self,
         request: pb::CombinedHeadersToSync,
     ) -> Result<pb::HeadersSyncedTo, prpc::server::Error> {
-        self.lock_phactory().sync_combined_headers(
+        self.lock_phactory(false)?.sync_combined_headers(
             request.decode_relaychain_headers()?,
             request.decode_authority_set_change()?,
             request.decode_parachain_headers()?,
@@ -1118,14 +1116,54 @@ impl<Platform: pal::Platform + Serialize + DeserializeOwned> PhactoryApi for Rpc
     /// Dispatch blocks (Sync storage changes)"
     async fn dispatch_blocks(&mut self, request: pb::Blocks) -> RpcResult<pb::SyncedTo> {
         let blocks = request.decode_blocks()?;
-        self.lock_phactory().dispatch_block(self.req_id, blocks)
+        let mut phactory = {
+            let mut phactory = self.lock_phactory(false)?;
+            if phactory.args.no_rcu || benchmark::syncing() {
+                // If RCU way is not suitable here, we do the traditional locked dispatch.
+                return phactory.dispatch_blocks(self.req_id, blocks);
+            }
+            // Otherwise, we clone the phactory and execute/dispatch the events of the blocks with
+            // a cloned phactory without locking(the lock would be released in the end of this scope)
+            // the singleton phactory. This way, we can avoid blocking the RPC server.
+            info!("Cloning Phactory to do RCU dispatch...");
+            let cloned = phactory.clone();
+            // We set rcu_dispatching = true to avoid a reentrant call to dispatch_blocks which
+            // would cause a state inconsistency.
+            phactory.rcu_dispatching = true;
+            cloned
+        };
+        // Start to dispatch the blocks with the cloned phactory without locking the singleton phactory.
+        info!("Unlocked Phactory, dispatching blocks...");
+        let req_id = self.req_id;
+        let span = tracing::Span::current();
+        let (synced_to, phactory) = tokio::task::spawn_blocking(move || {
+            let _guard = span.enter();
+            let synced_to = phactory.dispatch_blocks(req_id, blocks);
+            (synced_to, phactory)
+        })
+        .await
+        .expect("Dispatch blocks failed");
+        info!("Done, writing state back to Phactory...");
+        let mut guard = self.lock_phactory(true).unwrap();
+        let pending_effects = std::mem::take(&mut guard.pending_effects);
+        // While putting the cloned phactory back, the rcu_dispatching flag is also overwritten with
+        // its original value `false`.
+        **guard = phactory;
+        if !pending_effects.is_empty() {
+            // Apply pending effects with the singleton phactory locked.
+            tracing::info!(count = pending_effects.len(), "Applying pending effects");
+            for effects in pending_effects {
+                guard.apply_side_effects(effects);
+            }
+        }
+        synced_to
     }
 
     async fn init_runtime(
         &mut self,
         request: pb::InitRuntimeRequest,
     ) -> RpcResult<pb::InitRuntimeResponse> {
-        self.lock_phactory().init_runtime(
+        self.lock_phactory(false)?.init_runtime(
             request.is_parachain,
             request.decode_genesis_info()?,
             request.decode_genesis_state()?,
@@ -1139,12 +1177,12 @@ impl<Platform: pal::Platform + Serialize + DeserializeOwned> PhactoryApi for Rpc
         &mut self,
         req: pb::GetRuntimeInfoRequest,
     ) -> RpcResult<pb::InitRuntimeResponse> {
-        self.lock_phactory()
+        self.lock_phactory(true)?
             .get_runtime_info(req.force_refresh_ra, req.decode_operator()?)
     }
 
     async fn get_egress_messages(&mut self, _: ()) -> RpcResult<pb::GetEgressMessagesResponse> {
-        self.lock_phactory()
+        self.lock_phactory(true)?
             .get_egress_messages()
             .map(pb::GetEgressMessagesResponse::new)
     }
@@ -1153,24 +1191,37 @@ impl<Platform: pal::Platform + Serialize + DeserializeOwned> PhactoryApi for Rpc
         &mut self,
         request: pb::ContractQueryRequest,
     ) -> RpcResult<pb::ContractQueryResponse> {
-        let (tx, rx) = channel();
-        let phactory = self.phactory.clone();
-        tokio::spawn(async move {
-            if let Ok(effects) = rx.await {
-                phactory.lock().unwrap().apply_side_effects(effects);
-            }
-        });
         let query_fut = self
-            .lock_phactory()
-            .contract_query(self.req_id, request, tx)?;
-        query_fut.await
+            .lock_phactory(true)?
+            .contract_query(self.req_id, request)?;
+        let (response, effects) = query_fut.await?;
+        'apply_effects: {
+            let Some(effects) = effects else {
+                break 'apply_effects;
+            };
+            if effects.is_empty() {
+                break 'apply_effects;
+            }
+            let mut phactory = self.lock_phactory(true)?;
+            if phactory.rcu_dispatching {
+                const MAX_PENDING: usize = 64;
+                if phactory.pending_effects.len() >= MAX_PENDING {
+                    error!("Too many pending effects, dropping this");
+                    return Err(from_display("Too many pending effects"));
+                }
+                phactory.pending_effects.push(effects);
+            } else {
+                phactory.apply_side_effects(effects);
+            }
+        }
+        Ok(response)
     }
 
     async fn get_worker_state(
         &mut self,
         request: pb::GetWorkerStateRequest,
     ) -> RpcResult<pb::WorkerState> {
-        let mut phactory = self.lock_phactory();
+        let mut phactory = self.lock_phactory(true)?;
         let system = phactory.system()?;
         let gk = system
             .gatekeeper
@@ -1192,28 +1243,28 @@ impl<Platform: pal::Platform + Serialize + DeserializeOwned> PhactoryApi for Rpc
         &mut self,
         request: pb::AddEndpointRequest,
     ) -> RpcResult<pb::GetEndpointResponse> {
-        self.lock_phactory()
+        self.lock_phactory(false)?
             .add_endpoint(request.decode_endpoint_type()?, request.endpoint)
     }
 
     async fn refresh_endpoint_signing_time(&mut self, _: ()) -> RpcResult<pb::GetEndpointResponse> {
-        self.lock_phactory().sign_endpoints()
+        self.lock_phactory(false)?.sign_endpoints()
     }
 
     async fn get_endpoint_info(&mut self, _: ()) -> RpcResult<pb::GetEndpointResponse> {
-        self.lock_phactory().get_endpoint_info()
+        self.lock_phactory(true)?.get_endpoint_info()
     }
 
     async fn sign_endpoint_info(
         &mut self,
         request: pb::SignEndpointsRequest,
     ) -> Result<pb::GetEndpointResponse, prpc::server::Error> {
-        self.lock_phactory()
+        self.lock_phactory(true)?
             .sign_endpoint_info(VersionedWorkerEndpoints::V1(request.decode_endpoints()?))
     }
 
     async fn derive_phala_i2p_key(&mut self, _: ()) -> RpcResult<pb::DerivePhalaI2pKeyResponse> {
-        let mut phactory = self.lock_phactory();
+        let mut phactory = self.lock_phactory(true)?;
         let system = phactory.system()?;
         let derive_key = system
             .identity_key
@@ -1235,7 +1286,7 @@ impl<Platform: pal::Platform + Serialize + DeserializeOwned> PhactoryApi for Rpc
         &mut self,
         _request: (),
     ) -> RpcResult<pb::HandoverChallenge> {
-        let mut phactory = self.lock_phactory();
+        let mut phactory = self.lock_phactory(false)?;
         let (block, ts) = phactory.current_block()?;
         let system = phactory.system()?;
         let challenge = system.get_worker_key_challenge(block, ts);
@@ -1246,7 +1297,7 @@ impl<Platform: pal::Platform + Serialize + DeserializeOwned> PhactoryApi for Rpc
         &mut self,
         request: pb::HandoverChallengeResponse,
     ) -> RpcResult<pb::HandoverWorkerKey> {
-        let mut phactory = self.lock_phactory();
+        let mut phactory = self.lock_phactory(false)?;
         let attestation_provider = phactory.attestation_provider;
         let dev_mode = phactory.dev_mode;
         let in_sgx = attestation_provider == Some(AttestationProvider::Ias);
@@ -1405,7 +1456,7 @@ impl<Platform: pal::Platform + Serialize + DeserializeOwned> PhactoryApi for Rpc
         &mut self,
         request: pb::HandoverChallenge,
     ) -> RpcResult<pb::HandoverChallengeResponse> {
-        let mut phactory = self.lock_phactory();
+        let mut phactory = self.lock_phactory(false)?;
 
         // generate and save tmp key only for key handover encryption
         let handover_key = crate::new_sr25519_key();
@@ -1456,7 +1507,7 @@ impl<Platform: pal::Platform + Serialize + DeserializeOwned> PhactoryApi for Rpc
     }
 
     async fn handover_receive(&mut self, request: pb::HandoverWorkerKey) -> RpcResult<()> {
-        let mut phactory = self.lock_phactory();
+        let mut phactory = self.lock_phactory(false)?;
         let encrypted_worker_key = request.decode_worker_key().map_err(from_display)?;
 
         let dev_mode = encrypted_worker_key.dev_mode;
@@ -1516,7 +1567,7 @@ impl<Platform: pal::Platform + Serialize + DeserializeOwned> PhactoryApi for Rpc
         &mut self,
         request: super::NetworkConfig,
     ) -> Result<(), prpc::server::Error> {
-        self.lock_phactory().set_netconfig(request);
+        self.lock_phactory(false)?.set_netconfig(request);
         Ok(())
     }
 
@@ -1588,14 +1639,15 @@ impl<Platform: pal::Platform + Serialize + DeserializeOwned> PhactoryApi for Rpc
         &mut self,
         request: pb::GetContractInfoRequest,
     ) -> Result<pb::GetContractInfoResponse, prpc::server::Error> {
-        self.lock_phactory().get_contract_info(&request.contracts)
+        self.lock_phactory(true)?
+            .get_contract_info(&request.contracts)
     }
 
     async fn get_cluster_info(
         &mut self,
         _request: (),
     ) -> Result<pb::GetClusterInfoResponse, prpc::server::Error> {
-        self.lock_phactory().get_cluster_info()
+        self.lock_phactory(true)?.get_cluster_info()
     }
 
     async fn upload_sidevm_code(
@@ -1606,7 +1658,7 @@ impl<Platform: pal::Platform + Serialize + DeserializeOwned> PhactoryApi for Rpc
             .contract
             .try_into()
             .map_err(|_| from_display("Invalid contract id"))?;
-        self.lock_phactory()
+        self.lock_phactory(true)?
             .upload_sidevm_code(contract_id.into(), request.code)
     }
 
@@ -1630,7 +1682,7 @@ impl<Platform: pal::Platform + Serialize + DeserializeOwned> PhactoryApi for Rpc
         &mut self,
         _request: (),
     ) -> Result<pb::NetworkConfigResponse, prpc::server::Error> {
-        let phactory = self.lock_phactory();
+        let phactory = self.lock_phactory(true)?;
         Ok(pb::NetworkConfigResponse {
             public_rpc_port: phactory.args.public_port.map(Into::into),
             config: phactory.netconfig.clone(),
@@ -1640,29 +1692,34 @@ impl<Platform: pal::Platform + Serialize + DeserializeOwned> PhactoryApi for Rpc
         &mut self,
         request: pb::ChainState,
     ) -> Result<(), prpc::server::Error> {
-        self.lock_phactory()
+        self.lock_phactory(false)?
             .load_chain_state(request.block_number, request.decode_state()?)
             .map_err(from_display)
     }
     async fn stop(&mut self, request: pb::StopOptions) -> Result<(), prpc::server::Error> {
-        self.lock_phactory().stop(request.remove_checkpoints)
+        self.lock_phactory(true)?.stop(request.remove_checkpoints)
     }
     async fn load_storage_proof(
         &mut self,
         req: phactory_api::prpc::StorageProof,
     ) -> Result<(), prpc::server::Error> {
-        self.lock_phactory().load_storage_proof(req.proof)?;
+        self.lock_phactory(false)?.load_storage_proof(req.proof)?;
         Ok(())
     }
     async fn take_checkpoint(&mut self, _req: ()) -> Result<pb::SyncedTo, prpc::server::Error> {
-        let synced_to = self.lock_phactory().take_checkpoint().map_err(from_debug)?;
+        let synced_to = self
+            .lock_phactory(false)?
+            .take_checkpoint()
+            .map_err(from_debug)?;
         Ok(pb::SyncedTo { synced_to })
     }
     async fn statistics(
         &mut self,
         request: pb::StatisticsReqeust,
     ) -> Result<pb::StatisticsResponse, prpc::server::Error> {
-        self.lock_phactory().statistics(request).map_err(from_debug)
+        self.lock_phactory(true)?
+            .statistics(request)
+            .map_err(from_debug)
     }
 }
 
diff --git a/crates/phactory/src/secret_channel.rs b/crates/phactory/src/secret_channel.rs
index 1b4cfa1c14..1126426f71 100644
--- a/crates/phactory/src/secret_channel.rs
+++ b/crates/phactory/src/secret_channel.rs
@@ -125,6 +125,15 @@ mod receiver {
         _t: PhantomData<T>,
     }
 
+    impl<T> Clone for SecretPeeler<T> {
+        fn clone(&self) -> Self {
+            Self {
+                ecdh_key: self.ecdh_key.clone(),
+                _t: self._t,
+            }
+        }
+    }
+
     impl<T> SecretPeeler<T> {
         pub fn new(ecdh_key: ecdh::EcdhKey) -> Self {
             SecretPeeler {
@@ -158,6 +167,16 @@ mod receiver {
         _msg: PhantomData<Msg>,
     }
 
+    impl<Msg, Wrp, Plr: Clone> Clone for PeelingReceiver<Msg, Wrp, Plr> {
+        fn clone(&self) -> Self {
+            Self {
+                receiver: self.receiver.clone(),
+                peeler: self.peeler.clone(),
+                _msg: self._msg,
+            }
+        }
+    }
+
     impl<Msg, Wrp> PeelingReceiver<Msg, Wrp, PlainPeeler<Msg>> {
         #[allow(unused)]
         pub fn new_plain(receiver: TypedReceiver<Wrp>) -> Self {
diff --git a/crates/phactory/src/storage.rs b/crates/phactory/src/storage.rs
index 59eb9770a8..9c7fad9a26 100644
--- a/crates/phactory/src/storage.rs
+++ b/crates/phactory/src/storage.rs
@@ -50,6 +50,14 @@ mod storage_ext {
         trie_storage: TrieStorage<crate::RuntimeHasher>,
     }
 
+    impl Clone for ChainStorage {
+        fn clone(&self) -> Self {
+            Self {
+                trie_storage: self.trie_storage.snapshot(),
+            }
+        }
+    }
+
     impl From<TrieStorage<crate::RuntimeHasher>> for ChainStorage {
         fn from(value: TrieStorage<crate::RuntimeHasher>) -> Self {
             Self {
diff --git a/crates/phactory/src/system/gk.rs b/crates/phactory/src/system/gk.rs
index e0c978137f..28cc931637 100644
--- a/crates/phactory/src/system/gk.rs
+++ b/crates/phactory/src/system/gk.rs
@@ -23,7 +23,7 @@ use phala_types::{
 use serde::{Deserialize, Serialize};
 use sp_core::{hashing, sr25519, Pair};
 
-use crate::types::BlockInfo;
+use crate::{im_helpers::ordmap_for_each_mut, types::BlockInfo};
 
 use std::{
     collections::{BTreeMap, VecDeque},
@@ -69,7 +69,7 @@ fn get_cluster_key(master_key: &sr25519::Pair, cluster: &ContractClusterId) -> s
 }
 
 #[cfg(feature = "gk-stat")]
-#[derive(Debug, Default, Serialize, Deserialize)]
+#[derive(Debug, Default, Serialize, Deserialize, Clone)]
 struct WorkerStat {
     last_heartbeat_for_block: chain::BlockNumber,
     last_heartbeat_at_block: chain::BlockNumber,
@@ -77,7 +77,7 @@ struct WorkerStat {
     last_gk_responsive_event_at_block: chain::BlockNumber,
 }
 
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug, Serialize, Deserialize, Clone)]
 pub struct WorkerInfo {
     state: WorkerState,
     waiting_heartbeats: VecDeque<chain::BlockNumber>,
@@ -110,7 +110,7 @@ impl WorkerInfo {
     }
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Clone)]
 pub(crate) struct Gatekeeper<MsgChan> {
     /// The current master key in use
     #[serde(with = "more::key_bytes")]
@@ -626,19 +626,19 @@ impl<F: FnMut(EconomicEvent, &WorkerInfo)> EconomicEventListener for F {
     }
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Clone)]
 pub struct ComputingEconomics<MsgChan> {
     egress: MsgChan, // TODO.kevin: syncing the egress state while migrating.
     computing_events: TypedReceiver<WorkingReportEvent>,
     system_events: TypedReceiver<SystemEvent>,
     gatekeeper_events: TypedReceiver<GatekeeperEvent>,
-    workers: BTreeMap<WorkerPublicKey, WorkerInfo>,
+    workers: im::OrdMap<WorkerPublicKey, WorkerInfo>,
     tokenomic_params: tokenomic::Params,
     #[serde(skip, default)]
     eco_cache: EconomicCalcCache,
 }
 
-#[derive(Default)]
+#[derive(Default, Clone)]
 struct EconomicCalcCache {
     sum_share: FixedPoint,
     report: WorkingInfoUpdateEvent<chain::BlockNumber>,
@@ -713,9 +713,9 @@ impl<MsgChan: MessageChannel<Signer = Sr25519Signer>> ComputingEconomics<MsgChan
         let sum_share = self.sum_share();
         let report = WorkingInfoUpdateEvent::new(block.block_number, block.now_ms);
         self.eco_cache = EconomicCalcCache { sum_share, report };
-        for worker in self.workers.values_mut() {
+        ordmap_for_each_mut(&mut self.workers, |(_k, worker)| {
             worker.heartbeat_flag = false;
-        }
+        });
     }
 
     pub fn did_process_block(
@@ -723,7 +723,10 @@ impl<MsgChan: MessageChannel<Signer = Sr25519Signer>> ComputingEconomics<MsgChan
         block: &BlockInfo<'_>,
         event_listener: &mut impl EconomicEventListener,
     ) {
-        for worker_info in self.workers.values_mut() {
+        let eco_cache = &mut self.eco_cache;
+        let heartbeat_window = self.tokenomic_params.heartbeat_window;
+        let params = &self.tokenomic_params;
+        ordmap_for_each_mut(&mut self.workers, |(_k, worker_info)| {
             trace!(target: "gk_computing",
                 "[{}] block_post_process",
                 hex::encode(worker_info.state.pubkey)
@@ -737,7 +740,7 @@ impl<MsgChan: MessageChannel<Signer = Sr25519Signer>> ComputingEconomics<MsgChan
                     "[{}] Computing already stopped, do nothing.",
                     hex::encode(worker_info.state.pubkey)
                 );
-                continue;
+                return;
             }
 
             if worker_info.unresponsive {
@@ -748,7 +751,7 @@ impl<MsgChan: MessageChannel<Signer = Sr25519Signer>> ComputingEconomics<MsgChan
                         hex::encode(worker_info.state.pubkey)
                     );
                     worker_info.unresponsive = false;
-                    self.eco_cache
+                    eco_cache
                         .report
                         .recovered_to_online
                         .push(worker_info.state.pubkey);
@@ -761,7 +764,7 @@ impl<MsgChan: MessageChannel<Signer = Sr25519Signer>> ComputingEconomics<MsgChan
                     event_listener.emit_event(EconomicEvent::ExitUnresponsive, worker_info);
                 }
             } else if let Some(&hb_sent_at) = worker_info.waiting_heartbeats.get(0) {
-                if block.block_number - hb_sent_at > self.tokenomic_params.heartbeat_window {
+                if block.block_number - hb_sent_at > heartbeat_window {
                     trace!(
                         target: "gk_computing",
                         "[{}] case3: Idle, heartbeat failed, current={} waiting for {}.",
@@ -769,7 +772,7 @@ impl<MsgChan: MessageChannel<Signer = Sr25519Signer>> ComputingEconomics<MsgChan
                         block.block_number,
                         hb_sent_at
                     );
-                    self.eco_cache.report.offline.push(worker_info.state.pubkey);
+                    eco_cache.report.offline.push(worker_info.state.pubkey);
                     worker_info.unresponsive = true;
 
                     #[cfg(feature = "gk-stat")]
@@ -782,7 +785,6 @@ impl<MsgChan: MessageChannel<Signer = Sr25519Signer>> ComputingEconomics<MsgChan
                 }
             }
 
-            let params = &self.tokenomic_params;
             if worker_info.unresponsive {
                 trace!(
                     target: "gk_computing",
@@ -800,13 +802,13 @@ impl<MsgChan: MessageChannel<Signer = Sr25519Signer>> ComputingEconomics<MsgChan
                 );
                 worker_info.tokenomic.update_v_idle(params);
             }
-        }
+        });
 
-        let report = &self.eco_cache.report;
+        let report = core::mem::take(&mut self.eco_cache.report);
 
         if !report.is_empty() {
             debug!(target: "gk_computing", "Report: {:?}", report);
-            self.egress.push_message(report);
+            self.egress.push_message(&report);
         }
     }
 
@@ -1012,7 +1014,7 @@ impl<MsgChan: MessageChannel<Signer = Sr25519Signer>> ComputingEconomics<MsgChan
 
         let log_on = log::log_enabled!(log::Level::Debug);
         // TODO.kevin: Avoid unnecessary iteration for WorkerEvents.
-        for worker_info in self.workers.values_mut() {
+        ordmap_for_each_mut(&mut self.workers, |(_k, worker_info)| {
             // Replay the event on worker state, and collect the egressed heartbeat into waiting_heartbeats.
             let mut tracker = WorkerSMTracker::new(&mut worker_info.waiting_heartbeats);
             worker_info
@@ -1021,7 +1023,7 @@ impl<MsgChan: MessageChannel<Signer = Sr25519Signer>> ComputingEconomics<MsgChan
             if tracker.challenge_received {
                 event_listener.emit_event(EconomicEvent::HeartbeatChallenge, worker_info);
             }
-        }
+        });
 
         match &event {
             SystemEvent::WorkerEvent(e) => {
@@ -1260,7 +1262,7 @@ mod tokenomic {
         }
     }
 
-    #[derive(Debug, Serialize, Deserialize)]
+    #[derive(Debug, Serialize, Deserialize, Clone)]
     pub struct Params {
         #[serde(with = "serde_fp")]
         rho: FixedPoint,
diff --git a/crates/phactory/src/system/mod.rs b/crates/phactory/src/system/mod.rs
index 9ca2f5ffc4..31e8eb5bb3 100644
--- a/crates/phactory/src/system/mod.rs
+++ b/crates/phactory/src/system/mod.rs
@@ -125,7 +125,7 @@ impl From<String> for TransactionError {
     }
 }
 
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug, Serialize, Deserialize, Clone)]
 struct BenchState {
     start_block: chain::BlockNumber,
     start_time: u64,
@@ -133,13 +133,13 @@ struct BenchState {
     duration: u32,
 }
 
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug, Serialize, Deserialize, Clone)]
 enum WorkingState {
     Computing,
     Paused,
 }
 
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug, Serialize, Deserialize, Clone)]
 struct WorkingInfo {
     session_id: u32,
     state: WorkingState,
@@ -148,7 +148,7 @@ struct WorkingInfo {
 }
 
 // Minimum worker state machine can be reused to replay in GK.
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug, Serialize, Deserialize, Clone)]
 struct WorkerState {
     #[serde(with = "more::pubkey_bytes")]
     pubkey: WorkerPublicKey,
@@ -429,7 +429,7 @@ fn get_contract_key(cluster_key: &sr25519::Pair, contract_id: &ContractId) -> sr
         .expect("should not fail with valid info")
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Clone)]
 pub struct System<Platform> {
     platform: Platform,
     // Configuration
@@ -543,10 +543,10 @@ impl<Platform: pal::Platform> System<Platform> {
         }
     }
 
-    pub fn get_system_message_handler(&mut self) -> Option<CommandSender> {
+    pub fn get_system_message_handler(&self) -> Option<CommandSender> {
         let handler_contract_id = self
             .contract_cluster
-            .as_mut()?
+            .as_ref()?
             .config
             .log_handler
             .as_ref()?;
@@ -597,7 +597,7 @@ impl<Platform: pal::Platform> System<Platform> {
     }
 
     pub fn make_query(
-        &mut self,
+        &self,
         req_id: u64,
         contract_id: &AccountId,
         origin: Option<&chain::AccountId>,
@@ -610,11 +610,11 @@ impl<Platform: pal::Platform> System<Platform> {
     > {
         let contract = self
             .contracts
-            .get_mut(contract_id)
+            .get(contract_id)
             .ok_or(OpaqueError::ContractNotFound)?;
-        let mut cluster = self
+        let cluster = self
             .contract_cluster
-            .as_mut()
+            .as_ref()
             .expect("BUG: contract cluster should always exists")
             .snapshot();
         let sidevm_handle = contract.sidevm_handle();
diff --git a/crates/phala-mq/Cargo.toml b/crates/phala-mq/Cargo.toml
index 3509437f27..3f936ac29c 100644
--- a/crates/phala-mq/Cargo.toml
+++ b/crates/phala-mq/Cargo.toml
@@ -7,6 +7,7 @@ edition = "2018"
 
 [dependencies]
 log = "0.4.14"
+tracing = { version = "0.1", default-features = false }
 hex = { version =  "0.4.3", default-features = false, features = ['alloc'] }
 derive_more = { version = "0.99", default-features = false, features = ["display"] }
 parity-scale-codec = { version = "3.3", default-features = false, features = ["derive"] }
@@ -19,6 +20,7 @@ phala-serde-more = { path = "../phala-serde-more", default-features = false }
 
 # for checkpoint
 environmental = { version = "1.1.3", optional = true }
+im = "15"
 
 [features]
 default = ["dispatcher", "queue", "signers", "checkpoint"]
diff --git a/crates/phala-mq/src/dispatcher.rs b/crates/phala-mq/src/dispatcher.rs
index 8bc2a568d6..fe72cbdd3a 100644
--- a/crates/phala-mq/src/dispatcher.rs
+++ b/crates/phala-mq/src/dispatcher.rs
@@ -1,6 +1,6 @@
 use core::marker::PhantomData;
 
-use alloc::{collections::BTreeMap, vec::Vec};
+use alloc::vec::Vec;
 use serde::{Deserialize, Serialize};
 
 use crate::simple_mpsc::{channel, ReceiveError, Receiver as RawReceiver, Sender, Seq};
@@ -15,13 +15,14 @@ impl Seq for (u64, Message) {
     }
 }
 
-#[derive(Default)]
+#[derive(Default, Clone)]
 pub struct MessageDispatcher {
-    subscribers: BTreeMap<Path, Vec<Sender<(u64, Message)>>>,
+    subscribers: im::OrdMap<Path, Vec<Sender<(u64, Message)>>>,
     local_index: u64,
     //match_subscribers: Vec<Matcher, Vec<Sender<Message>>>,
 }
 
+#[derive(Clone)]
 pub struct Receiver<T> {
     inner: RawReceiver<(u64, T)>,
     topic: Vec<u8>,
@@ -79,7 +80,11 @@ impl MessageDispatcher {
                 if let Err(error) = receiver.send((sn, message.clone())) {
                     use crate::simple_mpsc::SendError::*;
                     match error {
-                        ReceiverGone => false,
+                        ReceiverGone => {
+                            let dst = String::from_utf8_lossy(message.destination.path());
+                            tracing::warn!(%dst, "ReceiverGone");
+                            false
+                        }
                     }
                 } else {
                     count += 1;
@@ -97,7 +102,7 @@ impl MessageDispatcher {
     /// Drop all unhandled messages.
     pub fn clear(&mut self) -> usize {
         let mut count = 0;
-        for subscriber in self.subscribers.values_mut().flatten() {
+        for subscriber in self.subscribers.values().flatten() {
             count += subscriber.clear();
         }
         count
@@ -112,6 +117,12 @@ pub enum TypedReceiveError {
     CodecError(CodecError),
 }
 
+impl TypedReceiveError {
+    pub fn is_sender_gone(&self) -> bool {
+        matches!(self, Self::SenderGone)
+    }
+}
+
 impl From<CodecError> for TypedReceiveError {
     fn from(e: CodecError) -> Self {
         Self::CodecError(e)
@@ -125,6 +136,15 @@ pub struct TypedReceiver<T> {
     _t: PhantomData<T>,
 }
 
+impl<T> Clone for TypedReceiver<T> {
+    fn clone(&self) -> Self {
+        Self {
+            queue: self.queue.clone(),
+            _t: self._t,
+        }
+    }
+}
+
 impl<T: Decode> TypedReceiver<T> {
     pub fn try_next(&mut self) -> Result<Option<(u64, T, MessageOrigin)>, TypedReceiveError> {
         let message = self.queue.try_next().map_err(|e| match e {
@@ -254,6 +274,10 @@ macro_rules! select_ignore_errors {
                             $block
                         }
                     }
+                    Err(err) if err.is_sender_gone() => {
+                        log::warn!("[{}] mq error: {:?}", $crate::function!(), err);
+                        panic!("mq error: {:?}", err);
+                    }
                     Err(err) => {
                         log::warn!("[{}] mq ignored error: {:?}", $crate::function!(), err);
                     }
diff --git a/crates/phala-mq/src/simple_mpsc.rs b/crates/phala-mq/src/simple_mpsc.rs
index ae9361b121..921d486d17 100644
--- a/crates/phala-mq/src/simple_mpsc.rs
+++ b/crates/phala-mq/src/simple_mpsc.rs
@@ -7,7 +7,7 @@ use derive_more::Display;
 struct Channel<T> {
     deque: VecDeque<T>,
     sender_count: usize,
-    receiver_gone: bool,
+    receiver_count: usize,
 }
 
 impl<T> Channel<T> {
@@ -19,7 +19,7 @@ impl<T> Channel<T> {
         Self {
             deque: VecDeque::with_capacity(cap),
             sender_count: 1,
-            receiver_gone: false,
+            receiver_count: 1,
         }
     }
 }
@@ -36,7 +36,7 @@ pub enum SendError {
 impl<T> Sender<T> {
     pub fn send(&self, value: T) -> Result<(), SendError> {
         let mut ch = self.0.lock();
-        if ch.receiver_gone {
+        if ch.receiver_count == 0 {
             Err(SendError::ReceiverGone)
         } else {
             ch.deque.push_back(value);
@@ -112,7 +112,14 @@ impl<T: Seq> Receiver<T> {
 
 impl<T> Drop for Receiver<T> {
     fn drop(&mut self) {
-        self.0.lock().receiver_gone = true;
+        self.0.lock().receiver_count -= 1;
+    }
+}
+
+impl<T> Clone for Receiver<T> {
+    fn clone(&self) -> Receiver<T> {
+        self.0.lock().receiver_count += 1;
+        Receiver(self.0.clone())
     }
 }
 
diff --git a/crates/pink/capi/src/types.rs b/crates/pink/capi/src/types.rs
index 15de25b4d2..45c79b354d 100644
--- a/crates/pink/capi/src/types.rs
+++ b/crates/pink/capi/src/types.rs
@@ -59,7 +59,7 @@ impl ExecutionMode {
     }
 }
 
-#[derive(Decode, Encode, Debug)]
+#[derive(Decode, Encode, Debug, Clone)]
 pub enum ExecSideEffects {
     V1 {
         pink_events: Vec<(AccountId, PinkEvent)>,
@@ -85,4 +85,14 @@ impl ExecSideEffects {
             },
         }
     }
+
+    pub fn is_empty(&self) -> bool {
+        match self {
+            ExecSideEffects::V1 {
+                pink_events,
+                ink_events,
+                instantiated,
+            } => pink_events.is_empty() && ink_events.is_empty() && instantiated.is_empty(),
+        }
+    }
 }
diff --git a/crates/pink/pink-extension/src/lib.rs b/crates/pink/pink-extension/src/lib.rs
index 4a8e0f7ddc..0efbaefc85 100644
--- a/crates/pink/pink-extension/src/lib.rs
+++ b/crates/pink/pink-extension/src/lib.rs
@@ -61,14 +61,14 @@ pub struct OspMessage {
     pub remote_pubkey: Option<EcdhPublicKey>,
 }
 
-#[derive(Encode, Decode, Debug)]
+#[derive(Encode, Decode, Debug, Clone)]
 #[cfg_attr(feature = "std", derive(scale_info::TypeInfo))]
 pub enum HookPoint {
     OnBlockEnd,
 }
 
 /// System Event used to communicate between the contract and the runtime.
-#[derive(Encode, Decode, Debug)]
+#[derive(Encode, Decode, Debug, Clone)]
 #[cfg_attr(feature = "std", derive(scale_info::TypeInfo))]
 pub enum PinkEvent {
     /// Set contract hook
@@ -147,7 +147,7 @@ impl PinkEvent {
     }
 }
 
-#[derive(Encode, Decode, Debug)]
+#[derive(Encode, Decode, Debug, Clone)]
 #[cfg_attr(feature = "std", derive(scale_info::TypeInfo))]
 pub enum CacheOp {
     Set { key: Vec<u8>, value: Vec<u8> },
diff --git a/standalone/justification-validate/src/validator.rs b/standalone/justification-validate/src/validator.rs
index dc9a8aff63..edabcf80cf 100644
--- a/standalone/justification-validate/src/validator.rs
+++ b/standalone/justification-validate/src/validator.rs
@@ -1,4 +1,4 @@
-use std::{collections::VecDeque, thread::JoinHandle};
+use std::thread::JoinHandle;
 
 use crossbeam::channel::{bounded, Sender};
 use log::{error, warn};
@@ -140,7 +140,7 @@ fn do_validate(
     let from = headers[0].header.number;
     let to = headers.last().unwrap().header.number;
 
-    let mut state_roots = VecDeque::new();
+    let mut state_roots = Default::default();
     let result = client.sync_header(headers, auth_change, &mut state_roots, 0);
     if let Err(err) = result {
         error!("Failed to sync header from={from:?} to={to:?}: {err:?}");
diff --git a/standalone/pruntime/Cargo.lock b/standalone/pruntime/Cargo.lock
index b8896878c8..fcaa520cf9 100644
--- a/standalone/pruntime/Cargo.lock
+++ b/standalone/pruntime/Cargo.lock
@@ -4279,6 +4279,7 @@ dependencies = [
  "hex",
  "hex-literal",
  "hex_fmt",
+ "im",
  "itertools",
  "log",
  "num-bigint",
@@ -4333,6 +4334,7 @@ dependencies = [
  "base64",
  "derive_more",
  "frame-system",
+ "im",
  "log",
  "parity-scale-codec",
  "phala-crypto",
@@ -4403,6 +4405,7 @@ dependencies = [
  "derive_more",
  "environmental",
  "hex",
+ "im",
  "log",
  "parity-scale-codec",
  "phala-serde-more",
@@ -4410,6 +4413,7 @@ dependencies = [
  "serde",
  "sp-core",
  "spin 0.9.4",
+ "tracing",
 ]
 
 [[package]]
diff --git a/standalone/pruntime/gramine-build/Rocket.toml b/standalone/pruntime/gramine-build/Rocket.toml
index 4cfc089200..d729a51bbf 100644
--- a/standalone/pruntime/gramine-build/Rocket.toml
+++ b/standalone/pruntime/gramine-build/Rocket.toml
@@ -2,6 +2,7 @@
 address = "0.0.0.0"
 port = 8000
 workers = 1
+max_blocking = 32
 cli_colors = false
 
 [global.limits]
diff --git a/standalone/pruntime/src/handover.rs b/standalone/pruntime/src/handover.rs
index 06383e5738..32caccc434 100644
--- a/standalone/pruntime/src/handover.rs
+++ b/standalone/pruntime/src/handover.rs
@@ -9,7 +9,7 @@ use tracing::info;
 
 pub(crate) async fn handover_from(url: &str, args: InitArgs) -> Result<()> {
     let mut this = RpcService::new(GraminePlatform);
-    this.lock_phactory().init(args);
+    this.lock_phactory(true).expect("Failed to lock Phactory").init(args);
 
     let from_pruntime = new_pruntime_client(url.into());
     info!("Requesting for challenge");
diff --git a/standalone/pruntime/src/main.rs b/standalone/pruntime/src/main.rs
index d1021a9f80..eafab9f91c 100644
--- a/standalone/pruntime/src/main.rs
+++ b/standalone/pruntime/src/main.rs
@@ -77,6 +77,10 @@ struct Args {
     #[arg(long)]
     #[arg(default_value_t = 0)]
     safe_mode_level: u8,
+
+    /// Disable the RCU policy to update the Phactory state.
+    #[arg(long)]
+    no_rcu: bool,
 }
 
 #[rocket::main]
@@ -141,6 +145,7 @@ async fn serve(sgx: bool) -> Result<(), rocket::Error> {
             cores,
             public_port: args.public_port,
             safe_mode_level: args.safe_mode_level,
+            no_rcu: args.no_rcu,
         }
     };
     info!("init_args: {:#?}", init_args);
diff --git a/standalone/pruntime/src/runtime.rs b/standalone/pruntime/src/runtime.rs
index 3e69ec8b96..9eab42c804 100644
--- a/standalone/pruntime/src/runtime.rs
+++ b/standalone/pruntime/src/runtime.rs
@@ -10,17 +10,21 @@ lazy_static::lazy_static! {
 }
 
 pub fn ecall_handle(req_id: u64, action: u8, input: &[u8]) -> Result<Vec<u8>> {
-    let mut factory = APPLICATION.lock_phactory();
+    let allow_rcu = action == phactory_api::actions::ACTION_GET_INFO;
+    let mut factory = APPLICATION.lock_phactory(allow_rcu)?;
     Ok(factory.handle_scale_api(req_id, action, input))
 }
 
 pub fn ecall_getinfo() -> String {
-    let info = APPLICATION.lock_phactory().get_info();
+    let Ok(guard) = APPLICATION.lock_phactory(true) else {
+        return r#"{"error": "Failed to lock Phactory""#.into();
+    };
+    let info = guard.get_info();
     serde_json::to_string_pretty(&info).unwrap_or_default()
 }
 
 pub fn ecall_sign_http_response(data: &[u8]) -> Option<String> {
-    APPLICATION.lock_phactory().sign_http_response(data)
+    APPLICATION.lock_phactory(true).ok()?.sign_http_response(data)
 }
 
 pub fn ecall_init(args: phactory_api::ecall_args::InitArgs) -> Result<()> {
@@ -33,7 +37,7 @@ pub fn ecall_init(args: phactory_api::ecall_args::InitArgs) -> Result<()> {
         match Phactory::restore_from_checkpoint(&GraminePlatform, &args) {
             Ok(Some(factory)) => {
                 info!("Loaded checkpoint");
-                **APPLICATION.lock_phactory() = factory;
+                **APPLICATION.lock_phactory(true).expect("Failed to lock Phactory") = factory;
                 return Ok(());
             }
             Err(err) => {
@@ -47,7 +51,7 @@ pub fn ecall_init(args: phactory_api::ecall_args::InitArgs) -> Result<()> {
         info!("Checkpoint disabled.");
     }
 
-    APPLICATION.lock_phactory().init(args);
+    APPLICATION.lock_phactory(true).unwrap().init(args);
 
     info!("Enclave init OK");
     Ok(())


Test patch: