Split out voting and banking threads in banking stage #27931

AshwinSekar · 2022-09-20T00:34:51Z

Additionally this allows us to aggressively prune the buffer for voting threads as with the new vote state only the latest vote from each validator is necessary.

Problem

Summary of Changes

Fixes #

nikhayes · 2022-09-20T22:59:26Z

They should maybe have separate block compute limits too? Right now it seems that there is a potential for non-vote transaction compute to starve out votes due to non votes filling up block compute (see recent discussion in mb planning channel)

AshwinSekar · 2022-09-23T21:03:56Z

They should maybe have separate block compute limits too? Right now it seems that there is a potential for non-vote transaction compute to starve out votes due to non votes filling up block compute (see recent discussion in mb planning channel)

The priority details are being ripped out of vote txs. With this change vote threads and bank threads would compete to add to block however having some sort of separation might be interesting cc @taozhu-chicago .

carllin · 2022-09-29T10:34:04Z

core/src/banking_stage.rs

@@ -648,12 +685,111 @@ impl BankingStage {
        (Ok(()), packet_vec_len, Some(leader_pubkey))
    }

+    #[allow(clippy::too_many_arguments)]
+    fn processing_function(


nit: processing_function -> do_process_packets

carllin · 2022-09-29T10:42:47Z

core/src/unprocessed_transaction_storage.rs

+            // Mark so we don't forward again
+            deserialized_packet.forwarded = true;


this seems like it should be handled inside filter_valid_packets_for_forwarding, was this a bug with the existing code?

If so, we should check this fix in separately so we don't block the fix behind conflicts with Tao's changes here: #26798

apfitzge · 2022-09-29T20:17:28Z

core/src/banking_stage.rs

@@ -2054,16 +2058,16 @@ impl BankingStage {
            .fetch_add(newly_buffered_packets_count, Ordering::Relaxed);
        banking_stage_stats
            .current_buffered_packet_batches_count
-            .swap(buffered_packet_batches.len(), Ordering::Relaxed);
+            .swap(unprocessed_transaction_storage.len(), Ordering::Relaxed);


what's going on here? this doesn't seem right.

It's not new in this change, but why not just remove the duplicate stat if we can't get an actual number of "packet_batches"

I agree, probably would want to divide by batch_size here, not sure what the original intent of this metric was.

seems at some point we had

banking_stage_stats.current_buffered_packets_count.swap( buffered_packets.iter().map(|packets| packets.1.len()).sum(), Ordering::Relaxed, );

because back then we stored the packets as batches, however now we use individual packets in the minmaxheap.

I think that means we can deprecate one of these metric points.

agree too, time to deprecate some metrics, can only back-fitting them for so long :)

yeah agreed, this can be deleted, as well as the corresponding field in the datapoint. The current_buffered_packets_count below this covers the number of packets in the buffer already

apfitzge · 2022-09-29T20:23:45Z

core/src/banking_stage.rs

@@ -1117,8 +1140,7 @@ impl BankingStage {
                        &socket,
                        poh_recorder,
                        cluster_info,
-                        &mut buffered_packet_batches,
-                        &forward_option,
+                        &mut unprocessed_transaction_storage,
                        transaction_status_sender.clone(),


can we use a reference here too?

what are you referring to?

ha sorry, that comment wasn't clear.
You changed a bunch of the transaction_status_sender args into references. But here we are cloning, so just wondering if you missed this one or if there's a reason we need to clone it here.

apfitzge · 2022-09-29T20:54:52Z

core/src/unprocessed_transaction_storage.rs

        )
    }

    fn filter_forwardable_packets_and_add_batches(
        &mut self,
        forward_packet_batches_by_accounts: &mut ForwardPacketBatchesByAccounts,
    ) -> FilterForwardingResults {
-        BankingStage::filter_valid_packets_for_forwarding(
+        let results = BankingStage::filter_valid_packets_for_forwarding(


nit: we've got a circular dependency with BankingStage and UnprocessedTransactionStorage. Might be better to move the filter_valid_packets_for_forwarding here. Since this is the only place it's used.

apfitzge · 2022-09-29T21:02:23Z

core/src/unprocessed_transaction_storage.rs

@@ -25,6 +23,9 @@ pub enum UnprocessedTransactionStorage {
    LocalTransactionStorage(ThreadLocalUnprocessedPackets),
 }

+unsafe impl Send for UnprocessedTransactionStorage {}


I'm not convinced we can actually do this w/o introducing race conditions on the reference counts for votes.

Even though we wrap in an RwLock in the VoteStorage structure, I think it's possible another thread could drop or clone the Rc from another reference to it. Then because it is non-atomically updated, we end up with a miscounted Rc.

Additionally, in my local changes for the tx-scheduler, I am actually sharing these references between threads, and converted the Rc to Arc.

It seems it'd be useful if I split that change out so we could use it here. I'll do that shortly.

^ I split out #28145 if we go that route here

I agree, this only works in this specific situation where there is only 1 thread dropping the Rc - even then it's pretty sketch. I would also prefer it to be an Arc. I can hold off and wait for #28145 .

@AshwinSekar, #28145 is merged

apfitzge · 2022-09-29T21:05:55Z

core/src/unprocessed_transaction_storage.rs

        &mut self,
-        packet_batch: &PacketBatch,
-        packet_indexes: &[usize],
+        deserialized_packets: Vec<ImmutableDeserializedPacket>,
    ) -> InsertPacketBatchesSummary {


naming-nit: we insert a "batch" but get "batches" summary back.

apfitzge · 2022-09-29T21:06:49Z

core/src/unprocessed_transaction_storage.rs

    ) -> InsertPacketBatchesSummary {
        match self {
            Self::VoteStorage(vote_storage) => {
                let VoteBatchInsertionMetrics {
                    num_dropped_gossip,
                    num_dropped_tpu,
-                } = vote_storage.deserialize_and_insert_batch(packet_batch, packet_indexes);
+                } = vote_storage.insert_batch(deserialized_packets);
                InsertPacketBatchesSummary {


why not have the respective insert_batch functions just return the summary struct directly?

core/src/leader_slot_banking_stage_metrics.rs

apfitzge · 2022-09-29T21:32:13Z

Was bench-tps run to compare before/after this change?

AshwinSekar · 2022-09-30T00:32:19Z

core/src/banking_stage.rs

-        }
-    }
-
-    /// Filter out packets that fail to sanitize, or are no longer valid (could be


All this code is moved (untouched) to unprocessed_transaction_storage for better encapsulation

This could be a separate PR as well I think

Caused some issues because we didn't have an unprocessed_transaction_storage to invoke on, we're still using packet batches on master.

AshwinSekar · 2022-09-30T00:32:54Z

core/src/banking_stage.rs

@@ -2057,35 +1804,6 @@ impl BankingStage {
            .collect_vec()
    }

-    /// Checks a batch of sanitized transactions again bank for age and status
-    fn bank_check_transactions(


This code was moved to the bank itself.

AshwinSekar · 2022-09-30T00:33:20Z

core/src/banking_stage.rs


        Self::filter_valid_transaction_indexes(&results, transaction_to_packet_indexes)
    }

-    fn filter_processed_packets<'a, F>(


Moved to unprocessed_transaction_storage

AshwinSekar · 2022-09-30T00:33:51Z

core/src/banking_stage.rs

@@ -3466,144 +3170,6 @@ mod tests {
        Blockstore::destroy(ledger_path.path()).unwrap();
    }

-    #[test]


test also moved to unprocessed_transaction_storage

AshwinSekar · 2022-09-30T00:35:08Z

core/src/unprocessed_packet_batches.rs

-// messages, and verifies secp256k1 instructions. A list of sanitized transactions are returned
-// with their packet indexes.
-#[allow(clippy::needless_collect)]
-pub fn transaction_from_deserialized_packet(


Moved this to ImmutableDeserializedPacket

wish this is in a separate PR that can be easily merged, so I can use it in my current PR now 😃 But all good, can be merge later

AshwinSekar · 2022-09-30T00:35:48Z

local-cluster/tests/local_cluster_slow_1.rs

                            .collect();
                            vote_slots.reverse();
-                            let vote_tx = vote_transaction::new_vote_transaction(


Updated to new vote instruction

AshwinSekar · 2022-09-30T01:31:34Z

Was bench-tps run to compare before/after this change?

Before

test bench_banking_stage_multi_accounts             ... bench:  11,967,033 ns/iter (+/- 4,120,951)
test bench_banking_stage_multi_accounts_with_voting ... bench:  15,190,062 ns/iter (+/- 1,918,300)
test bench_banking_stage_multi_programs             ... bench:  15,294,097 ns/iter (+/- 3,577,250)
test bench_banking_stage_multi_programs_with_voting ... bench:  22,088,729 ns/iter (+/- 2,597,948)

After

test bench_banking_stage_multi_accounts             ... bench:  11,092,034 ns/iter (+/- 2,780,082)
test bench_banking_stage_multi_accounts_with_voting ... bench:  15,460,632 ns/iter (+/- 1,716,954)
test bench_banking_stage_multi_programs             ... bench:  15,422,798 ns/iter (+/- 2,654,561)
test bench_banking_stage_multi_programs_with_voting ... bench:  21,099,046 ns/iter (+/- 1,589,215)

Additionally this allows us to aggressively prune the buffer for voting threads as with the new vote state only the latest vote from each validator is necessary.

apfitzge

Looks good. Only minor comments. Still wait for feedback from Carl or Tao - bigger change should have more than just my eyes on it

apfitzge · 2022-10-15T00:57:50Z

core/src/banking_stage.rs

+            .map(|bank_forks| {
+                let bank = bank_forks.root_bank();
+                bank.feature_set
+                    .is_active(&allow_votes_to_directly_update_vote_state::id())


is there any concern to use a feature check only at start up time to choose how we initialize? The feature gets activated, but then that changes how we restart.

yeah originally we were considering a hot swap approach where we check root bank every loop to decide however this seemed cumbersome. Not to mention that this hot swap would then require another feature in order to activate and with the rate that feature activations seem to be going it would be quicker to just wait for a restart.

apfitzge · 2022-10-15T01:11:03Z

core/src/immutable_deserialized_packet.rs

+
+    // This function deserializes packets into transactions, computes the blake3 hash of transaction
+    // messages, and verifies secp256k1 instructions.
+    pub fn compute_sanitized_transaction(


nit: compute seems like an odd name. build?

apfitzge · 2022-10-15T01:12:47Z

core/src/latest_unprocessed_votes.rs

@@ -240,12 +230,12 @@ impl LatestUnprocessedVotes {
    /// Votes from validators with 0 stakes are ignored
    pub fn get_and_insert_forwardable_packets(
        &self,
-        bank: Arc<Bank>,
+        bank: &Arc<Bank>,


nit: &Bank. don't need the Arc

tao-stones

looks good to me, wish some of refactoring were in their own PRs, but OK with this. There are other comments/nits to address, otherwise good to ge

tao-stones · 2022-10-16T18:57:36Z

core/src/unprocessed_packet_batches.rs

-// messages, and verifies secp256k1 instructions. A list of sanitized transactions are returned
-// with their packet indexes.
-#[allow(clippy::needless_collect)]
-pub fn transaction_from_deserialized_packet(


wish this is in a separate PR that can be easily merged, so I can use it in my current PR now 😃 But all good, can be merge later

tao-stones · 2022-10-16T19:07:12Z

core/benches/unprocessed_packet_batches.rs

@@ -104,7 +107,7 @@ fn insert_packet_batches(
 #[allow(clippy::unit_arg)]
 fn bench_packet_clone(bencher: &mut Bencher) {
    let batch_count = 1000;
-    let packet_per_batch_count = 128;
+    let packet_per_batch_count = UNPROCESSED_BUFFER_STEP_SIZE;


+1 for using proper const here

carllin · 2022-10-16T20:01:56Z

core/src/latest_unprocessed_votes.rs

+            | Ok(VoteInstruction::UpdateVoteStateSwitch(vote_state_update, _))
+            | Ok(VoteInstruction::CompactUpdateVoteState(vote_state_update))
+            | Ok(VoteInstruction::CompactUpdateVoteStateSwitch(vote_state_update, _)) => {


Can we introduce a is_single_vote_update() on VoteInstruction that returns true/false for each of these enum variations? That way we don't miss these cases in the future.

Can be a separate PR as well, since this seems like a bug in the existing code

will address in subsequent pr

carllin · 2022-10-16T20:03:00Z

core/src/latest_unprocessed_votes.rs

+    fn deserialize_packets<'a>(
+        packet_batch: &'a PacketBatch,
+        packet_indexes: &'a [usize],
+        vote_source: VoteSource,
+    ) -> impl Iterator<Item = LatestValidatorVotePacket> + 'a {
+        packet_indexes.iter().filter_map(move |packet_index| {
+            LatestValidatorVotePacket::new(packet_batch[*packet_index].clone(), vote_source).ok()
+        })
+    }


This function seems unchanged, can we move this back where it was before to make the diff smaller

core/src/leader_slot_banking_stage_metrics.rs

carllin · 2022-10-16T20:16:41Z

core/src/banking_stage.rs

-        }
-    }
-
-    /// Filter out packets that fail to sanitize, or are no longer valid (could be


This could be a separate PR as well I think

carllin · 2022-10-16T20:18:30Z

core/src/banking_stage.rs

@@ -2054,16 +2058,16 @@ impl BankingStage {
            .fetch_add(newly_buffered_packets_count, Ordering::Relaxed);
        banking_stage_stats
            .current_buffered_packet_batches_count
-            .swap(buffered_packet_batches.len(), Ordering::Relaxed);
+            .swap(unprocessed_transaction_storage.len(), Ordering::Relaxed);


yeah agreed, this can be deleted, as well as the corresponding field in the datapoint. The current_buffered_packets_count below this covers the number of packets in the buffer already

carllin · 2022-10-16T20:23:38Z

core/src/banking_stage.rs

+    fn test_unprocessed_transaction_storage_insert() -> Result<(), Box<dyn Error>> {
+        let keypair = Keypair::new();
+        let vote_keypair = Keypair::new();
+        let pubkey = solana_sdk::pubkey::new_rand();

-        let retryable_indexes = [0, 1, 2, 3, 5];
-        let mut non_retryable_indexes = vec![];
-        let f = |start, end| {
-            non_retryable_indexes.push((start, end));
-        };
-        BankingStage::filter_processed_packets(retryable_indexes.iter(), f);
-        assert_eq!(non_retryable_indexes, vec![(4, 5)]);
+        let small_transfer = Packet::from_data(
+            None,
+            system_transaction::transfer(&keypair, &pubkey, 1, Hash::new_unique()),
+        )?;
+        let mut vote = Packet::from_data(
+            None,
+            new_vote_state_update_transaction(
+                VoteStateUpdate::default(),
+                Hash::new_unique(),
+                &keypair,
+                &vote_keypair,
+                &vote_keypair,
+                None,
+            ),
+        )?;
+        vote.meta.flags.set(PacketFlags::SIMPLE_VOTE_TX, true);
+        let big_transfer = Packet::from_data(
+            None,
+            system_transaction::transfer(&keypair, &pubkey, 1000000, Hash::new_unique()),
+        )?;

-        let retryable_indexes = [1, 2, 3];
-        let mut non_retryable_indexes = vec![];
-        let f = |start, end| {
-            non_retryable_indexes.push((start, end));
-        };
-        BankingStage::filter_processed_packets(retryable_indexes.iter(), f);
-        assert_eq!(non_retryable_indexes, vec![(0, 1)]);
+        for thread_type in [
+            ThreadType::Transactions,
+            ThreadType::Voting(VoteSource::Gossip),
+            ThreadType::Voting(VoteSource::Tpu),
+        ] {
+            let mut transaction_storage = UnprocessedTransactionStorage::new_transaction_storage(
+                UnprocessedPacketBatches::with_capacity(100),
+                thread_type,
+            );
+            transaction_storage.insert_batch(vec![
+                ImmutableDeserializedPacket::new(small_transfer.clone(), None)?,
+                ImmutableDeserializedPacket::new(vote.clone(), None)?,
+                ImmutableDeserializedPacket::new(big_transfer.clone(), None)?,
+            ]);
+            let deserialized_packets = transaction_storage
+                .iter()
+                .map(|packet| packet.immutable_section().original_packet().clone())
+                .collect_vec();
+            assert_eq!(3, deserialized_packets.len());
+            assert!(deserialized_packets.contains(&small_transfer));
+            assert!(deserialized_packets.contains(&vote));
+            assert!(deserialized_packets.contains(&big_transfer));
+        }

-        let retryable_indexes = [1, 2, 3, 5];
-        let mut non_retryable_indexes = vec![];
-        let f = |start, end| {
-            non_retryable_indexes.push((start, end));
-        };
-        BankingStage::filter_processed_packets(retryable_indexes.iter(), f);
-        assert_eq!(non_retryable_indexes, vec![(0, 1), (4, 5)]);
+        for vote_source in [VoteSource::Gossip, VoteSource::Tpu] {
+            let mut transaction_storage = UnprocessedTransactionStorage::new_vote_storage(
+                Arc::new(LatestUnprocessedVotes::new()),
+                vote_source,
+            );
+            transaction_storage.insert_batch(vec![
+                ImmutableDeserializedPacket::new(small_transfer.clone(), None)?,
+                ImmutableDeserializedPacket::new(vote.clone(), None)?,
+                ImmutableDeserializedPacket::new(big_transfer.clone(), None)?,
+            ]);
+            assert_eq!(1, transaction_storage.len());
+        }
+        Ok(())
+    }


Seems like this test belongs in unprocessed_transaction_storage.rs since it's testing those internal methods

carllin · 2022-10-16T20:26:58Z

core/src/unprocessed_transaction_storage.rs

+        InsertPacketBatchSummary {
+            num_dropped_packets: num_dropped_gossip + num_dropped_tpu,
+            num_dropped_gossip_vote_packets: num_dropped_gossip,
+            num_dropped_tpu_vote_packets: num_dropped_tpu,
+            ..InsertPacketBatchSummary::default()
+        }


+1 for these structured return values

carllin · 2022-10-16T20:35:15Z

core/src/unprocessed_transaction_storage.rs

    pub(crate) num_dropped_gossip_vote_packets: usize,
    pub(crate) num_dropped_tpu_vote_packets: usize,


might be cleaner to just replace these two fields with a VoteBatchInsertionMetrics since that struct has the same fields.

Alternatively we can have InsertPacketBatchSummary be an enum of

enum InsertPacketBatchSummary { VoteBatchInsertionMetrics(metrics), PacketBatchMetrics(metrics), }

and just expose a fn total_dropped_packets() and fn total_dropped_tracer_packets() method on the enum

good idea also allows us to remove the (usize, usize) returns in unprocessed_packet_batches.

Pull request has been modified.

) * Split out voting and banking threads in banking stage Additionally this allows us to aggressively prune the buffer for voting threads as with the new vote state only the latest vote from each validator is necessary. * Update local cluster test to use new Vote ix * Encapsulate transaction storage filtering better * Address pr comments * Commit cargo lock change * clippy * Remove unsafe impls * pr comments * compute_sanitized_transaction -> build_sanitized_transaction * &Arc -> Arc * Move test * Refactor metrics enums * clippy

AshwinSekar force-pushed the take-last-vote-banking branch 4 times, most recently from c400a2a to 79a4b3a Compare September 20, 2022 19:39

AshwinSekar force-pushed the take-last-vote-banking branch from 79a4b3a to dd742cf Compare September 23, 2022 18:53

AshwinSekar requested review from carllin, tao-stones and apfitzge September 23, 2022 20:39

AshwinSekar marked this pull request as ready for review September 23, 2022 20:40

carllin reviewed Sep 29, 2022

View reviewed changes

apfitzge reviewed Sep 29, 2022

View reviewed changes

AshwinSekar force-pushed the take-last-vote-banking branch from dd742cf to 4e8c730 Compare September 30, 2022 00:30

AshwinSekar commented Sep 30, 2022

View reviewed changes

AshwinSekar force-pushed the take-last-vote-banking branch 2 times, most recently from cdc2706 to 53a11a0 Compare October 5, 2022 21:50

AshwinSekar requested review from carllin and apfitzge October 5, 2022 23:43

AshwinSekar force-pushed the take-last-vote-banking branch from 53a11a0 to 8d7f0c6 Compare October 11, 2022 18:20

AshwinSekar added 3 commits October 14, 2022 12:18

Split out voting and banking threads in banking stage

55d6ea7

Additionally this allows us to aggressively prune the buffer for voting threads as with the new vote state only the latest vote from each validator is necessary.

Update local cluster test to use new Vote ix

4f99760

Encapsulate transaction storage filtering better

c20faf0

AshwinSekar added 5 commits October 14, 2022 12:18

Address pr comments

5c3be7b

Commit cargo lock change

a9d1b22

clippy

a1957f3

Remove unsafe impls

a97c27e

pr comments

073de96

AshwinSekar force-pushed the take-last-vote-banking branch from 8d7f0c6 to 073de96 Compare October 14, 2022 19:37

apfitzge previously approved these changes Oct 15, 2022

View reviewed changes

tao-stones previously approved these changes Oct 16, 2022

View reviewed changes

carllin reviewed Oct 16, 2022

View reviewed changes

compute_sanitized_transaction -> build_sanitized_transaction

4ab13c7

AshwinSekar added 2 commits October 20, 2022 11:08

&Arc -> Arc

9f73732

Move test

5197eb4

AshwinSekar force-pushed the take-last-vote-banking branch from 24e8041 to f7a21bf Compare October 20, 2022 18:09

Refactor metrics enums

648c2ad

AshwinSekar force-pushed the take-last-vote-banking branch from f7a21bf to 648c2ad Compare October 20, 2022 19:17

clippy

907328f

AshwinSekar added the automerge Merge this Pull Request automatically once CI passes label Oct 20, 2022

mergify bot merged commit f207af7 into solana-labs:master Oct 20, 2022

		// Mark so we don't forward again
		deserialized_packet.forwarded = true;

		pub(crate) num_dropped_gossip_vote_packets: usize,
		pub(crate) num_dropped_tpu_vote_packets: usize,

Split out voting and banking threads in banking stage #27931

Split out voting and banking threads in banking stage #27931

Conversation

AshwinSekar commented Sep 20, 2022

Problem

Summary of Changes

nikhayes commented Sep 20, 2022

AshwinSekar commented Sep 23, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

apfitzge commented Sep 29, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

AshwinSekar commented Sep 30, 2022

apfitzge left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tao-stones left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment