shred: allow additional destination

Author: ptaffet-jump

src/app/fdctl/config/default.toml

@@ -1198,6 +1198,17 @@ dynamic_port_range = "8900-9000"
         # this one.
         shred_listen_port = 8003
 
+        # Shreds can also be forwarded to a specific address, for
+        # example to run an unstaked RPC or for archiving.  If
+        # additional_shred_destination is not empty, each new, valid
+        # shred that the validator receives will be forwarded to the
+        # specified address.  additional_shred_destination must be in
+        # the form "ip:port", for example "1.2.3.4:5566".  Shreds will
+        # be sent to this destination first, prior to sending to other
+        # validators.  This applies to leader slots and non-leader
+        # slots.
+        additional_shred_destination = ""
+
     # The metric tile receives metrics updates published from the rest
     # of the tiles and serves them via. a Prometheus compatible HTTP
     # endpoint.

src/app/fdctl/topology.c

@@ -440,6 +440,23 @@ fd_topo_initialize( config_t * config ) {
       tile->shred.expected_shred_version        = config->consensus.expected_shred_version;
       tile->shred.shred_listen_port             = config->tiles.shred.shred_listen_port;
       tile->shred.larger_shred_limits_per_block = config->development.bench.larger_shred_limits_per_block;
+      char   adtl_dest[ sizeof("255.255.255.255:65536") ];
+      memcpy( adtl_dest, config->tiles.shred.additional_shred_destination, sizeof(adtl_dest) );
+      if( FD_UNLIKELY( strcmp( adtl_dest, "" ) ) ) {
+        char * ip_end = strchr( adtl_dest, ':' );
+        if( FD_UNLIKELY( !ip_end ) ) FD_LOG_ERR(( "[tiles.shred.additional_shred_destination] must be empty or in the form ip:port" ));
+        *ip_end = '\0';
+
+        if( FD_UNLIKELY( !fd_cstr_to_ip4_addr( adtl_dest, &(tile->shred.adtl_dest.ip) ) ) ) {
+          FD_LOG_ERR(( "could not parse IP %s in [tiles.shred.additional_shred_destination]", adtl_dest ));
+        }
+
+        tile->shred.adtl_dest.port = fd_cstr_to_ushort( ip_end+1 );
+        if( FD_UNLIKELY( !tile->shred.adtl_dest.port ) ) FD_LOG_ERR(( "could not parse port %s in [tiles.shred.additional_shred_destination]", ip_end+1 ));
+      } else {
+        tile->shred.adtl_dest.ip   = 0U;
+        tile->shred.adtl_dest.port = 0;
+      }
 
     } else if( FD_UNLIKELY( !strcmp( tile->name, "store" ) ) ) {
       tile->store.disable_blockstore_from_slot = config->development.bench.disable_blockstore_from_slot;

src/app/shared/fd_config.h

@@ -371,6 +371,7 @@ struct fd_config {
     struct {
       uint   max_pending_shred_sets;
       ushort shred_listen_port;
+      char   additional_shred_destination[ sizeof("255.255.255.255:65536") ];
     } shred;
 
     struct {

src/app/shared/fd_config_parse.c

@@ -325,6 +325,7 @@ fdctl_pod_to_cfg( config_t * config,
 
   CFG_POP      ( uint,   tiles.shred.max_pending_shred_sets               );
   CFG_POP      ( ushort, tiles.shred.shred_listen_port                    );
+  CFG_POP      ( cstr,   tiles.shred.additional_shred_destination         );
 
   CFG_POP      ( cstr,   tiles.metric.prometheus_listen_address           );
   CFG_POP      ( ushort, tiles.metric.prometheus_listen_port              );

src/disco/shred/fd_shred_tile.c

@@ -145,6 +145,8 @@ typedef struct {
 
   int skip_frag;
 
+  fd_shred_dest_weighted_t adtl_dest[1];
+
   fd_ip4_udp_hdrs_t data_shred_net_hdr  [1];
   fd_ip4_udp_hdrs_t parity_shred_net_hdr[1];
 
@@ -509,12 +511,10 @@ during_frag( fd_shred_ctx_t * ctx,
 }
 
 static inline void
-send_shred( fd_shred_ctx_t *    ctx,
-            fd_shred_t const *  shred,
-            fd_shred_dest_t *   sdest,
-            fd_shred_dest_idx_t dest_idx,
-            ulong               tsorig ) {
-  fd_shred_dest_weighted_t * dest = fd_shred_dest_idx_to_dest( sdest, dest_idx );
+send_shred( fd_shred_ctx_t                 * ctx,
+            fd_shred_t const               * shred,
+            fd_shred_dest_weighted_t const * dest,
+            ulong                            tsorig ) {
 
   if( FD_UNLIKELY( !dest->ip4 ) ) return;
 
@@ -646,7 +646,8 @@ after_frag( fd_shred_ctx_t *    ctx,
         fd_shred_dest_idx_t * dests = fd_shred_dest_compute_children( sdest, &shred, 1UL, ctx->scratchpad_dests, 1UL, fanout, fanout, max_dest_cnt );
         if( FD_UNLIKELY( !dests ) ) break;
 
-        for( ulong j=0UL; j<*max_dest_cnt; j++ ) send_shred( ctx, *out_shred, sdest, dests[ j ], ctx->tsorig );
+        send_shred( ctx, *out_shred, ctx->adtl_dest, ctx->tsorig );
+        for( ulong j=0UL; j<*max_dest_cnt; j++ ) send_shred( ctx, *out_shred, fd_shred_dest_idx_to_dest( sdest, dests[ j ]), ctx->tsorig );
       } while( 0 );
 
       if( FD_LIKELY( ctx->blockstore && rv==FD_FEC_RESOLVER_SHRED_OKAY ) ) { /* optimize for the compiler - branch predictor will still be correct */
@@ -763,7 +764,10 @@ after_frag( fd_shred_ctx_t *    ctx,
   if( FD_UNLIKELY( !dests ) ) return;
 
   /* Send only the ones we didn't receive. */
-  for( ulong i=0UL; i<k; i++ ) for( ulong j=0UL; j<*max_dest_cnt; j++ ) send_shred( ctx, new_shreds[ i ], sdest, dests[ j*out_stride+i ], ctx->tsorig );
+  for( ulong i=0UL; i<k; i++ ) {
+    send_shred( ctx, new_shreds[ i ], ctx->adtl_dest, ctx->tsorig );
+    for( ulong j=0UL; j<*max_dest_cnt; j++ ) send_shred( ctx, new_shreds[ i ], fd_shred_dest_idx_to_dest( sdest, dests[ j*out_stride+i ]), ctx->tsorig );
+  }
 }
 
 static void
@@ -921,6 +925,9 @@ unprivileged_init( fd_topo_t *      topo,
   fd_ip4_udp_hdr_init( ctx->data_shred_net_hdr,   FD_SHRED_MIN_SZ, 0, tile->shred.shred_listen_port );
   fd_ip4_udp_hdr_init( ctx->parity_shred_net_hdr, FD_SHRED_MAX_SZ, 0, tile->shred.shred_listen_port );
 
+  ctx->adtl_dest->ip4  = tile->shred.adtl_dest.ip;
+  ctx->adtl_dest->port = tile->shred.adtl_dest.port;
+
   for( ulong i=0UL; i<tile->in_cnt; i++ ) {
     fd_topo_link_t const * link = &topo->links[ tile->in_link_id[ i ] ];
     fd_topo_wksp_t const * link_wksp = &topo->workspaces[ topo->objs[ link->dcache_obj_id ].wksp_id ];

src/disco/topo/fd_topo.h

@@ -230,6 +230,10 @@ typedef struct {
       ushort shred_listen_port;
       int    larger_shred_limits_per_block;
       ulong  expected_shred_version;
+      struct {
+        uint   ip;
+        ushort port;
+      } adtl_dest;
     } shred;
 
     struct {