feat(tpu): stake weighted QUIC QoS and flow control

ffi/rust/firedancer-sys/build.rs

@@ -62,7 +62,7 @@ fn main() {
             .clang_arg(format!("-I{prefix}/"))
             .header(&format!("wrapper_{lib}.h"))
             .blocklist_type("schar|uchar|ushort|uint|ulong")
-            .blocklist_item("SORT_QUICK_ORDER_STYLE|SORT_MERGE_THRESH|SORT_QUICK_THRESH|SORT_QUICK_ORDER_STYLE|SORT_QUICK_SWAP_MINIMIZE");
+            .blocklist_item("SORT_QUICK_ORDER_STYLE|SORT_MERGE_THRESH|SORT_QUICK_THRESH|SORT_QUICK_ORDER_STYLE|SORT_QUICK_SWAP_MINIMIZE|MAP_MEMOIZE|MAP_QUERY_OPT|MAP_KEY_EQUAL_IS_SLOW");
 
         // Well this is a complete mess. We want to only include, say, functions
         // declared in the `ballet` directory in the ballet bindgen output. If

ffi/rust/firedancer-sys/src/tango/mod.rs

@@ -5,6 +5,7 @@ mod fseq;
 mod mcache;
 mod tcache;
 mod xdp;
+mod stake;
 
 pub use cnc::*;
 pub use dcache::*;
@@ -13,3 +14,4 @@ pub use fseq::*;
 pub use mcache::*;
 pub use tcache::*;
 pub use xdp::*;
+pub use stake::*;

ffi/rust/firedancer-sys/src/tango/stake.rs

@@ -0,0 +1,13 @@
+pub use crate::generated::{
+    fd_stake_t,
+    fd_stake_align,
+    fd_stake_footprint,
+    fd_stake_join,
+    fd_stake_new,
+    fd_stake_version,
+    fd_stake_version_laddr,
+    fd_stake_write,
+    fd_stake_read,
+    fd_stake_dump,
+    FD_STAKE_ALIGN
+};

src/app/fdctl/config.h

@@ -127,12 +127,13 @@ typedef struct {
       ushort listen_port;
       char   xdp_mode[ 8 ];
 
-      uint max_concurrent_connections;
+      ulong max_concurrent_connections;
       uint max_concurrent_connection_ids_per_connection;
       uint max_concurrent_streams_per_connection;
       uint max_concurrent_handshakes;
       uint max_inflight_quic_packets;
       uint tx_buf_size;
+      int  stake_lg_slot_cnt;
       uint xdp_rx_queue_size;
       uint xdp_tx_queue_size;
       uint xdp_aio_depth;

src/app/fdctl/config/default.toml

@@ -377,6 +377,8 @@ dynamic_port_range = "8000-10000"
 
         # Maximum number of simultaneous QUIC connections which can be open. New
         # connections which would exceed this limit will not be accepted.
+        #
+        # Must be >=2 and a power of 2.
         max_concurrent_connections = 32
 
         # While in TCP a connection is identified by (Source IP, Source Port,
@@ -393,7 +395,9 @@ dynamic_port_range = "8000-10000"
         #
         # Currently this option does nothing, as we do not support creating
         # additional connection IDs.
-        max_concurrent_connection_ids_per_connection = 16
+        #
+        # Should be in sync with `fd_quic_conn.h`.
+        max_concurrent_connection_ids_per_connection = 4
 
         # QUIC allows for multiple streams to be multiplexed over a single
         # connection. This option sets the maximum number of simultaneous
@@ -438,6 +442,9 @@ dynamic_port_range = "8000-10000"
         # should be unused.
         tx_buf_size = 4096
 
+        # Lg number of stake nodes to track
+        stake_lg_slot_cnt = 10
+
         # Firedancer uses XDP for fast packet processing. XDP supports two
         # modes, XDP_SKB and XDP_DRV. XDP_DRV is preferred as it is faster, but
         # is not supported by all drivers.

src/app/fdctl/configure/workspace.c

@@ -2,12 +2,21 @@
 
 #include "../../../tango/fd_tango.h"
 #include "../../../tango/quic/fd_quic.h"
+#include "../../../tango/quic/fd_quic_qos.h"
 #include "../../../tango/xdp/fd_xsk_aio.h"
+#include "../../../tango/udpsock/fd_udpsock.h"
 
 #include <sys/stat.h>
 #include <linux/capability.h>
+#include <errno.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <sys/socket.h>
+#include <arpa/inet.h>
+#include <netinet/in.h>
 
 #define NAME "workspace"
+#define FD_HAS_XDP 0
 
 static void
 init_perm( security_t *     security,
@@ -81,6 +90,20 @@ static void quic( void * pod, char * fmt, fd_quic_limits_t * limits, ... ) {
             fd_quic_new      ( shmem, limits ) );
 }
 
+static void quic_qos( void * pod, char * fmt, fd_quic_qos_limits_t * limits, ... ) {
+  INSERTER( limits,
+            fd_quic_qos_align    (               ),
+            fd_quic_qos_footprint( limits        ),
+            fd_quic_qos_new      ( shmem, limits ) );
+}
+
+static void stake( void * pod, char * fmt, int lg_slot_cnt, ... ) {
+  INSERTER( lg_slot_cnt,
+            fd_stake_align    (                    ),
+            fd_stake_footprint( lg_slot_cnt        ),
+            fd_stake_new      ( shmem, lg_slot_cnt ) );
+}
+
 static void xsk( void * pod, char * fmt, ulong frame_sz, ulong rx_depth, ulong tx_depth, ... ) {
   INSERTER( tx_depth,
             fd_xsk_align    (                                                            ),
@@ -95,6 +118,13 @@ static void xsk_aio( void * pod, char * fmt, ulong tx_depth, ulong batch_count,
             fd_xsk_aio_new      ( shmem,    tx_depth, batch_count ) );
 }
 
+static void udpsock( void * pod, char * fmt, ulong frame_sz, ulong rx_depth, ulong tx_depth, ... ) {
+  INSERTER( tx_depth,
+            fd_udpsock_align    (                                     ),
+            fd_udpsock_footprint( frame_sz, rx_depth, tx_depth        ),
+            fd_udpsock_new      ( shmem, frame_sz, rx_depth, tx_depth ) );
+}
+
 static void alloc( void * pod, char * fmt, ulong align, ulong sz, ... ) {
   INSERTER( sz, align, sz, 1 );
 }
@@ -196,19 +226,6 @@ init( config_t * const config ) {
   if( FD_LIKELY( uid == 0 && seteuid( config->uid ) ) )
     FD_LOG_ERR(( "seteuid() failed (%i-%s)", errno, strerror( errno ) ));
 
-  fd_quic_limits_t limits = {
-    .conn_cnt                                      = config->tiles.quic.max_concurrent_connections,
-    .handshake_cnt                                 = config->tiles.quic.max_concurrent_handshakes,
-    .conn_id_cnt                                   = config->tiles.quic.max_concurrent_connection_ids_per_connection,
-    .conn_id_sparsity                              = 0.0,
-    .inflight_pkt_cnt                              = config->tiles.quic.max_inflight_quic_packets,
-    .tx_buf_sz                                     = config->tiles.quic.tx_buf_size,
-    .stream_cnt[ FD_QUIC_STREAM_TYPE_BIDI_CLIENT ] = 0,
-    .stream_cnt[ FD_QUIC_STREAM_TYPE_BIDI_SERVER ] = 0,
-    .stream_cnt[ FD_QUIC_STREAM_TYPE_UNI_CLIENT  ] = config->tiles.quic.max_concurrent_streams_per_connection,
-    .stream_cnt[ FD_QUIC_STREAM_TYPE_UNI_SERVER  ] = 0,
-  };
-
   for( ulong j=0; j<config->shmem.workspaces_cnt; j++ ) {
     workspace_config_t * wksp1 = &config->shmem.workspaces[ j ];
     WKSP_BEGIN( config, wksp1, 0 );
@@ -242,17 +259,64 @@ init( config_t * const config ) {
           fseq  ( pod, "fseq%lu", i );
         }
         break;
-      case wksp_quic:
-        cnc    ( pod, "cnc" );
-        quic   ( pod, "quic",    &limits );
-        xsk    ( pod, "xsk",     2048, config->tiles.quic.xdp_rx_queue_size, config->tiles.quic.xdp_tx_queue_size );
-        xsk_aio( pod, "xsk_aio", config->tiles.quic.xdp_tx_queue_size, config->tiles.quic.xdp_aio_depth );
-
+      case wksp_quic:;
+        fd_quic_limits_t quic_limits = {
+          .conn_cnt                                      = config->tiles.quic.max_concurrent_connections,
+          .handshake_cnt                                 = config->tiles.quic.max_concurrent_handshakes,
+          .conn_id_cnt                                   = config->tiles.quic.max_concurrent_connection_ids_per_connection,
+          .conn_id_sparsity                              = 0.0,
+          .inflight_pkt_cnt                              = config->tiles.quic.max_inflight_quic_packets,
+          .tx_buf_sz                                     = config->tiles.quic.tx_buf_size,
+          .stream_cnt[ FD_QUIC_STREAM_TYPE_BIDI_CLIENT ] = 0,
+          .stream_cnt[ FD_QUIC_STREAM_TYPE_BIDI_SERVER ] = 0,
+          .stream_cnt[ FD_QUIC_STREAM_TYPE_UNI_CLIENT  ] = config->tiles.quic.max_concurrent_streams_per_connection,
+          .stream_cnt[ FD_QUIC_STREAM_TYPE_UNI_SERVER  ] = 0,
+        };
+
+        int lg_max_conns = fd_ulong_find_msb( config->tiles.quic.max_concurrent_connections );
+        if ( lg_max_conns < 1 ) FD_LOG_ERR( ( "max_concurrent_connections must be at least 2." ) );
+        fd_quic_qos_limits_t quic_qos_limits = {
+            .min_streams    = FD_QUIC_QOS_DEFAULT_MIN_STREAMS,
+            .max_streams    = FD_QUIC_QOS_DEFAULT_MAX_STREAMS,
+            .total_streams  = FD_QUIC_QOS_DEFAULT_TOTAL_STREAMS,
+            .pq_lg_slot_cnt = lg_max_conns - 1,
+            .lru_depth      = config->tiles.quic.max_concurrent_connections >> 1,
+        };
+
+        cnc     ( pod, "cnc" );
+        quic    ( pod, "quic",    &quic_limits );
+        quic_qos( pod, "quic_qos", &quic_qos_limits );
+        stake   ( pod, "stake",   10 );  // FIXME slot cnt is not getting parsed
+
+        #if FD_HAS_XDP
+        (void)udpsock;
+        xsk     ( pod, "xsk",     2048, config->tiles.quic.xdp_rx_queue_size, config->tiles.quic.xdp_tx_queue_size );
+        xsk_aio ( pod, "xsk_aio", config->tiles.quic.xdp_tx_queue_size, config->tiles.quic.xdp_aio_depth );
         char const * quic_xsk_gaddr = fd_pod_query_cstr( pod, "xsk", NULL );
         void *       shmem          = fd_wksp_map      ( quic_xsk_gaddr );
         if( FD_UNLIKELY( !fd_xsk_bind( shmem, config->name, config->tiles.quic.interface, (uint)wksp1->kind_idx ) ) )
           FD_LOG_ERR(( "failed to bind xsk for quic tile %lu", wksp1->kind_idx ));
         fd_wksp_unmap( shmem );
+        #else
+        (void)xsk;
+        (void)xsk_aio;
+        int sock_fd = socket( AF_INET, SOCK_DGRAM, IPPROTO_UDP );
+        if( FD_UNLIKELY( sock_fd<0 ) ) {
+          FD_LOG_ERR( (
+              "socket(AF_INET,SOCK_DGRAM,IPPROTO_UDP) failed (%d-%s)", errno, strerror( errno ) ) );
+        }
+        struct sockaddr_in listen_addr = {
+          .sin_family = AF_INET,
+          .sin_addr   = { .s_addr = FD_IP4_ADDR(127, 0, 0, 1) },
+          .sin_port   = (ushort)fd_ushort_bswap( 8004 ),
+        };
+        if( FD_UNLIKELY( 0!=bind( sock_fd, (struct sockaddr const *)fd_type_pun_const( &listen_addr ), sizeof(struct sockaddr_in) ) ) ) {
+          close( sock_fd );
+          FD_LOG_ERR( ( "bind(sock_fd) failed (%d-%s)", errno, strerror( errno ) ) );
+        }
+        udpsock ( pod, "udpsock",     2048, config->tiles.quic.xdp_rx_queue_size, config->tiles.quic.xdp_tx_queue_size );
+        #endif
+
 
         uint1  ( pod, "ip_addr",                    config->tiles.quic.ip_addr );
         ushort1( pod, "listen_port",                config->tiles.quic.listen_port, 0 );

src/app/frank/fd_frank.h

@@ -4,6 +4,7 @@
 #include "../../disco/fd_disco.h"
 #include "../../ballet/fd_ballet.h" /* FIXME: CONSIDER HAVING THIS IN DISCO_BASE */
 #include "../../tango/xdp/fd_xsk.h"
+#include "../../tango/udpsock/fd_udpsock.h"
 
 /* FD_FRANK_CNC_DIAG_* are FD_CNC_DIAG_* style diagnostics and thus the
    same considerations apply.  Further they are harmonized with the
@@ -29,10 +30,11 @@ typedef struct {
    char *        tile_name;
    ulong         tile_idx;
    ulong         idx;
-   uchar const * tile_pod;
-   uchar const * in_pod;
-   uchar const * out_pod;
-   fd_xsk_t    * xsk;
+   uchar const  * tile_pod;
+   uchar const  * in_pod;
+   uchar const  * out_pod;
+   fd_xsk_t     * xsk;
+   fd_udpsock_t * udpsock;
 } fd_frank_args_t;
 
 typedef struct {

src/app/frank/fd_frank_quic.c

@@ -56,6 +56,14 @@ run( fd_frank_args_t * args ) {
   fd_quic_t * quic = fd_quic_join( fd_wksp_pod_map( args->tile_pod, "quic" ) );
   if( FD_UNLIKELY( !quic ) ) FD_LOG_ERR(( "fd_quic_join failed" ));
 
+  FD_LOG_INFO(( "loading quic" ));
+  fd_quic_qos_t * quic_qos = fd_quic_qos_join( fd_wksp_pod_map( args->tile_pod, "quic_qos" ) );
+  if( FD_UNLIKELY( !quic_qos ) ) FD_LOG_ERR(( "fd_quic_qos_join failed" ));
+
+  FD_LOG_INFO(( "loading stake" ));
+  fd_stake_t * stake = fd_stake_join( fd_wksp_pod_map( args->tile_pod, "stake" ) );
+  if( FD_UNLIKELY( !stake ) ) FD_LOG_ERR(( "fd_stake_join failed" ));
+
   FD_LOG_INFO(( "loading xsk_aio" ));
   fd_xsk_aio_t * xsk_aio = fd_xsk_aio_join( fd_wksp_pod_map( args->tile_pod, "xsk_aio" ), args->xsk );
   if( FD_UNLIKELY( !xsk_aio ) ) FD_LOG_ERR(( "fd_xsk_aio_join failed" ));
@@ -122,7 +130,12 @@ run( fd_frank_args_t * args ) {
   /* Start serving */
 
   FD_LOG_INFO(( "%s(%lu) run", args->tile_name, args->tile_idx ));
-  int err = fd_quic_tile( cnc, quic, xsk_aio, mcache, dcache, lazy, rng, scratch );
+  #if FD_HAS_XDP
+  #define SOCK args->xsk
+  #else
+  #define SOCK args->udpsock
+  #endif
+  int err = fd_quic_tile( cnc, quic, quic_qos, stake, SOCK, mcache, dcache, lazy, rng, scratch );
   if( FD_UNLIKELY( err ) ) FD_LOG_ERR(( "fd_quic_tile failed (%i)", err ));
 }
 

src/ballet/x509/fd_x509.c

@@ -50,14 +50,17 @@ fd_x509_gen_solana_cert( EVP_PKEY * pkey ) {
 
   /* Generate serial number */
   long serial;
-  if( FD_UNLIKELY( 1!=RAND_bytes( (uchar *)&serial, sizeof(long) ) ) ) {
+  if ( FD_UNLIKELY( 1 != RAND_bytes( (uchar *)&serial, sizeof( long ) ) ) ) {
     FD_LOG_WARNING(( "RAND_bytes() failed" ));
     goto cleanup1;
   }
   ASN1_INTEGER_set( X509_get_serialNumber(x), serial );
 
   /* Set public key (the only important part) */
-  X509_set_pubkey( x, pkey );
+  if ( FD_UNLIKELY( 1 != X509_set_pubkey( x, pkey ) ) ) {
+    FD_LOG_WARNING(( "X509_set_pubkey() failed" ));
+    goto cleanup1;
+  };
 
   /* Set very long expiration date */
   long not_before = 0L;            /* Jan  1 00:00:00 1975 GMT */

src/disco/quic/fd_quic.h

@@ -38,7 +38,10 @@
 
 #include "../fd_disco_base.h"
 #include "../../tango/quic/fd_quic.h"
+#include "../../tango/quic/fd_quic_qos.h"
+#include "../../tango/quic/tls/fd_quic_tls.h"
 #include "../../tango/xdp/fd_xdp.h"
+#include "../../tango/udpsock/fd_udpsock.h"
 #include "../../ballet/txn/fd_txn.h"
 
 #if FD_HAS_HOSTED
@@ -118,14 +121,20 @@ FD_FN_CONST ulong
 fd_quic_tile_scratch_footprint( ulong depth );
 
 int
-fd_quic_tile( fd_cnc_t *         cnc,        /* Local join to the tile's command-and-control */
-              fd_quic_t *        quic,       /* QUIC without active join */
-              fd_xsk_aio_t *     xsk_aio,    /* Local join to QUIC XSK aio */
-              fd_frag_meta_t *   mcache,     /* Local join to the tile's txn output mcache */
-              uchar *            dcache,     /* Local join to the tile's txn output dcache */
-              long               lazy,       /* Laziness, <=0 means use a reasonable default */
-              fd_rng_t *         rng,        /* Local join to the rng this tile should use */
-              void *             scratch );  /* Tile scratch memory */
+fd_quic_tile( fd_cnc_t *       cnc,  /* Local join to the tile's command-and-control */
+              fd_quic_t *      quic, /* QUIC without active join */
+              fd_quic_qos_t *  quic_qos,  /* Local join to QoS */
+              fd_stake_t *     stake,
+              #if FD_HAS_XDP
+              fd_xsk_aio_t *   xsk_aio, /* Local join to QUIC XSK aio */
+              #else
+              fd_udpsock_t *   udpsock, /* Local join to QUIC udp sock */
+              #endif
+              fd_frag_meta_t * mcache,  /* Local join to the tile's txn output mcache */
+              uchar *          dcache,  /* Local join to the tile's txn output dcache */
+              long             lazy,    /* Laziness, <=0 means use a reasonable default */
+              fd_rng_t *       rng,     /* Local join to the rng this tile should use */
+              void *           scratch );         /* Tile scratch memory */
 
 FD_PROTOTYPES_END