Add vinyl support to replay tile

Also add housekeep callback to vinyl tile

Author: riptl

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

@@ -20,6 +20,7 @@
 #include "../../shared/fd_config.h" /* config_t */
 #include "../../../disco/tiles.h"
 #include "../../../disco/topo/fd_topob.h"
+#include "../../../disco/topo/fd_topob_vinyl.h"
 #include "../../../util/pod/fd_pod_format.h"
 #include "../../../discof/replay/fd_replay_tile.h"
 #include "../../../discof/restore/utils/fd_ssctrl.h"
@@ -50,7 +51,7 @@ backtest_topo( config_t * config ) {
   ulong lta_tile_cnt    = config->firedancer.layout.snapla_tile_count;
 
   int disable_snap_loader      = !config->gossip.entrypoints_cnt;
-  int snap_vinyl               = !!config->firedancer.vinyl.enabled;
+  int vinyl_enabled            = !!config->firedancer.vinyl.enabled;
   int solcap_enabled           = strlen( config->capture.solcap_capture )>0;
   int snapshot_lthash_disabled = config->development.snapshots.disable_lthash_verification;
 
@@ -91,7 +92,7 @@ backtest_topo( config_t * config ) {
       config->firedancer.runtime.program_cache.heap_size_mib<<20 );
   fd_topob_tile_uses( topo, replay_tile, progcache_obj, FD_SHMEM_JOIN_MODE_READ_WRITE );
 
-  if( snap_vinyl ) {
+  if( vinyl_enabled ) {
     setup_topo_vinyl_meta( topo, &config->firedancer );
   }
 
@@ -105,7 +106,6 @@ backtest_topo( config_t * config ) {
   /**********************************************************************/
   /* Add the snapshot tiles to topo                                       */
   /**********************************************************************/
-  int vinyl_enabled = config->firedancer.vinyl.enabled;
   fd_topo_tile_t * snapin_tile = NULL;
   fd_topo_tile_t * snapwr_tile = NULL;
   if( FD_UNLIKELY( !disable_snap_loader ) ) {
@@ -352,7 +352,7 @@ backtest_topo( config_t * config ) {
   fd_topob_tile_uses( topo, replay_tile, txncache_obj, FD_SHMEM_JOIN_MODE_READ_WRITE );
   if( FD_LIKELY( !disable_snap_loader ) ) {
     fd_topob_tile_uses( topo, snapin_tile, txncache_obj, FD_SHMEM_JOIN_MODE_READ_WRITE );
-    if( snap_vinyl ) {
+    if( vinyl_enabled ) {
       ulong vinyl_map_obj_id  = fd_pod_query_ulong( topo->props, "vinyl.meta_map",  ULONG_MAX ); FD_TEST( vinyl_map_obj_id !=ULONG_MAX );
       ulong vinyl_pool_obj_id = fd_pod_query_ulong( topo->props, "vinyl.meta_pool", ULONG_MAX ); FD_TEST( vinyl_pool_obj_id!=ULONG_MAX );
       fd_topo_obj_t * vinyl_map_obj  = &topo->objs[ vinyl_map_obj_id ];
@@ -376,6 +376,10 @@ backtest_topo( config_t * config ) {
     fd_topob_tile_uses( topo, snapin_tile, funk_obj, FD_SHMEM_JOIN_MODE_READ_WRITE );
   }
 
+  if( vinyl_enabled ) {
+    fd_topob_vinyl_rq( topo, "replay", 0UL, "vinyl_replay", "replay", 1UL, 1UL, 1UL );
+  }
+
   for( ulong i=0UL; i<topo->tile_cnt; i++ ) {
     fd_topo_tile_t * tile = &topo->tiles[ i ];
     fd_topo_configure_tile( tile, config );

src/app/firedancer-dev/main.c

@@ -27,6 +27,9 @@ extern fd_topo_obj_callbacks_t fd_obj_cb_funk;
 extern fd_topo_obj_callbacks_t fd_obj_cb_vinyl_meta;
 extern fd_topo_obj_callbacks_t fd_obj_cb_vinyl_meta_ele;
 extern fd_topo_obj_callbacks_t fd_obj_cb_vinyl_data;
+extern fd_topo_obj_callbacks_t fd_obj_cb_vinyl_req_pool;
+extern fd_topo_obj_callbacks_t fd_obj_cb_vinyl_rq;
+extern fd_topo_obj_callbacks_t fd_obj_cb_vinyl_cq;
 
 fd_topo_obj_callbacks_t * CALLBACKS[] = {
   &fd_obj_cb_mcache,
@@ -47,6 +50,9 @@ fd_topo_obj_callbacks_t * CALLBACKS[] = {
   &fd_obj_cb_vinyl_meta,
   &fd_obj_cb_vinyl_meta_ele,
   &fd_obj_cb_vinyl_data,
+  &fd_obj_cb_vinyl_req_pool,
+  &fd_obj_cb_vinyl_rq,
+  &fd_obj_cb_vinyl_cq,
   NULL,
 };
 

src/app/firedancer/callbacks_vinyl.c

@@ -1,5 +1,6 @@
 #include "../../vinyl/fd_vinyl.h"
 #include "../../disco/topo/fd_topo.h"
+#include "../../flamenco/accdb/fd_vinyl_req_pool.h"
 #include "../../util/pod/fd_pod_format.h"
 
 #define VAL(name) (__extension__({                                                             \
@@ -111,3 +112,87 @@ fd_topo_obj_callbacks_t fd_obj_cb_vinyl_data = {
   .align     = vinyl_data_align,
   .new       = vinyl_data_new,
 };
+
+/* vinyl_req_pool: request allocator */
+
+static ulong
+vinyl_req_pool_align( fd_topo_t const *     topo,
+                      fd_topo_obj_t const * obj ) {
+  (void)topo; (void)obj;
+  return fd_vinyl_req_pool_align();
+}
+
+static ulong
+vinyl_req_pool_footprint( fd_topo_t const *     topo,
+                            fd_topo_obj_t const * obj ) {
+  return fd_vinyl_req_pool_footprint( VAL("batch_max"), VAL("batch_key_max") );
+}
+
+static void
+vinyl_req_pool_new( fd_topo_t const *     topo,
+                    fd_topo_obj_t const * obj ) {
+  FD_TEST( fd_vinyl_req_pool_new( fd_topo_obj_laddr( topo, obj->id ), VAL("batch_max"), VAL("batch_key_max") ) );
+}
+
+fd_topo_obj_callbacks_t fd_obj_cb_vinyl_req_pool = {
+  .name      = "vinyl_req_pool",
+  .footprint = vinyl_req_pool_footprint,
+  .align     = vinyl_req_pool_align,
+  .new       = vinyl_req_pool_new,
+};
+
+/* vinyl_rq: request queue */
+
+static ulong
+vinyl_rq_align( fd_topo_t const *     topo,
+                fd_topo_obj_t const * obj ) {
+  (void)topo; (void)obj;
+  return fd_vinyl_rq_align();
+}
+
+static ulong
+vinyl_rq_footprint( fd_topo_t const *     topo,
+                    fd_topo_obj_t const * obj ) {
+  return fd_vinyl_rq_footprint( VAL("req_cnt") );
+}
+
+static void
+vinyl_rq_new( fd_topo_t const *     topo,
+              fd_topo_obj_t const * obj ) {
+  FD_TEST( fd_vinyl_rq_new( fd_topo_obj_laddr( topo, obj->id ), VAL("req_cnt") ) );
+}
+
+fd_topo_obj_callbacks_t fd_obj_cb_vinyl_rq = {
+  .name      = "vinyl_rq",
+  .footprint = vinyl_rq_footprint,
+  .align     = vinyl_rq_align,
+  .new       = vinyl_rq_new,
+};
+
+/* vinyl_cq: completion queue */
+
+static ulong
+vinyl_cq_align( fd_topo_t const *     topo,
+                fd_topo_obj_t const * obj ) {
+  (void)topo; (void)obj;
+  return fd_vinyl_cq_align();
+}
+
+static ulong
+vinyl_cq_footprint( fd_topo_t const *     topo,
+                    fd_topo_obj_t const * obj ) {
+  return fd_vinyl_cq_footprint( VAL("comp_cnt") );
+}
+
+static void
+vinyl_cq_new( fd_topo_t const *     topo,
+              fd_topo_obj_t const * obj ) {
+  FD_TEST( fd_vinyl_cq_new( fd_topo_obj_laddr( topo, obj->id ), VAL("comp_cnt") ) );
+}
+
+fd_topo_obj_callbacks_t fd_obj_cb_vinyl_cq = {
+  .name      = "vinyl_cq",
+  .footprint = vinyl_cq_footprint,
+  .align     = vinyl_cq_align,
+  .new       = vinyl_cq_new,
+};

src/app/firedancer/topology.c

@@ -1241,6 +1241,12 @@ fd_topo_configure_tile( fd_topo_tile_t * tile,
     tile->replay.funk_obj_id      = fd_pod_query_ulong( config->topo.props, "funk",      ULONG_MAX ); FD_TEST( tile->replay.funk_obj_id     !=ULONG_MAX );
     tile->replay.progcache_obj_id = fd_pod_query_ulong( config->topo.props, "progcache", ULONG_MAX ); FD_TEST( tile->replay.progcache_obj_id!=ULONG_MAX );
 
+    if( config->firedancer.vinyl.enabled ) {
+      tile->replay.vinyl_data_wksp_id = fd_pod_query_ulong( config->topo.props, "vinyl.data", ULONG_MAX ); FD_TEST( tile->replay.vinyl_data_wksp_id!=ULONG_MAX );
+    } else {
+      tile->replay.vinyl_data_wksp_id = ULONG_MAX;
+    }
+
     tile->replay.max_live_slots = config->firedancer.runtime.max_live_slots;
 
     tile->replay.expected_shred_version = config->consensus.expected_shred_version;

src/disco/topo/fd_topo.h

@@ -366,6 +366,8 @@ struct fd_topo_tile {
       ulong max_vote_accounts;
 
       ulong funk_obj_id;
+      ulong vinyl_data_wksp_id;
+
       ulong txncache_obj_id;
       ulong progcache_obj_id;
 
@@ -621,6 +623,10 @@ struct fd_topo_tile {
 
       int  io_type; /* FD_VINYL_IO_TYPE_* */
       uint uring_depth;
+
+#     define FD_TOPO_VINYL_LINK_MAX 16UL
+      ulong rq_cnt;
+      ulong rq_obj_id[ FD_TOPO_VINYL_LINK_MAX ];
     } vinyl;
   };
 };
@@ -629,9 +635,13 @@ typedef struct fd_topo_tile fd_topo_tile_t;
 
 typedef struct {
   ulong id;
-  char  name[ 13UL ];
+  char  name[ 13UL ];  /* object type */
   ulong wksp_id;
 
+  /* Optional label for object */
+  char  label[ 13UL ];  /* object label */
+  ulong label_idx;      /* index of object for this label (ULONG_MAX if not labelled) */
+
   ulong offset;
   ulong footprint;
 } fd_topo_obj_t;
@@ -885,6 +895,49 @@ fd_topo_tile_producer_cnt( fd_topo_t const *     topo,
   return in_cnt;
 }
 
+FD_FN_PURE FD_FN_UNUSED static ulong
+fd_topo_obj_cnt( fd_topo_t const * topo,
+                 char const *      obj_type,
+                 char const *      label ) {
+  ulong cnt = 0UL;
+  for( ulong i=0UL; i<topo->obj_cnt; i++ ) {
+    fd_topo_obj_t const * obj = &topo->objs[ i ];
+    if( strncmp( obj->name, obj_type, sizeof(obj->name) ) ) continue;
+    if( label &&
+        strncmp( obj->label, label, sizeof(obj->label) ) ) continue;
+    cnt++;
+  }
+  return cnt;
+}
+
+FD_FN_PURE FD_FN_UNUSED static fd_topo_obj_t const *
+fd_topo_find_obj( fd_topo_t const * topo,
+                  char const *      obj_type,
+                  char const *      label,
+                  ulong             label_idx ) {
+  for( ulong i=0UL; i<topo->obj_cnt; i++ ) {
+    fd_topo_obj_t const * obj = &topo->objs[ i ];
+    if( strncmp( obj->name, obj_type, sizeof(obj->name) ) ) continue;
+    if( label &&
+        strncmp( obj->label, label, sizeof(obj->label) ) ) continue;
+    if( label_idx != ULONG_MAX && obj->label_idx != label_idx ) continue;
+    return obj;
+  }
+  return NULL;
+}
+
+FD_FN_PURE FD_FN_UNUSED static fd_topo_obj_t const *
+fd_topo_find_tile_obj( fd_topo_t const *      topo,
+                       fd_topo_tile_t const * tile,
+                       char const *           obj_type ) {
+  for( ulong i=0UL; i<(tile->uses_obj_cnt); i++ ) {
+    fd_topo_obj_t const * obj = &topo->objs[ tile->uses_obj_id[ i ] ];
+    if( strncmp( obj->name, obj_type, sizeof(obj->name) ) ) continue;
+    return obj;
+  }
+  return NULL;
+}
+
 /* Join (map into the process) all shared memory (huge/gigantic pages)
    needed by the tile, in the given topology.  All memory associated
    with the tile (aka. used by links that the tile either produces to or

src/disco/topo/fd_topob.c

@@ -57,14 +57,32 @@ fd_topob_obj( fd_topo_t *  topo,
   if( FD_UNLIKELY( wksp_id==ULONG_MAX ) ) FD_LOG_ERR(( "workspace not found: %s", wksp_name ));
 
   fd_topo_obj_t * obj = &topo->objs[ topo->obj_cnt ];
+  memset( obj, 0, sizeof(fd_topo_obj_t) );
   strncpy( obj->name, obj_name, sizeof(obj->name) );
-  obj->id      = topo->obj_cnt;
-  obj->wksp_id = wksp_id;
+  obj->id        = topo->obj_cnt;
+  obj->wksp_id   = wksp_id;
+  obj->label_idx = ULONG_MAX;
   topo->obj_cnt++;
 
   return obj;
 }
 
+fd_topo_obj_t *
+fd_topob_obj_named( fd_topo_t *  topo,
+                    char const * obj_type,
+                    char const * wksp_name,
+                    char const * label ) {
+  if( FD_UNLIKELY( !label ) ) FD_LOG_ERR(( "NULL args" ));
+  if( FD_UNLIKELY( strlen( label )>=sizeof(topo->objs[ topo->obj_cnt ].label ) ) ) FD_LOG_ERR(( "obj label too long: %s", label ));
+  fd_topo_obj_t * obj = fd_topob_obj( topo, obj_type, wksp_name );
+  if( FD_UNLIKELY( !obj ) ) return NULL;
+
+  fd_cstr_ncpy( obj->label, label, sizeof(obj->label) );
+  obj->label_idx = fd_topo_obj_cnt( topo, obj_type, label );
+
+  return obj;
+}
+
 fd_topo_link_t *
 fd_topob_link( fd_topo_t *  topo,
                char const * link_name,
@@ -106,10 +124,10 @@ fd_topob_link( fd_topo_t *  topo,
 }
 
 void
-fd_topob_tile_uses( fd_topo_t *      topo,
-                    fd_topo_tile_t * tile,
-                    fd_topo_obj_t *  obj,
-                    int              mode ) {
+fd_topob_tile_uses( fd_topo_t *           topo,
+                    fd_topo_tile_t *      tile,
+                    fd_topo_obj_t const * obj,
+                    int                   mode ) {
   (void)topo;
 
   if( FD_UNLIKELY( tile->uses_obj_cnt>=FD_TOPO_MAX_TILE_OBJS ) ) FD_LOG_ERR(( "tile `%s` uses too many objects", tile->name ));

src/disco/topo/fd_topob.h

@@ -40,7 +40,7 @@ fd_topo_wksp_t *
 fd_topob_wksp( fd_topo_t *  topo,
                char const * name );
 
-/* Add an object with the given name to the toplogy.  An object is
+/* Add an object with the given type to the toplogy.  An object is
    something that takes up space in memory, in a workspace.
 
    The workspace must exist and have been added to the topology.
@@ -50,9 +50,17 @@ fd_topob_wksp( fd_topo_t *  topo,
 
 fd_topo_obj_t *
 fd_topob_obj( fd_topo_t *  topo,
-              char const * obj_name,
+              char const * obj_type,
               char const * wksp_name );
 
+/* Same as fd_topo_obj, but labels the object. */
+
+fd_topo_obj_t *
+fd_topob_obj_named( fd_topo_t *  topo,
+                    char const * obj_type,
+                    char const * wksp_name,
+                    char const * label );
+
 /* Add a relationship saying that a certain tile uses a given object.
    This has the effect that when memory mapping required workspaces
    for a tile, it will map the workspace required for this object in
@@ -62,10 +70,10 @@ fd_topob_obj( fd_topo_t *  topo,
    FD_SHMEM_JOIN_MODE_READ_WRITE. */
 
 void
-fd_topob_tile_uses( fd_topo_t *      topo,
-                    fd_topo_tile_t * tile,
-                    fd_topo_obj_t *  obj,
-                    int              mode );
+fd_topob_tile_uses( fd_topo_t *           topo,
+                    fd_topo_tile_t *      tile,
+                    fd_topo_obj_t const * obj,
+                    int                   mode );
 
 /* Add a link to the toplogy.  The link will not have any producer or
    consumer(s) by default, and those need to be added after.  The link