use crate::poh_recorder::PohRecorder;
use solana_sdk::clock::DEFAULT_TICKS_PER_SLOT;
use solana_sdk::poh_config::PohConfig;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::thread::{self, sleep, Builder, JoinHandle};
use std::time::Instant;
pub struct PohService {
tick_producer: JoinHandle<()>,
}
pub const NUM_HASHES_PER_BATCH: u64 = 1;
impl PohService {
pub fn new(
poh_recorder: Arc<Mutex<PohRecorder>>,
poh_config: &Arc<PohConfig>,
poh_exit: &Arc<AtomicBool>,
) -> Self {
let poh_exit_ = poh_exit.clone();
let poh_config = poh_config.clone();
let tick_producer = Builder::new()
.name("solana-poh-service-tick_producer".to_string())
.spawn(move || {
solana_sys_tuner::request_realtime_poh();
if poh_config.hashes_per_tick.is_none() {
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 {
if let Some(cores) = core_affinity::get_core_ids() {
core_affinity::set_for_current(cores[0]);
}
Self::tick_producer(poh_recorder, &poh_exit_);
}
poh_exit_.store(true, Ordering::Relaxed);
})
.unwrap();
Self { tick_producer }
}
fn sleepy_tick_producer(
poh_recorder: Arc<Mutex<PohRecorder>>,
poh_config: &PohConfig,
poh_exit: &AtomicBool,
) {
while !poh_exit.load(Ordering::Relaxed) {
sleep(poh_config.target_tick_duration);
poh_recorder.lock().unwrap().tick();
}
}
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();
let mut now = Instant::now();
let mut num_ticks = 0;
let mut num_hashes = 0;
loop {
num_hashes += NUM_HASHES_PER_BATCH;
if poh.lock().unwrap().hash(NUM_HASHES_PER_BATCH) {
poh_recorder.lock().unwrap().tick();
num_ticks += 1;
if num_ticks >= DEFAULT_TICKS_PER_SLOT * 2 {
datapoint_info!(
"poh-service",
("ticks", num_ticks as i64, i64),
("hashes", num_hashes as i64, i64),
("elapsed_ms", now.elapsed().as_millis() as i64, i64),
);
num_ticks = 0;
num_hashes = 0;
now = Instant::now();
}
if poh_exit.load(Ordering::Relaxed) {
break;
}
}
}
}
pub fn join(self) -> thread::Result<()> {
self.tick_producer.join()
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::poh_recorder::WorkingBank;
use solana_ledger::genesis_utils::{create_genesis_config, GenesisConfigInfo};
use solana_ledger::leader_schedule_cache::LeaderScheduleCache;
use solana_ledger::{blockstore::Blockstore, get_tmp_ledger_path};
use solana_perf::test_tx::test_tx;
use solana_runtime::bank::Bank;
use solana_sdk::hash::hash;
use solana_sdk::pubkey::Pubkey;
use std::time::Duration;
#[test]
fn test_poh_service() {
let GenesisConfigInfo { genesis_config, .. } = create_genesis_config(2);
let bank = Arc::new(Bank::new(&genesis_config));
let prev_hash = bank.last_blockhash();
let ledger_path = get_tmp_ledger_path!();
{
let blockstore = Blockstore::open(&ledger_path)
.expect("Expected to be able to open database ledger");
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(),
prev_hash,
bank.slot(),
Some((4, 4)),
bank.ticks_per_slot(),
&Pubkey::default(),
&Arc::new(blockstore),
&Arc::new(LeaderScheduleCache::new_from_bank(&bank)),
&poh_config,
);
let poh_recorder = Arc::new(Mutex::new(poh_recorder));
let exit = Arc::new(AtomicBool::new(false));
let working_bank = WorkingBank {
bank: bank.clone(),
min_tick_height: bank.tick_height(),
max_tick_height: std::u64::MAX,
};
let entry_producer = {
let poh_recorder = poh_recorder.clone();
let exit = exit.clone();
Builder::new()
.name("solana-poh-service-entry_producer".to_string())
.spawn(move || {
loop {
let h1 = hash(b"hello world!");
let tx = test_tx();
let _ = poh_recorder
.lock()
.unwrap()
.record(bank.slot(), h1, vec![tx]);
if exit.load(Ordering::Relaxed) {
break;
}
}
})
.unwrap()
};
let poh_service = PohService::new(poh_recorder.clone(), &poh_config, &exit);
poh_recorder.lock().unwrap().set_working_bank(working_bank);
let mut hashes = 0;
let mut need_tick = true;
let mut need_entry = true;
let mut need_partial = true;
while need_tick || need_entry || need_partial {
let (_bank, (entry, _tick_height)) = entry_receiver.recv().unwrap();
if entry.is_tick() {
assert!(
entry.num_hashes <= poh_config.hashes_per_tick.unwrap(),
format!(
"{} <= {}",
entry.num_hashes,
poh_config.hashes_per_tick.unwrap()
)
);
if entry.num_hashes == poh_config.hashes_per_tick.unwrap() {
need_tick = false;
} else {
need_partial = false;
}
hashes += entry.num_hashes;
assert_eq!(hashes, poh_config.hashes_per_tick.unwrap());
hashes = 0;
} else {
assert!(entry.num_hashes >= 1);
need_entry = false;
hashes += entry.num_hashes;
}
}
exit.store(true, Ordering::Relaxed);
poh_service.join().unwrap();
entry_producer.join().unwrap();
}
Blockstore::destroy(&ledger_path).unwrap();
}
}