ForwardingStage #4508

apfitzge · 2025-01-16T21:39:21Z

Problem

Forwarding logic makes banking stage excessively more complicated than it needs to be.
Most nodes run without forwarding of non-vote transactions at all

Summary of Changes

Create a new ForwardingStage in the TPU that is intended to run parallel to banking-stage
Since the message types sent from Sigverify to BankingStage are wrapped in an Arc, we can (optionally) clone them and send to ForwardingStage.
This initial PR introduces ForwardingStage for simple receiving, buffering, forwarding logic, but does NOT hook it up yet.
A follow-up PR will be done to combine the hooking up of this ForwardingStage with the removal of all forwarding logic in BankingStage

Fixes #

apfitzge · 2025-01-17T14:07:37Z

@tao-stones (and other reviewers!) - f292ed1 (#4518) has an update of the documentation. Not including it in this PR since this PR does not actually hookup the ForwardingStage to SigVerify.

apfitzge · 2025-01-17T15:14:10Z

core/src/forwarding_stage.rs

+        let _res = conn.send_data_batch_async(batch);
+    }
+
+    fn send_non_vote_batch(&self, non_vote_batch: &mut Vec<(Vec<u8>, SocketAddr)>) {


@KirillLykov, this PR replaced the old one (adding ForwardingStage separatel from the hookup/removal).
Is this still the way to send packets? I recall you had built upon the old PR, please let me know if this should change here or we can/will do it in follow-up

I will try to move these changes on top of this PR Monday and will leave comments on the current PR.

KirillLykov · 2025-01-22T14:07:04Z

core/src/forwarding_stage.rs

+            forward_address_getter,
+            connection_cache,
+            data_budget: DataBudget::default(),
+            udp_socket: bind_to_unspecified().unwrap(),


I think we should stop binding on all interfaces @alexpyattaev

what should I use instead? I based the actual forwarding code in here on how we previously did it

Yeah, I know that we do this way but we should stop doing this. Maybe Alex has in mind some opinion about what function to use for this (since he told me that 0.0.0.0 is a problem). I think that it would make sense to pass udp_socket address to the constructor, so that on the level of validator the bind address is selected, which by default should be 127.0.0.1

For the stuff in agave you should prefer functions bind_in_range or bind_in_range_with_config. This will allow your code to respect bind-address CLI argument. Not respecting it will break firewall policies for some users.

Maybe I will abstract out all the network logic in this component, not sure, rebasing my old changes

core/src/forwarding_stage.rs

KirillLykov · 2025-01-27T11:59:01Z

core/src/forwarding_stage.rs

+        });
+    }
+
+    fn send_vote_batch(&self, addr: SocketAddr, vote_batch: &mut Vec<Vec<u8>>) {


I thought that votes are send using UDP while the rest using QUIC, looks like it is vise-versa now.

yeah - tests were all using UDP (non-votes should go out over whatever the connection cache uses), so didn't catch that I reversed these.

core/src/forwarding_stage.rs

alexpyattaev · 2025-01-27T13:39:52Z

core/src/forwarding_stage.rs

+    }
+}
+
+struct PacketContainer {


This should probably be its own file with unittests etc. Having it hanging with access to private fields is asking for trouble with silly bugs etc.

Module separation: fb06eef
Added tests: 24630c8

KirillLykov · 2025-01-27T14:45:28Z

core/src/forwarding_stage.rs

+    forward_address_getter: F,
+    connection_cache: Arc<ConnectionCache>,
+    data_budget: DataBudget,
+    udp_socket: UdpSocket,


I think that logic for networking should be split:

one simple struct for votes (UDP)

one structure for other transactions (TPU)

See apfitzge#6

I cherry-picked the commit but had to modify some stuff due to conflicts. Test was also failing with your bind_to instead of bind_to_unspecified, so I reverted that.

Interesting, why the test fails. Maybe leave this 0.0.0.0 as it is and I will try to change it to localhost in a separate PR.

at least for now I'd like to keep the actual networking changes to a minimum - I'm trying to just move this out of banking stage, not fix all the issues with it.

KirillLykov · 2025-01-29T16:21:09Z

core/src/forwarding_stage.rs

+
+    fn send_batch(&self, input_batch: &mut Vec<(Vec<u8>, SocketAddr)>) {
+        let mut batch = Vec::with_capacity(FORWARD_BATCH_SIZE);
+        core::mem::swap(&mut batch, input_batch); // why do we swap?


here you picked also my comment // why do we swap? which was referring to why can't we consume this vector. I later figured out that this problem might be resolved with adding a version of batch_send that takes SocketAddr and Vec<Vec<u8>> instead of Vec or pairs as it is now.

KirillLykov · 2025-01-29T16:27:29Z

core/src/forwarding_stage.rs

+    ) -> Self {
+        Self {
+            receiver,
+            packet_container: PacketContainer::with_capacity(4 * 4096),


Could you document the choice of this number? Or it has been taken from Forwarder and it is currently unclear how was chosen?

KirillLykov · 2025-01-29T16:28:23Z

core/src/forwarding_stage.rs

+                    }
+                };
+
+                // If timeout waas reached, prevent backup by draining all


Suggested change

// If timeout waas reached, prevent backup by draining all

// If timeout was reached, prevent backup by draining all

KirillLykov · 2025-01-29T16:29:30Z

core/src/forwarding_stage.rs

+                .filter(|p| initial_packet_meta_filter(p.meta()))
+            {
+                let Some(packet_data) = packet.data(..) else {
+                    // should never occur since we've already checked the


if it should never occur, shouldn't we assert?

KirillLykov · 2025-01-29T16:54:26Z

core/src/forwarding_stage.rs

+    }
+
+    fn new(
+        receiver: Receiver<(BankingPacketBatch, bool)>,


I wonder what will happen is this receiver will grow over time and ForwardingStage is not able to process packets with the desired speed. The processing seems to take O(num_txs_in_batch*tx_size) but I don't know if this might become a problem. One strategy would be just use bound channel and drop on the sender side. Do we write the size of the channel in some metric?

apfitzge added 9 commits January 16, 2025 15:34

ForwardingStage basic skeleton

78f8f01

receive_and_buffer

fec3c40

buffer_packet_batches

c263cdc

first pass, single buffer

a152b38

ForwardAddressGetter

6d6eed3

add forward_address_getter field

247143f

forwarding

2be12d0

explicit use instead of qualifying

bed4690

basic count metrics

e439f53

apfitzge self-assigned this Jan 16, 2025

apfitzge mentioned this pull request Jan 16, 2025

ForwardingStage #3222

Closed

apfitzge added 3 commits January 16, 2025 15:53

test_initial_packet_meta_filter

2679358

test_forwarding

3dce54a

clippy

a694e00

apfitzge mentioned this pull request Jan 17, 2025

ForwardingStage hookup, BankingStage forwarding removal #4518

Draft

apfitzge marked this pull request as ready for review January 17, 2025 15:12

apfitzge requested a review from tao-stones January 17, 2025 15:12

apfitzge commented Jan 17, 2025

View reviewed changes

KirillLykov mentioned this pull request Jan 17, 2025

poc for use tpu client next forwarding #3286

Open

apfitzge mentioned this pull request Jan 19, 2025

Plumb unified scheduler for BP minimally #4533

Open

KirillLykov reviewed Jan 22, 2025

View reviewed changes

core/src/forwarding_stage.rs Show resolved Hide resolved

KirillLykov reviewed Jan 22, 2025

View reviewed changes

core/src/forwarding_stage.rs Show resolved Hide resolved

KirillLykov reviewed Jan 22, 2025

View reviewed changes

core/src/forwarding_stage.rs Show resolved Hide resolved

documenting comments

d4799c8

KirillLykov reviewed Jan 27, 2025

View reviewed changes

alexpyattaev reviewed Jan 27, 2025

View reviewed changes

core/src/forwarding_stage.rs Show resolved Hide resolved

alexpyattaev reviewed Jan 27, 2025

View reviewed changes

KirillLykov reviewed Jan 27, 2025

View reviewed changes

apfitzge and others added 5 commits January 27, 2025 08:46

pass DataBudget

39717ab

bugfix: vote on udp

08b5361

packet_container module

fb06eef

packet_container tests

24630c8

add vote client

5218651

apfitzge mentioned this pull request Jan 28, 2025

Remove UnprocessedTransactionStorage #4604

Open

KirillLykov reviewed Jan 29, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

ForwardingStage #4508

ForwardingStage #4508

apfitzge commented Jan 16, 2025

apfitzge commented Jan 17, 2025

apfitzge Jan 17, 2025

KirillLykov Jan 17, 2025 •

edited

Loading

KirillLykov Jan 22, 2025

apfitzge Jan 22, 2025

KirillLykov Jan 22, 2025 •

edited

Loading

alexpyattaev Jan 22, 2025

KirillLykov Jan 22, 2025

KirillLykov Jan 27, 2025

apfitzge Jan 27, 2025

apfitzge Jan 27, 2025

alexpyattaev Jan 27, 2025 •

edited

Loading

apfitzge Jan 27, 2025

KirillLykov Jan 27, 2025

apfitzge Jan 27, 2025

KirillLykov Jan 27, 2025

apfitzge Jan 28, 2025

KirillLykov Jan 29, 2025

KirillLykov Jan 29, 2025

KirillLykov Jan 29, 2025

KirillLykov Jan 29, 2025

KirillLykov Jan 29, 2025

	// If timeout waas reached, prevent backup by draining all
	// If timeout was reached, prevent backup by draining all

ForwardingStage #4508

Are you sure you want to change the base?

ForwardingStage #4508

Conversation

apfitzge commented Jan 16, 2025

Problem

Summary of Changes

apfitzge commented Jan 17, 2025

Choose a reason for hiding this comment

KirillLykov Jan 17, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

KirillLykov Jan 22, 2025 • edited Loading

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

alexpyattaev Jan 27, 2025 • edited Loading

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

KirillLykov Jan 17, 2025 •

edited

Loading

KirillLykov Jan 22, 2025 •

edited

Loading

alexpyattaev Jan 27, 2025 •

edited

Loading