Skip to content

Commit 8af7620

Browse files
cali-jumptradingcali-jumptrading
committed
snapshots: highest manifest slot
1 parent 6130a9e commit 8af7620

File tree

12 files changed

+214
-99
lines changed

12 files changed

+214
-99
lines changed

‎src/discof/replay/fd_replay_tile.c

Lines changed: 16 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -792,7 +792,9 @@ on_snapshot_message( fd_replay_tile_ctx_t * ctx,
792792
fd_stem_context_t * stem,
793793
ulong in_idx,
794794
ulong chunk,
795-
ulong sig ) {
795+
ulong sig,
796+
ulong tsorig FD_PARAM_UNUSED,
797+
ulong tspub FD_PARAM_UNUSED) {
796798
ulong msg = fd_ssmsg_sig_message( sig );
797799
if( FD_LIKELY( msg==FD_SSMSG_DONE ) ) {
798800
/* An end of message notification indicates the snapshot is loaded.
@@ -801,19 +803,6 @@ on_snapshot_message( fd_replay_tile_ctx_t * ctx,
801803
state machine and set the state here accordingly. */
802804
FD_LOG_INFO(("Snapshot loaded, replay can start executing"));
803805
ctx->snapshot_init_done = 1;
804-
/* Kickoff repair orphans after the snapshots are done loading. If
805-
we kickoff repair after we receive a full manifest, we might try
806-
to repair a slot that is potentially huge amount of slots behind
807-
turbine causing our repair buffers to fill up. Instead, we should
808-
wait until we are done receiving all the snapshots.
809-
810-
TODO: Eventually, this logic should be cased out more:
811-
1. If we just have a full snapshot, load in the slot_ctx for the
812-
slot ctx and kickoff repair as soon as the manifest is
813-
received.
814-
2. If we are loading a full and incremental snapshot, we should
815-
only load in the slot_ctx and kickoff repair for the
816-
incremental snapshot. */
817806
kickoff_repair_orphans( ctx, stem );
818807
init_from_snapshot( ctx, stem );
819808
ulong curr_slot = fd_bank_slot_get( ctx->slot_ctx->bank );
@@ -824,6 +813,16 @@ on_snapshot_message( fd_replay_tile_ctx_t * ctx,
824813
}
825814

826815
switch( msg ) {
816+
case FD_SSMSG_HIGHEST_MANIFEST_SLOT: {
817+
/* snapin sends the highest manifest slot so far to notify any
818+
listeners of the highest rooted slot that will be loaded
819+
by the snapshot tiles. The highest manifest slot may change
820+
if the snapshot tiles retry snapshot loading, but it is
821+
guaranteed to be monotonically increasing. */
822+
/* TODO: currently a no-op. Repair cannot handle an receiving an
823+
incrementally changing rooted slot and stake weights yet. */
824+
break;
825+
}
827826
case FD_SSMSG_MANIFEST_FULL:
828827
case FD_SSMSG_MANIFEST_INCREMENTAL: {
829828
/* We may either receive a full snapshot manifest or an
@@ -944,8 +943,8 @@ after_frag( fd_replay_tile_ctx_t * ctx,
944943
ulong seq FD_PARAM_UNUSED,
945944
ulong sig,
946945
ulong sz FD_PARAM_UNUSED,
947-
ulong tsorig FD_PARAM_UNUSED,
948-
ulong tspub FD_PARAM_UNUSED,
946+
ulong tsorig,
947+
ulong tspub,
949948
fd_stem_context_t * stem FD_PARAM_UNUSED ) {
950949
if( FD_UNLIKELY( ctx->in_kind[ in_idx ]==IN_KIND_ROOT ) ) {
951950
ulong root = sig;
@@ -974,7 +973,7 @@ after_frag( fd_replay_tile_ctx_t * ctx,
974973

975974
fd_fseq_update( ctx->published_wmark, root );
976975
} else if( FD_UNLIKELY( ctx->in_kind[ in_idx ]==IN_KIND_SNAP ) ) {
977-
on_snapshot_message( ctx, stem, in_idx, ctx->_snap_out_chunk, sig );
976+
on_snapshot_message( ctx, stem, in_idx, ctx->_snap_out_chunk, sig, tsorig, tspub );
978977
} else if( FD_LIKELY( ctx->in_kind[ in_idx ]==IN_KIND_REPAIR ) ) {
979978

980979
/* Forks form a partial ordering over FEC sets. The Repair tile

‎src/discof/restore/fd_snapdc_tile.c

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -101,9 +101,11 @@ transition_malformed( fd_snapdc_tile_t * ctx,
101101
static inline void
102102
handle_control_frag( fd_snapdc_tile_t * ctx,
103103
fd_stem_context_t * stem,
104-
ulong sig ) {
104+
ulong sig,
105+
ulong tsorig,
106+
ulong tspub ) {
105107
/* 1. Pass the control message downstream to the next consumer. */
106-
fd_stem_publish( stem, 0UL, sig, ctx->out.chunk, 0UL, 0UL, 0UL, 0UL );
108+
fd_stem_publish( stem, 0UL, sig, ctx->out.chunk, 0UL, 0UL, tsorig, tspub );
107109
ulong error = ZSTD_DCtx_reset( ctx->zstd, ZSTD_reset_session_only );
108110
if( FD_UNLIKELY( ZSTD_isError( error ) ) ) FD_LOG_ERR(( "ZSTD_DCtx_reset failed (%lu-%s)", error, ZSTD_getErrorName( error ) ));
109111

@@ -148,6 +150,9 @@ handle_control_frag( fd_snapdc_tile_t * ctx,
148150
FD_TEST( ctx->state==FD_SNAPDC_STATE_DONE );
149151
ctx->state = FD_SNAPDC_STATE_SHUTDOWN;
150152
break;
153+
case FD_SNAPSHOT_MSG_HIGHEST_MANIFEST_SLOT:
154+
/* Informational control message forwarded to snapin */
155+
break;
151156
default:
152157
FD_LOG_ERR(( "unexpected control sig %lu", sig ));
153158
return;
@@ -253,7 +258,7 @@ returnable_frag( fd_snapdc_tile_t * ctx,
253258
FD_TEST( ctx->state!=FD_SNAPDC_STATE_SHUTDOWN );
254259

255260
if( FD_LIKELY( sig==FD_SNAPSHOT_MSG_DATA ) ) return handle_data_frag( ctx, stem, chunk, sz );
256-
else handle_control_frag( ctx,stem, sig );
261+
else handle_control_frag( ctx,stem, sig, tsorig, tspub );
257262

258263
return 0;
259264
}

‎src/discof/restore/fd_snapin_tile.c

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -214,7 +214,9 @@ handle_data_frag( fd_snapin_tile_t * ctx,
214214
static void
215215
handle_control_frag( fd_snapin_tile_t * ctx,
216216
fd_stem_context_t * stem,
217-
ulong sig ) {
217+
ulong sig,
218+
ulong tsorig,
219+
ulong tspub ) {
218220
switch( sig ) {
219221
case FD_SNAPSHOT_MSG_CTRL_RESET_FULL:
220222
ctx->full = 1;
@@ -257,6 +259,11 @@ handle_control_frag( fd_snapin_tile_t * ctx,
257259
case FD_SNAPSHOT_MSG_CTRL_SHUTDOWN:
258260
ctx->state = FD_SNAPIN_STATE_SHUTDOWN;
259261
break;
262+
case FD_SNAPSHOT_MSG_HIGHEST_MANIFEST_SLOT:
263+
/* Informational control message forwarded to snap_out */
264+
fd_stem_publish( stem, 0UL, FD_SSMSG_HIGHEST_MANIFEST_SLOT, 0UL, 0UL, 0UL, tsorig, tspub );
265+
/* No need to send an ack for an informational control message */
266+
return;
260267
default:
261268
FD_LOG_ERR(( "unexpected control sig %lu", sig ));
262269
return;
@@ -288,7 +295,7 @@ returnable_frag( fd_snapin_tile_t * ctx,
288295
FD_TEST( ctx->state!=FD_SNAPIN_STATE_SHUTDOWN );
289296

290297
if( FD_UNLIKELY( sig==FD_SNAPSHOT_MSG_DATA ) ) handle_data_frag( ctx, chunk, sz, stem );
291-
else handle_control_frag( ctx, stem, sig );
298+
else handle_control_frag( ctx, stem, sig, tsorig, tspub );
292299

293300
return 0;
294301
}

‎src/discof/restore/fd_snaprd_tile.c

Lines changed: 37 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22
#include "utils/fd_sshttp.h"
33
#include "utils/fd_ssctrl.h"
44
#include "utils/fd_ssarchive.h"
5+
#include "utils/fd_ssmsg.h"
56

67
#include "../../disco/topo/fd_topo.h"
78
#include "../../disco/metrics/fd_metrics.h"
@@ -74,10 +75,17 @@ struct fd_snaprd_tile {
7475
} local_in;
7576

7677
struct {
77-
char full_path[ PATH_MAX ];
78-
ulong full_len;
79-
char incremental_path[ PATH_MAX ];
80-
ulong incremental_len;
78+
79+
struct {
80+
char path[ PATH_MAX ];
81+
ulong len;
82+
} full;
83+
84+
struct {
85+
char path[ PATH_MAX ];
86+
ulong len;
87+
} incremental;
88+
8189
} http;
8290

8391
struct {
@@ -328,11 +336,11 @@ rename_snapshots( fd_snaprd_tile_t * ctx ) {
328336
if( FD_UNLIKELY( -1==ctx->local_out.dir_fd ) ) return;
329337

330338
if( FD_LIKELY( -1!=ctx->local_out.full_snapshot_fd ) ) {
331-
if( FD_UNLIKELY( -1==renameat( ctx->local_out.dir_fd, "snapshot.tar.bz2-partial", ctx->local_out.dir_fd, ctx->http.full_path ) ) )
339+
if( FD_UNLIKELY( -1==renameat( ctx->local_out.dir_fd, "snapshot.tar.bz2-partial", ctx->local_out.dir_fd, ctx->http.full.path ) ) )
332340
FD_LOG_ERR(( "renameat() failed (%i-%s)", errno, fd_io_strerror( errno ) ));
333341
}
334342
if( FD_LIKELY( -1!=ctx->local_out.incremental_snapshot_fd ) ) {
335-
if( FD_UNLIKELY( -1==renameat( ctx->local_out.dir_fd, "incremental-snapshot.tar.bz2-partial", ctx->local_out.dir_fd, ctx->http.incremental_path ) ) )
343+
if( FD_UNLIKELY( -1==renameat( ctx->local_out.dir_fd, "incremental-snapshot.tar.bz2-partial", ctx->local_out.dir_fd, ctx->http.incremental.path ) ) )
336344
FD_LOG_ERR(( "renameat() failed (%i-%s)", errno, fd_io_strerror( errno ) ));
337345
}
338346
}
@@ -384,17 +392,29 @@ after_credit( fd_snaprd_tile_t * ctx,
384392
} else {
385393
char path[ PATH_MAX ];
386394
char encoded_full_hash[ FD_BASE58_ENCODED_32_SZ ];
387-
char encoded_incremental_hash[ FD_BASE58_ENCODED_32_SZ ];
395+
396+
/* Generate the http paths */
388397
fd_base58_encode_32( best.snapshot_info->full.hash, NULL, encoded_full_hash );
389-
fd_base58_encode_32( best.snapshot_info->incremental.hash, NULL, encoded_incremental_hash );
390-
FD_TEST( fd_cstr_printf_check( ctx->http.full_path, PATH_MAX, &ctx->http.full_len, "snapshot-%lu-%s.tar.zst", best.snapshot_info->full.slot, encoded_full_hash ) );
391-
FD_TEST( fd_cstr_printf_check( ctx->http.incremental_path, PATH_MAX, &ctx->http.incremental_len, "incremental-snapshot-%lu-%lu-%s.tar.zst", best.snapshot_info->incremental.base_slot, best.snapshot_info->incremental.slot, encoded_incremental_hash ) );
392-
FD_TEST( fd_cstr_printf_check( path, PATH_MAX, NULL, "/%s", ctx->http.full_path ) );
398+
FD_TEST( fd_cstr_printf_check( ctx->http.full.path, PATH_MAX, &ctx->http.full.len, "snapshot-%lu-%s.tar.zst", best.snapshot_info->full.slot, encoded_full_hash ) );
399+
FD_TEST( fd_cstr_printf_check( path, PATH_MAX, NULL, "/%s", ctx->http.full.path ) );
400+
401+
if( ctx->config.incremental_snapshot_fetch ) {
402+
char encoded_incremental_hash[ FD_BASE58_ENCODED_32_SZ ];
403+
fd_base58_encode_32( best.snapshot_info->incremental.hash, NULL, encoded_incremental_hash );
404+
FD_TEST( fd_cstr_printf_check( ctx->http.incremental.path, PATH_MAX, &ctx->http.incremental.len, "incremental-snapshot-%lu-%lu-%s.tar.zst", best.snapshot_info->incremental.base_slot, best.snapshot_info->incremental.slot, encoded_incremental_hash ) );
405+
}
406+
407+
uint low;
408+
uint high;
409+
/* send the highest manifest slot */
410+
ulong highest_manifest_slot = ctx->config.incremental_snapshot_fetch ? best.snapshot_info->incremental.slot : best.snapshot_info->full.slot;
411+
fd_ssmsg_slot_to_frag( highest_manifest_slot, &low, &high );
412+
fd_stem_publish( stem, 0UL, FD_SNAPSHOT_MSG_HIGHEST_MANIFEST_SLOT, 0UL, 0UL, 0UL, low, high );
393413

394-
FD_LOG_NOTICE(( "downloading full snapshot from http://" FD_IP4_ADDR_FMT ":%hu/%s", FD_IP4_ADDR_FMT_ARGS( best.addr.addr ), best.addr.port, ctx->http.full_path ));
414+
FD_LOG_NOTICE(( "downloading full snapshot from http://" FD_IP4_ADDR_FMT ":%hu/%s", FD_IP4_ADDR_FMT_ARGS( best.addr.addr ), best.addr.port, ctx->http.full.path ));
395415
ctx->peer = best;
396416
ctx->state = FD_SNAPRD_STATE_READING_FULL_HTTP;
397-
fd_sshttp_init( ctx->sshttp, best.addr, path, ctx->http.full_len + 1UL, now );
417+
fd_sshttp_init( ctx->sshttp, best.addr, path, ctx->http.full.len + 1UL, now );
398418
}
399419
break;
400420
}
@@ -459,9 +479,9 @@ after_credit( fd_snaprd_tile_t * ctx,
459479
}
460480

461481
char path[ PATH_MAX ];
462-
FD_TEST( fd_cstr_printf_check( path, PATH_MAX, NULL, "/%s", ctx->http.incremental_path ) );
482+
FD_TEST( fd_cstr_printf_check( path, PATH_MAX, NULL, "/%s", ctx->http.incremental.path ) );
463483
FD_LOG_NOTICE(( "downloading incremental snapshot from http://" FD_IP4_ADDR_FMT ":%hu/%s", FD_IP4_ADDR_FMT_ARGS( ctx->peer.addr.addr ), ctx->peer.addr.port, path));
464-
fd_sshttp_init( ctx->sshttp, ctx->peer.addr, path, ctx->http.incremental_len + 1UL, fd_log_wallclock() );
484+
fd_sshttp_init( ctx->sshttp, ctx->peer.addr, path, ctx->http.incremental.len + 1UL, fd_log_wallclock() );
465485
ctx->state = FD_SNAPRD_STATE_READING_INCREMENTAL_HTTP;
466486
break;
467487
case FD_SNAPRD_STATE_FLUSHING_FULL_HTTP_RESET:
@@ -691,6 +711,8 @@ unprivileged_init( fd_topo_t * topo,
691711
ctx->out.wmark = fd_dcache_compact_wmark ( ctx->out.wksp, topo->links[ tile->out_link_id[ 0 ] ].dcache, topo->links[ tile->out_link_id[ 0 ] ].mtu );
692712
ctx->out.chunk = ctx->out.chunk0;
693713
ctx->out.mtu = topo->links[ tile->out_link_id[ 0 ] ].mtu;
714+
715+
fd_memset( &ctx->peer, 0, sizeof(ctx->peer) );
694716
}
695717

696718
#define STEM_BURST 2UL /* One control message, and one data message */

‎src/discof/restore/utils/fd_ssctrl.h

Lines changed: 18 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -40,13 +40,23 @@
4040

4141
#define FD_SNAPSHOT_MSG_DATA (0UL) /* Fragment represents some snapshot data */
4242

43-
#define FD_SNAPSHOT_MSG_CTRL_RESET_FULL (1UL) /* Reset to start loading a fresh full snapshot */
44-
#define FD_SNAPSHOT_MSG_CTRL_EOF_FULL (2UL) /* Full snapshot data is done, incremental data starting now */
45-
#define FD_SNAPSHOT_MSG_CTRL_RESET_INCREMENTAL (3UL) /* Incremental data being retried, start incremental over */
46-
#define FD_SNAPSHOT_MSG_CTRL_DONE (4UL) /* Snapshot load is over, data is finished for this tile */
47-
#define FD_SNAPSHOT_MSG_CTRL_SHUTDOWN (5UL) /* All tiles have acknowledged snapshot load is done, can now shutdown */
48-
49-
#define FD_SNAPSHOT_MSG_CTRL_ACK (6UL) /* Sent from tiles back to snaprd, meaning they ACK whatever control message was pending */
50-
#define FD_SNAPSHOT_MSG_CTRL_MALFORMED (7UL) /* Sent from tiles back to snaprd, meaning they consider the current snapshot malformed */
43+
/* The HIGHEST_MANIFEST_SLOT message is an informational control message
44+
that is forwarded from the snaprd tile to the snapin tile, which
45+
forwards the message to the snap_out link. The HIGHEST_MANIFEST_SLOT
46+
message contains the highest manifest slot so far. It is typically
47+
the incremental snapshot slot but can be the full snapshot slot if
48+
incremental snapshots are disabled. It is guaranteed to be
49+
monotonically increasing and is forwarded regardless of any snapshot
50+
loading error / retry. */
51+
#define FD_SNAPSHOT_MSG_HIGHEST_MANIFEST_SLOT (1UL) /* Fragment contains the highest manifest slot so far, guaranteed to be monotonically increasing */
52+
53+
#define FD_SNAPSHOT_MSG_CTRL_RESET_FULL (2UL) /* Reset to start loading a fresh full snapshot */
54+
#define FD_SNAPSHOT_MSG_CTRL_EOF_FULL (3UL) /* Full snapshot data is done, incremental data starting now */
55+
#define FD_SNAPSHOT_MSG_CTRL_RESET_INCREMENTAL (4UL) /* Incremental data being retried, start incremental over */
56+
#define FD_SNAPSHOT_MSG_CTRL_DONE (5UL) /* Snapshot load is over, data is finished for this tile */
57+
#define FD_SNAPSHOT_MSG_CTRL_SHUTDOWN (6UL) /* All tiles have acknowledged snapshot load is done, can now shutdown */
58+
59+
#define FD_SNAPSHOT_MSG_CTRL_ACK (7UL) /* Sent from tiles back to snaprd, meaning they ACK whatever control message was pending */
60+
#define FD_SNAPSHOT_MSG_CTRL_MALFORMED (8UL) /* Sent from tiles back to snaprd, meaning they consider the current snapshot malformed */
5161

5262
#endif /* HEADER_fd_src_discof_restore_utils_fd_ssctrl_h */

‎src/discof/restore/utils/fd_sshttp.c

Lines changed: 0 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -36,9 +36,6 @@ struct fd_sshttp_private {
3636
ulong response_len;
3737
char response[ USHORT_MAX ];
3838

39-
char full_snapshot_name[ PATH_MAX ];
40-
char incremental_snapshot_name[ PATH_MAX ];
41-
4239
ulong content_len;
4340

4441
ulong magic;
@@ -292,14 +289,6 @@ read_body( fd_sshttp_t * http,
292289
return FD_SSHTTP_ADVANCE_DATA;
293290
}
294291

295-
void
296-
fd_sshttp_snapshot_names( fd_sshttp_t * http,
297-
char const ** full_snapshot_name,
298-
char const ** incremental_snapshot_name ) {
299-
*full_snapshot_name = http->full_snapshot_name;
300-
*incremental_snapshot_name = http->incremental_snapshot_name;
301-
}
302-
303292
int
304293
fd_sshttp_advance( fd_sshttp_t * http,
305294
ulong * data_len,

‎src/discof/restore/utils/fd_sshttp.h

Lines changed: 0 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -24,12 +24,6 @@ fd_sshttp_new( void * shmem );
2424
fd_sshttp_t *
2525
fd_sshttp_join( void * sshttp );
2626

27-
/* Sets points to snapshot names */
28-
void
29-
fd_sshttp_snapshot_names( fd_sshttp_t * http,
30-
char const ** full_snapshot_name,
31-
char const ** incremental_snapshot_name );
32-
3327
void
3428
fd_sshttp_init( fd_sshttp_t * http,
3529
fd_ip4_port_t addr,

‎src/discof/restore/utils/fd_ssmsg.h

Lines changed: 22 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -3,9 +3,10 @@
33

44
#include "../../../flamenco/types/fd_types.h"
55

6-
#define FD_SSMSG_MANIFEST_FULL (0) /* A snapshot manifest message from the full snapshot */
7-
#define FD_SSMSG_MANIFEST_INCREMENTAL (1) /* A snapshot manifest message from the incremental snapshot */
8-
#define FD_SSMSG_DONE (2) /* Indicates the snapshot is fully loaded and tiles are shutting down */
6+
#define FD_SSMSG_MANIFEST_FULL (0) /* A snapshot manifest message from the full snapshot */
7+
#define FD_SSMSG_MANIFEST_INCREMENTAL (1) /* A snapshot manifest message from the incremental snapshot */
8+
#define FD_SSMSG_HIGHEST_MANIFEST_SLOT (2) /* The highest manifest slot so far, guaranteed to be monotonically increasing */
9+
#define FD_SSMSG_DONE (3) /* Indicates the snapshot is fully loaded and tiles are shutting down */
910

1011
/* This number is exhagerately high, but is consisent with Frankendancer.
1112
We need this to be about the number of validators, so roughly
@@ -24,6 +25,24 @@ fd_ssmsg_sig( ulong message,
2425
FD_FN_CONST static inline ulong fd_ssmsg_sig_manifest_size( ulong sig ) { return (sig >> 2); }
2526
FD_FN_CONST static inline ulong fd_ssmsg_sig_message( ulong sig ) { return (sig & 0x3UL); }
2627

28+
/* The FD_SSMSG_HIGHEST_MANIFEST_SLOT uses the tsorig and tspub fields
29+
of the fd_frag_meta_t struct to store the low and high 32 bits of
30+
the slot number. */
31+
static inline void
32+
fd_ssmsg_slot_to_frag( ulong slot,
33+
uint* low,
34+
uint* high ) {
35+
*low = (uint)(slot & 0xFFFFFFFFUL);
36+
*high = (uint)((slot >> 32UL) & 0xFFFFFFFFUL);
37+
}
38+
39+
static inline void
40+
fd_ssmsg_frag_to_slot( ulong low,
41+
ulong high,
42+
ulong* slot ) {
43+
*slot = (high << 32UL) | low;
44+
}
45+
2746
struct fd_snapshot_manifest_vote_account {
2847
/* The pubkey of the vote account */
2948
uchar vote_account_pubkey[ 32UL ];

0 commit comments

Comments
 (0)