fn tick(&mut self) -> bool {
        // Sanity check.
        assert_eq!(self.pending_values.len(), self.pending_values_map.len());

        // Process external requests sent to us through the interface channel.
        if !self.process_external_requests() {
            return false;
        }

        // See if network state thinks we're behind.
        let sync_service_is_behind = self.ledger_sync_service.is_behind(&self.network_state);

        match (self.ledger_sync_state.clone(), sync_service_is_behind) {
            // Fully in sync, nothing to do.
            (LedgerSyncState::InSync, false) => {}

            // Sync service reports we're behind and we are just finding out about it now.
            (LedgerSyncState::InSync, true) => {
                log::info!(self.logger, "sync_service reported we are behind, we're at slot {} and network state is {:?}", self.cur_slot, self.network_state.peer_to_current_slot());
                self.ledger_sync_state = LedgerSyncState::MaybeBehind(Instant::now());
            }

            // Sync wervice reports we're behind and we're maybe behind, see if enough time has
            // passed to move to IsBehind.
            (LedgerSyncState::MaybeBehind(behind_since), true) => {
                let is_behind_duration = Instant::now() - behind_since;
                if is_behind_duration > IS_BEHIND_GRACE_PERIOD {
                    log::warn!(
                        self.logger,
                        "sync_service reports we are behind, and we are past the grace period on slot {}!",
                        self.cur_slot,
                    );
                    counters::CATCHUP_INITIATED.inc();
                    self.is_behind.store(true, Ordering::SeqCst);
                    self.ledger_sync_state = LedgerSyncState::IsBehind {
                        attempt_sync_at: Instant::now(),
                        num_sync_attempts: 0,
                    };

                    // Continue on the next tick.
                    return true;
                }
            }

            // We think we might be behind but sync service reports we're good.
            (LedgerSyncState::MaybeBehind(_), false) => {
                self.ledger_sync_state = LedgerSyncState::InSync;
                log::info!(
                    self.logger,
                    "sync_service reported we're maybe behind but we caught up!"
                );
            }

            // We think we're behind and sync service confirms that, attempt to sync.
            (
                LedgerSyncState::IsBehind {
                    attempt_sync_at,
                    num_sync_attempts,
                },
                true,
            ) => {
                // See if it's time to attempt syncing.
                let now = Instant::now();
                if attempt_sync_at > now {
                    // Not yet. Continue on to the next tick and then try again. We sleep here to
                    // throttle the event loop as it won't be doing anything until we reach the next
                    // attempt_sync_at milestone.
                    log::trace!(
                        self.logger,
                        "sync_service reported we're behind, but deadline {:?} not reached yet (attempt {})!",
                        attempt_sync_at,
                        num_sync_attempts,
                    );

                    thread::sleep(Duration::from_secs(1));
                    return true;
                }

                log::info!(
                    self.logger,
                    "sync_service reported we're behind, attempting catchup (attempt {})!",
                    num_sync_attempts,
                );

                // Attempt incremental catch-up.
                let blocks_per_attempt = 100;
                if let Err(err) = self
                    .ledger_sync_service
                    .attempt_ledger_sync(&self.network_state, blocks_per_attempt)
                {
                    log::error!(self.logger, "Could not sync ledger: {:?}", err);

                    // The next time we attempt to sync is a linear back-off based on how many
                    // attempts we've done so far, capped at 60 seconds.
                    let next_sync_at = now + Duration::from_secs(min(num_sync_attempts + 1, 60));
                    self.ledger_sync_state = LedgerSyncState::IsBehind {
                        attempt_sync_at: next_sync_at,
                        num_sync_attempts: num_sync_attempts + 1,
                    };
                } else {
                    // We successfully synced a chunk of blocks, so reset our attempts to zero for the next chunk.
                    self.ledger_sync_state = LedgerSyncState::IsBehind {
                        attempt_sync_at,
                        num_sync_attempts: 0,
                    };
                }

                // Continue on the next tick.
                return true;
            }

            // We think we're behind but sync service indicates we're back to being in sync.
            (LedgerSyncState::IsBehind { .. }, false) => {
                log::info!(self.logger, "sync_service reports we are no longer behind!");

                // Reset scp state.
                self.scp.clear_pending_slots();

                // Clear any pending values that might no longer be valid.
                let tx_manager = self.tx_manager.clone();
                self.pending_values
                    .retain(|tx_hash| tx_manager.validate_tx_by_hash(tx_hash).is_ok());

                // Re-construct the BTreeMap with the remaining values, using the old timestamps.
                let mut new_pending_values_map = BTreeMap::new();
                for val in self.pending_values.iter() {
                    new_pending_values_map.insert(
                        val.clone(),
                        self.pending_values_map.get(val).unwrap().clone(),
                    );
                }
                self.pending_values_map = new_pending_values_map;

                // Nominate if needed.
                if !self.pending_values.is_empty() {
                    self.need_nominate = true;
                }

                // Update state.
                self.is_behind.store(false, Ordering::SeqCst);
                self.ledger_sync_state = LedgerSyncState::InSync;
                self.cur_slot = self.ledger.num_blocks().unwrap();
                self.prev_block_id = self.ledger.get_block(self.cur_slot - 1).unwrap().id;

                // Clear old entries from pending_consensus_msgs
                let cur_slot = self.cur_slot;
                self.pending_consensus_msgs
                    .retain(|&slot_index, _| slot_index >= cur_slot);
            }
        };

        // Sanity - code here should never run if we're behind.
        assert!(!self.is_behind.load(Ordering::SeqCst));
        if let LedgerSyncState::IsBehind { .. } = &self.ledger_sync_state {
            unreachable!();
        }

        // Nominate values for current slot.
        self.nominate_pending_values();

        // Process any queues consensus messages.
        self.process_consensus_msgs_for_cur_slot();

        // Process SCP timeouts.
        for outgoing_msg in self.scp.process_timeouts().into_iter() {
            (self.send_scp_message)(Some(outgoing_msg));
        }

        // See if we're done with the current slot.
        self.attempt_complete_cur_slot();

        // Update metrics.
        self.update_cur_metrics();

        true
    }
// The place where all the consensus work is actually done.
// Returns true until stop is requested.
The place where all the consensus work is actually done.
Returns true until stop is requested.
The place where all the consensus work is actually done . Returns true until stop is requested .
fn tick(&mut self) -> bool {
        // Sanity check.
        assert_eq!(self.pending_values.len(), self.pending_values_map.len());
        // Process external requests sent to us through the interface channel.
        if !self.process_external_requests() {
            return false;
        }
        // See if network state thinks we're behind.
        let sync_service_is_behind = self.ledger_sync_service.is_behind(&self.network_state);
        match (self.ledger_sync_state.clone(), sync_service_is_behind) {
            // Fully in sync, nothing to do.
            (LedgerSyncState::InSync, false) => {}
            // Sync service reports we're behind and we are just finding out about it now.
            (LedgerSyncState::InSync, true) => {
                log::info!(self.logger, "sync_service reported we are behind, we're at slot {} and network state is {:?}", self.cur_slot, self.network_state.peer_to_current_slot());
                self.ledger_sync_state = LedgerSyncState::MaybeBehind(Instant::now());
            }
            // Sync wervice reports we're behind and we're maybe behind, see if enough time has
            // passed to move to IsBehind.
            (LedgerSyncState::MaybeBehind(behind_since), true) => {
                let is_behind_duration = Instant::now() - behind_since;
                if is_behind_duration > IS_BEHIND_GRACE_PERIOD {
                    log::warn!(
                        self.logger,
                        "sync_service reports we are behind, and we are past the grace period on slot {}!",
                        self.cur_slot,
                    );
                    counters::CATCHUP_INITIATED.inc();
                    self.is_behind.store(true, Ordering::SeqCst);
                    self.ledger_sync_state = LedgerSyncState::IsBehind {
                        attempt_sync_at: Instant::now(),
                        num_sync_attempts: 0,
                    };
                    // Continue on the next tick.
                    return true;
                }
            }
            // We think we might be behind but sync service reports we're good.
            (LedgerSyncState::MaybeBehind(_), false) => {
                self.ledger_sync_state = LedgerSyncState::InSync;
                log::info!(
                    self.logger,
                    "sync_service reported we're maybe behind but we caught up!"
                );
            }
            // We think we're behind and sync service confirms that, attempt to sync.
            (
                LedgerSyncState::IsBehind {
                    attempt_sync_at,
                    num_sync_attempts,
                },
                true,
            ) => {
                // See if it's time to attempt syncing.
                let now = Instant::now();
                if attempt_sync_at > now {
                    // Not yet. Continue on to the next tick and then try again. We sleep here to
                    // throttle the event loop as it won't be doing anything until we reach the next
                    // attempt_sync_at milestone.
                    log::trace!(
                        self.logger,
                        "sync_service reported we're behind, but deadline {:?} not reached yet (attempt {})!",
                        attempt_sync_at,
                        num_sync_attempts,
                    );
                    thread::sleep(Duration::from_secs(1));
                    return true;
                }
                log::info!(
                    self.logger,
                    "sync_service reported we're behind, attempting catchup (attempt {})!",
                    num_sync_attempts,
                );
                // Attempt incremental catch-up.
                let blocks_per_attempt = 100;
                if let Err(err) = self
                    .ledger_sync_service
                    .attempt_ledger_sync(&self.network_state, blocks_per_attempt)
                {
                    log::error!(self.logger, "Could not sync ledger: {:?}", err);
                    // The next time we attempt to sync is a linear back-off based on how many
                    // attempts we've done so far, capped at 60 seconds.
                    let next_sync_at = now + Duration::from_secs(min(num_sync_attempts + 1, 60));
                    self.ledger_sync_state = LedgerSyncState::IsBehind {
                        attempt_sync_at: next_sync_at,
                        num_sync_attempts: num_sync_attempts + 1,
                    };
                } else {
                    // We successfully synced a chunk of blocks, so reset our attempts to zero for the next chunk.
                    self.ledger_sync_state = LedgerSyncState::IsBehind {
                        attempt_sync_at,
                        num_sync_attempts: 0,
                    };
                }
                // Continue on the next tick.
                return true;
            }
            // We think we're behind but sync service indicates we're back to being in sync.
            (LedgerSyncState::IsBehind { .. }, false) => {
                log::info!(self.logger, "sync_service reports we are no longer behind!");
                // Reset scp state.
                self.scp.clear_pending_slots();
                // Clear any pending values that might no longer be valid.
                let tx_manager = self.tx_manager.clone();
                self.pending_values
                    .retain(|tx_hash| tx_manager.validate_tx_by_hash(tx_hash).is_ok());
                // Re-construct the BTreeMap with the remaining values, using the old timestamps.
                let mut new_pending_values_map = BTreeMap::new();
                for val in self.pending_values.iter() {
                    new_pending_values_map.insert(
                        val.clone(),
                        self.pending_values_map.get(val).unwrap().clone(),
                    );
                }
                self.pending_values_map = new_pending_values_map;
                // Nominate if needed.
                if !self.pending_values.is_empty() {
                    self.need_nominate = true;
                }
                // Update state.
                self.is_behind.store(false, Ordering::SeqCst);
                self.ledger_sync_state = LedgerSyncState::InSync;
                self.cur_slot = self.ledger.num_blocks().unwrap();
                self.prev_block_id = self.ledger.get_block(self.cur_slot - 1).unwrap().id;
                // Clear old entries from pending_consensus_msgs
                let cur_slot = self.cur_slot;
                self.pending_consensus_msgs
                    .retain(|&slot_index, _| slot_index >= cur_slot);
            }
        };
        // Sanity - code here should never run if we're behind.
        assert!(!self.is_behind.load(Ordering::SeqCst));
        if let LedgerSyncState::IsBehind { .. } = &self.ledger_sync_state {
            unreachable!();
        }
        // Nominate values for current slot.
        self.nominate_pending_values();
        // Process any queues consensus messages.
        self.process_consensus_msgs_for_cur_slot();
        // Process SCP timeouts.
        for outgoing_msg in self.scp.process_timeouts().into_iter() {
            (self.send_scp_message)(Some(outgoing_msg));
        }
        // See if we're done with the current slot.
        self.attempt_complete_cur_slot();
        // Update metrics.
        self.update_cur_metrics();
        true
    }
fn tick ( & mut self ) -> bool { // Sanity check. assert_eq ! ( self pending_values len ( ) self pending_values_map len ( ) ) ; // Process external requests sent to us through the interface channel. if ! self . process_external_requests ( ) { return false ; } // See if network state thinks we're behind. let sync_service_is_behind = self . ledger_sync_service . is_behind ( & self . network_state ) ; match ( self . ledger_sync_state . clone ( ) , sync_service_is_behind ) { // Fully in sync, nothing to do. ( LedgerSyncState :: InSync , false ) => { } // Sync service reports we're behind and we are just finding out about it now. ( LedgerSyncState :: InSync , true ) => { log :: info ! ( self logger " " self cur_slot self network_state peer_to_current_slot ( ) ) ; self . ledger_sync_state = LedgerSyncState :: MaybeBehind ( Instant :: now ( ) ) ; } // Sync wervice reports we're behind and we're maybe behind, see if enough time has // passed to move to IsBehind. ( LedgerSyncState :: MaybeBehind ( behind_since ) , true ) => { let is_behind_duration = Instant :: now ( ) - behind_since ; if is_behind_duration > IS_BEHIND_GRACE_PERIOD { log :: warn ! ( self logger " " self cur_slot ) ; counters :: CATCHUP_INITIATED . inc ( ) ; self . is_behind . store ( true , Ordering :: SeqCst ) ; self . ledger_sync_state = LedgerSyncState :: IsBehind { attempt_sync_at : Instant :: now ( ) , num_sync_attempts : 0 , } ; // Continue on the next tick. return true ; } } // We think we might be behind but sync service reports we're good. ( LedgerSyncState :: MaybeBehind ( _ ) , false ) => { self . ledger_sync_state = LedgerSyncState :: InSync ; log :: info ! ( self logger " " ) ; } // We think we're behind and sync service confirms that, attempt to sync. ( LedgerSyncState :: IsBehind { attempt_sync_at , num_sync_attempts , } , true , ) => { // See if it's time to attempt syncing. let now = Instant :: now ( ) ; if attempt_sync_at > now { // Not yet. Continue on to the next tick and then try again. We sleep here to // throttle the event loop as it won't be doing anything until we reach the next // attempt_sync_at milestone. log :: trace ! ( self logger " " attempt_sync_at num_sync_attempts ) ; thread :: sleep ( Duration :: from_secs ( 1 ) ) ; return true ; } log :: info ! ( self logger " " num_sync_attempts ) ; // Attempt incremental catch-up. let blocks_per_attempt = 100 ; if let Err ( err ) = self . ledger_sync_service . attempt_ledger_sync ( & self . network_state , blocks_per_attempt ) { log :: error ! ( self logger " " err ) ; // The next time we attempt to sync is a linear back-off based on how many // attempts we've done so far, capped at 60 seconds. let next_sync_at = now + Duration :: from_secs ( min ( num_sync_attempts + 1 , 60 ) ) ; self . ledger_sync_state = LedgerSyncState :: IsBehind { attempt_sync_at : next_sync_at , num_sync_attempts : num_sync_attempts + 1 , } ; } else { // We successfully synced a chunk of blocks, so reset our attempts to zero for the next chunk. self . ledger_sync_state = LedgerSyncState :: IsBehind { attempt_sync_at , num_sync_attempts : 0 , } ; } // Continue on the next tick. return true ; } // We think we're behind but sync service indicates we're back to being in sync. ( LedgerSyncState :: IsBehind { .. } , false ) => { log :: info ! ( self logger " " ) ; // Reset scp state. self . scp . clear_pending_slots ( ) ; // Clear any pending values that might no longer be valid. let tx_manager = self . tx_manager . clone ( ) ; self . pending_values . retain ( | tx_hash | tx_manager . validate_tx_by_hash ( tx_hash ) . is_ok ( ) ) ; // Re-construct the BTreeMap with the remaining values, using the old timestamps. let mut new_pending_values_map = BTreeMap :: new ( ) ; for val in self . pending_values . iter ( ) { new_pending_values_map . insert ( val . clone ( ) , self . pending_values_map . get ( val ) . unwrap ( ) . clone ( ) , ) ; } self . pending_values_map = new_pending_values_map ; // Nominate if needed. if ! self . pending_values . is_empty ( ) { self . need_nominate = true ; } // Update state. self . is_behind . store ( false , Ordering :: SeqCst ) ; self . ledger_sync_state = LedgerSyncState :: InSync ; self . cur_slot = self . ledger . num_blocks ( ) . unwrap ( ) ; self . prev_block_id = self . ledger . get_block ( self . cur_slot - 1 ) . unwrap ( ) . id ; // Clear old entries from pending_consensus_msgs let cur_slot = self . cur_slot ; self . pending_consensus_msgs . retain ( | & slot_index , _ | slot_index >= cur_slot ) ; } } ; // Sanity - code here should never run if we're behind. assert ! ( self is_behind load ( Ordering SeqCst ) ) ; if let LedgerSyncState :: IsBehind { .. } = & self . ledger_sync_state { unreachable ! ( ) ; } // Nominate values for current slot. self . nominate_pending_values ( ) ; // Process any queues consensus messages. self . process_consensus_msgs_for_cur_slot ( ) ; // Process SCP timeouts. for outgoing_msg in self . scp . process_timeouts ( ) . into_iter ( ) { ( self . send_scp_message ) ( Some ( outgoing_msg ) ) ; } // See if we're done with the current slot. self . attempt_complete_cur_slot ( ) ; // Update metrics. self . update_cur_metrics ( ) ; true }
The place where all the consensus work is actually done.
The place where all the consensus work is actually done .
&mut self
&mut self