Stabilize some banking stage tests (#6251)

* Stabilize some banking stage tests

Fixes #5660

* Fix CI...

* clean up

* Fix ci

* Address review nits

* Use bank.max_tick_height due to off-by-one for no PohRecord's clearing bank

* Fix CI...

* Use bank.max_tick_height() instead for clarity

banking_bench/src/main.rs

@@ -155,7 +155,7 @@ fn main() {
             Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger"),
         );
         let (exit, poh_recorder, poh_service, signal_receiver) =
-            create_test_recorder(&bank, &blocktree);
+            create_test_recorder(&bank, &blocktree, None);
         let cluster_info = ClusterInfo::new_with_invalid_keypair(Node::new_localhost().info);
         let cluster_info = Arc::new(RwLock::new(cluster_info));
         let _banking_stage = BankingStage::new(

core/benches/banking_stage.rs

@@ -66,7 +66,7 @@ fn bench_consume_buffered(bencher: &mut Bencher) {
             Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger"),
         );
         let (exit, poh_recorder, poh_service, _signal_receiver) =
-            create_test_recorder(&bank, &blocktree);
+            create_test_recorder(&bank, &blocktree, None);
 
         let tx = test_tx();
         let len = 4096;
@@ -198,7 +198,7 @@ fn bench_banking(bencher: &mut Bencher, tx_type: TransactionType) {
             Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger"),
         );
         let (exit, poh_recorder, poh_service, signal_receiver) =
-            create_test_recorder(&bank, &blocktree);
+            create_test_recorder(&bank, &blocktree, None);
         let cluster_info = ClusterInfo::new_with_invalid_keypair(Node::new_localhost().info);
         let cluster_info = Arc::new(RwLock::new(cluster_info));
         let _banking_stage = BankingStage::new(

core/src/banking_stage.rs

@@ -936,6 +936,7 @@ impl Service for BankingStage {
 pub fn create_test_recorder(
     bank: &Arc<Bank>,
     blocktree: &Arc<Blocktree>,
+    poh_config: Option<PohConfig>,
 ) -> (
     Arc<AtomicBool>,
     Arc<Mutex<PohRecorder>>,
@@ -943,7 +944,7 @@ pub fn create_test_recorder(
     Receiver<WorkingBankEntry>,
 ) {
     let exit = Arc::new(AtomicBool::new(false));
-    let poh_config = Arc::new(PohConfig::default());
+    let poh_config = Arc::new(poh_config.unwrap_or_default());
     let (mut poh_recorder, entry_receiver) = PohRecorder::new(
         bank.tick_height(),
         bank.last_blockhash(),
@@ -994,7 +995,7 @@ mod tests {
                 Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger"),
             );
             let (exit, poh_recorder, poh_service, _entry_receiever) =
-                create_test_recorder(&bank, &blocktree);
+                create_test_recorder(&bank, &blocktree, None);
             let cluster_info = ClusterInfo::new_with_invalid_keypair(Node::new_localhost().info);
             let cluster_info = Arc::new(RwLock::new(cluster_info));
             let banking_stage = BankingStage::new(
@@ -1019,6 +1020,7 @@ mod tests {
             mut genesis_block, ..
         } = create_genesis_block(2);
         genesis_block.ticks_per_slot = 4;
+        let num_extra_ticks = 2;
         let bank = Arc::new(Bank::new(&genesis_block));
         let start_hash = bank.last_blockhash();
         let (verified_sender, verified_receiver) = unbounded();
@@ -1028,8 +1030,10 @@ mod tests {
             let blocktree = Arc::new(
                 Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger"),
             );
+            let mut poh_config = PohConfig::default();
+            poh_config.target_tick_count = Some(bank.max_tick_height() + num_extra_ticks);
             let (exit, poh_recorder, poh_service, entry_receiver) =
-                create_test_recorder(&bank, &blocktree);
+                create_test_recorder(&bank, &blocktree, Some(poh_config));
             let cluster_info = ClusterInfo::new_with_invalid_keypair(Node::new_localhost().info);
             let cluster_info = Arc::new(RwLock::new(cluster_info));
             let banking_stage = BankingStage::new(
@@ -1039,7 +1043,6 @@ mod tests {
                 vote_receiver,
             );
             trace!("sending bank");
-            sleep(Duration::from_millis(600));
             drop(verified_sender);
             drop(vote_sender);
             exit.store(true, Ordering::Relaxed);
@@ -1077,8 +1080,11 @@ mod tests {
             let blocktree = Arc::new(
                 Blocktree::open(&ledger_path).expect("Expected to be able to open database ledger"),
             );
+            let mut poh_config = PohConfig::default();
+            // limit tick count to avoid clearing working_bank at PohRecord then PohRecorderError(MaxHeightReached) at BankingStage
+            poh_config.target_tick_count = Some(bank.max_tick_height() - 1);
             let (exit, poh_recorder, poh_service, entry_receiver) =
-                create_test_recorder(&bank, &blocktree);
+                create_test_recorder(&bank, &blocktree, Some(poh_config));
             let cluster_info = ClusterInfo::new_with_invalid_keypair(Node::new_localhost().info);
             let cluster_info = Arc::new(RwLock::new(cluster_info));
             let banking_stage = BankingStage::new(
@@ -1128,6 +1134,9 @@ mod tests {
 
             drop(verified_sender);
             drop(vote_sender);
+            // wait until banking_stage to finish up all packets
+            banking_stage.join().unwrap();
+
             exit.store(true, Ordering::Relaxed);
             poh_service.join().unwrap();
             drop(poh_recorder);
@@ -1136,18 +1145,20 @@ mod tests {
             let bank = Bank::new(&genesis_block);
             bank.process_transaction(&fund_tx).unwrap();
             //receive entries + ticks
-            for _ in 0..10 {
+            loop {
                 let entries: Vec<Entry> = entry_receiver
                     .iter()
                     .map(|(_bank, (entry, _tick_height))| entry)
                     .collect();
 
                 assert!(entries.verify(&blockhash));
-                blockhash = entries.last().unwrap().hash;
-                for entry in entries {
-                    bank.process_transactions(&entry.transactions)
-                        .iter()
-                        .for_each(|x| assert_eq!(*x, Ok(())));
+                if !entries.is_empty() {
+                    blockhash = entries.last().unwrap().hash;
+                    for entry in entries {
+                        bank.process_transactions(&entry.transactions)
+                            .iter()
+                            .for_each(|x| assert_eq!(*x, Ok(())));
+                    }
                 }
 
                 if bank.get_balance(&to) == 1 {
@@ -1161,13 +1172,11 @@ mod tests {
             assert_eq!(bank.get_balance(&to2), 0);
 
             drop(entry_receiver);
-            banking_stage.join().unwrap();
         }
         Blocktree::destroy(&ledger_path).unwrap();
     }
 
     #[test]
-    #[ignore]
     fn test_banking_stage_entryfication() {
         solana_logger::setup();
         // In this attack we'll demonstrate that a verifier can interpret the ledger
@@ -1220,8 +1229,11 @@ mod tests {
                     Blocktree::open(&ledger_path)
                         .expect("Expected to be able to open database ledger"),
                 );
+                let mut poh_config = PohConfig::default();
+                // limit tick count to avoid clearing working_bank at PohRecord then PohRecorderError(MaxHeightReached) at BankingStage
+                poh_config.target_tick_count = Some(bank.max_tick_height() - 1);
                 let (exit, poh_recorder, poh_service, entry_receiver) =
-                    create_test_recorder(&bank, &blocktree);
+                    create_test_recorder(&bank, &blocktree, Some(poh_config));
                 let cluster_info =
                     ClusterInfo::new_with_invalid_keypair(Node::new_localhost().info);
                 let cluster_info = Arc::new(RwLock::new(cluster_info));

core/src/poh_service.rs

@@ -32,7 +32,15 @@ impl PohService {
             .name("solana-poh-service-tick_producer".to_string())
             .spawn(move || {
                 if poh_config.hashes_per_tick.is_none() {
-                    Self::sleepy_tick_producer(poh_recorder, &poh_config, &poh_exit_);
+                    if poh_config.target_tick_count.is_none() {
+                        Self::sleepy_tick_producer(poh_recorder, &poh_config, &poh_exit_);
+                    } else {
+                        Self::short_lived_sleepy_tick_producer(
+                            poh_recorder,
+                            &poh_config,
+                            &poh_exit_,
+                        );
+                    }
                 } else {
                     // PoH service runs in a tight loop, generating hashes as fast as possible.
                     // Let's dedicate one of the CPU cores to this thread so that it can gain
@@ -60,6 +68,22 @@ impl PohService {
         }
     }
 
+    fn short_lived_sleepy_tick_producer(
+        poh_recorder: Arc<Mutex<PohRecorder>>,
+        poh_config: &PohConfig,
+        poh_exit: &AtomicBool,
+    ) {
+        let mut warned = false;
+        for _ in 0..poh_config.target_tick_count.unwrap() {
+            sleep(poh_config.target_tick_duration);
+            poh_recorder.lock().unwrap().tick();
+            if poh_exit.load(Ordering::Relaxed) && !warned {
+                warned = true;
+                warn!("exit signal is ignored because PohService is scheduled to exit soon");
+            }
+        }
+    }
+
     fn tick_producer(poh_recorder: Arc<Mutex<PohRecorder>>, poh_exit: &AtomicBool) {
         let poh = poh_recorder.lock().unwrap().poh.clone();
         loop {
@@ -108,6 +132,7 @@ mod tests {
             let poh_config = Arc::new(PohConfig {
                 hashes_per_tick: Some(2),
                 target_tick_duration: Duration::from_millis(42),
+                target_tick_count: None,
             });
             let (poh_recorder, entry_receiver) = PohRecorder::new(
                 bank.tick_height(),

core/src/tvu.rs

@@ -256,7 +256,7 @@ pub mod tests {
         let blocktree = Arc::new(blocktree);
         let bank = bank_forks.working_bank();
         let (exit, poh_recorder, poh_service, _entry_receiver) =
-            create_test_recorder(&bank, &blocktree);
+            create_test_recorder(&bank, &blocktree, None);
         let voting_keypair = Keypair::new();
         let storage_keypair = Arc::new(Keypair::new());
         let leader_schedule_cache = Arc::new(LeaderScheduleCache::new_from_bank(&bank));

runtime/src/bank.rs

@@ -1755,6 +1755,7 @@ mod tests {
                         / DEFAULT_TICKS_PER_SLOT,
                 ),
                 hashes_per_tick: None,
+                target_tick_count: None,
             },
 
             ..GenesisBlock::default()

sdk/src/poh_config.rs

@@ -6,6 +6,9 @@ pub struct PohConfig {
     /// The target tick rate of the cluster.
     pub target_tick_duration: Duration,
 
+    /// The target total tick count to be produced; used for testing only
+    pub target_tick_count: Option<u64>,
+
     /// How many hashes to roll before emitting the next tick entry.
     /// None enables "Low power mode", which implies:
     /// * sleep for `target_tick_duration` instead of hashing
@@ -18,6 +21,7 @@ impl PohConfig {
         Self {
             target_tick_duration,
             hashes_per_tick: None,
+            target_tick_count: None,
         }
     }
 }