use {
crate::{
commitment::CommitmentAggregationData,
consensus_metrics::{
ConsensusMetrics, ConsensusMetricsEventReceiver, ConsensusMetricsEventSender,
},
consensus_pool_service::{ConsensusPoolContext, ConsensusPoolService},
consensus_rewards::ConsensusRewardsService,
event::{LeaderWindowInfo, RepairEventSender, VotorEventReceiver, VotorEventSender},
event_handler::{EventHandler, EventHandlerContext},
generated_cert_types::GeneratedCertTypes,
root_utils::RootContext,
timer_manager::TimerManager,
vote_history::VoteHistory,
vote_history_storage::VoteHistoryStorage,
voting_service::BLSOp,
voting_utils::VotingContext,
},
agave_votor_messages::{
consensus_message::ConsensusMessage,
reward_certificate::{AddVoteMessage, BuildRewardCertsRequest, BuildRewardCertsResponse},
},
crossbeam_channel::{Receiver, Sender},
parking_lot::RwLock as PlRwLock,
solana_clock::Slot,
solana_gossip::cluster_info::ClusterInfo,
solana_hash::Hash,
solana_keypair::Keypair,
solana_ledger::{blockstore::Blockstore, leader_schedule_cache::LeaderScheduleCache},
solana_pubkey::Pubkey,
solana_rpc::{
optimistically_confirmed_bank_tracker::BankNotificationSenderConfig,
rpc_subscriptions::RpcSubscriptions,
},
solana_runtime::{
bank_forks::BankForks, installed_scheduler_pool::BankWithScheduler,
snapshot_controller::SnapshotController,
validated_block_finalization::ValidatedBlockFinalizationCert,
},
std::{
collections::HashMap,
sync::{Arc, RwLock, atomic::AtomicBool},
thread::{self, JoinHandle},
time::Duration,
},
};
pub struct VotorConfig {
pub exit: Arc<AtomicBool>,
pub vote_account: Pubkey,
pub wait_to_vote_slot: Option<Slot>,
pub vote_history: VoteHistory,
pub vote_history_storage: Arc<dyn VoteHistoryStorage>,
pub generated_cert_types: Arc<GeneratedCertTypes>,
pub authorized_voter_keypairs: Arc<RwLock<Vec<Arc<Keypair>>>>,
pub blockstore: Arc<Blockstore>,
pub bank_forks: Arc<RwLock<BankForks>>,
pub cluster_info: Arc<ClusterInfo>,
pub leader_schedule_cache: Arc<LeaderScheduleCache>,
pub rpc_subscriptions: Option<Arc<RpcSubscriptions>>,
pub consensus_metrics_sender: ConsensusMetricsEventSender,
pub highest_finalized: Arc<RwLock<Option<ValidatedBlockFinalizationCert>>>,
pub snapshot_controller: Option<Arc<SnapshotController>>,
pub bls_sender: Sender<BLSOp>,
pub commitment_sender: Sender<CommitmentAggregationData>,
pub drop_bank_sender: Sender<Vec<BankWithScheduler>>,
pub bank_notification_sender: Option<BankNotificationSenderConfig>,
pub leader_window_info_sender: Sender<LeaderWindowInfo>,
pub highest_parent_ready: Arc<RwLock<(Slot, (Slot, Hash))>>,
pub event_sender: VotorEventSender,
pub own_vote_sender: Sender<Vec<ConsensusMessage>>,
pub reward_certs_sender: Sender<BuildRewardCertsResponse>,
pub repair_event_sender: RepairEventSender,
pub event_receiver: VotorEventReceiver,
pub consensus_message_receiver: Receiver<Vec<ConsensusMessage>>,
pub consensus_metrics_receiver: ConsensusMetricsEventReceiver,
pub reward_votes_receiver: Receiver<AddVoteMessage>,
pub build_reward_certs_receiver: Receiver<BuildRewardCertsRequest>,
}
pub(crate) struct SharedContext {
pub(crate) blockstore: Arc<Blockstore>,
pub(crate) bank_forks: Arc<RwLock<BankForks>>,
pub(crate) cluster_info: Arc<ClusterInfo>,
pub(crate) rpc_subscriptions: Option<Arc<RpcSubscriptions>>,
pub(crate) leader_window_info_sender: Sender<LeaderWindowInfo>,
pub(crate) highest_parent_ready: Arc<RwLock<(Slot, (Slot, Hash))>>,
pub(crate) vote_history_storage: Arc<dyn VoteHistoryStorage>,
pub(crate) repair_event_sender: RepairEventSender,
}
pub struct Votor {
event_handler: EventHandler,
consensus_pool_service: ConsensusPoolService,
timer_manager: Arc<PlRwLock<TimerManager>>,
consensus_rewards_service: ConsensusRewardsService,
metrics: JoinHandle<()>,
}
impl Votor {
pub fn new(config: VotorConfig) -> Self {
let VotorConfig {
exit,
vote_account,
wait_to_vote_slot,
vote_history,
vote_history_storage,
authorized_voter_keypairs,
blockstore,
bank_forks,
cluster_info,
leader_schedule_cache,
rpc_subscriptions,
snapshot_controller,
bls_sender,
commitment_sender,
drop_bank_sender,
bank_notification_sender,
leader_window_info_sender,
highest_parent_ready,
event_sender,
own_vote_sender,
repair_event_sender,
event_receiver,
consensus_message_receiver,
consensus_metrics_sender,
consensus_metrics_receiver,
reward_votes_receiver,
build_reward_certs_receiver,
reward_certs_sender,
generated_cert_types,
highest_finalized,
} = config;
let migration_status = bank_forks.read().unwrap().migration_status();
let identity_keypair = cluster_info.keypair();
let sharable_banks = bank_forks.read().unwrap().sharable_banks();
let shared_context = SharedContext {
blockstore: blockstore.clone(),
bank_forks,
cluster_info: cluster_info.clone(),
rpc_subscriptions,
highest_parent_ready,
leader_window_info_sender,
vote_history_storage,
repair_event_sender,
};
let voting_context = VotingContext {
vote_history,
vote_account_pubkey: vote_account,
identity_keypair,
authorized_voter_keypairs,
derived_bls_keypairs: HashMap::new(),
own_vote_sender,
bls_sender: bls_sender.clone(),
commitment_sender: commitment_sender.clone(),
wait_to_vote_slot,
sharable_banks: sharable_banks.clone(),
consensus_metrics_sender,
};
let root_context = RootContext {
leader_schedule_cache: leader_schedule_cache.clone(),
snapshot_controller,
bank_notification_sender,
drop_bank_sender,
};
let timer_manager = Arc::new(PlRwLock::new(TimerManager::new(
event_sender.clone(),
exit.clone(),
migration_status.clone(),
)));
let event_handler_context = EventHandlerContext {
exit: exit.clone(),
migration_status: migration_status.clone(),
event_receiver,
timer_manager: Arc::clone(&timer_manager),
shared_context,
voting_context,
root_context,
};
let epoch_schedule = sharable_banks.root().epoch_schedule().clone();
let consensus_pool_context = ConsensusPoolContext {
exit: exit.clone(),
migration_status,
generated_cert_types,
cluster_info: cluster_info.clone(),
my_vote_pubkey: vote_account,
blockstore,
sharable_banks: sharable_banks.clone(),
leader_schedule_cache: leader_schedule_cache.clone(),
consensus_message_receiver,
bls_sender,
event_sender,
commitment_sender,
highest_finalized,
};
let metrics = ConsensusMetrics::start_metrics_loop(
epoch_schedule,
consensus_metrics_receiver,
exit.clone(),
);
let event_handler = EventHandler::new(event_handler_context);
let consensus_pool_service = ConsensusPoolService::new(consensus_pool_context);
let consensus_rewards_service = ConsensusRewardsService::new(
cluster_info,
leader_schedule_cache,
sharable_banks,
exit,
reward_votes_receiver,
build_reward_certs_receiver,
reward_certs_sender,
);
Self {
event_handler,
consensus_pool_service,
consensus_rewards_service,
timer_manager,
metrics,
}
}
pub fn join(self) -> thread::Result<()> {
self.consensus_pool_service.join()?;
self.consensus_rewards_service.join()?;
let mut timer_manager = self.timer_manager;
loop {
match Arc::try_unwrap(timer_manager) {
Ok(manager) => {
manager.into_inner().join();
break;
}
Err(m) => {
timer_manager = m;
thread::sleep(Duration::from_millis(1));
}
}
}
self.metrics.join()?;
self.event_handler.join()
}
}