2021-12-03 09:00:31 -08:00
|
|
|
use {
|
|
|
|
crate::cluster_slots::ClusterSlots,
|
|
|
|
crossbeam_channel::{Receiver, RecvTimeoutError, Sender},
|
|
|
|
solana_gossip::cluster_info::ClusterInfo,
|
|
|
|
solana_ledger::blockstore::Blockstore,
|
|
|
|
solana_measure::measure::Measure,
|
|
|
|
solana_runtime::bank_forks::BankForks,
|
|
|
|
solana_sdk::clock::Slot,
|
|
|
|
std::{
|
|
|
|
sync::{
|
|
|
|
atomic::{AtomicBool, Ordering},
|
|
|
|
Arc, RwLock,
|
|
|
|
},
|
|
|
|
thread::{self, Builder, JoinHandle},
|
|
|
|
time::{Duration, Instant},
|
2020-08-11 12:48:13 -07:00
|
|
|
},
|
|
|
|
};
|
|
|
|
|
2021-06-02 17:20:00 -07:00
|
|
|
pub type ClusterSlotsUpdateReceiver = Receiver<Vec<Slot>>;
|
|
|
|
pub type ClusterSlotsUpdateSender = Sender<Vec<Slot>>;
|
|
|
|
|
2020-08-11 12:48:13 -07:00
|
|
|
#[derive(Default, Debug)]
|
|
|
|
struct ClusterSlotsServiceTiming {
|
|
|
|
pub lowest_slot_elapsed: u64,
|
2021-06-02 17:20:00 -07:00
|
|
|
pub process_cluster_slots_updates_elapsed: u64,
|
2020-08-11 12:48:13 -07:00
|
|
|
}
|
|
|
|
|
|
|
|
impl ClusterSlotsServiceTiming {
|
2021-06-02 17:20:00 -07:00
|
|
|
fn update(&mut self, lowest_slot_elapsed: u64, process_cluster_slots_updates_elapsed: u64) {
|
2020-08-11 12:48:13 -07:00
|
|
|
self.lowest_slot_elapsed += lowest_slot_elapsed;
|
2021-06-02 17:20:00 -07:00
|
|
|
self.process_cluster_slots_updates_elapsed += process_cluster_slots_updates_elapsed;
|
2020-08-11 12:48:13 -07:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
pub struct ClusterSlotsService {
|
|
|
|
t_cluster_slots_service: JoinHandle<()>,
|
|
|
|
}
|
|
|
|
|
|
|
|
impl ClusterSlotsService {
|
|
|
|
pub fn new(
|
|
|
|
blockstore: Arc<Blockstore>,
|
|
|
|
cluster_slots: Arc<ClusterSlots>,
|
|
|
|
bank_forks: Arc<RwLock<BankForks>>,
|
|
|
|
cluster_info: Arc<ClusterInfo>,
|
2021-06-02 17:20:00 -07:00
|
|
|
cluster_slots_update_receiver: ClusterSlotsUpdateReceiver,
|
2020-08-11 12:48:13 -07:00
|
|
|
exit: Arc<AtomicBool>,
|
|
|
|
) -> Self {
|
2021-07-13 15:32:59 -07:00
|
|
|
Self::initialize_lowest_slot(&blockstore, &cluster_info);
|
2021-06-02 17:20:00 -07:00
|
|
|
Self::initialize_epoch_slots(&bank_forks, &cluster_info);
|
2020-08-11 12:48:13 -07:00
|
|
|
let t_cluster_slots_service = Builder::new()
|
|
|
|
.name("solana-cluster-slots-service".to_string())
|
|
|
|
.spawn(move || {
|
|
|
|
Self::run(
|
|
|
|
blockstore,
|
|
|
|
cluster_slots,
|
|
|
|
bank_forks,
|
|
|
|
cluster_info,
|
2021-06-02 17:20:00 -07:00
|
|
|
cluster_slots_update_receiver,
|
2020-08-11 12:48:13 -07:00
|
|
|
exit,
|
|
|
|
)
|
|
|
|
})
|
|
|
|
.unwrap();
|
|
|
|
|
|
|
|
ClusterSlotsService {
|
|
|
|
t_cluster_slots_service,
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
pub fn join(self) -> thread::Result<()> {
|
|
|
|
self.t_cluster_slots_service.join()
|
|
|
|
}
|
|
|
|
|
|
|
|
fn run(
|
|
|
|
blockstore: Arc<Blockstore>,
|
|
|
|
cluster_slots: Arc<ClusterSlots>,
|
|
|
|
bank_forks: Arc<RwLock<BankForks>>,
|
|
|
|
cluster_info: Arc<ClusterInfo>,
|
2021-06-02 17:20:00 -07:00
|
|
|
cluster_slots_update_receiver: ClusterSlotsUpdateReceiver,
|
2020-08-11 12:48:13 -07:00
|
|
|
exit: Arc<AtomicBool>,
|
|
|
|
) {
|
|
|
|
let mut cluster_slots_service_timing = ClusterSlotsServiceTiming::default();
|
|
|
|
let mut last_stats = Instant::now();
|
|
|
|
loop {
|
|
|
|
if exit.load(Ordering::Relaxed) {
|
|
|
|
break;
|
|
|
|
}
|
2021-06-02 17:20:00 -07:00
|
|
|
let slots = match cluster_slots_update_receiver.recv_timeout(Duration::from_millis(200))
|
|
|
|
{
|
2021-03-12 05:44:06 -08:00
|
|
|
Ok(slots) => Some(slots),
|
|
|
|
Err(RecvTimeoutError::Timeout) => None,
|
|
|
|
Err(RecvTimeoutError::Disconnected) => {
|
|
|
|
warn!("Cluster slots service - sender disconnected");
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
};
|
2020-08-11 12:48:13 -07:00
|
|
|
let mut lowest_slot_elapsed = Measure::start("lowest_slot_elapsed");
|
|
|
|
let lowest_slot = blockstore.lowest_slot();
|
2021-07-13 15:32:59 -07:00
|
|
|
Self::update_lowest_slot(lowest_slot, &cluster_info);
|
2020-08-11 12:48:13 -07:00
|
|
|
lowest_slot_elapsed.stop();
|
2021-06-02 17:20:00 -07:00
|
|
|
let mut process_cluster_slots_updates_elapsed =
|
|
|
|
Measure::start("process_cluster_slots_updates_elapsed");
|
2021-03-12 05:44:06 -08:00
|
|
|
if let Some(slots) = slots {
|
2021-06-02 17:20:00 -07:00
|
|
|
Self::process_cluster_slots_updates(
|
|
|
|
slots,
|
|
|
|
&cluster_slots_update_receiver,
|
|
|
|
&cluster_info,
|
|
|
|
);
|
2021-03-12 05:44:06 -08:00
|
|
|
}
|
2021-08-04 08:11:33 -07:00
|
|
|
let root_bank = bank_forks.read().unwrap().root_bank();
|
|
|
|
cluster_slots.update(&root_bank, &cluster_info);
|
2021-06-02 17:20:00 -07:00
|
|
|
process_cluster_slots_updates_elapsed.stop();
|
2020-08-11 12:48:13 -07:00
|
|
|
|
|
|
|
cluster_slots_service_timing.update(
|
|
|
|
lowest_slot_elapsed.as_us(),
|
2021-06-02 17:20:00 -07:00
|
|
|
process_cluster_slots_updates_elapsed.as_us(),
|
2020-08-11 12:48:13 -07:00
|
|
|
);
|
|
|
|
|
|
|
|
if last_stats.elapsed().as_secs() > 2 {
|
|
|
|
datapoint_info!(
|
|
|
|
"cluster_slots_service-timing",
|
|
|
|
(
|
|
|
|
"lowest_slot_elapsed",
|
|
|
|
cluster_slots_service_timing.lowest_slot_elapsed,
|
|
|
|
i64
|
|
|
|
),
|
|
|
|
(
|
2021-06-02 17:20:00 -07:00
|
|
|
"process_cluster_slots_updates_elapsed",
|
|
|
|
cluster_slots_service_timing.process_cluster_slots_updates_elapsed,
|
2020-08-11 12:48:13 -07:00
|
|
|
i64
|
|
|
|
),
|
|
|
|
);
|
|
|
|
cluster_slots_service_timing = ClusterSlotsServiceTiming::default();
|
|
|
|
last_stats = Instant::now();
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-06-02 17:20:00 -07:00
|
|
|
fn process_cluster_slots_updates(
|
2021-03-12 05:44:06 -08:00
|
|
|
mut slots: Vec<Slot>,
|
2021-06-02 17:20:00 -07:00
|
|
|
cluster_slots_update_receiver: &ClusterSlotsUpdateReceiver,
|
2020-08-11 12:48:13 -07:00
|
|
|
cluster_info: &ClusterInfo,
|
|
|
|
) {
|
2021-06-02 17:20:00 -07:00
|
|
|
while let Ok(mut more) = cluster_slots_update_receiver.try_recv() {
|
2020-08-11 12:48:13 -07:00
|
|
|
slots.append(&mut more);
|
|
|
|
}
|
2020-12-13 17:26:34 -08:00
|
|
|
#[allow(clippy::stable_sort_primitive)]
|
2020-08-11 12:48:13 -07:00
|
|
|
slots.sort();
|
2021-03-12 05:44:06 -08:00
|
|
|
|
2020-08-11 12:48:13 -07:00
|
|
|
if !slots.is_empty() {
|
|
|
|
cluster_info.push_epoch_slots(&slots);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-07-13 15:32:59 -07:00
|
|
|
fn initialize_lowest_slot(blockstore: &Blockstore, cluster_info: &ClusterInfo) {
|
2020-08-11 12:48:13 -07:00
|
|
|
// Safe to set into gossip because by this time, the leader schedule cache should
|
|
|
|
// also be updated with the latest root (done in blockstore_processor) and thus
|
|
|
|
// will provide a schedule to window_service for any incoming shreds up to the
|
|
|
|
// last_confirmed_epoch.
|
2021-07-13 15:32:59 -07:00
|
|
|
cluster_info.push_lowest_slot(blockstore.lowest_slot());
|
2020-08-11 12:48:13 -07:00
|
|
|
}
|
|
|
|
|
2021-07-13 15:32:59 -07:00
|
|
|
fn update_lowest_slot(lowest_slot: Slot, cluster_info: &ClusterInfo) {
|
|
|
|
cluster_info.push_lowest_slot(lowest_slot);
|
2020-08-11 12:48:13 -07:00
|
|
|
}
|
|
|
|
|
2021-06-02 17:20:00 -07:00
|
|
|
fn initialize_epoch_slots(bank_forks: &RwLock<BankForks>, cluster_info: &ClusterInfo) {
|
|
|
|
// TODO: Should probably incorporate slots that were replayed on startup,
|
|
|
|
// and maybe some that were frozen < snapshot root in case validators restart
|
|
|
|
// from newer snapshots and lose history.
|
|
|
|
let frozen_banks = bank_forks.read().unwrap().frozen_banks();
|
|
|
|
let mut frozen_bank_slots: Vec<Slot> = frozen_banks.keys().cloned().collect();
|
|
|
|
frozen_bank_slots.sort_unstable();
|
2020-08-11 12:48:13 -07:00
|
|
|
|
2021-06-02 17:20:00 -07:00
|
|
|
if !frozen_bank_slots.is_empty() {
|
|
|
|
cluster_info.push_epoch_slots(&frozen_bank_slots);
|
2020-08-11 12:48:13 -07:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
#[cfg(test)]
|
|
|
|
mod test {
|
2021-07-13 15:32:59 -07:00
|
|
|
use {
|
|
|
|
super::*,
|
2021-07-21 05:16:26 -07:00
|
|
|
solana_gossip::{cluster_info::Node, crds_value::LowestSlot},
|
2021-07-23 08:25:03 -07:00
|
|
|
solana_sdk::{pubkey::Pubkey, signature::Keypair},
|
|
|
|
solana_streamer::socket::SocketAddrSpace,
|
2021-07-13 15:32:59 -07:00
|
|
|
};
|
2020-08-11 12:48:13 -07:00
|
|
|
|
|
|
|
#[test]
|
|
|
|
pub fn test_update_lowest_slot() {
|
2021-05-06 07:04:17 -07:00
|
|
|
let pubkey = Pubkey::new_unique();
|
|
|
|
let node_info = Node::new_localhost_with_pubkey(&pubkey);
|
2021-07-23 08:25:03 -07:00
|
|
|
let cluster_info = ClusterInfo::new(
|
|
|
|
node_info.info,
|
|
|
|
Arc::new(Keypair::new()),
|
|
|
|
SocketAddrSpace::Unspecified,
|
|
|
|
);
|
2021-07-13 15:32:59 -07:00
|
|
|
ClusterSlotsService::update_lowest_slot(5, &cluster_info);
|
2020-10-13 18:10:25 -07:00
|
|
|
cluster_info.flush_push_queue();
|
2021-05-06 07:04:17 -07:00
|
|
|
let lowest = {
|
2021-07-14 15:27:17 -07:00
|
|
|
let gossip_crds = cluster_info.gossip.crds.read().unwrap();
|
2021-07-21 05:16:26 -07:00
|
|
|
gossip_crds.get::<&LowestSlot>(pubkey).unwrap().clone()
|
2021-05-06 07:04:17 -07:00
|
|
|
};
|
2020-08-11 12:48:13 -07:00
|
|
|
assert_eq!(lowest.lowest, 5);
|
|
|
|
}
|
|
|
|
}
|