From c67ae6f835f5c1930b7095509afb8175fd68343f Mon Sep 17 00:00:00 2001
From: Peilun Li <peilun.li@confluxnetwork.org>
Date: Thu, 20 Jun 2024 16:57:15 +0800
Subject: [PATCH] Use spawn_blocking for storage task.

---
 node/miner/src/loader.rs              |  1 -
 node/miner/src/miner_id.rs            |  1 -
 node/miner/src/sealer.rs              |  4 +---
 node/miner/src/service.rs             |  2 +-
 node/miner/src/submitter.rs           |  2 +-
 node/storage-async/src/lib.rs         |  6 +++---
 node/sync/src/auto_sync/sync_store.rs | 10 ++--------
 node/sync/src/controllers/serial.rs   |  1 -
 node/sync/src/test_util.rs            |  2 --
 9 files changed, 8 insertions(+), 21 deletions(-)

diff --git a/node/miner/src/loader.rs b/node/miner/src/loader.rs
index afb2043..5133b60 100644
--- a/node/miner/src/loader.rs
+++ b/node/miner/src/loader.rs
@@ -2,7 +2,6 @@ use async_trait::async_trait;
 use std::sync::Arc;
 use storage::log_store::{MineLoadChunk, Store};
 
-
 #[async_trait]
 pub trait PoraLoader: Send + Sync {
     async fn load_sealed_data(&self, index: u64) -> Option<MineLoadChunk>;
diff --git a/node/miner/src/miner_id.rs b/node/miner/src/miner_id.rs
index 59d265d..aa0f9e5 100644
--- a/node/miner/src/miner_id.rs
+++ b/node/miner/src/miner_id.rs
@@ -8,7 +8,6 @@ use std::sync::Arc;
 use storage::log_store::{config::ConfigurableExt, Store};
 use storage::H256;
 
-
 const MINER_ID: &str = "mine.miner_id";
 
 pub fn load_miner_id(store: &dyn Store) -> storage::error::Result<Option<H256>> {
diff --git a/node/miner/src/sealer.rs b/node/miner/src/sealer.rs
index 2ffc43c..885cfee 100644
--- a/node/miner/src/sealer.rs
+++ b/node/miner/src/sealer.rs
@@ -1,9 +1,7 @@
 use std::{collections::BTreeMap, sync::Arc};
 
 use ethereum_types::H256;
-use tokio::{
-    time::{sleep, Duration, Instant},
-};
+use tokio::time::{sleep, Duration, Instant};
 
 use contract_interface::{EpochRangeWithContextDigest, ZgsFlow};
 use storage::{
diff --git a/node/miner/src/service.rs b/node/miner/src/service.rs
index 7d34101..7ef5d1a 100644
--- a/node/miner/src/service.rs
+++ b/node/miner/src/service.rs
@@ -6,8 +6,8 @@ use network::NetworkMessage;
 use std::sync::Arc;
 use storage::config::ShardConfig;
 use storage::log_store::Store;
+use tokio::sync::broadcast;
 use tokio::sync::mpsc;
-use tokio::sync::{broadcast};
 
 #[derive(Clone, Debug)]
 pub enum MinerMessage {
diff --git a/node/miner/src/submitter.rs b/node/miner/src/submitter.rs
index bf7b79e..9fdeee4 100644
--- a/node/miner/src/submitter.rs
+++ b/node/miner/src/submitter.rs
@@ -8,7 +8,7 @@ use shared_types::FlowRangeProof;
 use std::sync::Arc;
 use storage::log_store::Store;
 use task_executor::TaskExecutor;
-use tokio::sync::{mpsc};
+use tokio::sync::mpsc;
 
 use crate::config::{MineServiceMiddleware, MinerConfig};
 use crate::pora::AnswerWithoutProof;
diff --git a/node/storage-async/src/lib.rs b/node/storage-async/src/lib.rs
index a6884d5..9156cbc 100644
--- a/node/storage-async/src/lib.rs
+++ b/node/storage-async/src/lib.rs
@@ -6,7 +6,7 @@ use shared_types::{Chunk, ChunkArray, ChunkArrayWithProof, DataRoot, FlowProof,
 use std::sync::Arc;
 use storage::{error, error::Result, log_store::Store as LogStore, H256};
 use task_executor::TaskExecutor;
-use tokio::sync::{oneshot};
+use tokio::sync::oneshot;
 
 pub use storage::config::ShardConfig;
 
@@ -70,8 +70,8 @@ impl Store {
         let store = self.store.clone();
         let (tx, rx) = oneshot::channel();
 
-        self.executor.spawn(
-            async move {
+        self.executor.spawn_blocking(
+            move || {
                 // FIXME(zz): Not all functions need `write`. Refactor store usage.
                 let res = f(&*store);
 
diff --git a/node/sync/src/auto_sync/sync_store.rs b/node/sync/src/auto_sync/sync_store.rs
index ff885e4..e483d8c 100644
--- a/node/sync/src/auto_sync/sync_store.rs
+++ b/node/sync/src/auto_sync/sync_store.rs
@@ -71,10 +71,7 @@ impl SyncStore {
         let mut tx = ConfigTx::default();
 
         // not in pending queue
-        if !self
-            .pending_txs
-            .remove(store, Some(&mut tx), tx_seq)?
-        {
+        if !self.pending_txs.remove(store, Some(&mut tx), tx_seq)? {
             return Ok(false);
         }
 
@@ -92,10 +89,7 @@ impl SyncStore {
         let mut tx = ConfigTx::default();
 
         // not in ready queue
-        if !self
-            .ready_txs
-            .remove(store, Some(&mut tx), tx_seq)?
-        {
+        if !self.ready_txs.remove(store, Some(&mut tx), tx_seq)? {
             return Ok(false);
         }
 
diff --git a/node/sync/src/controllers/serial.rs b/node/sync/src/controllers/serial.rs
index b6ef02c..b41d999 100644
--- a/node/sync/src/controllers/serial.rs
+++ b/node/sync/src/controllers/serial.rs
@@ -651,7 +651,6 @@ mod tests {
     use storage::H256;
     use task_executor::{test_utils::TestRuntime, TaskExecutor};
     use tokio::sync::mpsc::{self, UnboundedReceiver};
-    
 
     #[test]
     fn test_status() {
diff --git a/node/sync/src/test_util.rs b/node/sync/src/test_util.rs
index c1154c4..d3a0981 100644
--- a/node/sync/src/test_util.rs
+++ b/node/sync/src/test_util.rs
@@ -9,7 +9,6 @@ use storage::{
     LogManager,
 };
 
-
 /// Creates stores for local node and peers with initialized transaction of specified chunk count.
 /// The first store is for local node, and data not stored. The second store is for peers, and all
 /// transactions are finalized for file sync.
@@ -101,7 +100,6 @@ pub mod tests {
     };
     use storage_async::Store;
     use task_executor::test_utils::TestRuntime;
-    
 
     pub struct TestStoreRuntime {
         pub runtime: TestRuntime,