[dag] dag rebootstrap #9967
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,133 +1,273 @@ | ||
| // Copyright © Aptos Foundation | ||
|
|
||
| use super::{ | ||
| adapter::Notifier, | ||
| anchor_election::RoundRobinAnchorElection, | ||
| dag_driver::DagDriver, | ||
| dag_fetcher::{DagFetcherService, FetchRequestHandler}, | ||
| dag_handler::NetworkHandler, | ||
| dag_network::TDAGNetworkSender, | ||
| dag_state_sync::{DagStateSynchronizer, StateSyncTrigger, DAG_WINDOW}, | ||
| dag_store::Dag, | ||
| order_rule::OrderRule, | ||
| rb_handler::NodeBroadcastHandler, | ||
| storage::DAGStorage, | ||
| types::{CertifiedNodeMessage, DAGMessage}, | ||
| }; | ||
| use crate::{ | ||
| dag::{adapter::NotifierAdapter, dag_fetcher::DagFetcher}, | ||
| experimental::buffer_manager::OrderedBlocks, | ||
| network::IncomingDAGRequest, | ||
| state_replication::{PayloadClient, StateComputer}, | ||
| }; | ||
| use aptos_channels::{ | ||
| aptos_channel::{self, Receiver}, | ||
| message_queues::QueueStyle, | ||
| }; | ||
| use aptos_consensus_types::common::Author; | ||
| use aptos_crypto::HashValue; | ||
| use aptos_infallible::RwLock; | ||
| use aptos_logger::error; | ||
| use aptos_reliable_broadcast::{RBNetworkSender, ReliableBroadcast}; | ||
| use aptos_types::{ | ||
| aggregate_signature::AggregateSignature, | ||
| block_info::BlockInfo, | ||
| epoch_state::EpochState, | ||
| ledger_info::{LedgerInfo, LedgerInfoWithSignatures}, | ||
| validator_signer::ValidatorSigner, | ||
| }; | ||
| use futures_channel::{ | ||
| mpsc::{UnboundedReceiver, UnboundedSender}, | ||
| oneshot, | ||
| }; | ||
| use std::{sync::Arc, time::Duration}; | ||
| use tokio::{select, task::JoinHandle}; | ||
| use tokio_retry::strategy::ExponentialBackoff; | ||
|
|
||
| struct DagBootstrapper { | ||
| self_peer: Author, | ||
| signer: Arc<ValidatorSigner>, | ||
| epoch_state: Arc<EpochState>, | ||
| storage: Arc<dyn DAGStorage>, | ||
| rb_network_sender: Arc<dyn RBNetworkSender<DAGMessage>>, | ||
| dag_network_sender: Arc<dyn TDAGNetworkSender>, | ||
| time_service: aptos_time_service::TimeService, | ||
| payload_client: Arc<dyn PayloadClient>, | ||
| state_computer: Arc<dyn StateComputer>, | ||
| } | ||
|
|
||
| impl DagBootstrapper { | ||
| fn new( | ||
| self_peer: Author, | ||
| signer: Arc<ValidatorSigner>, | ||
| epoch_state: Arc<EpochState>, | ||
| storage: Arc<dyn DAGStorage>, | ||
| rb_network_sender: Arc<dyn RBNetworkSender<DAGMessage>>, | ||
| dag_network_sender: Arc<dyn TDAGNetworkSender>, | ||
| time_service: aptos_time_service::TimeService, | ||
| payload_client: Arc<dyn PayloadClient>, | ||
| state_computer: Arc<dyn StateComputer>, | ||
| ) -> Self { | ||
| Self { | ||
| self_peer, | ||
| signer, | ||
| epoch_state, | ||
| storage, | ||
| rb_network_sender, | ||
| dag_network_sender, | ||
| time_service, | ||
| payload_client, | ||
| state_computer, | ||
| } | ||
| } | ||
|
|
||
| fn bootstrap_dag_store( | ||
| &self, | ||
| latest_ledger_info: LedgerInfo, | ||
| notifier: Arc<dyn Notifier>, | ||
| ) -> (Arc<RwLock<Dag>>, OrderRule) { | ||
| let dag = Arc::new(RwLock::new(Dag::new( | ||
| self.epoch_state.clone(), | ||
| self.storage.clone(), | ||
| latest_ledger_info.round(), | ||
| DAG_WINDOW, | ||
| ))); | ||
|
|
||
| let validators = self.epoch_state.verifier.get_ordered_account_addresses(); | ||
| let anchor_election = Box::new(RoundRobinAnchorElection::new(validators)); | ||
|
|
||
| let order_rule = OrderRule::new( | ||
| self.epoch_state.clone(), | ||
| latest_ledger_info, | ||
| dag.clone(), | ||
| anchor_election, | ||
| notifier, | ||
| self.storage.clone(), | ||
| ); | ||
|
|
||
| (dag, order_rule) | ||
| } | ||
|
|
||
| fn bootstrap_components( | ||
| &self, | ||
| dag: Arc<RwLock<Dag>>, | ||
| order_rule: OrderRule, | ||
| state_sync_trigger: StateSyncTrigger, | ||
| ) -> (NetworkHandler, DagFetcherService) { | ||
| let validators = self.epoch_state.verifier.get_ordered_account_addresses(); | ||
|
|
||
| // A backoff policy that starts at 100ms and doubles each iteration. | ||
| let rb_backoff_policy = ExponentialBackoff::from_millis(2).factor(50); | ||
| let rb = Arc::new(ReliableBroadcast::new( | ||
| validators.clone(), | ||
| self.rb_network_sender.clone(), | ||
| rb_backoff_policy, | ||
| self.time_service.clone(), | ||
| // TODO: add to config | ||
| Duration::from_millis(500), | ||
| )); | ||
|
|
||
| let (dag_fetcher, fetch_requester, node_fetch_waiter, certified_node_fetch_waiter) = | ||
| DagFetcherService::new( | ||
| self.epoch_state.clone(), | ||
| self.dag_network_sender.clone(), | ||
| dag.clone(), | ||
| self.time_service.clone(), | ||
| ); | ||
| let fetch_requester = Arc::new(fetch_requester); | ||
|
|
||
| let dag_driver = DagDriver::new( | ||
| self.self_peer, | ||
| self.epoch_state.clone(), | ||
| dag.clone(), | ||
| self.payload_client.clone(), | ||
| rb, | ||
| self.time_service.clone(), | ||
| self.storage.clone(), | ||
| order_rule, | ||
| fetch_requester.clone(), | ||
| ); | ||
| let rb_handler = NodeBroadcastHandler::new( | ||
| dag.clone(), | ||
| self.signer.clone(), | ||
| self.epoch_state.clone(), | ||
| self.storage.clone(), | ||
| fetch_requester, | ||
| ); | ||
| let fetch_handler = FetchRequestHandler::new(dag, self.epoch_state.clone()); | ||
|
|
||
| let dag_handler = NetworkHandler::new( | ||
| self.epoch_state.clone(), | ||
| rb_handler, | ||
| dag_driver, | ||
| fetch_handler, | ||
| node_fetch_waiter, | ||
| certified_node_fetch_waiter, | ||
| state_sync_trigger, | ||
| ); | ||
|
|
||
| (dag_handler, dag_fetcher) | ||
| } | ||
|
|
||
| async fn bootstrapper( | ||
| self, | ||
| mut dag_rpc_rx: Receiver<Author, IncomingDAGRequest>, | ||
| ordered_nodes_tx: UnboundedSender<OrderedBlocks>, | ||
| mut shutdown_rx: oneshot::Receiver<()>, | ||
| ) { | ||
| let adapter = Arc::new(NotifierAdapter::new(ordered_nodes_tx, self.storage.clone())); | ||
|
|
||
| let sync_manager = DagStateSynchronizer::new( | ||
| self.epoch_state.clone(), | ||
| adapter.clone(), | ||
| self.time_service.clone(), | ||
| self.state_computer.clone(), | ||
| self.storage.clone(), | ||
| ); | ||
|
|
||
| // TODO: fetch the correct block info | ||
| let ledger_info = LedgerInfoWithSignatures::new( | ||
| LedgerInfo::new(BlockInfo::empty(), HashValue::zero()), | ||
| AggregateSignature::empty(), | ||
| ); | ||
|
|
||
| loop { | ||
| let (dag_store, order_rule) = | ||
| self.bootstrap_dag_store(ledger_info.ledger_info().clone(), adapter.clone()); | ||
|
|
||
| let state_sync_trigger = StateSyncTrigger::new(dag_store.clone(), adapter.clone()); | ||
|
|
||
| let (handler, fetch_service) = | ||
| self.bootstrap_components(dag_store.clone(), order_rule, state_sync_trigger); | ||
|
|
||
| let df_handle = tokio::spawn(fetch_service.start()); | ||
|
There was a problem hiding this comment. nit: we can just have a drop guard like this created and avoid the abort/await lines in both branches? |
||
|
|
||
| // poll the network handler while waiting for rebootstrap notification or shutdown notification | ||
| select! { | ||
| biased; | ||
| _ = &mut shutdown_rx => { | ||
| df_handle.abort(); | ||
| let _ = df_handle.await; | ||
| return; | ||
| }, | ||
| certified_node_msg = handler.run(&mut dag_rpc_rx) => { | ||
| df_handle.abort(); | ||
| let _ = df_handle.await; | ||
|
There was a problem hiding this comment. probably should have a guard for these services |
||
|
|
||
| let dag_fetcher = DagFetcher::new(self.epoch_state.clone(), self.dag_network_sender.clone(), self.time_service.clone()); | ||
|
|
||
| if let Err(e) = sync_manager.sync_dag_to(&certified_node_msg, dag_fetcher, dag_store.clone()).await { | ||
|
There was a problem hiding this comment. i thought we're creating a new dag store instead of re-using the current one? There was a problem hiding this comment. yes, we are. I pass the existing dag store to do some assertion checks on whether to actually state sync. There was a problem hiding this comment. hmm, that sounds weird, the check should be done in the check function? There was a problem hiding this comment. yes, i am being paranoid. i check in the check function and i assert in the sync_to function. |
||
| error!(error = ?e, "unable to sync"); | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| pub(super) fn bootstrap_dag_for_test( | ||
| self_peer: Author, | ||
| signer: ValidatorSigner, | ||
| epoch_state: Arc<EpochState>, | ||
| latest_ledger_info: LedgerInfo, | ||
| storage: Arc<dyn DAGStorage>, | ||
| rb_network_sender: Arc<dyn RBNetworkSender<DAGMessage>>, | ||
| dag_network_sender: Arc<dyn TDAGNetworkSender>, | ||
| time_service: aptos_time_service::TimeService, | ||
| payload_client: Arc<dyn PayloadClient>, | ||
| state_computer: Arc<dyn StateComputer>, | ||
| ) -> ( | ||
| JoinHandle<CertifiedNodeMessage>, | ||
| JoinHandle<()>, | ||
| aptos_channel::Sender<Author, IncomingDAGRequest>, | ||
| UnboundedReceiver<OrderedBlocks>, | ||
| ) { | ||
| let bootstraper = DagBootstrapper::new( | ||
| self_peer, | ||
| signer.into(), | ||
| epoch_state, | ||
| storage.clone(), | ||
| rb_network_sender, | ||
| dag_network_sender, | ||
| time_service, | ||
| payload_client, | ||
| state_computer, | ||
| ); | ||
|
|
||
| let (ordered_nodes_tx, ordered_nodes_rx) = futures_channel::mpsc::unbounded(); | ||
| let adapter = Arc::new(NotifierAdapter::new(ordered_nodes_tx, storage.clone())); | ||
| let (dag_rpc_tx, mut dag_rpc_rx) = aptos_channel::new(QueueStyle::FIFO, 64, None); | ||
|
|
||
| let (dag_store, order_rule) = | ||
| bootstraper.bootstrap_dag_store(latest_ledger_info, adapter.clone()); | ||
|
|
||
| let state_sync_trigger = StateSyncTrigger::new(dag_store.clone(), adapter.clone()); | ||
|
|
||
| let (handler, fetch_service) = | ||
| bootstraper.bootstrap_components(dag_store.clone(), order_rule, state_sync_trigger); | ||
|
|
||
| let dh_handle = tokio::spawn(async move { handler.run(&mut dag_rpc_rx).await }); | ||
| let df_handle = tokio::spawn(fetch_service.start()); | ||
|
|
||
| (dh_handle, df_handle, dag_rpc_tx, ordered_nodes_rx) | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
nit: Why cannot you return arc from DagFetcherService::new?