Skip to content

Commit c1cbd7c

Browse files
riptlriptl
committed
net 2.0: generic userland RX flow steering
Fixes a design flaw where the xdp and sock tiles hardcode the flow steering rules for the app tiles. Defines a generic 'topo_net_rx' struct that maps UDP ports to output links (and metadata like DST_PROTO IDs). Adds a 'find_16x16' API for AVX-accelerated fast port matching. Temporarily undoes the 'repair ping-pong' flow steering hack.
1 parent 681cd12 commit c1cbd7c

File tree

20 files changed

+315
-279
lines changed

20 files changed

+315
-279
lines changed

‎src/app/fdctl/topology.c

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -380,14 +380,17 @@ fd_topo_initialize( config_t * config ) {
380380

381381
FOR(net_tile_cnt) fd_topos_net_tile_finish( topo, i );
382382

383+
fd_topo_net_rx_t rx_rules = {0};
384+
fd_topo_net_rx_rule_push( &rx_rules, DST_PROTO_SHRED, "net_shred", config->tiles.shred.shred_listen_port );
385+
fd_topo_net_rx_rule_push( &rx_rules, DST_PROTO_TPU_QUIC, "net_quic" , config->tiles.quic.quic_transaction_listen_port );
386+
fd_topo_net_rx_rule_push( &rx_rules, DST_PROTO_TPU_UDP, "net_quic" , config->tiles.quic.regular_transaction_listen_port );
387+
383388
for( ulong i=0UL; i<topo->tile_cnt; i++ ) {
384389
fd_topo_tile_t * tile = &topo->tiles[ i ];
385390

386391
if( FD_UNLIKELY( !strcmp( tile->name, "net" ) || !strcmp( tile->name, "sock" ) ) ) {
387392

388-
tile->net.shred_listen_port = config->tiles.shred.shred_listen_port;
389-
tile->net.quic_transaction_listen_port = config->tiles.quic.quic_transaction_listen_port;
390-
tile->net.legacy_transaction_listen_port = config->tiles.quic.regular_transaction_listen_port;
393+
tile->net.rx_rules = rx_rules;
391394

392395
} else if( FD_UNLIKELY( !strcmp( tile->name, "netlnk" ) ) ) {
393396

‎src/app/firedancer-dev/commands/backtest.c

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -323,7 +323,7 @@ backtest_topo( config_t * config ) {
323323

324324
for( ulong i=0UL; i<topo->tile_cnt; i++ ) {
325325
fd_topo_tile_t * tile = &topo->tiles[ i ];
326-
if( !fd_topo_configure_tile( tile, config ) ) {
326+
if( !fd_topo_configure_tile( tile, config, NULL ) ) {
327327
FD_LOG_ERR(( "unknown tile name %lu `%s`", i, tile->name ));
328328
}
329329

‎src/app/firedancer-dev/commands/gossip.c

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -42,7 +42,7 @@ gossip_topo( config_t * config ) {
4242
if( net_tile_id==ULONG_MAX ) net_tile_id = fd_topo_find_tile( topo, "sock", 0UL );
4343
if( FD_UNLIKELY( net_tile_id==ULONG_MAX ) ) FD_LOG_ERR(( "net tile not found" ));
4444
fd_topo_tile_t * net_tile = &topo->tiles[ net_tile_id ];
45-
net_tile->net.gossip_listen_port = config->gossip.port;
45+
fd_topo_net_rx_rule_push( &net_tile->net.rx_rules, DST_PROTO_GOSSIP, "net_gossip", config->gossip.port );
4646

4747
fd_topob_wksp( topo, "gossip" );
4848
fd_topo_tile_t * gossip_tile = fd_topob_tile( topo, "gossip", "gossip", "metric_in", 0UL, 0, 0 );

‎src/app/firedancer-dev/commands/repair.c

Lines changed: 15 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -58,9 +58,8 @@ static void
5858
repair_topo( config_t * config ) {
5959
resolve_gossip_entrypoints( config );
6060

61-
ulong net_tile_cnt = config->layout.net_tile_count;
62-
ulong shred_tile_cnt = config->layout.shred_tile_count;
63-
ulong quic_tile_cnt = config->layout.quic_tile_count;
61+
ulong net_tile_cnt = config->layout.net_tile_count;
62+
ulong shred_tile_cnt = config->layout.shred_tile_count;
6463

6564
fd_topo_t * topo = { fd_topob_new( &config->topo, config->name ) };
6665
topo->max_page_size = fd_cstr_to_shmem_page_sz( config->hugetlbfs.max_page_size );
@@ -71,7 +70,6 @@ repair_topo( config_t * config ) {
7170
fd_topob_wksp( topo, "net_shred" );
7271
fd_topob_wksp( topo, "net_gossip" );
7372
fd_topob_wksp( topo, "net_repair" );
74-
fd_topob_wksp( topo, "net_quic" );
7573

7674
fd_topob_wksp( topo, "shred_repair" );
7775
fd_topob_wksp( topo, "stake_out" );
@@ -93,8 +91,6 @@ repair_topo( config_t * config ) {
9391
fd_topob_wksp( topo, "sign_repair" );
9492

9593
fd_topob_wksp( topo, "repair_repla" );
96-
fd_topob_wksp( topo, "gossip_send" );
97-
fd_topob_wksp( topo, "send_txns" );
9894

9995
fd_topob_wksp( topo, "shred" );
10096
fd_topob_wksp( topo, "sign" );
@@ -111,7 +107,6 @@ repair_topo( config_t * config ) {
111107
ulong pending_fec_shreds_depth = fd_ulong_min( fd_ulong_pow2_up( config->tiles.shred.max_pending_shred_sets * FD_REEDSOL_DATA_SHREDS_MAX ), USHORT_MAX + 1 /* dcache max */ );
112108

113109
/* topo, link_name, wksp_name, depth, mtu, burst */
114-
FOR(quic_tile_cnt) fd_topob_link( topo, "quic_net", "net_quic", config->net.ingress_buffer_size, FD_NET_MTU, 1UL );
115110
FOR(shred_tile_cnt) fd_topob_link( topo, "shred_net", "net_shred", config->net.ingress_buffer_size, FD_NET_MTU, 1UL );
116111
117112
/**/ fd_topob_link( topo, "stake_out", "stake_out", 128UL, 40UL + 40200UL * 40UL, 1UL );
@@ -127,7 +122,6 @@ repair_topo( config_t * config ) {
127122
128123
/**/ fd_topob_link( topo, "crds_shred", "crds_shred", 128UL, 8UL + 40200UL * 38UL, 1UL );
129124
/**/ fd_topob_link( topo, "gossip_repai", "gossip_repai", 128UL, 40200UL * 38UL, 1UL );
130-
/**/ fd_topob_link( topo, "gossip_send", "gossip_send", 128UL, 40200UL * 38UL, 1UL );
131125
132126
/**/ fd_topob_link( topo, "gossip_net", "net_gossip", config->net.ingress_buffer_size, FD_NET_MTU, 1UL );
133127
@@ -140,8 +134,6 @@ repair_topo( config_t * config ) {
140134
/**/ fd_topob_link( topo, "repair_repla", "repair_repla", 65536UL, sizeof(fd_reasm_fec_t), 1UL );
141135
/**/ fd_topob_link( topo, "poh_shred", "poh_shred", 16384UL, USHORT_MAX, 1UL );
142136

143-
/**/ fd_topob_link( topo, "send_txns", "send_txns", 128UL, FD_TXN_MTU, 1UL );
144-
145137
FD_TEST( sizeof(fd_snapshot_manifest_t)<=(5UL*(1UL<<30UL)) );
146138
/**/ fd_topob_link( topo, "snap_out", "snap_out", 2UL, 5UL*(1UL<<30UL), 1UL );
147139

@@ -176,7 +168,6 @@ repair_topo( config_t * config ) {
176168

177169
FOR(net_tile_cnt) fd_topos_net_rx_link( topo, "net_gossip", i, config->net.ingress_buffer_size );
178170
FOR(net_tile_cnt) fd_topos_net_rx_link( topo, "net_repair", i, config->net.ingress_buffer_size );
179-
FOR(net_tile_cnt) fd_topos_net_rx_link( topo, "net_quic", i, config->net.ingress_buffer_size );
180171
FOR(net_tile_cnt) fd_topos_net_rx_link( topo, "net_shred", i, config->net.ingress_buffer_size );
181172

182173
/* topo, tile_name, tile_wksp, metrics_wksp, cpu_idx, is_agave, uses_keyswitch */
@@ -254,10 +245,6 @@ repair_topo( config_t * config ) {
254245
/* topo, tile_name, tile_kind_id, fseq_wksp, link_name, link_kind_id, reliable, polled */
255246
for( ulong j=0UL; j<shred_tile_cnt; j++ )
256247
fd_topos_tile_in_net( topo, "metric_in", "shred_net", j, FD_TOPOB_UNRELIABLE, FD_TOPOB_POLLED ); /* No reliable consumers of networking fragments, may be dropped or overrun */
257-
for( ulong j=0UL; j<quic_tile_cnt; j++ )
258-
fd_topos_tile_in_net( topo, "metric_in", "quic_net", j, FD_TOPOB_UNRELIABLE, FD_TOPOB_POLLED ); /* No reliable consumers of networking fragments, may be dropped or overrun */
259-
260-
/**/ fd_topob_tile_in( topo, "gossip", 0UL, "metric_in", "send_txns", 0UL, FD_TOPOB_RELIABLE, FD_TOPOB_POLLED );
261248

262249
/**/ fd_topos_tile_in_net( topo, "metric_in", "gossip_net", 0UL, FD_TOPOB_UNRELIABLE, FD_TOPOB_POLLED ); /* No reliable consumers of networking fragments, may be dropped or overrun */
263250
/**/ fd_topos_tile_in_net( topo, "metric_in", "repair_net", 0UL, FD_TOPOB_UNRELIABLE, FD_TOPOB_POLLED ); /* No reliable consumers of networking fragments, may be dropped or overrun */
@@ -294,7 +281,6 @@ repair_topo( config_t * config ) {
294281
/**/ fd_topob_tile_out( topo, "gossip", 0UL, "gossip_sign", 0UL );
295282
/**/ fd_topob_tile_in( topo, "gossip", 0UL, "metric_in", "sign_gossip", 0UL, FD_TOPOB_UNRELIABLE, FD_TOPOB_UNPOLLED );
296283
/**/ fd_topob_tile_out( topo, "sign", 0UL, "sign_gossip", 0UL );
297-
/**/ fd_topob_tile_out( topo, "gossip", 0UL, "gossip_send", 0UL );
298284
/**/ fd_topob_tile_out( topo, "gossip", 0UL, "gossip_tower", 0UL );
299285

300286
FOR(net_tile_cnt) fd_topob_tile_in( topo, "repair", 0UL, "metric_in", "net_repair", i, FD_TOPOB_UNRELIABLE, FD_TOPOB_POLLED ); /* No reliable consumers of networking fragments, may be dropped or overrun */
@@ -339,23 +325,25 @@ repair_topo( config_t * config ) {
339325
fd_topob_tile_out( topo, "scap", 0UL, "snap_out", 0UL );
340326
}
341327

342-
FD_TEST( link_permit_no_producers( topo, "quic_net" ) == quic_tile_cnt );
343-
FD_TEST( link_permit_no_producers( topo, "poh_shred" ) == 1UL );
344-
FD_TEST( link_permit_no_producers( topo, "send_txns" ) == 1UL );
345-
FD_TEST( link_permit_no_producers( topo, "repair_scap" ) == 1UL );
346-
FD_TEST( link_permit_no_producers( topo, "replay_scap" ) == 1UL );
328+
FD_TEST( link_permit_no_producers( topo, "poh_shred" ) == 1UL );
329+
FD_TEST( link_permit_no_producers( topo, "repair_scap" ) == 1UL );
330+
FD_TEST( link_permit_no_producers( topo, "replay_scap" ) == 1UL );
347331

348-
FD_TEST( link_permit_no_consumers( topo, "net_quic" ) == quic_tile_cnt );
349-
FD_TEST( link_permit_no_consumers( topo, "gossip_verif" ) == 1UL );
350-
FD_TEST( link_permit_no_consumers( topo, "gossip_tower" ) == 1UL );
351-
FD_TEST( link_permit_no_consumers( topo, "gossip_send" ) == 1UL );
352-
FD_TEST( link_permit_no_consumers( topo, "repair_repla" ) == 1UL );
332+
FD_TEST( link_permit_no_consumers( topo, "gossip_verif" ) == 1UL );
333+
FD_TEST( link_permit_no_consumers( topo, "gossip_tower" ) == 1UL );
334+
FD_TEST( link_permit_no_consumers( topo, "repair_repla" ) == 1UL );
353335

354336
FOR(net_tile_cnt) fd_topos_net_tile_finish( topo, i );
355337

338+
fd_topo_net_rx_t rx_rules = {0};
339+
fd_topo_net_rx_rule_push( &rx_rules, DST_PROTO_SHRED, "net_shred", config->tiles.shred.shred_listen_port );
340+
fd_topo_net_rx_rule_push( &rx_rules, DST_PROTO_GOSSIP, "net_gossip", config->gossip.port );
341+
fd_topo_net_rx_rule_push( &rx_rules, DST_PROTO_REPAIR, "net_repair", config->tiles.repair.repair_intake_listen_port );
342+
fd_topo_net_rx_rule_push( &rx_rules, DST_PROTO_REPAIR, "net_repair", config->tiles.repair.repair_serve_listen_port );
343+
356344
for( ulong i=0UL; i<topo->tile_cnt; i++ ) {
357345
fd_topo_tile_t * tile = &topo->tiles[ i ];
358-
if( !fd_topo_configure_tile( tile, config ) ) {
346+
if( !fd_topo_configure_tile( tile, config, &rx_rules ) ) {
359347
FD_LOG_ERR(( "unknown tile name %lu `%s`", i, tile->name ));
360348
}
361349
}

‎src/app/firedancer-dev/commands/sim.c

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -174,7 +174,7 @@ sim_topo( config_t * config ) {
174174
} else {
175175
FD_LOG_NOTICE(( "Found archive file from config: %s", tile->archiver.rocksdb_path ));
176176
}
177-
} else if( !fd_topo_configure_tile( tile, config ) ) {
177+
} else if( !fd_topo_configure_tile( tile, config, NULL ) ) {
178178
FD_LOG_ERR(( "unknown tile name %lu `%s`", i, tile->name ));
179179
}
180180

‎src/app/firedancer-dev/commands/snapshot_load.c

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -102,7 +102,7 @@ snapshot_load_topo( config_t * config,
102102

103103
for( ulong i=0UL; i<topo->tile_cnt; i++ ) {
104104
fd_topo_tile_t * tile = &topo->tiles[ i ];
105-
if( !fd_topo_configure_tile( tile, config ) ) {
105+
if( !fd_topo_configure_tile( tile, config, NULL ) ) {
106106
FD_LOG_ERR(( "unknown tile name %lu `%s`", i, tile->name ));
107107
}
108108
}

‎src/app/firedancer/topology.c

Lines changed: 16 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -825,9 +825,18 @@ fd_topo_initialize( config_t * config ) {
825825

826826
FOR(net_tile_cnt) fd_topos_net_tile_finish( topo, i );
827827

828+
fd_topo_net_rx_t rx_rules = {0};
829+
fd_topo_net_rx_rule_push( &rx_rules, DST_PROTO_TPU_QUIC, "net_quic", config->tiles.quic.quic_transaction_listen_port );
830+
fd_topo_net_rx_rule_push( &rx_rules, DST_PROTO_TPU_QUIC, "net_quic", config->tiles.quic.quic_transaction_listen_port );
831+
fd_topo_net_rx_rule_push( &rx_rules, DST_PROTO_SHRED, "net_shred", config->tiles.shred.shred_listen_port );
832+
fd_topo_net_rx_rule_push( &rx_rules, DST_PROTO_GOSSIP, "net_gossip", config->gossip.port );
833+
fd_topo_net_rx_rule_push( &rx_rules, DST_PROTO_REPAIR, "net_repair", config->tiles.repair.repair_intake_listen_port );
834+
fd_topo_net_rx_rule_push( &rx_rules, DST_PROTO_REPAIR, "net_repair", config->tiles.repair.repair_serve_listen_port );
835+
fd_topo_net_rx_rule_push( &rx_rules, DST_PROTO_SEND, "net_send", config->tiles.send.send_src_port );
836+
828837
for( ulong i=0UL; i<topo->tile_cnt; i++ ) {
829838
fd_topo_tile_t * tile = &topo->tiles[ i ];
830-
if( !fd_topo_configure_tile( tile, config ) ) {
839+
if( !fd_topo_configure_tile( tile, config, &rx_rules ) ) {
831840
FD_LOG_ERR(( "unknown tile name %lu `%s`", i, tile->name ));
832841
}
833842
}
@@ -858,17 +867,12 @@ fd_topo_initialize( config_t * config ) {
858867
}
859868

860869
int
861-
fd_topo_configure_tile( fd_topo_tile_t * tile,
862-
fd_config_t * config ) {
863-
if( FD_UNLIKELY( !strcmp( tile->name, "net" ) || !strcmp( tile->name, "sock" ) ) ) {
864-
865-
tile->net.shred_listen_port = config->tiles.shred.shred_listen_port;
866-
tile->net.quic_transaction_listen_port = config->tiles.quic.quic_transaction_listen_port;
867-
tile->net.legacy_transaction_listen_port = config->tiles.quic.regular_transaction_listen_port;
868-
tile->net.gossip_listen_port = config->gossip.port;
869-
tile->net.repair_intake_listen_port = config->tiles.repair.repair_intake_listen_port;
870-
tile->net.repair_serve_listen_port = config->tiles.repair.repair_serve_listen_port;
871-
tile->net.send_src_port = config->tiles.send.send_src_port;
870+
fd_topo_configure_tile( fd_topo_tile_t * tile,
871+
fd_config_t * config,
872+
fd_topo_net_rx_t const * rx_rules ) {
873+
if( FD_UNLIKELY( rx_rules && (!strcmp( tile->name, "net" ) || !strcmp( tile->name, "sock" )) ) ) {
874+
875+
tile->net.rx_rules = *rx_rules;
872876

873877
} else if( FD_UNLIKELY( !strcmp( tile->name, "netlnk" ) ) ) {
874878

‎src/app/firedancer/topology.h

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -49,8 +49,9 @@ setup_topo_txncache( fd_topo_t * topo,
4949
ulong max_txn_per_slot );
5050

5151
int
52-
fd_topo_configure_tile( fd_topo_tile_t * tile,
53-
fd_config_t * config );
52+
fd_topo_configure_tile( fd_topo_tile_t * tile,
53+
fd_config_t * config,
54+
fd_topo_net_rx_t const * rx_rules );
5455

5556
FD_PROTOTYPES_END
5657

‎src/app/shared_dev/commands/pktgen/pktgen.c

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -197,7 +197,7 @@ pktgen_cmd_fn( args_t * args FD_PARAM_UNUSED,
197197
fd_topo_tile_t * metric_tile = &topo->tiles[ fd_topo_find_tile( topo, "metric", 0UL ) ];
198198

199199
ushort const listen_port = 9000;
200-
net_tile->net.legacy_transaction_listen_port = listen_port;
200+
fd_topo_net_rx_rule_push( &net_tile->net.rx_rules, DST_PROTO_TPU_UDP, "net_quic", listen_port );
201201

202202
if( FD_UNLIKELY( !fd_cstr_to_ip4_addr( config->tiles.metric.prometheus_listen_address, &metric_tile->metric.prometheus_listen_addr ) ) )
203203
FD_LOG_ERR(( "failed to parse prometheus listen address `%s`", config->tiles.metric.prometheus_listen_address ));

‎src/app/shared_dev/commands/udpecho/udpecho.c

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -88,7 +88,7 @@ udpecho_cmd_fn( args_t * args,
8888
fd_topo_tile_t * net_tile = &topo->tiles[ fd_topo_find_tile( topo, "net", 0UL ) ];
8989
fd_topo_tile_t * metric_tile = &topo->tiles[ fd_topo_find_tile( topo, "metric", 0UL ) ];
9090

91-
net_tile->net.legacy_transaction_listen_port = args->udpecho.listen_port;
91+
fd_topo_net_rx_rule_push( &net_tile->net.rx_rules, DST_PROTO_TPU_UDP, "net_quic", args->udpecho.listen_port );
9292

9393
if( FD_UNLIKELY( !fd_cstr_to_ip4_addr( config->tiles.metric.prometheus_listen_address, &metric_tile->metric.prometheus_listen_addr ) ) )
9494
FD_LOG_ERR(( "failed to parse prometheus listen address `%s`", config->tiles.metric.prometheus_listen_address ));

0 commit comments

Comments
 (0)