fn replicate_requests(
        bank: &Arc<Bank>,
        cluster_info: &Arc<RwLock<ClusterInfo>>,
        window_receiver: &EntryReceiver,
        keypair: &Arc<Keypair>,
        vote_account_keypair: &Arc<Keypair>,
        vote_blob_sender: Option<&BlobSender>,
        ledger_entry_sender: &EntrySender,
        entry_height: &mut u64,
        last_entry_id: &mut Option<Hash>,
    ) -> Result<()> {
        let timer = Duration::new(1, 0);
        //coalesce all the available entries into a single vote
        let mut entries = window_receiver.recv_timeout(timer)?;
        while let Ok(mut more) = window_receiver.try_recv() {
            entries.append(&mut more);
        }

        metrics::submit(
            influxdb::Point::new("replicate-stage")
                .add_field("count", influxdb::Value::Integer(entries.len() as i64))
                .to_owned(),
        );

        let mut res = Ok(());
        let mut num_entries_to_write = entries.len();
        let (current_leader, _) = bank
            .get_current_leader()
            .expect("Scheduled leader id should never be unknown while processing entries");
        for (i, entry) in entries.iter().enumerate() {
            res = bank.process_entry(&entry);
            let my_id = keypair.pubkey();
            let (scheduled_leader, _) = bank
                .get_current_leader()
                .expect("Scheduled leader id should never be unknown while processing entries");

            // TODO: Remove this soon once we boot the leader from ClusterInfo
            if scheduled_leader != current_leader {
                cluster_info.write().unwrap().set_leader(scheduled_leader);
            }
            if my_id == scheduled_leader {
                num_entries_to_write = i + 1;
                break;
            }

            if res.is_err() {
                // TODO: This will return early from the first entry that has an erroneous
                // transaction, instad of processing the rest of the entries in the vector
                // of received entries. This is in line with previous behavior when
                // bank.process_entries() was used to process the entries, but doesn't solve the
                // issue that the bank state was still changed, leading to inconsistencies with the
                // leader as the leader currently should not be publishing erroneous transactions
                break;
            }
        }

        // If leader rotation happened, only write the entries up to leader rotation.
        entries.truncate(num_entries_to_write);
        *last_entry_id = Some(
            entries
                .last()
                .expect("Entries cannot be empty at this point")
                .id,
        );

        inc_new_counter_info!(
            "replicate-transactions",
            entries.iter().map(|x| x.transactions.len()).sum()
        );

        let entries_len = entries.len() as u64;
        // TODO: move this to another stage?
        // TODO: In line with previous behavior, this will write all the entries even if
        // an error occurred processing one of the entries (causing the rest of the entries to
        // not be processed).
        if entries_len != 0 {
            ledger_entry_sender.send(entries)?;
        }

        *entry_height += entries_len;
        res?;
        if let Some(sender) = vote_blob_sender {
            send_validator_vote(bank, vote_account_keypair, &cluster_info, sender)?;
        }

        Ok(())
    }
/// Process entry blobs, already in order
Process entry blobs, already in order
Process entry blobs already in order
fn replicate_requests(
        bank: &Arc<Bank>,
        cluster_info: &Arc<RwLock<ClusterInfo>>,
        window_receiver: &EntryReceiver,
        keypair: &Arc<Keypair>,
        vote_account_keypair: &Arc<Keypair>,
        vote_blob_sender: Option<&BlobSender>,
        ledger_entry_sender: &EntrySender,
        entry_height: &mut u64,
        last_entry_id: &mut Option<Hash>,
    ) -> Result<()> {
        let timer = Duration::new(1, 0);
        //coalesce all the available entries into a single vote
        let mut entries = window_receiver.recv_timeout(timer)?;
        while let Ok(mut more) = window_receiver.try_recv() {
            entries.append(&mut more);
        }
        metrics::submit(
            influxdb::Point::new("replicate-stage")
                .add_field("count", influxdb::Value::Integer(entries.len() as i64))
                .to_owned(),
        );
        let mut res = Ok(());
        let mut num_entries_to_write = entries.len();
        let (current_leader, _) = bank
            .get_current_leader()
            .expect("Scheduled leader id should never be unknown while processing entries");
        for (i, entry) in entries.iter().enumerate() {
            res = bank.process_entry(&entry);
            let my_id = keypair.pubkey();
            let (scheduled_leader, _) = bank
                .get_current_leader()
                .expect("Scheduled leader id should never be unknown while processing entries");
            // TODO: Remove this soon once we boot the leader from ClusterInfo
            if scheduled_leader != current_leader {
                cluster_info.write().unwrap().set_leader(scheduled_leader);
            }
            if my_id == scheduled_leader {
                num_entries_to_write = i + 1;
                break;
            }
            if res.is_err() {
                // TODO: This will return early from the first entry that has an erroneous
                // transaction, instad of processing the rest of the entries in the vector
                // of received entries. This is in line with previous behavior when
                // bank.process_entries() was used to process the entries, but doesn't solve the
                // issue that the bank state was still changed, leading to inconsistencies with the
                // leader as the leader currently should not be publishing erroneous transactions
                break;
            }
        }
        // If leader rotation happened, only write the entries up to leader rotation.
        entries.truncate(num_entries_to_write);
        *last_entry_id = Some(
            entries
                .last()
                .expect("Entries cannot be empty at this point")
                .id,
        );
        inc_new_counter_info!(
            "replicate-transactions",
            entries.iter().map(|x| x.transactions.len()).sum()
        );
        let entries_len = entries.len() as u64;
        // TODO: move this to another stage?
        // TODO: In line with previous behavior, this will write all the entries even if
        // an error occurred processing one of the entries (causing the rest of the entries to
        // not be processed).
        if entries_len != 0 {
            ledger_entry_sender.send(entries)?;
        }
        *entry_height += entries_len;
        res?;
        if let Some(sender) = vote_blob_sender {
            send_validator_vote(bank, vote_account_keypair, &cluster_info, sender)?;
        }
        Ok(())
    }
fn replicate_requests ( bank : & Arc < Bank > , cluster_info : & Arc < RwLock < ClusterInfo > > , window_receiver : & EntryReceiver , keypair : & Arc < Keypair > , vote_account_keypair : & Arc < Keypair > , vote_blob_sender : Option < & BlobSender > , ledger_entry_sender : & EntrySender , entry_height : & mut u64 , last_entry_id : & mut Option < Hash > , ) -> Result < ( ) > { let timer = Duration :: new ( 1 , 0 ) ; //coalesce all the available entries into a single vote let mut entries = window_receiver . recv_timeout ( timer ) ? ; while let Ok ( mut more ) = window_receiver . try_recv ( ) { entries . append ( & mut more ) ; } metrics :: submit ( influxdb :: Point :: new ( " " ) . add_field ( " " , influxdb :: Value :: Integer ( entries . len ( ) as i64 ) ) . to_owned ( ) , ) ; let mut res = Ok ( ( ) ) ; let mut num_entries_to_write = entries . len ( ) ; let ( current_leader , _ ) = bank . get_current_leader ( ) . expect ( " " ) ; for ( i , entry ) in entries . iter ( ) . enumerate ( ) { res = bank . process_entry ( & entry ) ; let my_id = keypair . pubkey ( ) ; let ( scheduled_leader , _ ) = bank . get_current_leader ( ) . expect ( " " ) ; // TODO: Remove this soon once we boot the leader from ClusterInfo if scheduled_leader != current_leader { cluster_info . write ( ) . unwrap ( ) . set_leader ( scheduled_leader ) ; } if my_id == scheduled_leader { num_entries_to_write = i + 1 ; break ; } if res . is_err ( ) { // TODO: This will return early from the first entry that has an erroneous // transaction, instad of processing the rest of the entries in the vector // of received entries. This is in line with previous behavior when // bank.process_entries() was used to process the entries, but doesn't solve the // issue that the bank state was still changed, leading to inconsistencies with the // leader as the leader currently should not be publishing erroneous transactions break ; } } // If leader rotation happened, only write the entries up to leader rotation. entries . truncate ( num_entries_to_write ) ; * last_entry_id = Some ( entries . last ( ) . expect ( " " ) . id , ) ; inc_new_counter_info ! ( " " entries iter ( ) map ( x x transactions len ( ) ) sum ( ) ) ; let entries_len = entries . len ( ) as u64 ; // TODO: move this to another stage? // TODO: In line with previous behavior, this will write all the entries even if // an error occurred processing one of the entries (causing the rest of the entries to // not be processed). if entries_len != 0 { ledger_entry_sender . send ( entries ) ? ; } * entry_height += entries_len ; res ? ; if let Some ( sender ) = vote_blob_sender { send_validator_vote ( bank , vote_account_keypair , & cluster_info , sender ) ? ; } Ok ( ( ) ) }
Process entry blobs, already in order
Process entry blobs already in order
bank
&Arc<Bank>
cluster_info
&Arc<RwLock<ClusterInfo>>
window_receiver
&EntryReceiver
keypair
&Arc<Keypair>
vote_account_keypair
&Arc<Keypair>
vote_blob_sender
Option<&BlobSender>
ledger_entry_sender
&EntrySender
entry_height
&mut u64
last_entry_id
&mut Option<Hash>
bank
&Arc<Bank>
cluster_info
&Arc<RwLock<ClusterInfo>>
window_receiver
&EntryReceiver
keypair
&Arc<Keypair>
vote_account_keypair
&Arc<Keypair>
vote_blob_sender
Option<&BlobSender>
ledger_entry_sender
&EntrySender
entry_height
&mut u64
last_entry_id
&mut Option<Hash>