pub fn spawn_notifier<T: BeaconChainTypes>(
    executor: task_executor::TaskExecutor,
    beacon_chain: Arc<BeaconChain<T>>,
    network: Arc<NetworkGlobals<T::EthSpec>>,
    seconds_per_slot: u64,
) -> Result<(), String> {
    let slot_duration = Duration::from_secs(seconds_per_slot);
    let duration_to_next_slot = beacon_chain
        .slot_clock
        .duration_to_next_slot()
        .ok_or("slot_notifier unable to determine time to next slot")?;

    // Run this half way through each slot.
    let start_instant = tokio::time::Instant::now() + duration_to_next_slot + (slot_duration / 2);

    // Run this each slot.
    let interval_duration = slot_duration;

    let speedo = Mutex::new(Speedo::default());
    let log = executor.log().clone();
    let mut interval = tokio::time::interval_at(start_instant, interval_duration);

    let interval_future = async move {
        // Perform pre-genesis logging.
        loop {
            match beacon_chain.slot_clock.duration_to_next_slot() {
                // If the duration to the next slot is greater than the slot duration, then we are
                // waiting for genesis.
                Some(next_slot) if next_slot > slot_duration => {
                    info!(
                        log,
                        "Waiting for genesis";
                        "peers" => peer_count_pretty(network.connected_peers()),
                        "wait_time" => estimated_time_pretty(Some(next_slot.as_secs() as f64)),
                    );
                    eth1_logging(&beacon_chain, &log);
                    sleep(slot_duration).await;
                }
                _ => break,
            }
        }

        // Perform post-genesis logging.
        loop {
            interval.tick().await;
            let connected_peer_count = network.connected_peers();
            let sync_state = network.sync_state();

            let head_info = match beacon_chain.head_info() {
                Ok(head_info) => head_info,
                Err(e) => {
                    error!(log, "Failed to get beacon chain head info"; "error" => format!("{:?}", e));
                    break;
                }
            };

            let head_slot = head_info.slot;
            let current_slot = match beacon_chain.slot() {
                Ok(slot) => slot,
                Err(e) => {
                    error!(
                        log,
                        "Unable to read current slot";
                        "error" => format!("{:?}", e)
                    );
                    break;
                }
            };

            let current_epoch = current_slot.epoch(T::EthSpec::slots_per_epoch());
            let finalized_epoch = head_info.finalized_checkpoint.epoch;
            let finalized_root = head_info.finalized_checkpoint.root;
            let head_root = head_info.block_root;

            let mut speedo = speedo.lock();
            speedo.observe(head_slot, Instant::now());

            metrics::set_gauge(
                &metrics::SYNC_SLOTS_PER_SECOND,
                speedo.slots_per_second().unwrap_or(0_f64) as i64,
            );

            // The next two lines take advantage of saturating subtraction on `Slot`.
            let head_distance = current_slot - head_slot;

            if connected_peer_count <= WARN_PEER_COUNT {
                warn!(log, "Low peer count"; "peer_count" => peer_count_pretty(connected_peer_count));
            }

            debug!(
                log,
                "Slot timer";
                "peers" => peer_count_pretty(connected_peer_count),
                "finalized_root" => format!("{}", finalized_root),
                "finalized_epoch" => finalized_epoch,
                "head_block" => format!("{}", head_root),
                "head_slot" => head_slot,
                "current_slot" => current_slot,
                "sync_state" =>format!("{}", sync_state)
            );

            // Log if we are syncing
            if sync_state.is_syncing() {
                let distance = format!(
                    "{} slots ({})",
                    head_distance.as_u64(),
                    slot_distance_pretty(head_distance, slot_duration)
                );

                let speed = speedo.slots_per_second();
                let display_speed = speed.map_or(false, |speed| speed != 0.0);

                if display_speed {
                    info!(
                        log,
                        "Syncing";
                        "peers" => peer_count_pretty(connected_peer_count),
                        "distance" => distance,
                        "speed" => sync_speed_pretty(speed),
                        "est_time" => estimated_time_pretty(speedo.estimated_time_till_slot(current_slot)),
                    );
                } else {
                    info!(
                        log,
                        "Syncing";
                        "peers" => peer_count_pretty(connected_peer_count),
                        "distance" => distance,
                        "est_time" => estimated_time_pretty(speedo.estimated_time_till_slot(current_slot)),
                    );
                }
            } else if sync_state.is_synced() {
                let block_info = if current_slot > head_slot {
                    "   …  empty".to_string()
                } else {
                    head_root.to_string()
                };
                info!(
                    log,
                    "Synced";
                    "peers" => peer_count_pretty(connected_peer_count),
                    "finalized_root" => format!("{}", finalized_root),
                    "finalized_epoch" => finalized_epoch,
                    "epoch" => current_epoch,
                    "block" => block_info,
                    "slot" => current_slot,
                );
            } else {
                info!(
                    log,
                    "Searching for peers";
                    "peers" => peer_count_pretty(connected_peer_count),
                    "finalized_root" => format!("{}", finalized_root),
                    "finalized_epoch" => finalized_epoch,
                    "head_slot" => head_slot,
                    "current_slot" => current_slot,
                );
            }

            eth1_logging(&beacon_chain, &log);
        }
    };

    // run the notifier on the current executor
    executor.spawn(interval_future, "notifier");

    Ok(())
}
/// Spawns a notifier service which periodically logs information about the node.
Spawns a notifier service which periodically logs information about the node.
Spawns a notifier service which periodically logs information about the node .
pub fn spawn_notifier<T: BeaconChainTypes>(
    executor: task_executor::TaskExecutor,
    beacon_chain: Arc<BeaconChain<T>>,
    network: Arc<NetworkGlobals<T::EthSpec>>,
    seconds_per_slot: u64,
) -> Result<(), String> {
    let slot_duration = Duration::from_secs(seconds_per_slot);
    let duration_to_next_slot = beacon_chain
        .slot_clock
        .duration_to_next_slot()
        .ok_or("slot_notifier unable to determine time to next slot")?;
    // Run this half way through each slot.
    let start_instant = tokio::time::Instant::now() + duration_to_next_slot + (slot_duration / 2);
    // Run this each slot.
    let interval_duration = slot_duration;
    let speedo = Mutex::new(Speedo::default());
    let log = executor.log().clone();
    let mut interval = tokio::time::interval_at(start_instant, interval_duration);
    let interval_future = async move {
        // Perform pre-genesis logging.
        loop {
            match beacon_chain.slot_clock.duration_to_next_slot() {
                // If the duration to the next slot is greater than the slot duration, then we are
                // waiting for genesis.
                Some(next_slot) if next_slot > slot_duration => {
                    info!(
                        log,
                        "Waiting for genesis";
                        "peers" => peer_count_pretty(network.connected_peers()),
                        "wait_time" => estimated_time_pretty(Some(next_slot.as_secs() as f64)),
                    );
                    eth1_logging(&beacon_chain, &log);
                    sleep(slot_duration).await;
                }
                _ => break,
            }
        }
        // Perform post-genesis logging.
        loop {
            interval.tick().await;
            let connected_peer_count = network.connected_peers();
            let sync_state = network.sync_state();
            let head_info = match beacon_chain.head_info() {
                Ok(head_info) => head_info,
                Err(e) => {
                    error!(log, "Failed to get beacon chain head info"; "error" => format!("{:?}", e));
                    break;
                }
            };
            let head_slot = head_info.slot;
            let current_slot = match beacon_chain.slot() {
                Ok(slot) => slot,
                Err(e) => {
                    error!(
                        log,
                        "Unable to read current slot";
                        "error" => format!("{:?}", e)
                    );
                    break;
                }
            };
            let current_epoch = current_slot.epoch(T::EthSpec::slots_per_epoch());
            let finalized_epoch = head_info.finalized_checkpoint.epoch;
            let finalized_root = head_info.finalized_checkpoint.root;
            let head_root = head_info.block_root;
            let mut speedo = speedo.lock();
            speedo.observe(head_slot, Instant::now());
            metrics::set_gauge(
                &metrics::SYNC_SLOTS_PER_SECOND,
                speedo.slots_per_second().unwrap_or(0_f64) as i64,
            );
            // The next two lines take advantage of saturating subtraction on `Slot`.
            let head_distance = current_slot - head_slot;
            if connected_peer_count <= WARN_PEER_COUNT {
                warn!(log, "Low peer count"; "peer_count" => peer_count_pretty(connected_peer_count));
            }
            debug!(
                log,
                "Slot timer";
                "peers" => peer_count_pretty(connected_peer_count),
                "finalized_root" => format!("{}", finalized_root),
                "finalized_epoch" => finalized_epoch,
                "head_block" => format!("{}", head_root),
                "head_slot" => head_slot,
                "current_slot" => current_slot,
                "sync_state" =>format!("{}", sync_state)
            );
            // Log if we are syncing
            if sync_state.is_syncing() {
                let distance = format!(
                    "{} slots ({})",
                    head_distance.as_u64(),
                    slot_distance_pretty(head_distance, slot_duration)
                );
                let speed = speedo.slots_per_second();
                let display_speed = speed.map_or(false, |speed| speed != 0.0);
                if display_speed {
                    info!(
                        log,
                        "Syncing";
                        "peers" => peer_count_pretty(connected_peer_count),
                        "distance" => distance,
                        "speed" => sync_speed_pretty(speed),
                        "est_time" => estimated_time_pretty(speedo.estimated_time_till_slot(current_slot)),
                    );
                } else {
                    info!(
                        log,
                        "Syncing";
                        "peers" => peer_count_pretty(connected_peer_count),
                        "distance" => distance,
                        "est_time" => estimated_time_pretty(speedo.estimated_time_till_slot(current_slot)),
                    );
                }
            } else if sync_state.is_synced() {
                let block_info = if current_slot > head_slot {
                    "   …  empty".to_string()
                } else {
                    head_root.to_string()
                };
                info!(
                    log,
                    "Synced";
                    "peers" => peer_count_pretty(connected_peer_count),
                    "finalized_root" => format!("{}", finalized_root),
                    "finalized_epoch" => finalized_epoch,
                    "epoch" => current_epoch,
                    "block" => block_info,
                    "slot" => current_slot,
                );
            } else {
                info!(
                    log,
                    "Searching for peers";
                    "peers" => peer_count_pretty(connected_peer_count),
                    "finalized_root" => format!("{}", finalized_root),
                    "finalized_epoch" => finalized_epoch,
                    "head_slot" => head_slot,
                    "current_slot" => current_slot,
                );
            }
            eth1_logging(&beacon_chain, &log);
        }
    };
    // run the notifier on the current executor
    executor.spawn(interval_future, "notifier");
    Ok(())
}
pub fn spawn_notifier < T : BeaconChainTypes > ( executor : task_executor :: TaskExecutor , beacon_chain : Arc < BeaconChain < T > > , network : Arc < NetworkGlobals < T :: EthSpec > > , seconds_per_slot : u64 , ) -> Result < ( ) , String > { let slot_duration = Duration :: from_secs ( seconds_per_slot ) ; let duration_to_next_slot = beacon_chain . slot_clock . duration_to_next_slot ( ) . ok_or ( " " ) ? ; // Run this half way through each slot. let start_instant = tokio :: time :: Instant :: now ( ) + duration_to_next_slot + ( slot_duration / 2 ) ; // Run this each slot. let interval_duration = slot_duration ; let speedo = Mutex :: new ( Speedo :: default ( ) ) ; let log = executor . log ( ) . clone ( ) ; let mut interval = tokio :: time :: interval_at ( start_instant , interval_duration ) ; let interval_future = async move { // Perform pre-genesis logging. loop { match beacon_chain . slot_clock . duration_to_next_slot ( ) { // If the duration to the next slot is greater than the slot duration, then we are // waiting for genesis. Some ( next_slot ) if next_slot > slot_duration => { info ! ( log " " " " peer_count_pretty ( network connected_peers ( ) ) " " estimated_time_pretty ( Some ( next_slot as_secs ( ) as f64 ) ) ) ; eth1_logging ( & beacon_chain , & log ) ; sleep ( slot_duration ) . await ; } _ => break , } } // Perform post-genesis logging. loop { interval . tick ( ) . await ; let connected_peer_count = network . connected_peers ( ) ; let sync_state = network . sync_state ( ) ; let head_info = match beacon_chain . head_info ( ) { Ok ( head_info ) => head_info , Err ( e ) => { error ! ( log " " " " format ( " " e ) ) ; break ; } } ; let head_slot = head_info . slot ; let current_slot = match beacon_chain . slot ( ) { Ok ( slot ) => slot , Err ( e ) => { error ! ( log " " " " format ( " " e ) ) ; break ; } } ; let current_epoch = current_slot . epoch ( T :: EthSpec :: slots_per_epoch ( ) ) ; let finalized_epoch = head_info . finalized_checkpoint . epoch ; let finalized_root = head_info . finalized_checkpoint . root ; let head_root = head_info . block_root ; let mut speedo = speedo . lock ( ) ; speedo . observe ( head_slot , Instant :: now ( ) ) ; metrics :: set_gauge ( & metrics :: SYNC_SLOTS_PER_SECOND , speedo . slots_per_second ( ) . unwrap_or ( 0_f64 ) as i64 , ) ; // The next two lines take advantage of saturating subtraction on `Slot`. let head_distance = current_slot - head_slot ; if connected_peer_count <= WARN_PEER_COUNT { warn ! ( log " " " " peer_count_pretty ( connected_peer_count ) ) ; } debug ! ( log " " " " peer_count_pretty ( connected_peer_count ) " " format ( " " finalized_root ) " " finalized_epoch " " format ( " " head_root ) " " head_slot " " current_slot " " format ( " " sync_state ) ) ; // Log if we are syncing if sync_state . is_syncing ( ) { let distance = format ! ( " " head_distance as_u64 ( ) slot_distance_pretty ( head_distance slot_duration ) ) ; let speed = speedo . slots_per_second ( ) ; let display_speed = speed . map_or ( false , | speed | speed != 0.0 ) ; if display_speed { info ! ( log " " " " peer_count_pretty ( connected_peer_count ) " " distance " " sync_speed_pretty ( speed ) " " estimated_time_pretty ( speedo estimated_time_till_slot ( current_slot ) ) ) ; } else { info ! ( log " " " " peer_count_pretty ( connected_peer_count ) " " distance " " estimated_time_pretty ( speedo estimated_time_till_slot ( current_slot ) ) ) ; } } else if sync_state . is_synced ( ) { let block_info = if current_slot > head_slot { " t o _string() } else { head_root . to_string ( ) } ; info ! ( log " " " " peer_count_pretty ( connected_peer_count ) " " format ( " " finalized_root ) " " finalized_epoch " " current_epoch " " block_info " " current_slot ) ; } else { info ! ( log " " " " peer_count_pretty ( connected_peer_count ) " " format ( " " finalized_root ) " " finalized_epoch " " head_slot " " current_slot ) ; } eth1_logging ( & beacon_chain , & log ) ; } } ; // run the notifier on the current executor executor . spawn ( interval_future , " " ) ; Ok ( ( ) ) }
Spawns a notifier service which periodically logs information about the node.
Spawns a notifier service which periodically logs information about the node .
executor
task_executor::TaskExecutor
beacon_chain
Arc<BeaconChain<T>>
network
Arc<NetworkGlobals<T::EthSpec>>
seconds_per_slot
u64