Add --jobs parameter to customize concurrency

Enables users to tradeoff cpu/memory usage for performance
Made `out_of_order` blocks bounded by restricting the number of workers being spawned

Author: buffrr

node/src/bin/spaced.rs

@@ -60,6 +60,7 @@ impl Composer {
             spaced.chain.state.clone(),
             rx,
             self.shutdown.clone(),
+            spaced.num_workers,
         );
 
         self.services.spawn(async move {

node/src/config.rs

@@ -47,6 +47,9 @@ pub struct Args {
     /// Network to use
     #[arg(long, env = "SPACED_CHAIN")]
     chain: ExtendedNetwork,
+    /// Number of concurrent workers allowed during syncing
+    #[arg(short, long, env = "SPACED_JOBS", default_value = "8")]
+    jobs: u8,
     /// Bitcoin RPC URL
     #[arg(long, env = "SPACED_BITCOIN_RPC_URL")]
     bitcoin_rpc_url: Option<String>,
@@ -68,7 +71,6 @@ pub struct Args {
     #[arg(long, help_heading = Some(RPC_OPTIONS), default_values = ["127.0.0.1", "::1"], env = "SPACED_RPC_BIND")]
     rpc_bind: Vec<String>,
     /// Listen for JSON-RPC connections on <port>
-    /// (default: 22220, testnet: 22221, signet: 22224, regtest: 22226)
     #[arg(long, help_heading = Some(RPC_OPTIONS), env = "SPACED_RPC_PORT")]
     rpc_port: Option<u16>,
 }
@@ -183,6 +185,7 @@ impl Args {
             bind: rpc_bind_addresses,
             chain,
             block_index,
+            num_workers: args.jobs as usize,
         })
     }
 

node/src/source.rs

@@ -38,6 +38,7 @@ pub struct BlockFetcher {
     rpc: Arc<BitcoinRpc>,
     job_id: Arc<AtomicUsize>,
     sender: std::sync::mpsc::SyncSender<BlockEvent>,
+    num_workers: usize,
 }
 
 pub enum BlockEvent {
@@ -304,6 +305,7 @@ impl BlockFetcher {
     pub fn new(
         rpc: BitcoinRpc,
         client: reqwest::blocking::Client,
+        num_workers: usize,
     ) -> (Self, std::sync::mpsc::Receiver<BlockEvent>) {
         let (tx, rx) = std::sync::mpsc::sync_channel(12);
         (
@@ -312,6 +314,7 @@ impl BlockFetcher {
                 rpc: Arc::new(rpc),
                 job_id: Arc::new(AtomicUsize::new(0)),
                 sender: tx,
+                num_workers,
             },
             rx,
         )
@@ -329,6 +332,7 @@ impl BlockFetcher {
         let task_rpc = self.rpc.clone();
         let current_task = self.job_id.clone();
         let task_sender = self.sender.clone();
+        let num_workers = self.num_workers;
 
         _ = std::thread::spawn(move || {
             let mut last_check = Instant::now() - Duration::from_secs(2);
@@ -354,16 +358,14 @@ impl BlockFetcher {
                     };
 
                 if tip > start_block.height {
-                    let concurrency = std::cmp::min(tip - start_block.height, 8);
-
                     let res = Self::run_workers(
                         job_id,
                         current_task.clone(),
                         task_rpc.clone(),
                         task_sender.clone(),
                         start_block,
                         tip,
-                        concurrency as usize,
+                        num_workers,
                     );
 
                     match res {
@@ -388,7 +390,7 @@ impl BlockFetcher {
         sender: std::sync::mpsc::SyncSender<BlockEvent>,
         start_block: ChainAnchor,
         end_height: u32,
-        concurrency: usize,
+        num_workers: usize,
     ) -> Result<ChainAnchor, BlockFetchError> {
         let mut workers = Workers {
             current_job,
@@ -399,8 +401,8 @@ impl BlockFetcher {
             end_height,
             ordered_sender: sender,
             rpc,
-            concurrency,
-            pool: ThreadPool::new(concurrency),
+            num_workers,
+            pool: ThreadPool::new(num_workers),
         };
 
         workers.run()
@@ -464,7 +466,7 @@ struct Workers {
     end_height: u32,
     ordered_sender: std::sync::mpsc::SyncSender<BlockEvent>,
     rpc: Arc<BitcoinRpc>,
-    concurrency: usize,
+    num_workers: usize,
     pool: ThreadPool,
 }
 
@@ -508,7 +510,9 @@ impl Workers {
 
     #[inline(always)]
     fn can_add_workers(&self) -> bool {
-        self.pool.queued_count() < self.concurrency && !self.queued_all()
+        self.out_of_order.len() < self.num_workers
+            && self.pool.queued_count() < self.num_workers
+            && !self.queued_all()
     }
 
     #[inline(always)]
@@ -518,7 +522,7 @@ impl Workers {
 
     fn run(&mut self) -> Result<ChainAnchor, BlockFetchError> {
         let client = reqwest::blocking::Client::new();
-        let (tx, rx) = std::sync::mpsc::sync_channel(self.concurrency);
+        let (tx, rx) = std::sync::mpsc::sync_channel(self.num_workers);
 
         'queue_blocks: while !self.queued_all() {
             if self.should_stop() {
@@ -559,15 +563,15 @@ impl Workers {
                 }
             }
 
-            std::thread::sleep(Duration::from_millis(10));
+            std::thread::sleep(Duration::from_millis(1));
         }
 
         // Wait for all blocks to be processed
         while self.pool.active_count() > 0 || self.last_emitted.height != self.end_height {
             while self.try_emit_next_block(&rx)? {
                 // do nothing
             }
-            std::thread::sleep(Duration::from_millis(10));
+            std::thread::sleep(Duration::from_millis(1));
         }
         Ok(self.last_emitted)
     }
@@ -753,7 +757,7 @@ mod test {
         let start_block_hash: BlockHash =
             rpc.send_json_blocking(&client, &rpc.get_block_hash(start_block))?;
 
-        let (fetcher, receiver) = BlockFetcher::new(rpc, client);
+        let (fetcher, receiver) = BlockFetcher::new(rpc, client, 8);
 
         println!("fetcher starting from block {}", start_block);
 

node/src/sync.rs

@@ -36,6 +36,7 @@ pub struct Spaced {
     pub rpc: BitcoinRpc,
     pub data_dir: PathBuf,
     pub bind: Vec<SocketAddr>,
+    pub num_workers: usize,
 }
 
 impl Spaced {
@@ -155,7 +156,7 @@ impl Spaced {
 
         let rpc = source.rpc.clone();
         let client = reqwest::blocking::Client::new();
-        let (fetcher, receiver) = BlockFetcher::new(rpc.clone(), client.clone());
+        let (fetcher, receiver) = BlockFetcher::new(rpc.clone(), client.clone(), self.num_workers);
         fetcher.start(start_block);
 
         let mut shutdown_signal = shutdown.subscribe();

node/src/wallets.rs

@@ -282,8 +282,10 @@ impl RpcWallet {
         mut wallet: SpacesWallet,
         mut commands: Receiver<WalletCommand>,
         mut shutdown: broadcast::Receiver<()>,
+        num_workers: usize,
     ) -> anyhow::Result<()> {
-        let (fetcher, receiver) = BlockFetcher::new(source.rpc.clone(), source.client.clone());
+        let (fetcher, receiver) =
+            BlockFetcher::new(source.rpc.clone(), source.client.clone(), num_workers);
 
         let mut wallet_tip = {
             let tip = wallet.coins.local_chain().tip();
@@ -706,6 +708,7 @@ impl RpcWallet {
         store: LiveSnapshot,
         mut channel: Receiver<LoadedWallet>,
         shutdown: broadcast::Sender<()>,
+        num_workers: usize,
     ) -> anyhow::Result<()> {
         let mut shutdown_signal = shutdown.subscribe();
         let mut wallet_results = FuturesUnordered::new();
@@ -734,7 +737,8 @@ impl RpcWallet {
                                 wallet_chain,
                                 loaded.wallet,
                                 loaded.rx,
-                                wallet_shutdown
+                                wallet_shutdown,
+                                num_workers
                             ));
                         });
                         wallet_results.push(named_future(wallet_name, rx));