ledger-tool: Make joining AccountsBackgroundService optional (#1673)

AccountsBackgroundService performs several operations that can take a
long time to complete and do not check the exit flag mid-operation.
Thus, ledger-tool can get hung up for a while waiting for ABS to
finish. However, many ledger-tool command do not ABS to have finished.

So, return a handle to the ABS thread and allow the caller to decide
whether to join ABS or not. As of right now, create-snapshot is the
only command that requires ABS to have finished before continuing.

ledger-tool/src/ledger_utils.rs

@@ -54,6 +54,19 @@ use {
     thiserror::Error,
 };
 
+pub struct LoadAndProcessLedgerOutput {
+    pub bank_forks: Arc<RwLock<BankForks>>,
+    pub starting_snapshot_hashes: Option<StartingSnapshotHashes>,
+    // Typically, we would want to join all threads before returning. However,
+    // AccountsBackgroundService (ABS) performs several long running operations
+    // that don't respond to the exit flag. Blocking on these operations could
+    // significantly delay getting results that do not need ABS to finish. So,
+    // skip joining ABS and instead let the caller decide whether to block or
+    // not. It is safe to let ABS continue in the background, and ABS will stop
+    // if/when it finally checks the exit flag
+    pub accounts_background_service: AccountsBackgroundService,
+}
+
 const PROCESS_SLOTS_HELP_STRING: &str =
     "The starting slot is either the latest found snapshot slot, or genesis (slot 0) if the \
      --no-snapshot flag was specified or if no snapshots were found. \
@@ -99,7 +112,7 @@ pub fn load_and_process_ledger_or_exit(
     blockstore: Arc<Blockstore>,
     process_options: ProcessOptions,
     transaction_status_sender: Option<TransactionStatusSender>,
-) -> (Arc<RwLock<BankForks>>, Option<StartingSnapshotHashes>) {
+) -> LoadAndProcessLedgerOutput {
     load_and_process_ledger(
         arg_matches,
         genesis_config,
@@ -119,7 +132,7 @@ pub fn load_and_process_ledger(
     blockstore: Arc<Blockstore>,
     process_options: ProcessOptions,
     transaction_status_sender: Option<TransactionStatusSender>,
-) -> Result<(Arc<RwLock<BankForks>>, Option<StartingSnapshotHashes>), LoadAndProcessLedgerError> {
+) -> Result<LoadAndProcessLedgerOutput, LoadAndProcessLedgerError> {
     let bank_snapshots_dir = if blockstore.is_primary_access() {
         blockstore.ledger_path().join("snapshot")
     } else {
@@ -402,11 +415,14 @@ pub fn load_and_process_ledger(
         None, // Maybe support this later, though
         &accounts_background_request_sender,
     )
-    .map(|_| (bank_forks, starting_snapshot_hashes))
+    .map(|_| LoadAndProcessLedgerOutput {
+        bank_forks,
+        starting_snapshot_hashes,
+        accounts_background_service,
+    })
     .map_err(LoadAndProcessLedgerError::ProcessBlockstoreFromRoot);
 
     exit.store(true, Ordering::Relaxed);
-    accounts_background_service.join().unwrap();
     accounts_hash_verifier.join().unwrap();
     if let Some(service) = transaction_status_service {
         service.join().unwrap();

ledger-tool/src/main.rs

@@ -1439,13 +1439,14 @@ fn main() {
                         arg_matches,
                         get_access_type(&process_options),
                     );
-                    let (bank_forks, _) = load_and_process_ledger_or_exit(
-                        arg_matches,
-                        &genesis_config,
-                        Arc::new(blockstore),
-                        process_options,
-                        None,
-                    );
+                    let LoadAndProcessLedgerOutput { bank_forks, .. } =
+                        load_and_process_ledger_or_exit(
+                            arg_matches,
+                            &genesis_config,
+                            Arc::new(blockstore),
+                            process_options,
+                            None,
+                        );
 
                     println!(
                         "{}",
@@ -1629,13 +1630,14 @@ fn main() {
                         arg_matches,
                         get_access_type(&process_options),
                     );
-                    let (bank_forks, _) = load_and_process_ledger_or_exit(
-                        arg_matches,
-                        &genesis_config,
-                        Arc::new(blockstore),
-                        process_options,
-                        transaction_status_sender,
-                    );
+                    let LoadAndProcessLedgerOutput { bank_forks, .. } =
+                        load_and_process_ledger_or_exit(
+                            arg_matches,
+                            &genesis_config,
+                            Arc::new(blockstore),
+                            process_options,
+                            transaction_status_sender,
+                        );
 
                     let working_bank = bank_forks.read().unwrap().working_bank();
                     if print_accounts_stats {
@@ -1695,13 +1697,14 @@ fn main() {
                         arg_matches,
                         get_access_type(&process_options),
                     );
-                    let (bank_forks, _) = load_and_process_ledger_or_exit(
-                        arg_matches,
-                        &genesis_config,
-                        Arc::new(blockstore),
-                        process_options,
-                        None,
-                    );
+                    let LoadAndProcessLedgerOutput { bank_forks, .. } =
+                        load_and_process_ledger_or_exit(
+                            arg_matches,
+                            &genesis_config,
+                            Arc::new(blockstore),
+                            process_options,
+                            None,
+                        );
 
                     let dot = graph_forks(&bank_forks.read().unwrap(), &graph_config);
                     let extension = Path::new(&output_file).extension();
@@ -1864,13 +1867,23 @@ fn main() {
                         output_directory.display()
                     );
 
-                    let (bank_forks, starting_snapshot_hashes) = load_and_process_ledger_or_exit(
+                    let LoadAndProcessLedgerOutput {
+                        bank_forks,
+                        starting_snapshot_hashes,
+                        accounts_background_service,
+                    } = load_and_process_ledger_or_exit(
                         arg_matches,
                         &genesis_config,
                         blockstore.clone(),
                         process_options,
                         None,
                     );
+                    // Snapshot creation will implicitly perform AccountsDb
+                    // flush and clean operations. These operations cannot be
+                    // run concurrently, so ensure ABS is stopped to avoid that
+                    // possibility.
+                    accounts_background_service.join().unwrap();
+
                     let mut bank = bank_forks
                         .read()
                         .unwrap()
@@ -2252,13 +2265,14 @@ fn main() {
                         arg_matches,
                         get_access_type(&process_options),
                     );
-                    let (bank_forks, _) = load_and_process_ledger_or_exit(
-                        arg_matches,
-                        &genesis_config,
-                        Arc::new(blockstore),
-                        process_options,
-                        None,
-                    );
+                    let LoadAndProcessLedgerOutput { bank_forks, .. } =
+                        load_and_process_ledger_or_exit(
+                            arg_matches,
+                            &genesis_config,
+                            Arc::new(blockstore),
+                            process_options,
+                            None,
+                        );
                     let bank = bank_forks.read().unwrap().working_bank();
 
                     let include_sysvars = arg_matches.is_present("include_sysvars");
@@ -2303,13 +2317,14 @@ fn main() {
                         arg_matches,
                         get_access_type(&process_options),
                     );
-                    let (bank_forks, _) = load_and_process_ledger_or_exit(
-                        arg_matches,
-                        &genesis_config,
-                        Arc::new(blockstore),
-                        process_options,
-                        None,
-                    );
+                    let LoadAndProcessLedgerOutput { bank_forks, .. } =
+                        load_and_process_ledger_or_exit(
+                            arg_matches,
+                            &genesis_config,
+                            Arc::new(blockstore),
+                            process_options,
+                            None,
+                        );
                     let bank_forks = bank_forks.read().unwrap();
                     let slot = bank_forks.working_bank().slot();
                     let bank = bank_forks.get(slot).unwrap_or_else(|| {

ledger-tool/src/program.rs

@@ -79,7 +79,7 @@ fn load_blockstore(ledger_path: &Path, arg_matches: &ArgMatches<'_>) -> Arc<Bank
     let genesis_config = open_genesis_config_by(ledger_path, arg_matches);
     info!("genesis hash: {}", genesis_config.hash());
     let blockstore = open_blockstore(ledger_path, arg_matches, AccessType::Secondary);
-    let (bank_forks, ..) = load_and_process_ledger_or_exit(
+    let LoadAndProcessLedgerOutput { bank_forks, .. } = load_and_process_ledger_or_exit(
         arg_matches,
         &genesis_config,
         Arc::new(blockstore),