Banking tracer with simulator wip

core/src/banking_stage.rs

@@ -420,14 +420,16 @@ impl BankingStage {
             TOTAL_BUFFERED_PACKETS / ((num_threads - NUM_VOTE_PROCESSING_THREADS) as usize);
         // Keeps track of extraneous vote transactions for the vote threads
         let latest_unprocessed_votes = Arc::new(LatestUnprocessedVotes::new());
-        let should_split_voting_threads = bank_forks
+        let mut should_split_voting_threads = bank_forks
             .read()
             .map(|bank_forks| {
                 let bank = bank_forks.root_bank();
                 bank.feature_set
                     .is_active(&allow_votes_to_directly_update_vote_state::id())
             })
             .unwrap_or(false);
+        info!("should_split_voting_threads: {}", should_split_voting_threads);
+        //should_split_voting_threads = false;
         // Many banks that process transactions in parallel.
         let bank_thread_hdls: Vec<JoinHandle<()>> = (0..num_threads)
             .map(|i| {
@@ -479,8 +481,9 @@ impl BankingStage {
                 let data_budget = data_budget.clone();
                 let connection_cache = connection_cache.clone();
                 let bank_forks = bank_forks.clone();
+                let thread_name = format!("solBanknStgTx{i:02}");
                 Builder::new()
-                    .name(format!("solBanknStgTx{i:02}"))
+                    .name(thread_name.clone())
                     .spawn(move || {
                         Self::process_loop(
                             &mut packet_deserializer,
@@ -496,6 +499,7 @@ impl BankingStage {
                             &bank_forks,
                             unprocessed_transaction_storage,
                         );
+                        info!("banking thread ended: {}!", thread_name);
                     })
                     .unwrap()
             })

core/src/packet_deserializer.rs

@@ -50,7 +50,7 @@ impl PacketDeserializer {
 
     /// Deserialize packet batches, aggregates tracer packet stats, and collect
     /// them into ReceivePacketResults
-    fn deserialize_and_collect_packets(
+    pub fn deserialize_and_collect_packets(
         packet_count: usize,
         banking_batches: &[BankingPacketBatch],
     ) -> ReceivePacketResults {

core/src/replay_stage.rs

@@ -1776,7 +1776,7 @@ impl ReplayStage {
                 root_slot,
                 my_pubkey,
                 rpc_subscriptions,
-                NewBankOptions { vote_only_bank },
+                NewBankOptions { vote_only_bank, ..Default::default() },
             );
             // make sure parent is frozen for finalized hashes via the above
             // new()-ing of its child bank

gossip/src/cluster_info.rs

@@ -406,7 +406,7 @@ impl ClusterInfo {
         keypair: Arc<Keypair>,
         socket_addr_space: SocketAddrSpace,
     ) -> Self {
-        assert_eq!(contact_info.id, keypair.pubkey());
+        //assert_eq!(contact_info.id, keypair.pubkey());
         let id = contact_info.id;
         let me = Self {
             gossip: CrdsGossip::default(),

ledger-tool/src/main.rs

@@ -24,7 +24,10 @@ use {
         },
     },
     solana_cli_output::{CliAccount, CliAccountNewConfig, OutputFormat},
-    solana_core::system_monitor_service::{SystemMonitorService, SystemMonitorStatsReportConfig},
+    solana_core::{
+        banking_trace::BankingSimulator,
+        system_monitor_service::{SystemMonitorService, SystemMonitorStatsReportConfig},
+    },
     solana_entry::entry::Entry,
     solana_geyser_plugin_manager::geyser_plugin_service::GeyserPluginService,
     solana_ledger::{
@@ -1196,7 +1199,7 @@ fn load_bank_forks(
     .map(|_| (bank_forks, starting_snapshot_hashes));
 
     exit.store(true, Ordering::Relaxed);
-    accounts_background_service.join().unwrap();
+    //accounts_background_service.join().unwrap();
 
     result
 }
@@ -2027,6 +2030,11 @@ fn main() {
                     .help("Snapshot archive format to use.")
                     .conflicts_with("no_snapshot")
             )
+        ).subcommand(
+            SubCommand::with_name("simulate-leader-blocks")
+            .about("Simulate recreating blocks with banking trace as if a leader")
+            .arg(&halt_at_slot_arg)
+            .arg(&max_genesis_archive_unpacked_size_arg)
         ).subcommand(
             SubCommand::with_name("accounts")
             .about("Print account stats and contents after processing the ledger")
@@ -3358,6 +3366,90 @@ fn main() {
                     }
                 }
             }
+            ("simulate-leader-blocks", Some(arg_matches)) => {
+                let simulator = BankingSimulator::new(PathBuf::new().join("/dev/stdin"));
+
+                if std::env::var("DUMP").is_ok() {
+                    simulator.dump(None);
+                    return
+                }
+
+                let mut accounts_index_config = AccountsIndexConfig::default();
+                if let Some(bins) = value_t!(arg_matches, "accounts_index_bins", usize).ok() {
+                    accounts_index_config.bins = Some(bins);
+                }
+
+                accounts_index_config.index_limit_mb = if let Some(limit) =
+                    value_t!(arg_matches, "accounts_index_memory_limit_mb", usize).ok()
+                {
+                    IndexLimitMb::Limit(limit)
+                } else if arg_matches.is_present("disable_accounts_disk_index") {
+                    IndexLimitMb::InMemOnly
+                } else {
+                    IndexLimitMb::Unspecified
+                };
+
+                {
+                    let mut accounts_index_paths: Vec<PathBuf> =
+                        if arg_matches.is_present("accounts_index_path") {
+                            values_t_or_exit!(arg_matches, "accounts_index_path", String)
+                                .into_iter()
+                                .map(PathBuf::from)
+                                .collect()
+                        } else {
+                            vec![]
+                        };
+                    if accounts_index_paths.is_empty() {
+                        accounts_index_paths = vec![ledger_path.join("accounts_index")];
+                    }
+                    accounts_index_config.drives = Some(accounts_index_paths);
+                }
+
+                let accounts_db_config = Some(AccountsDbConfig {
+                    ancient_append_vec_offset: value_t!(
+                        matches,
+                        "accounts_db_ancient_append_vecs",
+                        i64
+                    )
+                    .ok(),
+                    skip_initial_hash_calc: arg_matches
+                        .is_present("accounts_db_skip_initial_hash_calculation"),
+                    ..AccountsDbConfig::default()
+                });
+
+                let process_options = ProcessOptions {
+                    poh_verify: false,
+                    halt_at_slot: value_t!(arg_matches, "halt_at_slot", Slot).ok(),
+                    accounts_db_config,
+                    ..ProcessOptions::default()
+                };
+
+                let blockstore = open_blockstore(
+                    &ledger_path,
+                    AccessType::Secondary,
+                    wal_recovery_mode,
+                    &shred_storage_type,
+                    force_update_to_open,
+                );
+                let genesis_config = open_genesis_config_by(&ledger_path, arg_matches);
+                let (bank_forks, ..) = load_bank_forks(
+                    arg_matches,
+                    &genesis_config,
+                    &blockstore,
+                    process_options,
+                    snapshot_archive_path,
+                    incremental_snapshot_archive_path,
+                )
+                .unwrap_or_else(|err| {
+                    eprintln!("Ledger verification failed: {:?}", err);
+                    exit(1);
+                });
+
+                //simulator.seek(bank); => Ok or Err("no BankStart")
+                simulator.simulate(&genesis_config, bank_forks, Arc::new(blockstore));
+
+                println!("Ok");
+            }
             ("accounts", Some(arg_matches)) => {
                 let halt_at_slot = value_t!(arg_matches, "halt_at_slot", Slot).ok();
                 let process_options = ProcessOptions {

ledger/src/blockstore.rs

@@ -229,6 +229,10 @@ impl Blockstore {
         self.ledger_path.join("banking_trace")
     }
 
+    pub fn banking_retracer_path(&self) -> PathBuf {
+        self.ledger_path.join("banking_retrace")
+    }
+
     /// Opens a Ledger in directory, provides "infinite" window of shreds
     pub fn open(ledger_path: &Path) -> Result<Blockstore> {
         Self::do_open(ledger_path, BlockstoreOptions::default())

poh/src/poh_recorder.rs

@@ -61,12 +61,21 @@ type Result<T> = std::result::Result<T, PohRecorderError>;
 
 pub type WorkingBankEntry = (Arc<Bank>, (Entry, u64));
 
-#[derive(Debug, Clone)]
+#[derive(Clone)]
 pub struct BankStart {
     pub working_bank: Arc<Bank>,
     pub bank_creation_time: Arc<Instant>,
 }
 
+impl std::fmt::Debug for BankStart {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("BankStart")
+            .field("working_bank", &format!("Bank slot: {}", &self.working_bank.slot()))
+            .field("bank_creation_time", &self.bank_creation_time)
+            .finish_non_exhaustive()
+    }
+}
+
 impl BankStart {
     fn get_working_bank_if_not_expired(&self) -> Option<&Arc<Bank>> {
         if self.should_working_bank_still_be_processing_txs() {
@@ -351,6 +360,10 @@ impl PohRecorder {
         self.ticks_per_slot
     }
 
+    pub fn slot(&self) -> Slot {
+        self.tick_height() / self.ticks_per_slot()
+    }
+
     pub fn recorder(&self) -> TransactionRecorder {
         TransactionRecorder::new(self.record_sender.clone(), self.is_exited.clone())
     }

poh/src/poh_service.rs

@@ -342,6 +342,7 @@ impl PohService {
         let poh = poh_recorder.read().unwrap().poh.clone();
         let mut timing = PohTiming::new();
         let mut next_record = None;
+        info!("starting tick producer at {}ns/(1slot={}ticks):", target_ns_per_tick * ticks_per_slot, ticks_per_slot);
         loop {
             let should_tick = Self::record_or_hash(
                 &mut next_record,

runtime/src/accounts.rs

@@ -543,6 +543,7 @@ impl Accounts {
         feature_set: &FeatureSet,
         fee_structure: &FeeStructure,
         account_overrides: Option<&AccountOverrides>,
+        lamports_per_signature_override: Option<u64>,
     ) -> Vec<TransactionLoadResult> {
         txs.iter()
             .zip(lock_results)
@@ -554,7 +555,9 @@ impl Accounts {
                         .unwrap_or_else(|| {
                             hash_queue.get_lamports_per_signature(tx.message().recent_blockhash())
                         });
-                    let fee = if let Some(lamports_per_signature) = lamports_per_signature {
+                    let fee = if let Some(lamports_per_signature) =
+                        lamports_per_signature.or(lamports_per_signature_override)
+                    {
                         Bank::calculate_fee(
                             tx.message(),
                             lamports_per_signature,
@@ -1422,6 +1425,7 @@ mod tests {
             feature_set,
             fee_structure,
             None,
+            None,
         )
     }
 
@@ -3078,6 +3082,7 @@ mod tests {
             &FeatureSet::all_enabled(),
             &FeeStructure::default(),
             account_overrides,
+            None,
         )
     }