restore: async O_DIRECT via worker thread

Author: riptl

book/api/metrics-generated.md

@@ -1093,3 +1093,14 @@
 | <span class="metrics-name">exec_&#8203;progcache_&#8203;invalidations</span> | counter | Number of program cache invalidations |
 
 </div>
+
+## Snapwr Tile
+
+<div class="metrics">
+
+| Metric | Type | Description |
+|--------|------|-------------|
+| <span class="metrics-name">snapwr_&#8203;state</span> | gauge | State of the tile. 0=IDLE, 1=PROCESSING, 4=SHUTDOWN |
+| <span class="metrics-name">snapwr_&#8203;vinyl_&#8203;bytes_&#8203;written</span> | gauge | Number of bytes written so far to the vinyl snapshot file. Might decrease if snapshot creation is aborted and restarted |
+
+</div>

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

@@ -333,6 +333,8 @@ backtest_cmd_topo( config_t * config ) {
   backtest_topo( config );
 }
 
+extern configure_stage_t fd_cfg_stage_vinyl;
+
 static args_t
 configure_args( void ) {
   args_t args = {
@@ -342,6 +344,7 @@ configure_args( void ) {
   ulong stage_idx = 0UL;
   args.configure.stages[ stage_idx++ ] = &fd_cfg_stage_hugetlbfs;
   args.configure.stages[ stage_idx++ ] = &fd_cfg_stage_snapshots;
+  args.configure.stages[ stage_idx++ ] = &fd_cfg_stage_vinyl;
   args.configure.stages[ stage_idx++ ] = NULL;
 
   return args;

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

@@ -7,7 +7,6 @@
 #include "../../../disco/topo/fd_topob.h"
 #include "../../../disco/pack/fd_pack.h"
 #include "../../../disco/pack/fd_pack_cost.h"
-#include "../../../util/tile/fd_tile_private.h"
 #include "../../../util/pod/fd_pod_format.h"
 #include "../../../discof/restore/utils/fd_ssctrl.h"
 #include "../../../discof/restore/utils/fd_ssmsg.h"
@@ -25,8 +24,7 @@ fd_topo_run_tile_t
 fdctl_tile_run( fd_topo_tile_t const * tile );
 
 static void
-snapshot_load_topo( config_t *     config,
-                    args_t const * args ) {
+snapshot_load_topo( config_t * config ) {
   fd_topo_t * topo = &config->topo;
   fd_topob_new( &config->topo, config->name );
   topo->max_page_size = fd_cstr_to_shmem_page_sz( config->hugetlbfs.max_page_size );
@@ -47,50 +45,56 @@ snapshot_load_topo( config_t *     config,
     setup_topo_vinyl( topo, &config->firedancer );
   }
 
-  static ushort tile_to_cpu[ FD_TILE_MAX ] = {0};
-  if( args->snapshot_load.tile_cpus[0] ) {
-    ulong cpu_cnt = fd_tile_private_cpus_parse( args->snapshot_load.tile_cpus, tile_to_cpu );
-    if( FD_UNLIKELY( cpu_cnt<4UL ) ) FD_LOG_ERR(( "--tile-cpus specifies %lu CPUs, but need at least 4", cpu_cnt ));
-  }
-
   /* metrics tile *****************************************************/
   fd_topob_wksp( topo, "metric_in" );
   fd_topob_wksp( topo, "metric" );
-  fd_topob_tile( topo, "metric",  "metric", "metric_in", tile_to_cpu[0], 0, 0 );
+  fd_topob_tile( topo, "metric",  "metric", "metric_in", ULONG_MAX, 0, 0 );
 
   /* read() tile */
   fd_topob_wksp( topo, "snapct" );
-  fd_topo_tile_t * snapct_tile = fd_topob_tile( topo, "snapct", "snapct", "snapct", tile_to_cpu[1], 0, 0 );
+  fd_topo_tile_t * snapct_tile = fd_topob_tile( topo, "snapct", "snapct", "metric_in", ULONG_MAX, 0, 0 );
   snapct_tile->allow_shutdown = 1;
 
   /* load tile */
   fd_topob_wksp( topo, "snapld" );
-  fd_topo_tile_t * snapld_tile = fd_topob_tile( topo, "snapld", "snapld", "snapld", tile_to_cpu[2], 0, 0 );
+  fd_topo_tile_t * snapld_tile = fd_topob_tile( topo, "snapld", "snapld", "metric_in", ULONG_MAX, 0, 0 );
   snapld_tile->allow_shutdown = 1;
 
   /* "snapdc": Zstandard decompress tile */
   fd_topob_wksp( topo, "snapdc" );
-  fd_topo_tile_t * snapdc_tile = fd_topob_tile( topo, "snapdc", "snapdc", "snapdc", tile_to_cpu[3], 0, 0 );
+  fd_topo_tile_t * snapdc_tile = fd_topob_tile( topo, "snapdc", "snapdc", "metric_in", ULONG_MAX, 0, 0 );
   snapdc_tile->allow_shutdown = 1;
 
   /* "snapin": Snapshot parser tile */
   fd_topob_wksp( topo, "snapin" );
-  fd_topo_tile_t * snapin_tile = fd_topob_tile( topo, "snapin", "snapin", "snapin", tile_to_cpu[4], 0, 0 );
+  fd_topo_tile_t * snapin_tile = fd_topob_tile( topo, "snapin", "snapin", "metric_in", ULONG_MAX, 0, 0 );
   snapin_tile->allow_shutdown = 1;
 
+  /* "snapwr": Snapshot writer tile */
+  int vinyl_enabled = config->firedancer.vinyl.enabled;
+  if( vinyl_enabled ) {
+    fd_topob_wksp( topo, "snapwr" );
+    fd_topo_tile_t * snapwr_tile = fd_topob_tile( topo, "snapwr", "snapwr", "metric_in", ULONG_MAX, 0, 0 );
+    snapwr_tile->allow_shutdown = 1;
+  }
+
   fd_topob_wksp( topo, "snapct_ld"    );
   fd_topob_wksp( topo, "snapld_dc"    );
   fd_topob_wksp( topo, "snapdc_in"    );
   fd_topob_wksp( topo, "snapin_ct"    );
   fd_topob_wksp( topo, "snapin_manif" );
   fd_topob_wksp( topo, "snapct_repr"  );
+  if( vinyl_enabled ) fd_topob_wksp( topo, "snapin_wr" );
 
   fd_topob_link( topo, "snapct_ld",   "snapct_ld",     128UL,   sizeof(fd_ssctrl_init_t),       1UL );
   fd_topob_link( topo, "snapld_dc",   "snapld_dc",     16384UL, USHORT_MAX,                     1UL );
   fd_topob_link( topo, "snapdc_in",   "snapdc_in",     16384UL, USHORT_MAX,                     1UL );
   fd_topob_link( topo, "snapin_ct",   "snapin_ct",     128UL,   0UL,                            1UL );
   fd_topob_link( topo, "snapin_manif", "snapin_manif", 2UL,     sizeof(fd_snapshot_manifest_t), 1UL )->permit_no_consumers = 1;
   fd_topob_link( topo, "snapct_repr", "snapct_repr",   128UL,   0UL,                            1UL )->permit_no_consumers = 1;
+  if( vinyl_enabled ) {
+    fd_topob_link( topo, "snapin_wr", "snapin_wr", 4UL, 16UL<<20, 1UL );
+  }
 
   fd_topob_tile_in ( topo, "snapct",  0UL, "metric_in", "snapin_ct",    0UL, FD_TOPOB_RELIABLE,   FD_TOPOB_POLLED );
   fd_topob_tile_in ( topo, "snapct",  0UL, "metric_in", "snapld_dc",    0UL, FD_TOPOB_RELIABLE,   FD_TOPOB_POLLED );
@@ -103,6 +107,10 @@ snapshot_load_topo( config_t *     config,
   fd_topob_tile_in ( topo, "snapin",  0UL, "metric_in", "snapdc_in",    0UL, FD_TOPOB_RELIABLE,   FD_TOPOB_POLLED );
   fd_topob_tile_out( topo, "snapin",  0UL,              "snapin_ct",    0UL                                       );
   fd_topob_tile_out( topo, "snapin",  0UL,              "snapin_manif", 0UL                                       );
+  if( vinyl_enabled ) {
+    fd_topob_tile_out( topo, "snapin", 0UL,              "snapin_wr", 0UL );
+    fd_topob_tile_in ( topo, "snapwr", 0UL, "metric_in", "snapin_wr", 0UL, FD_TOPOB_RELIABLE, FD_TOPOB_POLLED );
+  }
 
   /* snapin funk / txncache access */
   fd_topob_tile_uses( topo, snapin_tile, funk_obj, FD_SHMEM_JOIN_MODE_READ_WRITE );
@@ -127,34 +135,19 @@ snapshot_load_topo( config_t *     config,
     fd_topo_configure_tile( tile, config );
   }
 
-  if( !args->snapshot_load.tile_cpus[0] ) {
-    fd_topob_auto_layout( topo, 0 );
-  }
+  fd_topob_auto_layout( topo, 0 );
   fd_topob_finish( topo, CALLBACKS );
 }
 
-static void
-snapshot_load_cmd_args( int *    pargc,
-                        char *** pargv,
-                        args_t * args ) {
-  char const * tile_cpus                = fd_env_strip_cmdline_cstr( pargc, pargv,  "--tile-cpus",     "FD_TILE_CPUS", NULL );
-
-  if( tile_cpus ) {
-    ulong tile_cpus_strlen = strlen( tile_cpus );
-    if( FD_UNLIKELY( tile_cpus_strlen>=sizeof(args->snapshot_load.tile_cpus) ) ) FD_LOG_ERR(( "--tile-cpus: flag too long" ));
-    fd_cstr_fini( fd_cstr_append_text( fd_cstr_init( args->snapshot_load.tile_cpus ), tile_cpus, tile_cpus_strlen ) );
-  }
-}
-
 extern int * fd_log_private_shared_lock;
 
 static void
 snapshot_load_cmd_fn( args_t *   args,
                       config_t * config ) {
+  (void)args;
   if( FD_UNLIKELY( config->firedancer.snapshots.sources.gossip.allow_any || 0UL!=config->firedancer.snapshots.sources.gossip.allow_list_cnt ) ) {
     FD_LOG_ERR(( "snapshot-load command is incompatible with gossip snapshot sources" ));
   }
-  snapshot_load_topo( config, args );
   fd_topo_t * topo = &config->topo;
 
   args_t configure_args = {
@@ -181,14 +174,18 @@ snapshot_load_cmd_fn( args_t *   args,
   fd_topo_tile_t * snapld_tile = &topo->tiles[ fd_topo_find_tile( topo, "snapld", 0UL ) ];
   fd_topo_tile_t * snapdc_tile = &topo->tiles[ fd_topo_find_tile( topo, "snapdc", 0UL ) ];
   fd_topo_tile_t * snapin_tile = &topo->tiles[ fd_topo_find_tile( topo, "snapin", 0UL ) ];
+  ulong            snapwr_idx  =               fd_topo_find_tile( topo, "snapwr", 0UL );
+  fd_topo_tile_t * snapwr_tile = snapwr_idx!=ULONG_MAX ? &topo->tiles[ snapwr_idx ] : NULL;
 
   ulong volatile * const snapct_metrics = fd_metrics_tile( snapct_tile->metrics );
   ulong volatile * const snapld_metrics = fd_metrics_tile( snapld_tile->metrics );
   ulong volatile * const snapdc_metrics = fd_metrics_tile( snapdc_tile->metrics );
   ulong volatile * const snapin_metrics = fd_metrics_tile( snapin_tile->metrics );
+  ulong volatile * const snapwr_metrics = snapwr_tile ? fd_metrics_tile( snapwr_tile->metrics ) : NULL;
 
   ulong total_off_old    = 0UL;
   ulong decomp_off_old   = 0UL;
+  ulong vinyl_off_old    = 0UL;
   ulong snapct_backp_old = 0UL;
   ulong snapct_wait_old  = 0UL;
   ulong snapld_backp_old = 0UL;
@@ -197,6 +194,7 @@ snapshot_load_cmd_fn( args_t *   args,
   ulong snapdc_wait_old  = 0UL;
   ulong snapin_backp_old = 0UL;
   ulong snapin_wait_old  = 0UL;
+  ulong snapwr_wait_old  = 0UL;
   ulong acc_cnt_old      = 0UL;
   sleep( 1 );
   puts( "" );
@@ -207,7 +205,11 @@ snapshot_load_cmd_fn( args_t *   args,
   puts( "- stall: Waiting on upstream tile"         );
   puts( "- acc:   Number of accounts"               );
   puts( "" );
-  puts( "--------------------------------------------[ct],[ld],[dc],[in]--------[ct],[ld],[dc],[in]--------------" );
+  fputs( "--------------------------------------------", stdout );
+  if( snapwr_tile ) fputs( "--------------", stdout );
+  fputs( "[ct],[ld],[dc],[in]--------[ct],[ld],[dc],[in]", stdout );
+  if( snapwr_tile ) fputs( ",[wr]" , stdout );
+  puts( "--------------" );
   long next = start+1000L*1000L*1000L;
   for(;;) {
     ulong snapct_status = FD_VOLATILE_CONST( snapct_metrics[ MIDX( GAUGE, TILE, STATUS ) ] );
@@ -228,6 +230,7 @@ snapshot_load_cmd_fn( args_t *   args,
                          snapct_metrics[ MIDX( GAUGE, SNAPCT, INCREMENTAL_BYTES_READ ) ];
     ulong decomp_off   = snapdc_metrics[ MIDX( GAUGE, SNAPDC, FULL_DECOMPRESSED_BYTES_READ ) ] +
                          snapdc_metrics[ MIDX( GAUGE, SNAPDC, INCREMENTAL_DECOMPRESSED_BYTES_READ ) ];
+    ulong vinyl_off    = snapwr_tile ? snapwr_metrics[ MIDX( GAUGE, SNAPWR, VINYL_BYTES_WRITTEN ) ] : 0UL;
     ulong snapct_backp = snapct_metrics[ MIDX( COUNTER, TILE, REGIME_DURATION_NANOS_BACKPRESSURE_PREFRAG ) ];
     ulong snapct_wait  = snapct_metrics[ MIDX( COUNTER, TILE, REGIME_DURATION_NANOS_CAUGHT_UP_POSTFRAG   ) ] + snapct_backp;
     ulong snapld_backp = snapld_metrics[ MIDX( COUNTER, TILE, REGIME_DURATION_NANOS_BACKPRESSURE_PREFRAG ) ];
@@ -236,26 +239,42 @@ snapshot_load_cmd_fn( args_t *   args,
     ulong snapdc_wait  = snapdc_metrics[ MIDX( COUNTER, TILE, REGIME_DURATION_NANOS_CAUGHT_UP_POSTFRAG   ) ] + snapdc_backp;
     ulong snapin_backp = snapin_metrics[ MIDX( COUNTER, TILE, REGIME_DURATION_NANOS_BACKPRESSURE_PREFRAG ) ];
     ulong snapin_wait  = snapin_metrics[ MIDX( COUNTER, TILE, REGIME_DURATION_NANOS_CAUGHT_UP_POSTFRAG   ) ] + snapin_backp;
+    ulong snapwr_wait  = 0UL;
+    if( snapwr_tile ) {
+      snapwr_wait      = snapwr_metrics[ MIDX( COUNTER, TILE, REGIME_DURATION_NANOS_CAUGHT_UP_POSTFRAG   ) ] +
+                         snapwr_metrics[ MIDX( COUNTER, TILE, REGIME_DURATION_NANOS_BACKPRESSURE_PREFRAG ) ];
+    }
 
     double progress = 100.0 * (double)snapct_metrics[ MIDX( GAUGE, SNAPCT, FULL_BYTES_READ ) ] / (double)snapct_metrics[ MIDX( GAUGE, SNAPCT, FULL_BYTES_TOTAL ) ];
 
     ulong acc_cnt      = snapin_metrics[ MIDX( GAUGE, SNAPIN, ACCOUNTS_INSERTED    ) ];
-    printf( "%5.1f %% comp=%4.0f MB/s raw=%4.0f MB/s backp=(%3.0f%%,%3.0f%%,%3.0f%%,%3.0f%%) busy=(%3.0f%%,%3.0f%%,%3.0f%%,%3.0f%%) acc=%4.1f M/s\n",
+    printf( "%5.1f %% comp=%4.0fMB/s snap=%4.0fMB/s",
             progress,
             (double)( total_off -total_off_old  )/1e6,
-            (double)( decomp_off-decomp_off_old )/1e6,
+            (double)( decomp_off-decomp_off_old )/1e6 );
+    if( snapwr_tile ) {
+      printf( " vinyl=%4.0fMB/s", (double)( vinyl_off - vinyl_off_old )/1e6 );
+    }
+    printf( " backp=(%3.0f%%,%3.0f%%,%3.0f%%,%3.0f%%",
             ( (double)( snapct_backp-snapct_backp_old )*ns_per_tick )/1e7,
             ( (double)( snapld_backp-snapld_backp_old )*ns_per_tick )/1e7,
             ( (double)( snapdc_backp-snapdc_backp_old )*ns_per_tick )/1e7,
-            ( (double)( snapin_backp-snapin_backp_old )*ns_per_tick )/1e7,
+            ( (double)( snapin_backp-snapin_backp_old )*ns_per_tick )/1e7 );
+    printf( ") busy=(%3.0f%%,%3.0f%%,%3.0f%%,%3.0f%%",
             100-( ( (double)( snapct_wait-snapct_wait_old  )*ns_per_tick )/1e7 ),
             100-( ( (double)( snapld_wait-snapld_wait_old  )*ns_per_tick )/1e7 ),
             100-( ( (double)( snapdc_wait-snapdc_wait_old  )*ns_per_tick )/1e7 ),
-            100-( ( (double)( snapin_wait-snapin_wait_old  )*ns_per_tick )/1e7 ),
+            100-( ( (double)( snapin_wait-snapin_wait_old  )*ns_per_tick )/1e7 ) );
+    if( snapwr_tile ) {
+      printf( ",%3.0f%%",
+            100-( ( (double)( snapwr_wait-snapwr_wait_old  )*ns_per_tick )/1e7 ) );
+    }
+    printf( ") acc=%4.1f M/s\n",
             (double)( acc_cnt-acc_cnt_old  )/1e6 );
     fflush( stdout );
     total_off_old    = total_off;
     decomp_off_old   = decomp_off;
+    vinyl_off_old    = vinyl_off;
     snapct_backp_old = snapct_backp;
     snapct_wait_old  = snapct_wait;
     snapld_backp_old = snapld_backp;
@@ -264,6 +283,7 @@ snapshot_load_cmd_fn( args_t *   args,
     snapdc_wait_old  = snapdc_wait;
     snapin_backp_old = snapin_backp;
     snapin_wait_old  = snapin_wait;
+    snapwr_wait_old  = snapwr_wait;
     acc_cnt_old      = acc_cnt;
 
     next+=1000L*1000L*1000L;
@@ -272,7 +292,7 @@ snapshot_load_cmd_fn( args_t *   args,
 
 action_t fd_action_snapshot_load = {
   .name = NAME,
-  .args = snapshot_load_cmd_args,
+  .topo = snapshot_load_topo,
   .perm = dev_cmd_perm,
   .fn   = snapshot_load_cmd_fn
 };

src/app/firedancer-dev/main.c

@@ -50,6 +50,8 @@ fd_topo_obj_callbacks_t * CALLBACKS[] = {
   NULL,
 };
 
+extern configure_stage_t fd_cfg_stage_vinyl;
+
 configure_stage_t * STAGES[] = {
   &fd_cfg_stage_kill,
   &fd_cfg_stage_netns,
@@ -61,6 +63,7 @@ configure_stage_t * STAGES[] = {
   &fd_cfg_stage_keys,
   &fd_cfg_stage_genesis,
   &fd_cfg_stage_snapshots,
+  &fd_cfg_stage_vinyl,
   NULL,
 };
 
@@ -108,6 +111,7 @@ extern fd_topo_run_tile_t fd_tile_snapct;
 extern fd_topo_run_tile_t fd_tile_snapld;
 extern fd_topo_run_tile_t fd_tile_snapdc;
 extern fd_topo_run_tile_t fd_tile_snapin;
+extern fd_topo_run_tile_t fd_tile_snapwr;
 
 fd_topo_run_tile_t * TILES[] = {
   &fd_tile_net,
@@ -154,6 +158,7 @@ fd_topo_run_tile_t * TILES[] = {
   &fd_tile_snapld,
   &fd_tile_snapdc,
   &fd_tile_snapin,
+  &fd_tile_snapwr,
   &fd_tile_genesi,
   &fd_tile_ipecho,
   NULL,

src/app/firedancer/main.c

@@ -90,6 +90,7 @@ extern fd_topo_run_tile_t fd_tile_snapct;
 extern fd_topo_run_tile_t fd_tile_snapld;
 extern fd_topo_run_tile_t fd_tile_snapdc;
 extern fd_topo_run_tile_t fd_tile_snapin;
+extern fd_topo_run_tile_t fd_tile_snapwr;
 
 fd_topo_run_tile_t * TILES[] = {
   &fd_tile_net,
@@ -121,6 +122,7 @@ fd_topo_run_tile_t * TILES[] = {
   &fd_tile_snapld,
   &fd_tile_snapdc,
   &fd_tile_snapin,
+  &fd_tile_snapwr,
   &fd_tile_genesi,
   &fd_tile_ipecho,
   NULL,