ledger-tool: stream output of accounts subcommand

ledger-tool/src/main.rs

@@ -11,7 +11,10 @@ use {
     itertools::Itertools,
     log::*,
     regex::Regex,
-    serde::Serialize,
+    serde::{
+        ser::{SerializeSeq, Serializer},
+        Serialize,
+    },
     serde_json::json,
     solana_account_decoder::{UiAccount, UiAccountData, UiAccountEncoding},
     solana_clap_utils::{
@@ -20,6 +23,7 @@ use {
             is_parsable, is_pow2, is_pubkey, is_pubkey_or_keypair, is_slot, is_valid_percentage,
         },
     },
+    solana_cli_output::{CliAccount, OutputFormat},
     solana_core::{
         system_monitor_service::SystemMonitorService, validator::move_and_async_delete_path,
     },
@@ -39,14 +43,15 @@ use {
     },
     solana_measure::{measure, measure::Measure},
     solana_runtime::{
+        accounts::Accounts,
         accounts_background_service::{
             AbsRequestHandlers, AbsRequestSender, AccountsBackgroundService,
             PrunedBanksRequestHandler, SnapshotRequestHandler,
         },
         accounts_db::{AccountsDbConfig, FillerAccountsConfig},
         accounts_index::{AccountsIndexConfig, IndexLimitMb, ScanConfig},
         accounts_update_notifier_interface::AccountsUpdateNotifier,
-        bank::{Bank, RewardCalculationEvent},
+        bank::{Bank, RewardCalculationEvent, TotalAccountsStats},
         bank_forks::BankForks,
         cost_model::CostModel,
         cost_tracker::CostTracker,
@@ -3369,52 +3374,63 @@ fn main() {
                 });
 
                 let bank = bank_forks.read().unwrap().working_bank();
-                let mut measure = Measure::start("getting accounts");
-                let accounts: BTreeMap<_, _> = bank
-                    .get_all_accounts_with_modified_slots()
-                    .unwrap()
-                    .into_iter()
-                    .filter(|(pubkey, _account, _slot)| {
-                        include_sysvars || !solana_sdk::sysvar::is_sysvar_id(pubkey)
-                    })
-                    .map(|(pubkey, account, slot)| (pubkey, (account, slot)))
-                    .collect();
-                measure.stop();
-                info!("{}", measure);
+                let mut serializer = serde_json::Serializer::new(stdout());
+                let (summarize, mut json_serializer) =
+                    match OutputFormat::from_matches(arg_matches, "output_format", false) {
+                        OutputFormat::Json | OutputFormat::JsonCompact => {
+                            (false, Some(serializer.serialize_seq(None).unwrap()))
+                        }
+                        _ => (true, None),
+                    };
+                let mut total_accounts_stats = TotalAccountsStats::default();
+                let rent_collector = bank.rent_collector();
+                let print_account_contents = !arg_matches.is_present("no_account_contents");
+                let print_account_data = !arg_matches.is_present("no_account_data");
+                let print_encoding_format = match arg_matches.value_of("encoding") {
+                    Some("jsonParsed") => UiAccountEncoding::JsonParsed,
+                    Some("base64") => UiAccountEncoding::Base64,
+                    Some("base64+zstd") => UiAccountEncoding::Base64Zstd,
+                    _ => UiAccountEncoding::Base58,
+                };
+                let scan_func = |some_account_tuple: Option<(&Pubkey, AccountSharedData, Slot)>| {
+                    if let Some((pubkey, account, slot)) = some_account_tuple
+                        .filter(|(_, account, _)| Accounts::is_loadable(account.lamports()))
+                    {
+                        if !include_sysvars && solana_sdk::sysvar::is_sysvar_id(pubkey) {
+                            return;
+                        }
 
-                let mut measure = Measure::start("calculating total accounts stats");
-                let total_accounts_stats = bank.calculate_total_accounts_stats(
-                    accounts
-                        .iter()
-                        .map(|(pubkey, (account, _slot))| (pubkey, account)),
-                );
-                measure.stop();
-                info!("{}", measure);
+                        total_accounts_stats.accumulate_account(pubkey, &account, rent_collector);
 
-                let print_account_contents = !arg_matches.is_present("no_account_contents");
-                if print_account_contents {
-                    let print_account_data = !arg_matches.is_present("no_account_data");
-                    let print_encoding_format = match arg_matches.value_of("encoding") {
-                        Some("jsonParsed") => UiAccountEncoding::JsonParsed,
-                        Some("base64") => UiAccountEncoding::Base64,
-                        Some("base64+zstd") => UiAccountEncoding::Base64Zstd,
-                        _ => UiAccountEncoding::Base64,
-                    };
-                    let mut measure = Measure::start("printing account contents");
-                    for (pubkey, (account, slot)) in accounts.into_iter() {
-                        output_account(
-                            &pubkey,
-                            &account,
-                            Some(slot),
-                            print_account_data,
-                            print_encoding_format,
-                        );
+                        if print_account_contents {
+                            if let Some(json_serializer) = json_serializer.as_mut() {
+                                let cli_account = CliAccount::new(
+                                    pubkey, &account, false, /*use_lamports_unit*/
+                                );
+                                json_serializer.serialize_element(&cli_account).unwrap();
+                            } else {
+                                output_account(
+                                    pubkey,
+                                    &account,
+                                    Some(slot),
+                                    print_account_data,
+                                    print_encoding_format,
+                                );
+                            }
+                        }
                     }
-                    measure.stop();
-                    info!("{}", measure);
+                };
+                let mut measure = Measure::start("scanning accounts");
+                bank.scan_all_accounts_with_modified_slots(scan_func)
+                    .unwrap();
+                measure.stop();
+                info!("{}", measure);
+                if let Some(json_serializer) = json_serializer {
+                    json_serializer.end().unwrap();
+                }
+                if summarize {
+                    println!("\n{:#?}", total_accounts_stats);
                 }
-
-                println!("{:#?}", total_accounts_stats);
             }
             ("capitalization", Some(arg_matches)) => {
                 let halt_at_slot = value_t!(arg_matches, "halt_at_slot", Slot).ok();

runtime/src/accounts.rs

@@ -840,7 +840,7 @@ impl Accounts {
         }
     }
 
-    fn is_loadable(lamports: u64) -> bool {
+    pub fn is_loadable(lamports: u64) -> bool {
         // Don't ever load zero lamport accounts into runtime because
         // the existence of zero-lamport accounts are never deterministic!!
         lamports > 0
@@ -1008,6 +1008,23 @@ impl Accounts {
         )
     }
 
+    pub fn scan_all<F>(
+        &self,
+        ancestors: &Ancestors,
+        bank_id: BankId,
+        scan_func: F,
+    ) -> ScanResult<()>
+    where
+        F: FnMut(Option<(&Pubkey, AccountSharedData, Slot)>),
+    {
+        self.accounts_db.scan_accounts_no_default(
+            ancestors,
+            bank_id,
+            scan_func,
+            &ScanConfig::default(),
+        )
+    }
+
     pub fn hold_range_in_memory<R>(
         &self,
         range: &R,

runtime/src/accounts_db.rs

@@ -4637,7 +4637,25 @@ impl AccountsDb {
         A: Default,
     {
         let mut collector = A::default();
+        self.scan_accounts_no_default(
+            ancestors,
+            bank_id,
+            |maybe_account_info| scan_func(&mut collector, maybe_account_info),
+            config,
+        )
+        .map(|_| collector)
+    }
 
+    pub fn scan_accounts_no_default<F>(
+        &self,
+        ancestors: &Ancestors,
+        bank_id: BankId,
+        mut scan_func: F,
+        config: &ScanConfig,
+    ) -> ScanResult<()>
+    where
+        F: FnMut(Option<(&Pubkey, AccountSharedData, Slot)>),
+    {
         // This can error out if the slots being scanned over are aborted
         self.accounts_index.scan_accounts(
             ancestors,
@@ -4647,12 +4665,12 @@ impl AccountsDb {
                     .get_account_accessor(slot, pubkey, &account_info.storage_location())
                     .get_loaded_account()
                     .map(|loaded_account| (pubkey, loaded_account.take_account(), slot));
-                scan_func(&mut collector, account_slot)
+                scan_func(account_slot)
             },
             config,
         )?;
 
-        Ok(collector)
+        Ok(())
     }
 
     pub fn unchecked_scan_accounts<F, A>(

runtime/src/bank.rs

@@ -6553,6 +6553,15 @@ impl Bank {
         self.rc.accounts.load_all(&self.ancestors, self.bank_id)
     }
 
+    pub fn scan_all_accounts_with_modified_slots<F>(&self, scan_func: F) -> ScanResult<()>
+    where
+        F: FnMut(Option<(&Pubkey, AccountSharedData, Slot)>),
+    {
+        self.rc
+            .accounts
+            .scan_all(&self.ancestors, self.bank_id, scan_func)
+    }
+
     pub fn get_program_accounts_modified_since_parent(
         &self,
         program_id: &Pubkey,