Add config option to control JSONRPC request parallelism

Blockstream · May 23, 2024 · a9be78d · a9be78d
1 parent 0ac2dbc
commit a9be78d
Show file tree

Hide file tree

Showing 5 changed files with 27 additions and 4 deletions.
diff --git a/src/bin/electrs.rs b/src/bin/electrs.rs
@@ -49,6 +49,7 @@ fn run_server(config: Arc<Config>) -> Result<()> {
         &config.daemon_dir,
         &config.blocks_dir,
         config.daemon_rpc_addr,
+        config.daemon_parallelism,
         config.cookie_getter(),
         config.network_type,
         signal.clone(),

diff --git a/src/bin/tx-fingerprint-stats.rs b/src/bin/tx-fingerprint-stats.rs
@@ -33,6 +33,7 @@ fn main() {
             &config.daemon_dir,
             &config.blocks_dir,
             config.daemon_rpc_addr,
+            config.daemon_parallelism,
             config.cookie_getter(),
             config.network_type,
             signal,

diff --git a/src/config.rs b/src/config.rs
@@ -25,6 +25,7 @@ pub struct Config {
     pub daemon_dir: PathBuf,
     pub blocks_dir: PathBuf,
     pub daemon_rpc_addr: SocketAddr,
+    pub daemon_parallelism: usize,
     pub cookie: Option<String>,
     pub electrum_rpc_addr: SocketAddr,
     pub http_addr: SocketAddr,
@@ -132,6 +133,12 @@ impl Config {
                     .help("Bitcoin daemon JSONRPC 'addr:port' to connect (default: 127.0.0.1:8332 for mainnet, 127.0.0.1:18332 for testnet and 127.0.0.1:18443 for regtest)")
                     .takes_value(true),
             )
+            .arg(
+                Arg::with_name("daemon_parallelism")
+                    .long("daemon-parallelism")
+                    .help("Number of JSONRPC requests to send in parallel")
+                    .default_value("10")
+            )
             .arg(
                 Arg::with_name("monitoring_addr")
                     .long("monitoring-addr")
@@ -386,6 +393,7 @@ impl Config {
             daemon_dir,
             blocks_dir,
             daemon_rpc_addr,
+            daemon_parallelism: value_t_or_exit!(m, "daemon_parallelism", usize),
             cookie,
             utxos_limit: value_t_or_exit!(m, "utxos_limit", usize),
             electrum_rpc_addr,

diff --git a/src/daemon.rs b/src/daemon.rs
@@ -260,6 +260,7 @@ impl Counter {
 
 pub struct Daemon {
     daemon_dir: PathBuf,
+    daemon_parallelism: usize,
     blocks_dir: PathBuf,
     network: Network,
     conn: Mutex<Connection>,
@@ -276,13 +277,15 @@ impl Daemon {
         daemon_dir: &PathBuf,
         blocks_dir: &PathBuf,
         daemon_rpc_addr: SocketAddr,
+        daemon_parallelism: usize,
         cookie_getter: Arc<dyn CookieGetter>,
         network: Network,
         signal: Waiter,
         metrics: &Metrics,
     ) -> Result<Daemon> {
         let daemon = Daemon {
             daemon_dir: daemon_dir.clone(),
+            daemon_parallelism,
             blocks_dir: blocks_dir.clone(),
             network,
             conn: Mutex::new(Connection::new(
@@ -335,6 +338,7 @@ impl Daemon {
     pub fn reconnect(&self) -> Result<Daemon> {
         Ok(Daemon {
             daemon_dir: self.daemon_dir.clone(),
+            daemon_parallelism: self.daemon_parallelism,
             blocks_dir: self.blocks_dir.clone(),
             network: self.network,
             conn: Mutex::new(self.conn.lock().unwrap().reconnect()?),
@@ -406,10 +410,17 @@ impl Daemon {
     fn requests(&self, method: &str, params_list: &[Value]) -> Result<Vec<Value>> {
         // Send in parallel as individual JSONRPC requests, with no batching.
         // See https://github.com/Blockstream/electrs/pull/33
-        params_list
-            .par_iter()
-            .map(|params| self.retry_request(method, params))
-            .collect()
+        let pool = rayon::ThreadPoolBuilder::new()
+            .num_threads(self.daemon_parallelism)
+            .thread_name(|i| format!("rpc-requests-{}", i))
+            .build()
+            .unwrap();
+        pool.install(|| {
+            params_list
+                .par_iter()
+                .map(|params| self.retry_request(method, params))
+                .collect()
+        })
     }
 
     // bitcoind JSONRPC API:

diff --git a/tests/common.rs b/tests/common.rs
@@ -89,6 +89,7 @@ impl TestRunner {
             network_type,
             db_path: electrsdb.path().to_path_buf(),
             daemon_dir: daemon_subdir.clone(),
+            daemon_parallelism: 3,
             blocks_dir: daemon_subdir.join("blocks"),
             daemon_rpc_addr: params.rpc_socket.into(),
             cookie: None,
@@ -127,6 +128,7 @@ impl TestRunner {
             &config.daemon_dir,
             &config.blocks_dir,
             config.daemon_rpc_addr,
+            config.daemon_parallelism,
             config.cookie_getter(),
             config.network_type,
             signal.clone(),