diff --git a/packages/rs-drive-abci/src/rpc/core.rs b/packages/rs-drive-abci/src/rpc/core.rs index efcbbeac3ca..f81ee32032d 100644 --- a/packages/rs-drive-abci/src/rpc/core.rs +++ b/packages/rs-drive-abci/src/rpc/core.rs @@ -1,3 +1,4 @@ +use crate::rpc::prefetch::CorePrefetcher; use dpp::dashcore::ephemerealdata::chain_lock::ChainLock; use dpp::dashcore::{Block, BlockHash, QuorumHash, Transaction, Txid}; use dpp::dashcore::{Header, InstantLock}; @@ -131,6 +132,9 @@ pub trait CoreRPCLike { /// Default implementation of Dash Core RPC using DashCoreRPC client pub struct DefaultCoreRPC { inner: Client, + /// Speculative fetcher for the next core height, on its own connection. + /// `None` when a second connection could not be opened. + prefetcher: Option, } // TODO: Create errors for these error codes in dashcore_rpc @@ -206,8 +210,13 @@ macro_rules! retry { impl DefaultCoreRPC { /// Create new instance pub fn open(url: &str, username: String, password: String) -> Result { + let prefetcher = CorePrefetcher::new(url, username.clone(), password.clone()); + if prefetcher.is_none() { + tracing::warn!("could not open a second Core RPC connection; block sync will fetch masternode and quorum updates on the critical path"); + } Ok(DefaultCoreRPC { inner: Client::new(url, Auth::UserPass(username, password))?, + prefetcher, }) } } @@ -270,7 +279,24 @@ impl CoreRPCLike for DefaultCoreRPC { &self, height: Option, ) -> Result { - retry!(self.inner.get_quorum_listextended_reversed(height)) + // Block sync walks core heights in order, so the next call is almost + // always for height + 1. Take the speculative answer when it is for the + // height we were asked about, and start the next guess either way. + let prefetched = height + .zip(self.prefetcher.as_ref()) + .and_then(|(height, prefetcher)| prefetcher.take_quorum_list(height)); + + let result = match prefetched { + Some(list) => Ok(list), + None => retry!(self.inner.get_quorum_listextended_reversed(height)), + }; + + if let (Ok(_), Some(height), Some(prefetcher)) = (&result, height, self.prefetcher.as_ref()) + { + prefetcher.start_quorum_list(height + 1); + } + + result } fn get_quorum_info( @@ -289,9 +315,25 @@ impl CoreRPCLike for DefaultCoreRPC { base_block: Option, block: u32, ) -> Result { - retry!(self - .inner - .get_protx_listdiff(base_block.unwrap_or(1), block)) + let base = base_block.unwrap_or(1); + + // Same reasoning as get_quorum_listextended: the next diff a syncing + // node asks for is from this block to the one after it. + let prefetched = self + .prefetcher + .as_ref() + .and_then(|prefetcher| prefetcher.take_protx_diff(base, block)); + + let result = match prefetched { + Some(diff) => Ok(diff), + None => retry!(self.inner.get_protx_listdiff(base, block)), + }; + + if let (Ok(_), Some(prefetcher)) = (&result, self.prefetcher.as_ref()) { + prefetcher.start_protx_diff(block, block + 1); + } + + result } /// Verify Instant Lock signature diff --git a/packages/rs-drive-abci/src/rpc/mod.rs b/packages/rs-drive-abci/src/rpc/mod.rs index 53bee285fd6..8688f07f04e 100644 --- a/packages/rs-drive-abci/src/rpc/mod.rs +++ b/packages/rs-drive-abci/src/rpc/mod.rs @@ -1,4 +1,7 @@ /// Dash Core RPC pub mod core; + +/// Speculative Core RPC fetching for consecutive core heights +pub mod prefetch; /// Core signatures verification pub mod signature; diff --git a/packages/rs-drive-abci/src/rpc/prefetch.rs b/packages/rs-drive-abci/src/rpc/prefetch.rs new file mode 100644 index 00000000000..9658bcafe87 --- /dev/null +++ b/packages/rs-drive-abci/src/rpc/prefetch.rs @@ -0,0 +1,143 @@ +//! Speculative fetching of the two Core RPC responses a block needs when the +//! core chain-locked height advances. +//! +//! Replaying mainnet history, roughly every other Platform block advances the +//! core height by one, and each of those blocks blocks on `protx listdiff` and +//! `quorum listextended` in turn — about a millisecond of the seven a block +//! costs. The heights are consecutive, so the answer for the next one can be +//! fetched while the current block is still executing. +//! +//! Two things keep this from misbehaving at the tip, where the next core block +//! does not exist yet: the speculative call runs on its own connection, so it +//! never delays a real one, and a failed guess backs the prefetcher off for a +//! while instead of asking Core for a block it does not have on every block. + +use std::sync::atomic::{AtomicU32, Ordering}; +use std::sync::mpsc::{sync_channel, Receiver}; +use std::sync::{Arc, Mutex}; + +use dpp::dashcore_rpc::dashcore_rpc_json::{ExtendedQuorumListResult, MasternodeListDiff}; +use dpp::dashcore_rpc::{Auth, Client, Error as CoreError, RpcApi}; + +/// How many calls to skip after a speculative fetch fails. A failure means the +/// guessed height is not on Core's chain yet, which is the steady state at the +/// tip, and asking again on the next block would just repeat the error. +const BACKOFF_CALLS: u32 = 32; + +struct Pending { + key: K, + result: Receiver>, +} + +/// Holds one in-flight speculative fetch of each kind. +pub struct CorePrefetcher { + client: Arc, + quorum_list: Mutex>>, + protx_diff: Mutex>>, + /// Calls left to skip before speculating again after a failure. + backoff: AtomicU32, +} + +impl std::fmt::Debug for CorePrefetcher { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("CorePrefetcher") + } +} + +impl CorePrefetcher { + /// Opens a second connection to Core for speculative calls. Returns `None` + /// if it cannot be opened; prefetching is an optimisation, and a node that + /// cannot open it should still run. + pub fn new(url: &str, username: String, password: String) -> Option { + let client = Client::new(url, Auth::UserPass(username, password)).ok()?; + Some(CorePrefetcher { + client: Arc::new(client), + quorum_list: Mutex::new(None), + protx_diff: Mutex::new(None), + backoff: AtomicU32::new(0), + }) + } + + fn may_speculate(&self) -> bool { + let left = self.backoff.load(Ordering::Relaxed); + if left == 0 { + return true; + } + self.backoff.store(left - 1, Ordering::Relaxed); + false + } + + fn note_failure(&self) { + self.backoff.store(BACKOFF_CALLS, Ordering::Relaxed); + } + + /// Takes the speculative quorum list for `height`, if one was started and + /// succeeded. Blocks until the in-flight call finishes. + pub fn take_quorum_list(&self, height: u32) -> Option { + let pending = self.quorum_list.lock().ok()?.take()?; + if pending.key != height { + return None; + } + match pending.result.recv().ok()? { + Ok(list) => Some(list), + Err(_) => { + self.note_failure(); + None + } + } + } + + /// Starts fetching the quorum list for `height` in the background. + pub fn start_quorum_list(&self, height: u32) { + if !self.may_speculate() { + return; + } + let Ok(mut slot) = self.quorum_list.lock() else { + return; + }; + let (tx, rx) = sync_channel(1); + let client = Arc::clone(&self.client); + std::thread::spawn(move || { + let _ = tx.send(client.get_quorum_listextended_reversed(Some(height))); + }); + *slot = Some(Pending { + key: height, + result: rx, + }); + } + + /// Takes the speculative masternode list diff for `base -> block`, if one + /// was started and succeeded. + pub fn take_protx_diff(&self, base: u32, block: u32) -> Option { + let pending = self.protx_diff.lock().ok()?.take()?; + if pending.key != (base, block) { + return None; + } + match pending.result.recv().ok()? { + Ok(diff) => Some(diff), + Err(_) => { + self.note_failure(); + None + } + } + } + + /// Starts fetching the masternode list diff `base -> block` in the background. + pub fn start_protx_diff(&self, base: u32, block: u32) { + if !self.may_speculate() { + return; + } + let Ok(mut slot) = self.protx_diff.lock() else { + return; + }; + let (tx, rx) = sync_channel(1); + let client = Arc::clone(&self.client); + std::thread::spawn(move || { + let _ = tx.send(client.get_protx_listdiff(base, block)); + }); + *slot = Some(Pending { + key: (base, block), + result: rx, + }); + } +}