Reduce downtime during shutdown (#742)

Author: taco-paco

Cargo.lock

@@ -219,7 +219,8 @@ features = ["lz4"]
 solana-account = { git = "https://github.com/magicblock-labs/solana-account.git", rev = "1beed4c" }
 solana-storage-proto = { path = "./storage-proto" }
 solana-svm = { git = "https://github.com/magicblock-labs/magicblock-svm.git", rev = "3e9456ec4" }
-# using parent commit d4e9e16 of rocksdb 0.23.0
+# Fork is used to enable `disable_manual_compaction` usage
+# Fork is based on commit d4e9e16 of rocksdb (parent commit of 0.23.0 release)
 # without patching update isn't possible due to conflict with solana deps
-# TODO(edwin): remove once solana version is raised to 2.3
-rocksdb = { git = "https://github.com/rust-rocksdb/rust-rocksdb.git", rev = "d4e9e16" }
+# TODO(edwin): remove once solana deps upgraded and are using rust-rocksdb 0.25.0(likely)
+rocksdb = { git = "https://github.com/magicblock-labs/rust-rocksdb.git", rev = "6d975197" }

Cargo.toml

@@ -662,7 +662,6 @@ impl MagicValidator {
             committor_service.stop();
         }
 
-        self.ledger_truncator.stop();
         self.claim_fees_task.stop();
 
         if self.config.chain_operation.is_some()
@@ -687,6 +686,15 @@ impl MagicValidator {
     pub fn ledger(&self) -> &Ledger {
         &self.ledger
     }
+
+    /// Prepares RocksDB for shutdown by cancelling all Manual compactions
+    /// This speeds up `stop` as it doesn't have to await for compaction cancellation
+    /// Calling this still allows to write or read from DB
+    pub fn prepare_ledger_for_shutdown(&mut self) {
+        self.ledger_truncator.stop();
+        // Calls & awaits until manual compaction is canceled
+        self.ledger.cancel_manual_compactions();
+    }
 }
 
 fn programs_to_load(programs: &[LoadableProgram]) -> Vec<(Pubkey, PathBuf)> {

magicblock-api/src/magic_validator.rs

@@ -11,6 +11,10 @@ use std::{
 use bincode::{deserialize, serialize};
 use log::*;
 use magicblock_core::link::blocks::BlockHash;
+use magicblock_metrics::metrics::{
+    start_ledger_disable_compactions_timer, start_ledger_shutdown_timer,
+    HistogramTimer,
+};
 use rocksdb::{Direction as IteratorDirection, FlushOptions};
 use scc::HashCache;
 use solana_clock::{Slot, UnixTimestamp};
@@ -1262,12 +1266,28 @@ impl Ledger {
 
     /// Graceful db shutdown
     pub fn shutdown(&self, wait: bool) -> LedgerResult<()> {
+        let _guard = MeasureGuard {
+            measure: Measure::start("Ledger shutdown"),
+            _timer: start_ledger_shutdown_timer(),
+        };
         self.flush()?;
         self.db.backend.db.cancel_all_background_work(wait);
 
         Ok(())
     }
 
+    /// Cancels manual compaction
+    /// Here we utilize the internal of `disable_manual_compaction`
+    /// Which not only disables future manual compaction,
+    /// but also cancels all the running one
+    pub fn cancel_manual_compactions(&self) {
+        let _guard = MeasureGuard {
+            measure: Measure::start("Compaction cancellation"),
+            _timer: start_ledger_disable_compactions_timer(),
+        };
+        self.db.backend.db.disable_manual_compaction();
+    }
+
     /// Cached latest block data
     pub fn latest_block(&self) -> &LatestBlock {
         &self.latest_block
@@ -1311,6 +1331,19 @@ impl_has_column!(TransactionMemos, transaction_memos_cf);
 impl_has_column!(PerfSamples, perf_samples_cf);
 impl_has_column!(AccountModDatas, account_mod_datas_cf);
 
+struct MeasureGuard {
+    measure: Measure,
+    _timer: HistogramTimer,
+}
+
+impl Drop for MeasureGuard {
+    fn drop(&mut self) {
+        self.measure.stop();
+        // We print it in case metrics wouldn't have time to be scraped
+        info!("{}", self.measure);
+    }
+}
+
 // -----------------
 // Tests
 // -----------------

magicblock-config/src/validator.rs

@@ -63,6 +63,7 @@ async fn test_truncator_not_purged_size() {
             signature.unwrap()
         })
         .collect::<Vec<_>>();
+    ledger.flush().unwrap();
 
     ledger_truncator.start();
     tokio::time::sleep(Duration::from_millis(10)).await;
@@ -87,6 +88,7 @@ async fn test_truncator_non_empty_ledger() {
         })
         .collect::<Vec<_>>();
 
+    ledger.flush().unwrap();
     let mut ledger_truncator =
         LedgerTruncator::new(ledger.clone(), TEST_TRUNCATION_TIME_INTERVAL, 0);
 

magicblock-ledger/src/store/api.rs

@@ -118,6 +118,25 @@ lazy_static::lazy_static! {
         ),
     ).unwrap();
 
+    static ref LEDGER_DISABLE_COMPACTIONS_TIME: Histogram = Histogram::with_opts(
+        HistogramOpts::new(
+            "ledger_disable_compactions_time",
+            "Time in seconds spent on disabling manual compaction"
+        )
+        .buckets(
+            vec![0.1, 3.0, 10.0, 60.0, (10 * 60) as f64, (30 * 60) as f64]
+        ),
+        ).unwrap();
+
+    static ref LEDGER_SHUTDOWN_TIME: Histogram = Histogram::with_opts(
+        HistogramOpts::new(
+            "ledger_shutdown_time",
+            "Time taken for ledger to shutdown"
+        )
+        .buckets(
+            vec![0.1, 1.0, 2.0, 3.0, 10.0, 60.0]
+        ),
+        ).unwrap();
 
     // -----------------
     // Accounts
@@ -377,6 +396,8 @@ pub(crate) fn register() {
         register!(LEDGER_COLUMNS_COUNT_DURATION_SECONDS);
         register!(LEDGER_TRUNCATOR_COMPACTION_SECONDS);
         register!(LEDGER_TRUNCATOR_DELETE_SECONDS);
+        register!(LEDGER_DISABLE_COMPACTIONS_TIME);
+        register!(LEDGER_SHUTDOWN_TIME);
         register!(ACCOUNTS_SIZE_GAUGE);
         register!(ACCOUNTS_COUNT_GAUGE);
         register!(PENDING_ACCOUNT_CLONES_GAUGE);
@@ -483,6 +504,14 @@ pub fn observe_ledger_truncator_delete<T, F: FnOnce() -> T>(f: F) -> T {
     LEDGER_TRUNCATOR_DELETE_SECONDS.observe_closure_duration(f)
 }
 
+pub fn start_ledger_disable_compactions_timer() -> HistogramTimer {
+    LEDGER_DISABLE_COMPACTIONS_TIME.start_timer()
+}
+
+pub fn start_ledger_shutdown_timer() -> HistogramTimer {
+    LEDGER_SHUTDOWN_TIME.start_timer()
+}
+
 pub fn set_accounts_size(value: i64) {
     ACCOUNTS_SIZE_GAUGE.set(value)
 }

magicblock-ledger/tests/test_ledger_truncator.rs

@@ -138,5 +138,7 @@ async fn run() {
     if let Err(err) = Shutdown::wait().await {
         error!("Failed to gracefully shutdown: {}", err);
     }
+
+    api.prepare_ledger_for_shutdown();
     api.stop().await;
 }