Line data Source code
1 : /* The snapmk tile coordinates snapshot production.
2 :
3 : snapmk discovers accounts (zero copy) and generates compression jobs
4 : for downstream snapzp tiles (load balanced).
5 :
6 : snapmk uses snaprd as a worker thread. The snapmk's snaprd_out fseq
7 : has an application region, which indicates when snaprd should spring
8 : into action. Once activated, it reads all accdb partitions into a
9 : ring buffer (mcache/dcache). snaprd is a separate worker thread
10 : because otherwise, snapmk would be blocked on synchronous I/O and
11 : waste time copying data.
12 :
13 : Finally, snapmk interfaces with the replay tile. The replay tile
14 : instructs when to produce a snapshot, snapmk takes ownership of the
15 : pipeline / state machine, and notifies the replay tile when snap
16 : production is done.
17 :
18 : ### File management
19 :
20 : All snap producer tile processes (or threads) in a Firedancer
21 : instance share a fixed pool of file descriptors for snapshot file
22 : descriptors.
23 :
24 : These FDs are visible under proper snapshot file names
25 : (e.g. "snapshot-433132190-8cSwtWwmkj3oaXPsyUxhL1QkuaAVY1S4vRhqL6EJQ99Z.tar.zst")
26 : or placeholders (e.g. ".snapshot-x67.partial"). Old snapshots
27 : eventually get recycled.
28 :
29 : The snapmk tile manages these file descriptors (decides which snaps
30 : to select for new creations, which ones to recycle, etc).
31 :
32 : ### Zero copy snaprd->mk->zp path
33 :
34 : The snaprd tile reads accdb data into memory at high bandwidth.
35 : The snapmk tile does streaming parsing of this input data, and then
36 : distributes compression jobs to downstream snapzp tiles. Those
37 : compression jobs are mere pointers to the snaprd data. Thus, snaprd
38 : must not only backpressure on slow snapmk but slow snapzps too.
39 :
40 : The snapmk tile achieves this by tracking the snapzp job -> snaprd
41 : frag seq in rd_shadow. */
42 :
43 : #define _GNU_SOURCE
44 : #define ZSTD_STATIC_LINKING_ONLY
45 : #include <zstd.h>
46 : #include <errno.h>
47 : #include <fcntl.h>
48 : #include <stdio.h>
49 : #include <linux/futex.h>
50 : #include <sys/syscall.h>
51 : #include <sys/types.h>
52 : #include <sys/stat.h>
53 : #include <unistd.h>
54 : #include <stdatomic.h>
55 :
56 : #include "fd_snapmk_tile.h"
57 :
58 : #include "fd_backup.h"
59 : #include "fd_snap_pool.h"
60 : #include "fd_backup_cache.h"
61 : #include "fd_backup_disk.h"
62 : #include "fd_backup_shmem.h"
63 : #include "fd_ssmanifest_writer.h"
64 : #include "fd_txncache_writer.h"
65 : #include "../fd_startup.h"
66 : #include "../replay/fd_replay_tile.h"
67 : #include "../restore/utils/fd_ssarchive.h"
68 : #include "../../disco/metrics/fd_metrics.h"
69 : #include "../../disco/stem/fd_stem.h"
70 : #include "../../disco/topo/fd_topo.h"
71 : #include "../../disco/events/generated/fd_event_gen.h"
72 : #include "../../flamenco/runtime/fd_system_ids.h"
73 : #include "../../tango/fseq/fd_fseq.h"
74 :
75 : #include <time.h> /* CLOCK_REALTIME */
76 : #include "generated/fd_snapmk_tile_seccomp.h"
77 :
78 : /* Raw state values exported by the metrics (states= enums) */
79 : FD_STATIC_ASSERT( SNAPMK_STATE_IDLE==FD_METRICS_ENUM_SNAPMK_JOB_STATE_V_IDLE_IDX, metrics_state );
80 : FD_STATIC_ASSERT( SNAPMK_STATE_START==FD_METRICS_ENUM_SNAPMK_JOB_STATE_V_START_IDX, metrics_state );
81 : FD_STATIC_ASSERT( SNAPMK_STATE_TAR_HEADERS==FD_METRICS_ENUM_SNAPMK_JOB_STATE_V_TAR_HEADERS_IDX, metrics_state );
82 : FD_STATIC_ASSERT( SNAPMK_STATE_MANIFEST==FD_METRICS_ENUM_SNAPMK_JOB_STATE_V_MANIFEST_IDX, metrics_state );
83 : FD_STATIC_ASSERT( SNAPMK_STATE_ACCDB_CACHE==FD_METRICS_ENUM_SNAPMK_JOB_STATE_V_ACCDB_CACHE_IDX, metrics_state );
84 : FD_STATIC_ASSERT( SNAPMK_STATE_ACCDB_CACHE_FLUSH==FD_METRICS_ENUM_SNAPMK_JOB_STATE_V_ACCDB_CACHE_FLUSH_IDX, metrics_state );
85 : FD_STATIC_ASSERT( SNAPMK_STATE_ACCDB_CACHE_FINISH==FD_METRICS_ENUM_SNAPMK_JOB_STATE_V_ACCDB_CACHE_FINISH_IDX, metrics_state );
86 : FD_STATIC_ASSERT( SNAPMK_STATE_ACCDB_DISK==FD_METRICS_ENUM_SNAPMK_JOB_STATE_V_ACCDB_DISK_IDX, metrics_state );
87 : FD_STATIC_ASSERT( SNAPMK_STATE_ACCDB_DISK_FLUSH==FD_METRICS_ENUM_SNAPMK_JOB_STATE_V_ACCDB_DISK_FLUSH_IDX, metrics_state );
88 : FD_STATIC_ASSERT( SNAPMK_STATE_ACCDB_DISK_FINISH==FD_METRICS_ENUM_SNAPMK_JOB_STATE_V_ACCDB_DISK_FINISH_IDX, metrics_state );
89 : FD_STATIC_ASSERT( SNAPMK_STATE_ACCDB_DELTA==FD_METRICS_ENUM_SNAPMK_JOB_STATE_V_ACCDB_DELTA_IDX, metrics_state );
90 : FD_STATIC_ASSERT( SNAPMK_STATE_ACCDB_DELTA_FLUSH==FD_METRICS_ENUM_SNAPMK_JOB_STATE_V_ACCDB_DELTA_FLUSH_IDX, metrics_state );
91 : FD_STATIC_ASSERT( SNAPMK_STATE_ACCDB_DELTA_FINISH==FD_METRICS_ENUM_SNAPMK_JOB_STATE_V_ACCDB_DELTA_FINISH_IDX, metrics_state );
92 : FD_STATIC_ASSERT( SNAPMK_STATE_STATUS_CACHE==FD_METRICS_ENUM_SNAPMK_JOB_STATE_V_STATUS_CACHE_IDX, metrics_state );
93 : FD_STATIC_ASSERT( SNAPMK_STATE_EOF_MARKER==FD_METRICS_ENUM_SNAPMK_JOB_STATE_V_EOF_MARKER_IDX, metrics_state );
94 : FD_STATIC_ASSERT( SNAPMK_STATE_DONE==FD_METRICS_ENUM_SNAPMK_JOB_STATE_V_DONE_IDX, metrics_state );
95 : FD_STATIC_ASSERT( SNAPMK_STATE_FAIL==FD_METRICS_ENUM_SNAPMK_JOB_STATE_V_FAIL_IDX, metrics_state );
96 : FD_STATIC_ASSERT( SNAPMK_STATE_SLEEP==FD_METRICS_ENUM_SNAPMK_JOB_STATE_V_SLEEP_IDX, metrics_state );
97 : FD_STATIC_ASSERT( SNAPMK_STATE_STARTUP==FD_METRICS_ENUM_SNAPMK_JOB_STATE_V_STARTUP_IDX, metrics_state );
98 : FD_STATIC_ASSERT( SNAPMK_STATE_STARTUP_BURST==FD_METRICS_ENUM_SNAPMK_JOB_STATE_V_STARTUP_BURST_IDX, metrics_state );
99 :
100 0 : #define RAW_BUF_SZ (32UL<<20)
101 0 : #define COMP_BUF_SZ ZSTD_COMPRESSBOUND( RAW_BUF_SZ )
102 :
103 : /* FD_SNAPMK_ZP_DEPTH must match the snapmk_zp link depth in topology.c.
104 : Asserted at init. Sizes the per-link snaprd-seq shadow rings. */
105 0 : #define FD_SNAPMK_ZP_DEPTH 1024
106 :
107 : /* Max number of reliable snapmk_out consumers */
108 : #define SNAPMK_OUT_CONS_MAX 64
109 :
110 : /* SNAPMK_STEM_BURST: number of snapmk_out frags published in one event
111 : loop cycle (snapmk_zp links are exempt) */
112 0 : #define SNAPMK_STEM_BURST 3UL
113 0 : #define SNAPMK_STEM_LAZY 8700UL
114 :
115 : /* power saving (sleeping) */
116 : #define IDLE_THRES (16384UL) /* no of idle busy loop iters before sleeping */
117 0 : #define IDLE_SLEEP ((long)1e6) /* sleep duration (nanoseconds) */
118 :
119 : struct fd_snapmk {
120 : uint state;
121 :
122 : fd_backup_cache_t acc_cache[1];
123 : visited_set_t * visited_set;
124 :
125 : /* snapshot files */
126 :
127 : int snap_fd; /* current snapshot (-1 if idle) */
128 : int snap_dir_fd; /* dirfd to prevent hijacking */
129 : char snap_dir [ PATH_MAX ];
130 : char final_name[ FD_SNAP_NAME_MAX ];
131 : uint snap_idx; /* in pool */
132 : uint snap_max;
133 : uint snap_full_max; /* <=snap_max */
134 : fd_backup_inode_t pool [ FD_SNAP_MAX ];
135 : ulong pool_sz[ FD_SNAP_MAX ];
136 : ulong final_sz;
137 :
138 : /* snapzp worker threads */
139 :
140 : ulong zp_cnt; /* [0,zp_cnt] out links are to zp */
141 : ulong const * zp_cons_fseq[ SNAPZP_TILE_MAX ];
142 : atomic_ulong * file_off_p;
143 : atomic_ulong * appendvec_slot_ticket_p;
144 :
145 : /* snaprd worker thread */
146 :
147 : atomic_ulong * rd_fseq;
148 : ulong rd_tile_id; /* snaprd, rung when a credit return finds it parked on backpressure */
149 : atomic_ulong * rd_ctl;
150 : ulong rd_seq; /* seq of the snaprd frag last parsed */
151 : ulong rd_seq_cache; /* last watermark published to snaprd */
152 : fd_wksp_t * rd_in_mem;
153 : ulong rd_in_mtu;
154 : ulong rd_fseq_dummy;
155 :
156 : /* per zp out link shadow ring: rd_shadow[i][seq%depth] = snaprd seq
157 : referenced by the mk_zp frag published at seq on link i */
158 : ulong * rd_shadow[ SNAPZP_TILE_MAX ];
159 :
160 : struct {
161 : ulong out_idx;
162 : void * mem;
163 : ulong chunk;
164 : ulong chunk0;
165 : ulong wmark;
166 :
167 : ulong const * cons_fseq[ SNAPMK_OUT_CONS_MAX ];
168 : ulong cons_cnt;
169 : ulong * seq_prod;
170 : } out;
171 :
172 : ulong zp_rr_idx; /* round-robin cursor over zp out links */
173 : ulong zp_ready; /* bit set */
174 : ulong zp_flush_pending; /* bit set */
175 : ulong zp_barrier[ SNAPZP_TILE_MAX ];
176 :
177 : fd_banks_t * banks;
178 : fd_bank_t * bank;
179 : fd_pubkey_t leader; /* slot leader of bank */
180 : fd_txncache_t * txncache;
181 : fd_ssmanifest_writer_t manifest_writer[1];
182 : fd_txncache_writer_t txncache_writer[1];
183 :
184 : ulong manifest_pad;
185 : ulong manifest_sz;
186 :
187 : ulong status_cache_sz; /* bytes serialized so far */
188 : ulong status_cache_expected_sz; /* size the writer counted up front, written into the TAR header */
189 : void * status_cache_writer_arena;
190 : ulong status_cache_writer_arena_sz;
191 :
192 : int zstd_frame_open; /* compressor has started a frame that an e_end has not yet closed */
193 : ulong zstd_data_frame_cnt;
194 : ulong zstd_padding_sz;
195 : ulong uncompressed_sz;
196 : long start_time;
197 : long accounts_start_time;
198 : long end_time;
199 : uchar snap_hash[ 32 ];
200 : ulong snapshot_compress_ticks0;
201 : ulong snapshot_io_blocked_ticks0;
202 : ulong last_snapshot_create_slot;
203 :
204 : int incremental;
205 : ulong base_slot;
206 :
207 : struct {
208 : fd_accdb_delta_t const * pool;
209 : uint const * chain;
210 : ulong chain_cnt;
211 : ulong chain_idx;
212 : uint ele_idx;
213 : ulong ele_cnt;
214 : ulong ele_total;
215 : } delta;
216 :
217 : /* replay in link */
218 :
219 : fd_wksp_t * replay_in_mem;
220 : ulong const * replay_in_seq_prod; /* replay_snapmk producer seq */
221 : ulong replay_in_seq_cons; /* next expected replay_snapmk seq */
222 : ulong idle_iter; /* busy loop iters spent in IDLE */
223 :
224 : /* IPC */
225 : struct {
226 : void * mem;
227 : ulong chunk0;
228 : ulong wmark;
229 : ulong chunk;
230 : } zp_out[ FD_TOPO_MAX_TILE_OUT_LINKS ];
231 : fd_backup_cache_msg_t scan_batch[1];
232 : ushort in_kind[ FD_TOPO_MAX_TILE_IN_LINKS ];
233 :
234 : fd_backup_overrun_t * overrun;
235 : fd_backup_stats_t * stats;
236 : fd_snapmk_accparse_t accparse[1];
237 :
238 : /* disk batch staging (FD_BACKUP_ORIG_ACC_DISK_BATCH). A batch is
239 : staged out of the parser, then flushed to a zp tile once an output
240 : link has credit. disk_batch_pending guards against re-staging while
241 : a staged batch is awaiting credit. */
242 : fd_backup_disk_batch_msg_t disk_batch[1];
243 : ulong disk_batch_base_gaddr;
244 : int disk_batch_pending;
245 : int disk_out_idx; /* snapzp output */
246 :
247 : /* account data cache */
248 : uchar * cache [ FD_ACCDB_CACHE_CLASS_CNT ];
249 : ulong cache_max[ FD_ACCDB_CACHE_CLASS_CNT ];
250 :
251 : /* accdb shared memory */
252 : fd_accdb_t * accdb;
253 : fd_accdb_shmem_t * accdb_shmem;
254 : fd_accdb_fork_shmem_t const * accdb_shfork;
255 : fd_accdb_fork_id_t const * accdb_root_fork;
256 : ulong * accdb_snapshot_sync;
257 :
258 : fd_sleep_t * accdb_sleep;
259 : ulong accdb_tile_id;
260 :
261 : /* output buffer */
262 : ZSTD_CCtx * zst;
263 : ZSTD_inBuffer raw_buf;
264 : ZSTD_outBuffer comp_buf;
265 : uchar raw [ RAW_BUF_SZ ];
266 : uchar comp[ COMP_BUF_SZ ];
267 :
268 : /* startup related */
269 : fd_startup_gate_t startup_gate[1];
270 : ulong startup_pool_idx;
271 :
272 : struct {
273 : ulong snapshots_started[ FD_METRICS_ENUM_SNAP_TYPE_CNT ];
274 : ulong snapshots_finished[ FD_METRICS_ENUM_SNAP_TYPE_CNT ];
275 : ulong last_snapshot_slot_started_full;
276 : ulong last_snapshot_slot_started_incremental;
277 : ulong last_snapshot_slot_finished_full;
278 : ulong last_snapshot_slot_finished_incremental;
279 : ulong bytes_compressed;
280 : ulong bytes_written;
281 : ulong io_blocked_ticks;
282 : ulong compress_ticks;
283 : } metrics;
284 : };
285 :
286 : typedef struct fd_snapmk fd_snapmk_t;
287 :
288 0 : #define IN_KIND_REPLAY 1
289 0 : #define IN_KIND_SNAPRD 2
290 :
291 : FD_FN_CONST static inline ulong
292 0 : scratch_align( void ) {
293 0 : return fd_ulong_max( fd_ulong_max( alignof(fd_snapmk_t), 32UL ), fd_txncache_align() );
294 0 : }
295 :
296 : FD_FN_PURE static inline ulong
297 0 : scratch_footprint( fd_topo_tile_t const * tile ) {
298 0 : ulong max_live_slots = tile->snapmk.max_live_slots;
299 :
300 0 : ulong zp_cnt = tile->out_cnt - 1UL; /* last out link is snapmk_out */
301 :
302 0 : ulong l = FD_LAYOUT_INIT;
303 0 : l = FD_LAYOUT_APPEND( l, alignof(fd_snapmk_t), sizeof(fd_snapmk_t) );
304 0 : l = FD_LAYOUT_APPEND( l, 32UL, ZSTD_estimateCStreamSize( FD_BACKUP_ZSTD_LEVEL ) );
305 0 : l = FD_LAYOUT_APPEND( l, fd_txncache_align(), fd_txncache_footprint( max_live_slots ) );
306 0 : l = FD_LAYOUT_APPEND( l, alignof(ulong), zp_cnt*FD_SNAPMK_ZP_DEPTH*sizeof(ulong) );
307 0 : l = FD_LAYOUT_APPEND( l, fd_txncache_writer_arena_align(), fd_txncache_writer_arena_sz( tile->snapmk.max_txn_per_slot ) );
308 0 : l = FD_LAYOUT_APPEND( l, fd_accdb_align(), fd_accdb_footprint( max_live_slots, 0 ) );
309 0 : return FD_LAYOUT_FINI( l, scratch_align() );
310 0 : }
311 :
312 : static void
313 : privileged_init( fd_topo_t const * topo,
314 0 : fd_topo_tile_t const * tile ) {
315 0 : FD_SCRATCH_ALLOC_INIT( l, fd_topo_obj_laddr( topo, tile->tile_obj_id ) );
316 0 : fd_snapmk_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_snapmk_t), sizeof(fd_snapmk_t) );
317 0 : memset( ctx, 0, sizeof(fd_snapmk_t) );
318 0 : fd_memset( ctx->pool_sz, 0, sizeof(ctx->pool_sz) );
319 :
320 0 : fd_cstr_ncpy( ctx->snap_dir, tile->snapmk.snapshots_path, PATH_MAX );
321 :
322 0 : int dir_fd = open( ctx->snap_dir, O_RDONLY|O_DIRECTORY );
323 0 : if( FD_UNLIKELY( dir_fd<0 ) ) {
324 0 : FD_LOG_ERR(( "open(%s) failed: %s", ctx->snap_dir, fd_io_strerror( errno ) ));
325 0 : }
326 0 : ctx->snap_dir_fd = dir_fd;
327 0 : ctx->snap_fd = -1;
328 :
329 0 : ctx->snap_full_max = tile->snapmk.max_full_snapshots_to_keep;
330 0 : ctx->snap_max = tile->snapmk.max_full_snapshots_to_keep+
331 0 : tile->snapmk.max_incremental_snapshots_to_keep;
332 0 : ctx->snap_idx = UINT_MAX;
333 0 : }
334 :
335 : static ulong
336 : populate_allowed_fds( fd_topo_t const * topo,
337 : fd_topo_tile_t const * tile,
338 : ulong out_fds_cnt,
339 0 : int * out_fds ) {
340 0 : fd_snapmk_t * ctx = fd_topo_obj_laddr( topo, tile->tile_obj_id );
341 0 : if( FD_UNLIKELY( out_fds_cnt<5UL+(ulong)ctx->snap_max ) ) FD_LOG_ERR(( "out_fds_cnt %lu", out_fds_cnt ));
342 0 : ulong out_cnt = 0UL;
343 0 : out_fds[ out_cnt++ ] = 2; /* stderr */
344 0 : if( FD_LIKELY( -1!=fd_log_private_logfile_fd() ) )
345 0 : out_fds[ out_cnt++ ] = fd_log_private_logfile_fd(); /* logfile */
346 0 : out_fds[ out_cnt++ ] = ctx->snap_dir_fd;
347 0 : out_fds[ out_cnt++ ] = FD_ACCDB_FD_RO;
348 0 : out_fds[ out_cnt++ ] = FD_VOTE_STAKES_FD; /* vote stakes disk spill */
349 0 : for( uint i=0U; i<ctx->snap_max; i++ )
350 0 : out_fds[ out_cnt++ ] = FD_SNAP_FD( i ); /* snapshot pool */
351 0 : return out_cnt;
352 0 : }
353 :
354 : static ulong
355 : populate_allowed_seccomp( fd_topo_t const * topo,
356 : fd_topo_tile_t const * tile,
357 : ulong out_cnt,
358 0 : struct sock_filter * out ) {
359 0 : fd_snapmk_t * ctx = fd_topo_obj_laddr( topo, tile->tile_obj_id );
360 0 : populate_sock_filter_policy_fd_snapmk_tile(
361 0 : out_cnt, out,
362 0 : (uint)fd_log_private_logfile_fd(),
363 0 : (uint)ctx->snap_dir_fd,
364 0 : (uint)FD_SNAP_FD( 0 ), (uint)FD_SNAP_FD( ctx->snap_max-1U ),
365 0 : (uint)FD_ACCDB_FD_RO,
366 0 : (uint)FD_VOTE_STAKES_FD );
367 0 : return sock_filter_policy_fd_snapmk_tile_instr_cnt;
368 0 : }
369 :
370 : static void
371 : unprivileged_init( fd_topo_t const * topo,
372 0 : fd_topo_tile_t const * tile ) {
373 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
374 0 : ulong max_live_slots = tile->snapmk.max_live_slots;
375 :
376 0 : ulong zp_cnt = tile->out_cnt - 1UL;
377 :
378 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
379 0 : fd_snapmk_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_snapmk_t), sizeof(fd_snapmk_t) );
380 0 : void * _zstd = FD_SCRATCH_ALLOC_APPEND( l, 32UL, ZSTD_estimateCStreamSize( FD_BACKUP_ZSTD_LEVEL ) );
381 0 : void * _txnc_lj = FD_SCRATCH_ALLOC_APPEND( l, fd_txncache_align(), fd_txncache_footprint( max_live_slots ) );
382 0 : ulong * _rd_shdw = FD_SCRATCH_ALLOC_APPEND( l, alignof(ulong), zp_cnt*FD_SNAPMK_ZP_DEPTH*sizeof(ulong) );
383 0 : void * _arena = FD_SCRATCH_ALLOC_APPEND( l, fd_txncache_writer_arena_align(), fd_txncache_writer_arena_sz( tile->snapmk.max_txn_per_slot ) );
384 0 : void * _accdb = FD_SCRATCH_ALLOC_APPEND( l, fd_accdb_align(), fd_accdb_footprint( max_live_slots, 0 ) );
385 0 : ulong end = FD_SCRATCH_ALLOC_FINI( l, scratch_align() );
386 0 : ctx->status_cache_writer_arena = _arena;
387 0 : ctx->status_cache_writer_arena_sz = fd_txncache_writer_arena_sz( tile->snapmk.max_txn_per_slot );
388 0 : FD_CHECK_CRIT( end==(ulong)scratch + scratch_footprint( tile ), "bug when calculating tile memory layout" );
389 :
390 0 : for( ulong i=0UL; i<zp_cnt; i++ ) {
391 0 : ctx->rd_shadow[ i ] = _rd_shdw + i*FD_SNAPMK_ZP_DEPTH;
392 0 : }
393 :
394 0 : ctx->state = SNAPMK_STATE_STARTUP;
395 0 : ctx->replay_in_seq_cons = ULONG_MAX;
396 0 : ctx->idle_iter = 0UL;
397 :
398 0 : fd_startup_gate_init( ctx->startup_gate, topo, tile->in_cnt );
399 0 : ctx->startup_pool_idx = 0UL;
400 :
401 0 : ctx->incremental = 0;
402 0 : ctx->base_slot = ULONG_MAX;
403 0 : void * _backup = fd_topo_obj_laddr( topo, tile->snapmk.visited_set_obj_id );
404 0 : ctx->visited_set = fd_backup_set ( _backup );
405 0 : ctx->overrun = fd_backup_overrun( _backup );
406 0 : ctx->stats = fd_backup_stats ( _backup );
407 0 : FD_TEST( ctx->visited_set );
408 0 : FD_TEST( ctx->overrun );
409 0 : FD_TEST( ctx->stats );
410 :
411 0 : ulong banks_obj_id = tile->snapmk.banks_obj_id;
412 0 : FD_TEST( banks_obj_id!=ULONG_MAX );
413 0 : ctx->banks = fd_banks_join( fd_topo_obj_laddr( topo, banks_obj_id ) );
414 0 : FD_TEST( ctx->banks );
415 :
416 0 : fd_txncache_shmem_t * tc_shmem = fd_txncache_shmem_join( fd_topo_obj_laddr( topo, tile->snapmk.txncache_obj_id ) );
417 0 : FD_TEST( tc_shmem );
418 0 : ctx->txncache = fd_txncache_join( fd_txncache_new( _txnc_lj, tc_shmem ) );
419 0 : FD_TEST( ctx->txncache );
420 :
421 0 : ulong * zp_fseq = fd_fseq_join( fd_topo_obj_laddr( topo, tile->snapmk.zp_fseq_id ) ); FD_TEST( zp_fseq );
422 0 : atomic_ulong * zp_app = fd_fseq_app_laddr( zp_fseq );
423 0 : ctx->file_off_p = &zp_app[ 0 ];
424 0 : ctx->appendvec_slot_ticket_p = &zp_app[ 1 ];
425 :
426 0 : void * _accdb_shmem = fd_topo_obj_laddr( topo, tile->snapmk.accdb_obj_id );
427 0 : fd_accdb_shmem_t * accdb_shmem_ro = fd_accdb_shmem_join( _accdb_shmem );
428 0 : FD_TEST( accdb_shmem_ro );
429 0 : ctx->accdb_shmem = accdb_shmem_ro;
430 0 : ctx->accdb_snapshot_sync = &accdb_shmem_ro->snapshot_sync;
431 0 : ctx->accdb_sleep = topo->sleep_obj_id!=ULONG_MAX ? fd_sleep_join( fd_topo_obj_laddr( topo, topo->sleep_obj_id ) ) : NULL;
432 0 : ctx->accdb_tile_id = fd_topo_find_tile( topo, "accdb", 0UL );
433 0 : ulong * epoch_fseq = fd_fseq_join( fd_topo_obj_laddr( topo, tile->snapmk.accdb_epoch_obj_id ) );
434 0 : FD_TEST( epoch_fseq );
435 0 : ctx->accdb = fd_accdb_join_readonly( _accdb, accdb_shmem_ro, epoch_fseq, FD_ACCDB_FD_RO );
436 0 : FD_TEST( ctx->accdb );
437 0 : fd_backup_cache_join( ctx->acc_cache, accdb_shmem_ro, epoch_fseq );
438 0 : {
439 0 : FD_SCRATCH_ALLOC_INIT( l, accdb_shmem_ro );
440 0 : FD_SCRATCH_ALLOC_APPEND( l, FD_ACCDB_SHMEM_ALIGN, sizeof(fd_accdb_shmem_t) );
441 0 : ctx->accdb_shfork = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_accdb_fork_shmem_t), max_live_slots*sizeof(fd_accdb_fork_shmem_t) );
442 0 : }
443 0 : ctx->accdb_root_fork = &accdb_shmem_ro->root_fork_id;
444 :
445 0 : ctx->delta.chain = (uint const *)( (uchar const *)accdb_shmem_ro + accdb_shmem_ro->delta.chain_off );
446 0 : ctx->delta.pool = (fd_accdb_delta_t const *)( (uchar const *)accdb_shmem_ro + accdb_shmem_ro->delta.ele_off );
447 0 : ctx->delta.chain_cnt = accdb_shmem_ro->delta.chain_cnt;
448 0 : ctx->delta.chain_idx = 0UL;
449 0 : ctx->delta.ele_idx = UINT_MAX;
450 0 : ctx->delta.ele_cnt = 0UL;
451 0 : ctx->delta.ele_total = 0UL;
452 :
453 0 : for( ulong i=0UL; i < tile->in_cnt; i++ ) {
454 0 : fd_topo_link_t const * link = &topo->links[ tile->in_link_id[ i ] ];
455 0 : if( 0==strcmp( link->name, "replay_snapmk" ) ) {
456 0 : FD_TEST( !ctx->in_kind[ i ] );
457 0 : ctx->in_kind[ i ] = IN_KIND_REPLAY;
458 0 : fd_topo_wksp_t const * link_wksp = &topo->workspaces[ topo->objs[ link->dcache_obj_id ].wksp_id ];
459 0 : ctx->replay_in_mem = link_wksp->wksp;
460 0 : ctx->replay_in_seq_prod = fd_mcache_seq_laddr_const( link->mcache );
461 0 : } else if( 0==strcmp( link->name, "snaprd_out" ) ) {
462 0 : FD_TEST( !ctx->in_kind[ i ] );
463 0 : ctx->in_kind[ i ] = IN_KIND_SNAPRD;
464 0 : fd_topo_wksp_t const * link_wksp = &topo->workspaces[ topo->objs[ link->dcache_obj_id ].wksp_id ];
465 0 : FD_CHECK_CRIT( link->mtu<=UINT_MAX, "oob MTU" );
466 0 : ctx->rd_in_mem = link_wksp->wksp;
467 0 : ctx->rd_in_mtu = link->mtu;
468 : /* fseq used for cnc and flow control of snaprd tile */
469 0 : ulong * fseq = tile->in_link_fseq[ i ];
470 0 : FD_CHECK_ERR( fseq, "no fseq for snaprd_out link" );
471 0 : ctx->rd_fseq = (atomic_ulong *)fseq;
472 0 : ctx->rd_ctl = fd_fseq_app_laddr( fseq );
473 0 : ctx->rd_tile_id = fd_topo_find_link_producer( topo, link );
474 0 : FD_STATIC_ASSERT( sizeof(ulong)<=FD_FSEQ_APP_FOOTPRINT, fseq_app_space );
475 0 : } else {
476 0 : FD_LOG_ERR(( "Unexpected input link \"%s\"", link->name ));
477 0 : }
478 0 : }
479 0 : FD_CHECK_ERR( ctx->replay_in_mem, "missing replay_snapmk link" );
480 0 : ctx->replay_in_seq_cons = __atomic_load_n( ctx->replay_in_seq_prod, __ATOMIC_ACQUIRE );
481 0 : FD_CHECK_ERR( ctx->rd_in_mem, "missing snaprd_out link" );
482 :
483 0 : FD_TEST( tile->out_cnt >= 2 );
484 0 : FD_TEST( tile->out_cnt <= FD_TOPO_MAX_TILE_OUT_LINKS );
485 0 : ctx->zp_cnt = tile->out_cnt - 1UL;
486 0 : for( ulong i=0UL; i < ctx->zp_cnt; i++ ) {
487 0 : fd_topo_link_t const * link = &topo->links[ tile->out_link_id[ i ] ];
488 0 : if( 0!=strcmp( link->name, "snapmk_zp" ) ) {
489 0 : FD_LOG_ERR(( "Unexpected output link \"%s\"", link->name ));
490 0 : }
491 0 : FD_TEST( link->mcache );
492 0 : FD_TEST( fd_mcache_depth( link->mcache )==FD_SNAPMK_ZP_DEPTH );
493 0 : ctx->zp_out[ i ].mem = topo->workspaces[ topo->objs[ link->dcache_obj_id ].wksp_id ].wksp;
494 0 : ctx->zp_out[ i ].chunk0 = fd_dcache_compact_chunk0( ctx->zp_out[ i ].mem, link->dcache );
495 0 : ctx->zp_out[ i ].wmark = fd_dcache_compact_wmark ( ctx->zp_out[ i ].mem, link->dcache, link->mtu );
496 0 : ctx->zp_out[ i ].chunk = ctx->zp_out[ i ].chunk0;
497 :
498 0 : for( ulong j=0UL; j<topo->tile_cnt; j++ ) {
499 0 : fd_topo_tile_t const * consumer = &topo->tiles[ j ];
500 0 : for( ulong k=0UL; k<consumer->in_cnt; k++ ) {
501 0 : if( FD_LIKELY( consumer->in_link_id[ k ]!=tile->out_link_id[ i ] ) ) continue;
502 0 : if( FD_UNLIKELY( !consumer->in_link_reliable[ k ] ) ) continue;
503 0 : FD_TEST( !ctx->zp_cons_fseq[ i ] );
504 0 : ctx->zp_cons_fseq[ i ] = consumer->in_link_fseq[ k ];
505 0 : }
506 0 : }
507 0 : FD_TEST( ctx->zp_cons_fseq[ i ] );
508 0 : }
509 :
510 0 : ctx->out.out_idx = tile->out_cnt - 1UL;
511 0 : fd_topo_link_t const * out_link = &topo->links[ tile->out_link_id[ ctx->out.out_idx ] ];
512 0 : if( 0!=strcmp( out_link->name, "snapmk_out" ) ) {
513 0 : FD_LOG_ERR(( "Unexpected output link \"%s\"", out_link->name ));
514 0 : }
515 0 : FD_CHECK_ERR( out_link->mtu >= sizeof(fd_snapmk_msg_t), "snapmk_out link MTU too small" );
516 0 : ctx->out.mem = fd_wksp_containing( out_link->dcache );
517 0 : ctx->out.chunk0 = fd_dcache_compact_chunk0( ctx->out.mem, out_link->dcache );
518 0 : ctx->out.wmark = fd_dcache_compact_wmark ( ctx->out.mem, out_link->dcache, out_link->mtu );
519 0 : ctx->out.chunk = ctx->out.chunk0;
520 :
521 0 : FD_TEST( out_link->mcache );
522 0 : ctx->out.seq_prod = fd_mcache_seq_laddr( out_link->mcache );
523 0 : ctx->out.cons_cnt = 0UL;
524 0 : for( ulong j=0UL; j<topo->tile_cnt; j++ ) {
525 0 : fd_topo_tile_t const * consumer = &topo->tiles[ j ];
526 0 : for( ulong k=0UL; k<consumer->in_cnt; k++ ) {
527 0 : if( FD_LIKELY( consumer->in_link_id[ k ]!=tile->out_link_id[ ctx->out.out_idx ] ) ) continue;
528 0 : if( FD_UNLIKELY( !consumer->in_link_reliable[ k ] ) ) continue;
529 0 : FD_CHECK_ERR( ctx->out.cons_cnt<SNAPMK_OUT_CONS_MAX, "too many snapmk_out consumers" );
530 0 : FD_TEST( consumer->in_link_fseq[ k ] );
531 0 : ctx->out.cons_fseq[ ctx->out.cons_cnt++ ] = consumer->in_link_fseq[ k ];
532 0 : }
533 0 : }
534 :
535 0 : ctx->zst = ZSTD_initStaticCStream( _zstd, ZSTD_estimateCStreamSize( FD_BACKUP_ZSTD_LEVEL ) );
536 0 : FD_TEST( ctx->zst );
537 0 : ulong zst_err;
538 0 : zst_err = ZSTD_CCtx_setParameter( ctx->zst, ZSTD_c_compressionLevel, FD_BACKUP_ZSTD_LEVEL );
539 0 : if( FD_UNLIKELY( ZSTD_isError( zst_err ) ) ) {
540 0 : FD_LOG_ERR(( "ZSTD_CCtx_setParameter(ZSTD_c_compressionLevel) failed: %s", ZSTD_getErrorName( zst_err ) ));
541 0 : }
542 0 : ctx->raw_buf = (ZSTD_inBuffer ){ .src = ctx->raw, .size = 0UL };
543 0 : ctx->comp_buf = (ZSTD_outBuffer){ .dst = ctx->comp, .size = COMP_BUF_SZ };
544 :
545 0 : ctx->rd_fseq[0] = 0UL;
546 0 : ctx->rd_seq = 0UL;
547 0 : ctx->rd_seq_cache = ULONG_MAX;
548 0 : }
549 :
550 : /* zip_reset discards any buffered data and assumes that the compression
551 : stream is clean (last frame was finished). */
552 :
553 : static void
554 0 : zip_reset( fd_snapmk_t * ctx ) {
555 0 : ctx->raw_buf.size = 0UL;
556 0 : ctx->raw_buf.pos = 0UL;
557 0 : ctx->zstd_frame_open = 0;
558 0 : }
559 :
560 : /* zip_append adds bytes into the input buffer (opens a new frame if
561 : none is open). Panics if raw_buf is out of buf space; it is the
562 : caller's responsibility to guarantee that data_sz is small enough. */
563 :
564 : static void
565 : zip_append( fd_snapmk_t * ctx,
566 : void const * data,
567 0 : ulong data_sz ) {
568 0 : if( FD_UNLIKELY( !data_sz ) ) return;
569 0 : FD_CHECK_CRIT( ctx->raw_buf.size + data_sz <= RAW_BUF_SZ, "insufficient raw buffer space" );
570 0 : fd_memcpy( ctx->raw + ctx->raw_buf.size, data, data_sz );
571 0 : ctx->raw_buf.size += data_sz;
572 0 : }
573 :
574 : /* zip_flush provides the input buffer to the Zstandard compressor.
575 : Depending on directive, it ...
576 : (ZSTD_e_continue) ... optimistically does compression work
577 : (ZSTD_e_flush) ... drains/empties the input buffer without ending
578 : the current frame
579 : (ZSTD_e_end) ... ends the current frame.
580 : Does not sync the underlying file descriptor. */
581 :
582 : static void
583 : zip_flush( fd_snapmk_t * ctx,
584 0 : ZSTD_EndDirective directive ) {
585 :
586 : /* Compress chunk */
587 0 : ulong raw_pos = ctx->raw_buf.pos;
588 0 : long t0 = fd_tickcount();
589 0 : ulong ret = ZSTD_compressStream2( ctx->zst, &ctx->comp_buf, &ctx->raw_buf, directive );
590 0 : long t1 = fd_tickcount();
591 0 : if( FD_UNLIKELY( ZSTD_isError( ret ) ) ) {
592 0 : FD_LOG_ERR(( "ZSTD_compressStream2 failed: %s", ZSTD_getErrorName( ret ) ));
593 0 : }
594 : /* A non-zero return after ZSTD_e_end means the frame epilogue is
595 : still pending in the compressor (comp_buf too small), which would
596 : corrupt the stream once raw frames or the next frame are written
597 : after it. */
598 0 : FD_CHECK_ERR( !( directive==ZSTD_e_end && ret ), "ZSTD_e_end left compressed data pending" );
599 0 : ctx->zstd_frame_open = directive!=ZSTD_e_end && ( ctx->zstd_frame_open || ctx->raw_buf.pos>raw_pos );
600 0 : ctx->metrics.bytes_compressed += ctx->raw_buf.pos - raw_pos;
601 0 : ctx->uncompressed_sz += ctx->raw_buf.pos - raw_pos;
602 0 : ctx->metrics.compress_ticks += (ulong)( t1-t0 );
603 :
604 : /* Move uncompressed bytes to left */
605 0 : if( ctx->raw_buf.pos < ctx->raw_buf.size ) {
606 0 : memmove( ctx->raw,
607 0 : ctx->raw + ctx->raw_buf.pos,
608 0 : ctx->raw_buf.size - ctx->raw_buf.pos );
609 0 : ctx->raw_buf.size -= ctx->raw_buf.pos;
610 0 : ctx->raw_buf.pos = 0UL;
611 0 : } else {
612 0 : ctx->raw_buf.size = 0UL;
613 0 : ctx->raw_buf.pos = 0UL;
614 0 : }
615 :
616 : /* Write compressed bytes to file */
617 0 : ulong comp_wr_;
618 0 : ulong comp_sz = ctx->comp_buf.pos;
619 0 : t0 = fd_tickcount();
620 0 : int wr_err = fd_io_write(
621 0 : ctx->snap_fd,
622 0 : ctx->comp,
623 0 : comp_sz, comp_sz,
624 0 : &comp_wr_ );
625 0 : t1 = fd_tickcount();
626 0 : if( FD_UNLIKELY( wr_err ) ) {
627 0 : FD_LOG_ERR(( "fd_io_write failed: %s", fd_io_strerror( wr_err ) ));
628 0 : }
629 0 : if( FD_UNLIKELY( comp_wr_ != comp_sz ) ) {
630 0 : FD_LOG_ERR(( "fd_io_write did not write full buffer (expected %lu bytes, wrote %lu bytes)", comp_sz, comp_wr_ ));
631 0 : }
632 0 : ctx->metrics.bytes_written += comp_wr_;
633 0 : ctx->metrics.io_blocked_ticks += (ulong)( t1-t0 );
634 0 : ctx->pool_sz[ ctx->snap_idx ] += comp_wr_;
635 0 : ctx->comp_buf.pos = 0UL;
636 0 : ctx->comp_buf.size = COMP_BUF_SZ;
637 0 : ctx->zstd_data_frame_cnt += directive==ZSTD_e_end;
638 0 : }
639 :
640 : /* zip_end_frame ends the current frame, if any is open or any input
641 : is buffered. Skips the call otherwise, which would emit an empty
642 : frame. */
643 :
644 : static void
645 0 : zip_end_frame( fd_snapmk_t * ctx ) {
646 0 : if( ctx->zstd_frame_open || ctx->raw_buf.size ) zip_flush( ctx, ZSTD_e_end );
647 0 : }
648 :
649 : /* zip_align aligns the Zstandard compressed stream by 512 bytes using
650 : skippable frames. */
651 :
652 : static void
653 0 : zip_align( fd_snapmk_t * ctx ) {
654 0 : long off = lseek( ctx->snap_fd, 0L, SEEK_CUR );
655 0 : if( FD_UNLIKELY( off<0L ) ) {
656 0 : FD_LOG_ERR(( "lseek failed: %i-%s", errno, fd_io_strerror( errno ) ));
657 0 : }
658 0 : ulong uoff = (ulong)off;
659 : /* Align using skippable frame */
660 0 : ulong aoff = fd_ulong_align_up( uoff, 4096UL );
661 0 : ulong pad_sz = aoff - uoff;
662 0 : if( FD_UNLIKELY( pad_sz>0UL && pad_sz<8UL ) ) {
663 0 : aoff += 4096UL;
664 0 : pad_sz += 4096UL;
665 0 : }
666 0 : if( pad_sz>0UL ) {
667 0 : long t0 = fd_tickcount();
668 0 : uchar frame_hdr[ 8 ];
669 0 : FD_STORE( uint, frame_hdr, ZSTD_MAGIC_SKIPPABLE_START );
670 0 : FD_STORE( uint, frame_hdr+4, (uint)( pad_sz-8 ) );
671 0 : ulong wr_sz_;
672 0 : int err = fd_io_write( ctx->snap_fd, frame_hdr, 8UL, 8UL, &wr_sz_ );
673 0 : if( FD_UNLIKELY( err ) ) {
674 0 : FD_LOG_ERR(( "fd_io_write failed: %i-%s", err, fd_io_strerror( err ) ));
675 0 : }
676 0 : static uchar const zero[ 4096UL ] = {0};
677 0 : err = fd_io_write( ctx->snap_fd, zero, pad_sz-8UL, pad_sz-8UL, &wr_sz_ );
678 0 : if( FD_UNLIKELY( err ) ) {
679 0 : FD_LOG_ERR(( "fd_io_write failed: %i-%s", err, fd_io_strerror( err ) ));
680 0 : }
681 0 : long t1 = fd_tickcount();
682 0 : ctx->metrics.bytes_written += pad_sz;
683 0 : ctx->metrics.io_blocked_ticks += (ulong)( t1-t0 );
684 0 : ctx->pool_sz[ ctx->snap_idx ] += pad_sz;
685 0 : ctx->zstd_padding_sz += pad_sz;
686 0 : }
687 0 : atomic_store_explicit( ctx->file_off_p, aoff, memory_order_release );
688 0 : }
689 :
690 : /* snapmk_status_cache_tar_hdr builds the TAR header of the status cache
691 : file for a payload of sz bytes. */
692 :
693 : static fd_tar_meta_t *
694 : snapmk_status_cache_tar_hdr( fd_tar_meta_t * meta,
695 0 : ulong sz ) {
696 0 : fd_backup_tar_file_hdr( meta, sz );
697 0 : if( FD_UNLIKELY( !fd_tar_meta_set_size( meta, sz ) ) ) {
698 0 : FD_LOG_ERR(( "status cache (%lu bytes) too large for a TAR size field", sz ));
699 0 : }
700 0 : fd_cstr_ncpy( meta->name, "snapshots/status_cache", sizeof(meta->name) );
701 0 : fd_tar_meta_set_chksum( meta );
702 0 : return meta;
703 0 : }
704 :
705 : /* snapmk_status_cache_prepare starts status cache serialization and
706 : writes its TAR header. The writer counts the status cache up front,
707 : so its size is known before the first byte is streamed. */
708 :
709 : static void
710 0 : snapmk_status_cache_prepare( fd_snapmk_t * ctx ) {
711 0 : fd_bank_t * bank = ctx->bank;
712 0 : ulong slot_history_sz = 0UL;
713 0 : uchar const * slot_history =
714 0 : fd_sysvar_cache_data_query( &bank->f.sysvar_cache, fd_sysvar_slot_history_id.uc, &slot_history_sz );
715 0 : if( FD_UNLIKELY( !fd_txncache_writer_init( ctx->txncache_writer, ctx->txncache, bank->txncache_fork_id, bank->f.slot, slot_history, slot_history_sz, ctx->status_cache_writer_arena, ctx->status_cache_writer_arena_sz ) ) ) {
716 0 : FD_LOG_CRIT(( "cannot serialize txncache fork %hu of snapshot bank at slot %lu because the root moved or SlotHistory lacks execution slots", bank->txncache_fork_id.val, bank->f.slot ));
717 0 : }
718 0 : ctx->status_cache_sz = 0UL;
719 0 : ctx->status_cache_expected_sz = fd_txncache_writer_serialized_sz( ctx->txncache_writer );
720 :
721 0 : zip_reset( ctx );
722 0 : fd_tar_meta_t meta;
723 0 : snapmk_status_cache_tar_hdr( &meta, ctx->status_cache_expected_sz );
724 0 : zip_append( ctx, &meta, sizeof(fd_tar_meta_t) );
725 0 : }
726 :
727 : /* snapmk_status_cache does a unit of status cache serialization and
728 : compression work. Returns 0 once status cache compression is fully
729 : done (TAR/Zstandard out stream clean). Otherwise, returns 1, which
730 : implies another work call is needed. */
731 :
732 : static int
733 0 : snapmk_status_cache( fd_snapmk_t * ctx ) {
734 0 : if( FD_UNLIKELY( ctx->raw_buf.size + FD_TXNCACHE_WRITER_BUF_MIN > RAW_BUF_SZ ) ) {
735 0 : zip_flush( ctx, ZSTD_e_continue );
736 0 : return 1;
737 0 : }
738 0 : ulong buf_rem = RAW_BUF_SZ - ctx->raw_buf.size;
739 0 : uchar * chunk = ctx->raw + ctx->raw_buf.size;
740 0 : ulong chunk_sz = fd_txncache_writer_serialize( ctx->txncache_writer, chunk, buf_rem );
741 0 : if( FD_UNLIKELY( !chunk_sz ) ) { /* done serializing? */
742 0 : FD_CHECK_CRIT( ctx->status_cache_sz==ctx->status_cache_expected_sz, "status cache size does not match its TAR header" );
743 0 : zip_flush( ctx, ZSTD_e_continue );
744 0 : ulong pad = fd_ulong_align_up( ctx->status_cache_sz, sizeof(fd_tar_meta_t) ) - ctx->status_cache_sz;
745 0 : if( pad ) {
746 0 : static uchar const zero[ sizeof(fd_tar_meta_t) ] = {0};
747 0 : zip_append( ctx, zero, pad );
748 0 : }
749 0 : zip_end_frame( ctx );
750 0 : ctx->state = SNAPMK_STATE_EOF_MARKER;
751 0 : return 0;
752 0 : }
753 0 : ctx->raw_buf.size += chunk_sz;
754 0 : ctx->status_cache_sz += chunk_sz;
755 0 : zip_flush( ctx, ZSTD_e_continue );
756 0 : return 1;
757 0 : }
758 :
759 : /* snapmk_eof_marker writes a compressed "end of TAR stream" marker.
760 : Assumes clean tar and Zstandard stream. */
761 :
762 : static void
763 0 : snapmk_eof_marker( fd_snapmk_t * ctx ) {
764 0 : FD_CHECK_ERR( ctx->raw_buf.size==0UL, "Zstandard stream unclean" );
765 0 : ctx->raw_buf.pos = 0UL;
766 0 : ctx->raw_buf.size = 1024UL;
767 0 : fd_memset( ctx->raw, 0, 1024UL );
768 0 : zip_flush( ctx, ZSTD_e_end );
769 0 : }
770 :
771 : /* snapmk_done_rename renames the "partial" snapshot file to a proper
772 : "snapshot-*.tar.zst" or "incremental-snapshot-*-*.tar.zst" file. */
773 :
774 : static void
775 0 : snapmk_done_rename( fd_snapmk_t * ctx ) {
776 0 : long file_sz = lseek( ctx->snap_fd, 0L, SEEK_END );
777 0 : if( FD_UNLIKELY( file_sz<0L ) ) {
778 0 : FD_LOG_ERR(( "lseek failed: %s", fd_io_strerror( errno ) ));
779 0 : }
780 0 : ctx->final_sz = (ulong)file_sz;
781 0 : ctx->pool_sz[ ctx->snap_idx ] = ctx->final_sz;
782 :
783 0 : fd_backup_inode_t * inode = &ctx->pool[ ctx->snap_idx ];
784 0 : struct flock lock = {
785 0 : .l_type = F_UNLCK,
786 0 : .l_whence = SEEK_SET
787 0 : };
788 0 : if( FD_UNLIKELY( fcntl( ctx->snap_fd, F_SETLK, &lock ) ) ) {
789 0 : FD_LOG_ERR(( "fcntl(F_UNLCK, %s) failed: %i-%s",
790 0 : inode->name, errno, fd_io_strerror( errno ) ));
791 0 : }
792 0 : if( FD_UNLIKELY( syscall( SYS_renameat2, ctx->snap_dir_fd, inode->name,
793 0 : ctx->snap_dir_fd, ctx->final_name, 0U ) ) ) {
794 0 : FD_LOG_ERR(( "renameat2(%s, %s) failed: %s", inode->name, ctx->final_name, fd_io_strerror( errno ) ));
795 0 : }
796 0 : fd_cstr_ncpy( inode->name, ctx->final_name, sizeof(inode->name) );
797 :
798 0 : if( FD_UNLIKELY( ctx->incremental ) ) {
799 0 : inode->full_slot = ctx->base_slot;
800 0 : inode->incr_slot = ctx->bank->f.slot;
801 0 : ctx->metrics.last_snapshot_slot_finished_incremental = ctx->bank->f.slot;
802 0 : ctx->metrics.snapshots_finished[ FD_METRICS_ENUM_SNAP_TYPE_V_INCREMENTAL_IDX ]++;
803 0 : } else {
804 0 : inode->full_slot = ctx->bank->f.slot;
805 0 : inode->incr_slot = ULONG_MAX;
806 0 : ctx->base_slot = ctx->bank->f.slot;
807 0 : ctx->metrics.last_snapshot_slot_finished_full = ctx->bank->f.slot;
808 0 : ctx->metrics.snapshots_finished[ FD_METRICS_ENUM_SNAP_TYPE_V_FULL_IDX ]++;
809 0 : }
810 :
811 0 : ctx->snap_fd = -1;
812 :
813 0 : ctx->end_time = fd_log_wallclock();
814 0 : FD_LOG_INFO(( "%s snapshot created in %.3f seconds (%s/%s, %.3f GB)",
815 0 : ctx->incremental ? "incremental" : "full",
816 0 : (double)( ctx->end_time - ctx->start_time )/1e9,
817 0 : ctx->snap_dir, ctx->final_name,
818 0 : (double)file_sz/1e9 ));
819 :
820 0 : ctx->state = SNAPMK_STATE_DONE;
821 0 : }
822 :
823 : /* snapshot_sync_advance requests replay to advance the snapshot sync
824 : state machine. */
825 :
826 : static void
827 : snapshot_sync_transition( fd_snapmk_t * ctx,
828 : ulong state_from,
829 : ulong state_req,
830 0 : ulong state_to ) {
831 0 : while( FD_UNLIKELY( fd_accdb_snapshot_sync_state( ctx->accdb_snapshot_sync )!=state_from ) ) FD_YIELD();
832 0 : fd_accdb_snapshot_sync_advance( ctx->accdb_snapshot_sync, state_req );
833 0 : if( FD_UNLIKELY( ctx->accdb_sleep ) ) fd_sleep_ring( ctx->accdb_sleep, ctx->accdb_tile_id );
834 0 : while( FD_UNLIKELY( fd_accdb_snapshot_sync_state( ctx->accdb_snapshot_sync )!=state_to ) ) FD_YIELD();
835 0 : }
836 :
837 : static ulong
838 : snapshot_sync_request( fd_snapmk_t * ctx,
839 : ulong state_from,
840 0 : ulong state_req ) {
841 0 : while( FD_UNLIKELY( fd_accdb_snapshot_sync_state( ctx->accdb_snapshot_sync )!=state_from ) ) FD_YIELD();
842 0 : fd_accdb_snapshot_sync_advance( ctx->accdb_snapshot_sync, state_req );
843 0 : if( FD_UNLIKELY( ctx->accdb_sleep ) ) fd_sleep_ring( ctx->accdb_sleep, ctx->accdb_tile_id );
844 0 : for(;;) {
845 0 : ulong state = fd_accdb_snapshot_sync_state( ctx->accdb_snapshot_sync );
846 0 : if( FD_LIKELY( state!=state_req ) ) return state;
847 0 : FD_YIELD();
848 0 : }
849 0 : }
850 :
851 : static inline ulong
852 : zp_publish( fd_snapmk_t * ctx,
853 : fd_stem_context_t * stem,
854 : ulong out_idx,
855 : ulong sig,
856 : ulong chunk,
857 : ulong sz,
858 : ulong ctl,
859 : ulong tsorig,
860 0 : ulong tspub ) {
861 0 : FD_TEST( out_idx<ctx->zp_cnt );
862 0 : fd_frag_meta_t * mcache = stem->mcaches[ out_idx ];
863 0 : ulong depth = stem->depths [ out_idx ];
864 0 : ulong * seqp = &stem->seqs [ out_idx ];
865 0 : ulong seq = *seqp;
866 0 : # if FD_HAS_AVX
867 0 : fd_mcache_publish_avx( mcache, depth, seq, sig, chunk, sz, ctl, tsorig, tspub );
868 : # elif FD_HAS_ARM
869 : fd_mcache_publish_arm( mcache, depth, seq, sig, chunk, sz, ctl, tsorig, tspub );
870 : # else
871 : fd_mcache_publish ( mcache, depth, seq, sig, chunk, sz, ctl, tsorig, tspub );
872 : # endif
873 0 : ulong cr_avail = fd_ulong_sat_sub( stem->cr_avail[ out_idx ], 1UL );
874 0 : stem->cr_avail[ out_idx ] = cr_avail;
875 0 : *stem->min_cr_avail = fd_ulong_min( cr_avail, *stem->min_cr_avail );
876 0 : if( FD_UNLIKELY( !cr_avail ) ) ctx->zp_ready &= ~fd_ulong_mask_bit( (int)out_idx );
877 0 : *seqp = fd_seq_inc( seq, 1UL );
878 0 : if( FD_UNLIKELY( stem->sleep ) ) fd_sleep_wake_check( stem->sleep, stem->wake+stem->wake_off[ out_idx ], (ulong)(stem->wake_off[ out_idx+1UL ]-stem->wake_off[ out_idx ]) );
879 0 : return seq;
880 0 : }
881 :
882 : /* broadcast is called repeatedly until a message has been sent to all
883 : snapzp tiles. Returns 1 if the message was sent to all tiles, 0
884 : otherwise (call again).
885 :
886 : broadcast_prepare must be called once before attempting to broadcast.
887 :
888 : Typically, a barrier is installed before the broadcast (wait for all
889 : snapzp tiles to catch up before broadcasting). Then, another
890 : barrier is installed at the broadcast (wait for all snapzp tiles to
891 : ACK the broadcast before continuing). */
892 :
893 : static void
894 0 : broadcast_prepare( fd_snapmk_t * ctx ) {
895 0 : ctx->zp_flush_pending = fd_ulong_mask( 0, (int)ctx->zp_cnt-1 );
896 0 : }
897 :
898 : static int
899 : broadcast( fd_snapmk_t * ctx,
900 : fd_stem_context_t * stem,
901 : ulong ctl,
902 0 : int * charge_busy ) {
903 0 : int did_work = 0;
904 0 : ulong zp_cnt = ctx->zp_cnt;
905 0 : for( ulong i=0UL; i<zp_cnt; i++ ) {
906 0 : if( !fd_ulong_extract_bit( ctx->zp_flush_pending, (int)i ) ) continue;
907 0 : if( !stem->cr_avail[ i ] ) continue;
908 0 : zp_publish( ctx, stem, i, 0UL, 0UL, 0UL, ctl, 0UL, 0UL );
909 0 : ctx->zp_barrier[ i ] = stem->seqs[ i ]; /* FINISH barrier */
910 0 : ctx->zp_flush_pending &= ~fd_ulong_mask_bit( (int)i );
911 0 : *charge_busy = 1;
912 0 : did_work = 1;
913 0 : }
914 0 : if( (!ctx->zp_flush_pending) & (!did_work) ) {
915 0 : return 1;
916 0 : }
917 0 : return 0;
918 0 : }
919 :
920 : /* barrier_install blocks this tile until all snapzp tiles have caught
921 : up with the last published messages. */
922 :
923 : static void
924 : barrier_install( fd_snapmk_t * ctx,
925 0 : fd_stem_context_t const * stem ) {
926 0 : ulong zp_cnt = ctx->zp_cnt;
927 0 : for( ulong i=0UL; i<zp_cnt; i++ ) {
928 0 : ctx->zp_barrier[ i ] = stem->seqs[ i ]; /* FLUSH barrier */
929 0 : }
930 0 : }
931 :
932 : /* zp_rr_next picks a snapzp tile for a new job (ULONG_MAX if none are
933 : ready). FIXME rewrite this O(1) with rotate+find_lsb. */
934 :
935 : static ulong
936 0 : zp_rr_next( fd_snapmk_t * ctx ) {
937 0 : ulong n = ctx->zp_cnt;
938 0 : for( ulong k=0UL; k<n; k++ ) {
939 0 : ulong idx = ctx->zp_rr_idx;
940 0 : ctx->zp_rr_idx = fd_ulong_if( ctx->zp_rr_idx+1UL>=n, 0UL, ctx->zp_rr_idx+1UL );
941 0 : if( ctx->zp_ready & (1UL<<idx) ) return idx;
942 0 : }
943 0 : return ULONG_MAX;
944 0 : }
945 :
946 : /* zp_alloc allocates a message payload on snapzp_mk[ out_idx ]. */
947 :
948 : static inline void *
949 : zp_alloc( fd_snapmk_t * ctx,
950 : ulong out_idx,
951 : ulong sz,
952 0 : ulong * chunk ) {
953 0 : FD_TEST( sz );
954 0 : FD_TEST( out_idx<ctx->zp_cnt );
955 0 : *chunk = ctx->zp_out[ out_idx ].chunk;
956 0 : void * laddr = fd_chunk_to_laddr( ctx->zp_out[ out_idx ].mem, *chunk );
957 0 : ctx->zp_out[ out_idx ].chunk =
958 0 : fd_dcache_compact_next( *chunk, sz, ctx->zp_out[ out_idx ].chunk0, ctx->zp_out[ out_idx ].wmark );
959 0 : return laddr;
960 0 : }
961 :
962 : /* rd_ack sends read acknowledgements to snaprd. */
963 :
964 : static void
965 : rd_ack( fd_snapmk_t * ctx,
966 0 : fd_stem_context_t const * stem ) {
967 0 : ulong zp_cnt = ctx->zp_cnt;
968 0 : ulong rd_seq = ctx->rd_seq;
969 0 : for( ulong i=0UL; i<zp_cnt; i++ ) {
970 : /* Must be the live fseq, not stem's cached cons_seq. Publishes are
971 : gated on the live consumer position (zp_sync_cr_avail), so pub can
972 : outrun stem's cached cons_seq by more than the link depth. Indexing
973 : rd_shadow with a stale cons would then read a slot that a newer
974 : publish has already wrapped onto, yielding a floor that is too new
975 : and releasing snaprd buffers that snapzp is still reading. */
976 0 : ulong cons = fd_fseq_query( ctx->zp_cons_fseq[ i ] );
977 0 : ulong pub = stem->seqs[ i ];
978 0 : if( FD_UNLIKELY( !fd_seq_lt( cons, pub ) ) ) continue;
979 : /* snapzp ack for 'cons' means that this snapzp has fully consumed
980 : up to snaprd seq 'floor'. */
981 0 : ulong floor = ctx->rd_shadow[ i ][ cons & (FD_SNAPMK_ZP_DEPTH-1) ];
982 0 : rd_seq = fd_seq_lt( floor, rd_seq ) ? floor : rd_seq;
983 0 : }
984 0 : if( rd_seq != ctx->rd_seq_cache ) {
985 0 : ctx->rd_seq_cache = rd_seq;
986 0 : atomic_store_explicit( ctx->rd_fseq, rd_seq, memory_order_release );
987 : /* early credit return: ring snaprd if it parked on our credits */
988 0 : if( FD_UNLIKELY( stem->sleep ) ) {
989 0 : __atomic_thread_fence( __ATOMIC_SEQ_CST );
990 0 : if( FD_UNLIKELY( FD_VOLATILE_CONST( stem->sleep->credit_bits[ ctx->rd_tile_id>>6 ] ) & (1UL<<(ctx->rd_tile_id&63UL)) ) ) fd_sleep_ring( stem->sleep, ctx->rd_tile_id );
991 0 : }
992 0 : }
993 0 : }
994 :
995 : /* recv_credit is called whenever consumer flow control credits are
996 : refreshed. */
997 :
998 : static void
999 : recv_credit( fd_snapmk_t * ctx,
1000 : ulong out_idx,
1001 : ulong out_seq,
1002 0 : ulong cons_seq ) {
1003 0 : if( out_idx < ctx->zp_cnt ) {
1004 0 : long in_flight = fd_long_max( fd_seq_diff( out_seq, cons_seq ), 0L );
1005 0 : long cr_avail = FD_SNAPMK_ZP_DEPTH - in_flight;
1006 0 : ctx->zp_ready |= fd_ulong_if( cr_avail>0L, 1UL<<out_idx, 0UL );
1007 0 : }
1008 0 : }
1009 :
1010 : /* zp_sync_cr_avail syncs zp_ready and stem credit accounting. */
1011 :
1012 : static void
1013 : zp_sync_cr_avail( fd_snapmk_t * ctx,
1014 0 : fd_stem_context_t const * stem ) {
1015 0 : ulong zp_cnt = ctx->zp_cnt;
1016 0 : for( ulong i=0UL; i<zp_cnt; i++ ) {
1017 0 : if( FD_LIKELY( stem->cr_avail[ i ] ) ) continue; /* stem already agrees */
1018 0 : if( FD_LIKELY( !fd_ulong_extract_bit( ctx->zp_ready, (int)i ) ) ) continue;
1019 0 : ulong cons = fd_fseq_query( ctx->zp_cons_fseq[ i ] );
1020 0 : long in_flight = fd_long_max( fd_seq_diff( stem->seqs[ i ], cons ), 0L );
1021 0 : ulong cr_avail = (ulong)fd_long_max( (long)FD_SNAPMK_ZP_DEPTH-in_flight, 0L );
1022 0 : stem->cr_avail[ i ] = cr_avail;
1023 0 : if( FD_UNLIKELY( !cr_avail ) ) ctx->zp_ready &= ~fd_ulong_mask_bit( (int)i );
1024 0 : }
1025 0 : }
1026 :
1027 : /* clean_overruns recovers from torn cache reads by downstream snapzp tiles.
1028 : Marks the affected accounts as 'not visited' so they are retried later. */
1029 :
1030 : static void
1031 0 : clean_overruns( fd_snapmk_t * ctx ) {
1032 0 : fd_backup_overrun_t * q = ctx->overrun;
1033 0 : for(;;) {
1034 0 : uint pos = q->tail;
1035 0 : fd_backup_overrun_slot_t * slot = &q->slot[ pos & (FD_BACKUP_OVERRUN_DEPTH-1U) ];
1036 0 : if( FD_LIKELY( __atomic_load_n( &slot->seq, __ATOMIC_ACQUIRE )!=pos ) ) break;
1037 :
1038 0 : uint acc_idx = slot->acc_idx;
1039 0 : __atomic_store_n( &slot->seq, pos+FD_BACKUP_OVERRUN_DEPTH-1U, __ATOMIC_RELEASE );
1040 0 : q->tail = pos+1U;
1041 :
1042 0 : fd_backup_visited_remove( ctx->visited_set, (ulong)acc_idx );
1043 0 : }
1044 0 : }
1045 :
1046 : /* check_credit runs every run loop iteration. It specifies custom flow
1047 : control behavior. */
1048 :
1049 : static void
1050 : check_credit( fd_snapmk_t * ctx,
1051 : fd_stem_context_t * stem,
1052 : int * charge_busy,
1053 0 : int * is_backpressured ) {
1054 0 : (void)stem; (void)charge_busy; (void)is_backpressured;
1055 :
1056 0 : if( FD_LIKELY( ctx->state!=SNAPMK_STATE_ACCDB_DISK ) ) {
1057 0 : clean_overruns( ctx );
1058 0 : }
1059 :
1060 0 : if( FD_LIKELY( ctx->state!=SNAPMK_STATE_IDLE &&
1061 0 : ctx->state!=SNAPMK_STATE_SLEEP ) ) {
1062 0 : zp_sync_cr_avail( ctx, stem );
1063 0 : }
1064 :
1065 0 : switch( ctx->state ) {
1066 0 : case SNAPMK_STATE_IDLE:
1067 0 : case SNAPMK_STATE_SLEEP:
1068 0 : break;
1069 :
1070 : /* these state send jobs to snapzp tiles */
1071 0 : case SNAPMK_STATE_ACCDB_DISK:
1072 0 : rd_ack( ctx, stem );
1073 0 : if( FD_UNLIKELY( ctx->disk_out_idx>=0 ) ) {
1074 0 : if( FD_UNLIKELY( !fd_ulong_extract_bit( ctx->zp_ready, ctx->disk_out_idx ) ) ) {
1075 0 : *is_backpressured = 1;
1076 0 : return;
1077 0 : }
1078 0 : }
1079 0 : __attribute__((fallthrough));
1080 0 : case SNAPMK_STATE_START:
1081 0 : case SNAPMK_STATE_ACCDB_CACHE:
1082 0 : case SNAPMK_STATE_ACCDB_DELTA:
1083 0 : if( FD_UNLIKELY( !ctx->zp_ready ) ) {
1084 0 : *is_backpressured = 1;
1085 0 : return;
1086 0 : }
1087 0 : *is_backpressured = 0; /* undo stem backpressure */
1088 0 : break;
1089 :
1090 : /* these states broadcast */
1091 0 : case SNAPMK_STATE_ACCDB_CACHE_FLUSH:
1092 0 : case SNAPMK_STATE_ACCDB_CACHE_FINISH:
1093 0 : case SNAPMK_STATE_ACCDB_DISK_FLUSH:
1094 0 : case SNAPMK_STATE_ACCDB_DISK_FINISH:
1095 0 : case SNAPMK_STATE_ACCDB_DELTA_FLUSH:
1096 0 : case SNAPMK_STATE_ACCDB_DELTA_FINISH: {
1097 : /* wait for snapzp tiles to acknowledge zp_barrier[*] */
1098 0 : *is_backpressured = 0;
1099 0 : for( ulong i=0UL; i < ctx->zp_cnt; i++ ) {
1100 0 : if( FD_UNLIKELY( fd_seq_lt( fd_fseq_query( ctx->zp_cons_fseq[ i ] ), ctx->zp_barrier[ i ] ) ) ) {
1101 0 : *is_backpressured = 1;
1102 0 : return;
1103 0 : }
1104 0 : }
1105 0 : break;
1106 0 : }
1107 0 : default:
1108 : /* use default backpressure mechanism */
1109 0 : break;
1110 0 : }
1111 0 : }
1112 :
1113 : /* snapmk_tar_headers writes out the first few fixed parts of a snapshot
1114 : file. */
1115 :
1116 : static void
1117 0 : snapmk_tar_headers( fd_snapmk_t * ctx ) {
1118 0 : ulong slot = ctx->bank->f.slot;
1119 :
1120 0 : ctx->raw_buf.pos = ctx->raw_buf.size = 0UL;
1121 0 : uchar * p = ctx->raw;
1122 0 : fd_tar_meta_t meta;
1123 :
1124 0 : fd_backup_tar_file_hdr( &meta, 5UL );
1125 0 : fd_cstr_ncpy( meta.name, "version", sizeof(meta.name) );
1126 0 : fd_tar_meta_set_chksum( &meta );
1127 0 : memcpy( p, &meta, sizeof(fd_tar_meta_t) );
1128 0 : p += sizeof(fd_tar_meta_t);
1129 :
1130 0 : memcpy( p, "1.2.0", 5UL );
1131 0 : memset( p+5, 0, 512UL-5UL );
1132 0 : p += 512UL;
1133 :
1134 0 : fd_backup_tar_dir_hdr( &meta );
1135 0 : fd_cstr_ncpy( meta.name, "snapshots/", sizeof(meta.name) );
1136 0 : fd_tar_meta_set_chksum( &meta );
1137 0 : memcpy( p, &meta, sizeof(fd_tar_meta_t) );
1138 0 : p += sizeof(fd_tar_meta_t);
1139 :
1140 0 : fd_backup_tar_dir_hdr( &meta );
1141 0 : fd_cstr_printf_check( meta.name, sizeof(meta.name), NULL, "snapshots/%lu/", slot );
1142 0 : fd_tar_meta_set_chksum( &meta );
1143 0 : memcpy( p, &meta, sizeof(fd_tar_meta_t) );
1144 0 : p += sizeof(fd_tar_meta_t);
1145 :
1146 0 : ctx->manifest_sz = ctx->manifest_writer->serialized_sz;
1147 0 : fd_backup_tar_file_hdr( &meta, ctx->manifest_sz );
1148 0 : fd_cstr_printf_check( meta.name, sizeof(meta.name), NULL, "snapshots/%lu/%lu", slot, slot );
1149 0 : fd_tar_meta_set_chksum( &meta );
1150 0 : memcpy( p, &meta, sizeof(fd_tar_meta_t) );
1151 0 : p += sizeof(fd_tar_meta_t);
1152 0 : ctx->raw_buf.size = (ulong)( p - ctx->raw );
1153 0 : ctx->manifest_pad = fd_ulong_align_up( ctx->manifest_sz, 512UL ) - ctx->manifest_sz;
1154 :
1155 0 : zip_flush( ctx, ZSTD_e_end );
1156 0 : }
1157 :
1158 : /* snapmk_manifest_chunk writes out a chunk of snapshot manifest data.
1159 : Returns 1 if there is more work to do, 0 if the snapshot manifest was
1160 : fully written. */
1161 :
1162 : static int
1163 0 : snapmk_manifest_chunk( fd_snapmk_t * ctx ) {
1164 0 : if( FD_UNLIKELY( ctx->raw_buf.size + FD_SSMANIFEST_BUF_MIN > RAW_BUF_SZ ) ) {
1165 0 : zip_flush( ctx, ZSTD_e_continue );
1166 0 : return 1;
1167 0 : }
1168 0 : ulong buf_rem = RAW_BUF_SZ - ctx->raw_buf.size;
1169 0 : ulong chunk_sz = fd_snap_manifest_serialize(
1170 0 : ctx->manifest_writer,
1171 0 : ctx->raw + ctx->raw_buf.size,
1172 0 : buf_rem );
1173 0 : ctx->raw_buf.size += chunk_sz;
1174 0 : if( FD_LIKELY( chunk_sz ) ) return 1;
1175 :
1176 : /* Done compressing manifest */
1177 0 : zip_flush( ctx, ZSTD_e_continue );
1178 0 : if( ctx->manifest_pad ) {
1179 0 : fd_memset( ctx->raw, 0, ctx->manifest_pad );
1180 0 : ctx->raw_buf.size = ctx->manifest_pad;
1181 0 : }
1182 0 : zip_flush( ctx, ZSTD_e_end );
1183 0 : zip_align( ctx );
1184 0 : ctx->accounts_start_time = fd_log_wallclock();
1185 0 : return 0;
1186 0 : }
1187 :
1188 : /* snapmk_accdb_cache schedules accdb cache work. Returns 1 if there is
1189 : more work to do, 0 otherwise. */
1190 :
1191 : static int
1192 : snapmk_accdb_cache( fd_snapmk_t * ctx,
1193 0 : fd_stem_context_t * stem ) {
1194 0 : ulong out_idx = zp_rr_next( ctx );
1195 0 : fd_backup_cache_msg_t * frag = ctx->scan_batch;
1196 0 : frag = fd_backup_cache_scan( ctx->acc_cache, frag );
1197 0 : if( FD_UNLIKELY( !frag ) ) return 0;
1198 :
1199 : /* remove duplicates
1200 : first pass (fast), ILP-friendly/vectorizable check */
1201 0 : for( ulong i=0UL; i<FD_BACKUP_CACHE_PARA; i++ ) {
1202 0 : uint acc_idx = frag->acc_idx[ i ];
1203 0 : if( acc_idx==UINT_MAX ) continue;
1204 0 : if( FD_UNLIKELY( fd_backup_visited_test( ctx->visited_set, (ulong)acc_idx ) ) ) {
1205 0 : frag->acc_idx[ i ] = UINT_MAX;
1206 0 : }
1207 0 : }
1208 :
1209 : /* second pass: intra-batch conflict detect */
1210 0 : for( ulong i=0UL; i<FD_BACKUP_CACHE_PARA; i++ ) {
1211 0 : uint acc_idx = frag->acc_idx[ i ];
1212 0 : if( acc_idx==UINT_MAX ) continue;
1213 0 : if( FD_UNLIKELY( fd_backup_visited_test( ctx->visited_set, (ulong)acc_idx ) ) ) {
1214 0 : frag->acc_idx[ i ] = UINT_MAX;
1215 0 : memset( frag->pubkey[ i ].uc, 0, sizeof(fd_pubkey_t) );
1216 0 : continue;
1217 0 : }
1218 0 : fd_backup_visited_insert( ctx->visited_set, (ulong)acc_idx );
1219 0 : }
1220 :
1221 : /* publish a batch of cached accounts */
1222 0 : ulong chunk;
1223 0 : void * payload = zp_alloc( ctx, (ulong)out_idx, sizeof(fd_backup_cache_msg_t), &chunk );
1224 0 : fd_memcpy( payload, frag, sizeof(fd_backup_cache_msg_t) );
1225 0 : ulong ctl = fd_frag_meta_ctl( FD_BACKUP_ORIG_ACC_CACHE, 0, 0, 0 );
1226 0 : zp_publish( ctx, stem, (ulong)out_idx, 0UL, chunk, sizeof(fd_backup_cache_msg_t), ctl, 0UL, 0UL );
1227 :
1228 0 : return 1;
1229 0 : }
1230 :
1231 : /* snapmk_accdb_delta drains a batch of incremental snapshot accounts
1232 : and passes them to snapzp for a compression job. */
1233 :
1234 : static int
1235 : snapmk_accdb_delta( fd_snapmk_t * ctx,
1236 0 : fd_stem_context_t * stem ) {
1237 0 : ulong out_idx = zp_rr_next( ctx );
1238 :
1239 0 : fd_backup_delta_msg_t batch = {0};
1240 0 : while( batch.cnt<FD_BACKUP_CACHE_PARA ) {
1241 0 : if( FD_UNLIKELY( ctx->delta.ele_idx==UINT_MAX ) ) {
1242 0 : if( FD_UNLIKELY( ctx->delta.chain_idx>=ctx->delta.chain_cnt ) ) break;
1243 0 : ctx->delta.ele_idx = __atomic_load_n( &ctx->delta.chain[ ctx->delta.chain_idx++ ], __ATOMIC_ACQUIRE );
1244 0 : continue;
1245 0 : }
1246 0 : fd_accdb_delta_t const * cur = &ctx->delta.pool[ ctx->delta.ele_idx ];
1247 0 : ctx->delta.ele_idx = __atomic_load_n( &cur->next, __ATOMIC_RELAXED );
1248 0 : fd_memcpy( &batch.pubkey[ batch.cnt ], cur->pubkey, sizeof(fd_pubkey_t) );
1249 0 : batch.cnt++;
1250 0 : }
1251 0 : if( FD_UNLIKELY( !batch.cnt ) ) return 0;
1252 0 : ctx->delta.ele_cnt += batch.cnt;
1253 :
1254 0 : ulong chunk;
1255 0 : void * payload = zp_alloc( ctx, out_idx, sizeof(fd_backup_delta_msg_t), &chunk );
1256 0 : fd_memcpy( payload, &batch, sizeof(fd_backup_delta_msg_t) );
1257 0 : ulong ctl = fd_frag_meta_ctl( FD_BACKUP_ORIG_ACC_DELTA, 0, 0, 0 );
1258 0 : zp_publish( ctx, stem, out_idx, 0UL, chunk, sizeof(fd_backup_delta_msg_t), ctl, 0UL, 0UL );
1259 :
1260 0 : return 1;
1261 0 : }
1262 :
1263 : /* snapmk_replay_sleep sleeps until the replay tile publishes a new
1264 : snapshot command or IDLE_SLEEP nanoseconds pass. */
1265 :
1266 : static void
1267 0 : snapmk_replay_sleep( fd_snapmk_t * ctx ) {
1268 0 : struct timespec const ts = { .tv_sec = (IDLE_SLEEP)/(long)1e9, .tv_nsec = (IDLE_SLEEP)%(long)1e9 };
1269 0 : long res = syscall( SYS_futex, (uint *)ctx->replay_in_seq_prod, FUTEX_WAIT, (uint)ctx->replay_in_seq_cons, &ts );
1270 0 : if( res==0 || (res==-1 && errno==EAGAIN) ) {
1271 : /* stop sleeping */
1272 0 : ctx->state = SNAPMK_STATE_IDLE;
1273 0 : ctx->idle_iter = 0UL;
1274 0 : } else if( res==-1 && errno!=ETIMEDOUT ) {
1275 0 : FD_LOG_ERR(( "FUTEX_WAIT failed (%i-%s)", errno, fd_io_strerror( errno ) ));
1276 0 : }
1277 0 : }
1278 :
1279 : /* snapmk_msg_alloc allocates space for a snapmk_out payload. */
1280 :
1281 : static fd_snapmk_msg_t *
1282 0 : snapmk_msg_alloc( fd_snapmk_t * ctx ) {
1283 0 : return fd_chunk_to_laddr( ctx->out.mem, ctx->out.chunk );
1284 0 : }
1285 :
1286 : /* wake all reliable consumers by unconditionally waking them (snapsv
1287 : tiles). This is inefficient (does a FUTEX_WAKE syscall), but
1288 : acceptable given the very low frag production rate. */
1289 :
1290 : static void
1291 : snapmk_out_wake( fd_snapmk_t * ctx,
1292 0 : fd_stem_context_t * stem ) {
1293 0 : fd_mcache_seq_update( ctx->out.seq_prod, stem->seqs[ ctx->out.out_idx ] );
1294 0 : if( FD_UNLIKELY( -1==syscall( SYS_futex, (uint *)ctx->out.seq_prod, FUTEX_WAKE, INT_MAX, NULL, NULL, 0 ) ) ) {
1295 0 : FD_LOG_ERR(( "FUTEX_WAKE failed (%i-%s)", errno, fd_io_strerror( errno ) ));
1296 0 : }
1297 0 : }
1298 :
1299 : /* snapmk_msg_publish publishes a msg and frag on snapmk_out. */
1300 :
1301 : static void
1302 : snapmk_msg_publish( fd_snapmk_t * ctx,
1303 : fd_stem_context_t * stem,
1304 0 : ulong msg_type ) {
1305 0 : ulong sz;
1306 0 : switch( msg_type ) { /* known at compile time */
1307 0 : case FD_SNAPMK_MSG_CREATED: sz = sizeof(fd_snapmk_msg_created_t); break;
1308 0 : case FD_SNAPMK_MSG_DELETED: sz = sizeof(fd_snapmk_msg_deleted_t); break;
1309 0 : case FD_SNAPMK_MSG_STARTED: sz = sizeof(fd_snapmk_msg_started_t); break;
1310 0 : case FD_SNAPMK_MSG_FAILED: sz = sizeof(fd_snapmk_msg_failed_t); break;
1311 0 : case FD_SNAPMK_MSG_FOUND: sz = sizeof(fd_snapmk_msg_found_t); break;
1312 0 : default:
1313 0 : FD_LOG_CRIT(( "invalid msg_type %lu", msg_type ));
1314 0 : }
1315 0 : ulong chunk = ctx->out.chunk;
1316 0 : ulong tspub = fd_frag_meta_ts_comp( fd_tickcount() );
1317 0 : fd_stem_publish( stem, ctx->out.out_idx, msg_type, chunk, sz, 0UL, 0UL, tspub );
1318 0 : ctx->out.chunk = fd_dcache_compact_next( chunk, sz, ctx->out.chunk0, ctx->out.wmark );
1319 0 : snapmk_out_wake( ctx, stem );
1320 0 : }
1321 :
1322 : static int
1323 0 : snapshot_zstd_strategy( ZSTD_strategy strategy ) {
1324 0 : switch( strategy ) {
1325 0 : case ZSTD_fast: return FD_EVENT_SNAPSHOT_CREATED_ZSTD_STRATEGY_FAST;
1326 0 : case ZSTD_dfast: return FD_EVENT_SNAPSHOT_CREATED_ZSTD_STRATEGY_DFAST;
1327 0 : case ZSTD_greedy: return FD_EVENT_SNAPSHOT_CREATED_ZSTD_STRATEGY_GREEDY;
1328 0 : case ZSTD_lazy: return FD_EVENT_SNAPSHOT_CREATED_ZSTD_STRATEGY_LAZY;
1329 0 : case ZSTD_lazy2: return FD_EVENT_SNAPSHOT_CREATED_ZSTD_STRATEGY_LAZY2;
1330 0 : case ZSTD_btlazy2: return FD_EVENT_SNAPSHOT_CREATED_ZSTD_STRATEGY_BTLAZY2;
1331 0 : case ZSTD_btopt: return FD_EVENT_SNAPSHOT_CREATED_ZSTD_STRATEGY_BTOPT;
1332 0 : case ZSTD_btultra: return FD_EVENT_SNAPSHOT_CREATED_ZSTD_STRATEGY_BTULTRA;
1333 0 : case ZSTD_btultra2: return FD_EVENT_SNAPSHOT_CREATED_ZSTD_STRATEGY_BTULTRA2;
1334 0 : default: FD_LOG_CRIT(( "unexpected Zstandard strategy %i", (int)strategy ));
1335 0 : }
1336 0 : }
1337 :
1338 : static void
1339 : snapshot_event_init( fd_snapmk_t const * ctx,
1340 : int result,
1341 0 : fd_event_snapshot_created_t * ev ) {
1342 0 : *ev = (fd_event_snapshot_created_t) {
1343 0 : .result = result,
1344 0 : .slot = ctx->bank->f.slot,
1345 0 : .base_slot = ctx->incremental ? ctx->base_slot : 0UL,
1346 0 : .epoch = ctx->bank->f.epoch,
1347 0 : .block_height = ctx->bank->f.block_height,
1348 0 : .capitalization = ctx->bank->f.capitalization,
1349 0 : .transaction_count = ctx->bank->f.parent_txn_count + ctx->bank->f.txn_count,
1350 0 : .start_time = (ulong)ctx->start_time,
1351 0 : .accounts_start_time = (ulong)ctx->accounts_start_time,
1352 0 : .end_time = (ulong)ctx->end_time
1353 0 : };
1354 0 : if( FD_LIKELY( result==FD_EVENT_SNAPSHOT_CREATED_RESULT_SUCCESS ) ) {
1355 0 : ev->filename_len = fd_cstr_nlen( ctx->final_name, sizeof(ctx->final_name) );
1356 0 : fd_memcpy( ev->filename, ctx->final_name, ev->filename_len );
1357 0 : }
1358 0 : fd_memcpy( ev->hash, ctx->snap_hash, sizeof(ev->hash) );
1359 0 : fd_memcpy( ev->bank_hash, ctx->bank->f.bank_hash.uc, sizeof(ev->bank_hash) );
1360 0 : fd_memcpy( ev->block_id, ctx->bank->f.block_id.uc, sizeof(ev->block_id) );
1361 0 : fd_memcpy( ev->bank_accounts_lthash, ctx->bank->f.lthash.bytes, sizeof(ev->bank_accounts_lthash) );
1362 0 : ev->bank_accounts_lthash_len = sizeof(ev->bank_accounts_lthash);
1363 0 : ZSTD_compressionParameters zstd_params = ZSTD_getCParams( FD_BACKUP_ZSTD_LEVEL, ZSTD_CONTENTSIZE_UNKNOWN, 0UL );
1364 0 : ev->zstd_window_size = 1UL << zstd_params.windowLog;
1365 0 : ev->zstd_strategy = snapshot_zstd_strategy( zstd_params.strategy );
1366 0 : }
1367 :
1368 : /* after_credit runs every run loop iteration, provided that all out
1369 : links have at least STEM_BURST credit available, or check_credit
1370 : passed. */
1371 :
1372 : static void
1373 : after_credit( fd_snapmk_t * ctx,
1374 : fd_stem_context_t * stem,
1375 : int * poll_in,
1376 0 : int * charge_busy ) {
1377 0 : (void)poll_in;
1378 :
1379 0 : switch( ctx->state ) {
1380 0 : case SNAPMK_STATE_IDLE:
1381 0 : if( FD_UNLIKELY( ++ctx->idle_iter >= IDLE_THRES ) ) {
1382 0 : ctx->state = SNAPMK_STATE_SLEEP;
1383 0 : ctx->idle_iter = 0UL;
1384 0 : }
1385 0 : *charge_busy = 0;
1386 0 : break;
1387 0 : case SNAPMK_STATE_SLEEP: {
1388 0 : snapmk_replay_sleep( ctx );
1389 0 : *charge_busy = 0;
1390 0 : break;
1391 0 : }
1392 0 : case SNAPMK_STATE_START: {
1393 0 : ulong zp_cnt = ctx->zp_cnt;
1394 0 : int did_work = 0;
1395 0 : for( ulong i=0UL; i<zp_cnt; i++ ) {
1396 : /* FIXME use find_lsb? */
1397 0 : if( !fd_ulong_extract_bit( ctx->zp_flush_pending, (int)i ) ) continue;
1398 0 : if( !stem->cr_avail[ i ] ) continue;
1399 0 : ulong chunk;
1400 0 : fd_backup_start_msg_t * frag = zp_alloc( ctx, i, sizeof(fd_backup_start_msg_t), &chunk );
1401 0 : memset( frag, 0, sizeof(fd_backup_start_msg_t) );
1402 0 : frag->slot = ctx->bank->f.slot;
1403 0 : frag->slot_lo = ctx->incremental ? ctx->base_slot+1UL : 0UL;
1404 0 : frag->snap_idx = ctx->snap_idx;
1405 0 : frag->fork_id = ctx->bank->accdb_fork_id.val;
1406 0 : ulong ctl = fd_frag_meta_ctl( FD_BACKUP_ORIG_START, 0, 0, 0 );
1407 0 : zp_publish( ctx, stem, i, 0UL, chunk, sizeof(fd_backup_start_msg_t), ctl, 0UL, 0UL );
1408 0 : ctx->zp_flush_pending &= ~fd_ulong_mask_bit( (int)i );
1409 0 : ctx->zp_ready &= ~fd_ulong_mask_bit( (int)i );
1410 0 : *charge_busy = 1;
1411 0 : did_work = 1;
1412 0 : }
1413 : /* all snapzp tiles have been told to start; begin writing the tar */
1414 0 : if( (!ctx->zp_flush_pending) & (!did_work) ) {
1415 0 : ctx->state = SNAPMK_STATE_TAR_HEADERS;
1416 0 : }
1417 0 : break;
1418 0 : }
1419 0 : case SNAPMK_STATE_TAR_HEADERS:
1420 0 : *charge_busy = 1;
1421 0 : snapmk_tar_headers( ctx );
1422 0 : ctx->state = SNAPMK_STATE_MANIFEST;
1423 0 : break;
1424 0 : case SNAPMK_STATE_MANIFEST:
1425 0 : *charge_busy = 1;
1426 0 : if( FD_UNLIKELY( !snapmk_manifest_chunk( ctx ) ) ) {
1427 0 : ctx->state = ctx->incremental ? SNAPMK_STATE_ACCDB_DELTA : SNAPMK_STATE_ACCDB_CACHE;
1428 0 : }
1429 0 : break;
1430 0 : case SNAPMK_STATE_ACCDB_CACHE: {
1431 0 : *charge_busy = 1;
1432 0 : if( FD_UNLIKELY( !snapmk_accdb_cache( ctx, stem ) ) ) {
1433 0 : barrier_install( ctx, stem );
1434 0 : broadcast_prepare( ctx );
1435 0 : ctx->state = SNAPMK_STATE_ACCDB_CACHE_FLUSH;
1436 0 : }
1437 0 : break;
1438 0 : }
1439 0 : case SNAPMK_STATE_ACCDB_CACHE_FLUSH: {
1440 : /* done reading from cache; now tell snapzp workers to end their
1441 : Zstandard frames */
1442 0 : ulong ctl = fd_frag_meta_ctl( FD_BACKUP_ORIG_FLUSH, 0, 0, 0 );
1443 0 : if( broadcast( ctx, stem, ctl, charge_busy ) ) {
1444 0 : ctx->state = SNAPMK_STATE_ACCDB_CACHE_FINISH;
1445 0 : }
1446 0 : break;
1447 0 : }
1448 0 : case SNAPMK_STATE_ACCDB_CACHE_FINISH:
1449 0 : *charge_busy = 1;
1450 0 : clean_overruns( ctx );
1451 : /* done snapshotting accdb cache;
1452 : now instruct snaprd tile to start reading accdb disk data */
1453 0 : ctx->state = SNAPMK_STATE_ACCDB_DISK;
1454 0 : atomic_fetch_add_explicit( ctx->rd_ctl, 1UL, memory_order_release );
1455 0 : break;
1456 0 : case SNAPMK_STATE_ACCDB_DISK:
1457 : /* driven by returnable_frag */
1458 0 : break;
1459 0 : case SNAPMK_STATE_ACCDB_DISK_FLUSH: {
1460 : /* done reading from disk; now tell snapzp workers to end their
1461 : Zstandard frames */
1462 0 : ulong ctl = fd_frag_meta_ctl( FD_BACKUP_ORIG_FLUSH, 0, 0, 0 );
1463 0 : if( broadcast( ctx, stem, ctl, charge_busy ) ) {
1464 0 : ctx->state = SNAPMK_STATE_ACCDB_DISK_FINISH;
1465 0 : }
1466 0 : break;
1467 0 : }
1468 0 : case SNAPMK_STATE_ACCDB_DELTA: {
1469 0 : *charge_busy = 1;
1470 0 : if( FD_UNLIKELY( !snapmk_accdb_delta( ctx, stem ) ) ) {
1471 0 : barrier_install( ctx, stem );
1472 0 : broadcast_prepare( ctx );
1473 0 : ctx->state = SNAPMK_STATE_ACCDB_DELTA_FLUSH;
1474 0 : }
1475 0 : break;
1476 0 : }
1477 0 : case SNAPMK_STATE_ACCDB_DELTA_FLUSH: {
1478 0 : ulong ctl = fd_frag_meta_ctl( FD_BACKUP_ORIG_FLUSH, 0, 0, 0 );
1479 0 : if( broadcast( ctx, stem, ctl, charge_busy ) ) {
1480 0 : ctx->state = SNAPMK_STATE_ACCDB_DELTA_FINISH;
1481 0 : }
1482 0 : break;
1483 0 : }
1484 0 : case SNAPMK_STATE_ACCDB_DISK_FINISH:
1485 0 : case SNAPMK_STATE_ACCDB_DELTA_FINISH: {
1486 : /* accounts done, snapzp workers idle; now process status cache */
1487 0 : long file_sz = lseek( ctx->snap_fd, 0L, SEEK_END );
1488 0 : if( FD_UNLIKELY( file_sz<0L ) ) {
1489 0 : FD_LOG_ERR(( "lseek failed: %i-%s", errno, fd_io_strerror( errno ) ));
1490 0 : }
1491 0 : ctx->pool_sz[ ctx->snap_idx ] = (ulong)file_sz;
1492 0 : snapmk_status_cache_prepare( ctx );
1493 0 : ctx->state = SNAPMK_STATE_STATUS_CACHE;
1494 0 : break;
1495 0 : }
1496 0 : case SNAPMK_STATE_STATUS_CACHE:
1497 : /* process status cache piece wise */
1498 0 : *charge_busy = 1;
1499 0 : if( FD_UNLIKELY( !snapmk_status_cache( ctx ) ) ) {
1500 0 : ctx->state = SNAPMK_STATE_EOF_MARKER;
1501 0 : }
1502 0 : break;
1503 0 : case SNAPMK_STATE_EOF_MARKER:
1504 : /* all data written to snapshot, file not yet structurally clean;
1505 : now write end-of-snapshot marker */
1506 0 : *charge_busy = 1;
1507 0 : snapmk_eof_marker ( ctx );
1508 0 : snapmk_done_rename( ctx );
1509 0 : broadcast_prepare ( ctx );
1510 0 : ctx->state = SNAPMK_STATE_DONE;
1511 0 : break;
1512 0 : case SNAPMK_STATE_DONE: {
1513 : /* snapshot file complete; now broadcast "done" signal to all worker
1514 : tiles, and notify accdb/replay to resume */
1515 0 : ulong ctl = fd_frag_meta_ctl( FD_BACKUP_ORIG_DONE, 0, 1, 0 );
1516 0 : if( broadcast( ctx, stem, ctl, charge_busy ) ) {
1517 0 : snapshot_sync_transition( ctx, FD_ACCDB_SNAPSHOT_SYNC_RUNNING, FD_ACCDB_SNAPSHOT_SYNC_DONE, FD_ACCDB_SNAPSHOT_SYNC_IDLE );
1518 0 : fd_snapmk_msg_created_t * msg = &snapmk_msg_alloc( ctx )->created;
1519 0 : *msg = (fd_snapmk_msg_created_t) {
1520 0 : .slot = ctx->bank->f.slot,
1521 0 : .base_slot = ctx->incremental ? ctx->base_slot : ULONG_MAX,
1522 0 : .sz = ctx->final_sz,
1523 0 : .pool_idx = ctx->snap_idx
1524 0 : };
1525 0 : fd_cstr_ncpy( msg->name, ctx->final_name, sizeof(msg->name) );
1526 :
1527 0 : ulong account_count = 0UL;
1528 0 : ulong accounts_size = 0UL;
1529 0 : ulong tombstone_count = 0UL;
1530 0 : ulong cached_accounts_count = 0UL;
1531 0 : ulong disk_accounts_count = 0UL;
1532 0 : ulong zstd_data_frame_count = ctx->zstd_data_frame_cnt;
1533 0 : ulong zstd_padding_size = ctx->zstd_padding_sz;
1534 0 : ulong uncompressed_size = ctx->uncompressed_sz;
1535 0 : ulong compress_ticks = ctx->metrics.compress_ticks - ctx->snapshot_compress_ticks0;
1536 0 : ulong io_blocked_ticks = ctx->metrics.io_blocked_ticks - ctx->snapshot_io_blocked_ticks0;
1537 0 : for( ulong i=0UL; i<ctx->zp_cnt; i++ ) {
1538 0 : fd_backup_worker_stats_t const * stats = &ctx->stats->worker[ i ];
1539 0 : uncompressed_size += __atomic_load_n( &stats->uncompressed_sz, __ATOMIC_ACQUIRE );
1540 0 : account_count += __atomic_load_n( &stats->account_cnt, __ATOMIC_RELAXED );
1541 0 : accounts_size += __atomic_load_n( &stats->account_sz, __ATOMIC_RELAXED );
1542 0 : tombstone_count += __atomic_load_n( &stats->tombstone_cnt, __ATOMIC_RELAXED );
1543 0 : cached_accounts_count += __atomic_load_n( &stats->cached_account_cnt, __ATOMIC_RELAXED );
1544 0 : disk_accounts_count += __atomic_load_n( &stats->disk_account_cnt, __ATOMIC_RELAXED );
1545 0 : zstd_data_frame_count += __atomic_load_n( &stats->zstd_data_frame_cnt, __ATOMIC_RELAXED );
1546 0 : zstd_padding_size += __atomic_load_n( &stats->zstd_padding_sz, __ATOMIC_RELAXED );
1547 0 : compress_ticks += __atomic_load_n( &stats->compress_ticks, __ATOMIC_RELAXED );
1548 0 : io_blocked_ticks += __atomic_load_n( &stats->io_blocked_ticks, __ATOMIC_RELAXED );
1549 0 : }
1550 0 : FD_TEST( account_count==tombstone_count+cached_accounts_count+disk_accounts_count );
1551 0 : fd_event_snapshot_created_t ev[1];
1552 0 : snapshot_event_init( ctx, FD_EVENT_SNAPSHOT_CREATED_RESULT_SUCCESS, ev );
1553 0 : ev->account_count = account_count;
1554 0 : ev->tombstone_count = tombstone_count;
1555 0 : ev->cached_accounts_count = cached_accounts_count;
1556 0 : ev->disk_accounts_count = disk_accounts_count;
1557 0 : ev->manifest_size = ctx->manifest_sz;
1558 0 : ev->accounts_size = accounts_size;
1559 0 : ev->status_cache_size = ctx->status_cache_sz;
1560 0 : ev->compressed_size = ctx->final_sz;
1561 0 : ev->uncompressed_size = uncompressed_size;
1562 0 : ev->zstd_data_frame_count = zstd_data_frame_count;
1563 0 : ev->zstd_padding_size = zstd_padding_size;
1564 0 : ev->duration_compress_nanos = fd_metrics_convert_ticks_to_nanoseconds( compress_ticks );
1565 0 : ev->duration_io_blocked_nanos = fd_metrics_convert_ticks_to_nanoseconds( io_blocked_ticks );
1566 :
1567 0 : snapmk_msg_publish( ctx, stem, FD_SNAPMK_MSG_CREATED );
1568 0 : fd_event_report_snapshot_created( ev );
1569 :
1570 0 : ctx->state = SNAPMK_STATE_SLEEP;
1571 0 : ctx->snap_idx = UINT_MAX;
1572 0 : }
1573 0 : break;
1574 0 : }
1575 0 : case SNAPMK_STATE_FAIL: {
1576 0 : fd_event_snapshot_created_t ev[1];
1577 0 : snapshot_event_init( ctx, FD_EVENT_SNAPSHOT_CREATED_RESULT_TOO_MANY_INCREMENTAL_ACCOUNTS, ev );
1578 0 : snapmk_msg_alloc( ctx )->failed = (fd_snapmk_msg_failed_t) {
1579 0 : .slot = ctx->bank->f.slot,
1580 0 : .base_slot = ctx->base_slot
1581 0 : };
1582 0 : snapmk_msg_publish( ctx, stem, FD_SNAPMK_MSG_FAILED );
1583 0 : fd_event_report_snapshot_created( ev );
1584 0 : ctx->snap_idx = UINT_MAX;
1585 0 : ctx->state = SNAPMK_STATE_SLEEP;
1586 0 : *charge_busy = 1;
1587 0 : break;
1588 0 : }
1589 0 : case SNAPMK_STATE_STARTUP: /* wait for startup */
1590 0 : if( FD_UNLIKELY( fd_startup_gate_idle( ctx->startup_gate ) ) ) {
1591 0 : ctx->state = SNAPMK_STATE_STARTUP_BURST;
1592 0 : fd_snap_pool_recover( ctx->snap_dir_fd, ctx->snap_dir, ctx->pool, ctx->snap_max );
1593 0 : }
1594 0 : break;
1595 0 : case SNAPMK_STATE_STARTUP_BURST: { /* burst publish pre-existing snaps */
1596 0 : if( FD_UNLIKELY( ctx->startup_pool_idx >= ctx->snap_max ) ) {
1597 0 : ctx->state = SNAPMK_STATE_SLEEP;
1598 0 : break;
1599 0 : }
1600 0 : ulong snap_idx = ctx->startup_pool_idx++;
1601 0 : fd_backup_inode_t * inode = &ctx->pool[ snap_idx ];
1602 0 : struct stat st;
1603 0 : if( FD_UNLIKELY( 0!=syscall( SYS_fstat, FD_SNAP_FD( snap_idx ), &st ) ) ) break;
1604 0 : ctx->pool_sz[ snap_idx ] = (ulong)st.st_size;
1605 0 : if( FD_UNLIKELY( inode->full_slot==ULONG_MAX ) ) break;
1606 0 : fd_snapmk_msg_found_t * msg = &snapmk_msg_alloc( ctx )->found;
1607 0 : *msg = (fd_snapmk_msg_found_t) {
1608 0 : .slot = inode->incr_slot!=ULONG_MAX ? inode->incr_slot : inode->full_slot,
1609 0 : .base_slot = inode->incr_slot!=ULONG_MAX ? inode->full_slot : ULONG_MAX,
1610 0 : .pool_idx = (uint)snap_idx,
1611 0 : .fs_timestamp = LONG_MAX
1612 0 : };
1613 0 : fd_cstr_ncpy( msg->name, inode->name, sizeof(msg->name) );
1614 0 : msg->sz = (ulong)st.st_size;
1615 0 : msg->fs_timestamp = ((long)st.st_mtim.tv_sec*(long)1e9) + (long)st.st_mtim.tv_nsec;
1616 0 : snapmk_msg_publish( ctx, stem, FD_SNAPMK_MSG_FOUND );
1617 0 : break;
1618 0 : }
1619 0 : default:
1620 0 : FD_LOG_CRIT(( "invalid state %u", ctx->state ));
1621 0 : }
1622 0 : }
1623 :
1624 : /* snap_pool_select finds a free snapshot file descriptor for
1625 : production. */
1626 :
1627 : static uint
1628 0 : snap_pool_select( fd_snapmk_t * ctx ) {
1629 :
1630 0 : uint slot0 = ctx->incremental ? ctx->snap_full_max : 0U;
1631 0 : uint slot1 = ctx->incremental ? ctx->snap_max : ctx->snap_full_max;
1632 0 : FD_CHECK_ERR( slot0<slot1, "no snapshot file descriptors reserved" );
1633 :
1634 : /* if this snapshot already exists, recreate it */
1635 0 : for( uint i=slot0; i<slot1; i++ ) {
1636 0 : if( FD_UNLIKELY( !strcmp( ctx->pool[ i ].name, ctx->final_name ) ) ) return i;
1637 0 : }
1638 :
1639 0 : uint slot_idx = UINT_MAX;
1640 0 : ulong oldest = ULONG_MAX;
1641 0 : for( uint i=slot0; i<slot1; i++ ) {
1642 0 : if( FD_UNLIKELY( ctx->pool[ i ].full_slot==ULONG_MAX ) ) return i; /* free */
1643 0 : ulong slot = ctx->incremental ? ctx->pool[ i ].incr_slot : ctx->pool[ i ].full_slot;
1644 0 : if( slot<oldest ) {
1645 0 : oldest = slot;
1646 0 : slot_idx = i;
1647 0 : }
1648 0 : }
1649 0 : FD_CHECK_ERR( slot_idx!=UINT_MAX, "no snapshot file descriptor to recycle" );
1650 :
1651 0 : return slot_idx;
1652 0 : }
1653 :
1654 : /* snap_pool_acquire picks a new file descriptor to hold a snapshot
1655 : file. May recycle an existing snapshot. */
1656 :
1657 : static uint
1658 : snap_pool_acquire( fd_snapmk_t * ctx,
1659 0 : fd_stem_context_t * stem ) {
1660 0 : uint snap_pool_idx = snap_pool_select( ctx );
1661 :
1662 0 : if( FD_UNLIKELY( ctx->pool[ snap_pool_idx ].full_slot==ULONG_MAX ) ) return snap_pool_idx; /* free */
1663 :
1664 : /* recycle (signals consumers to unlock) */
1665 :
1666 0 : FD_CHECK_ERR( snap_pool_idx < ctx->snap_max, "invalid snap_pool_idx" );
1667 :
1668 0 : fd_backup_inode_t * inode = &ctx->pool[ snap_pool_idx ];
1669 0 : fd_snapmk_msg_deleted_t * msg = &snapmk_msg_alloc( ctx )->deleted;
1670 0 : *msg = (fd_snapmk_msg_deleted_t) {
1671 0 : .slot = inode->incr_slot!=ULONG_MAX ? inode->incr_slot : inode->full_slot,
1672 0 : .base_slot = inode->incr_slot!=ULONG_MAX ? inode->full_slot : ULONG_MAX,
1673 0 : .pool_idx = snap_pool_idx
1674 0 : };
1675 0 : fd_cstr_ncpy( msg->name, inode->name, sizeof(msg->name) );
1676 0 : snapmk_msg_publish( ctx, stem, FD_SNAPMK_MSG_DELETED );
1677 :
1678 : /* do a blocking wait for the file to become free */
1679 :
1680 0 : int snap_fd = FD_SNAP_FD( snap_pool_idx );
1681 0 : struct flock lock = {
1682 0 : .l_type = F_WRLCK,
1683 0 : .l_whence = SEEK_SET
1684 0 : };
1685 0 : if( FD_UNLIKELY( fcntl( snap_fd, F_SETLKW, &lock ) ) ) {
1686 0 : FD_LOG_ERR(( "fcntl(F_SETLKW, %s) failed: %i-%s",
1687 0 : ctx->pool[ snap_pool_idx ].name, errno, fd_io_strerror( errno ) ));
1688 0 : }
1689 :
1690 0 : FD_LOG_INFO(( "evicting old snapshot file: %s", inode->name ));
1691 0 : if( FD_UNLIKELY( ftruncate( snap_fd, 0L ) ) ) {
1692 0 : FD_LOG_ERR(( "ftruncate(%s) failed: %s", inode->name, fd_io_strerror( errno ) ));
1693 0 : }
1694 0 : ctx->pool_sz[ snap_pool_idx ] = 0UL;
1695 :
1696 0 : char partial_name[ sizeof(inode->name) ];
1697 0 : fd_snap_pool_partial_name( partial_name, snap_pool_idx );
1698 0 : if( FD_UNLIKELY( syscall( SYS_renameat2, ctx->snap_dir_fd, inode->name,
1699 0 : ctx->snap_dir_fd, partial_name, 0U ) ) ) {
1700 0 : FD_LOG_ERR(( "renameat2(%s, %s) failed: %s", inode->name, partial_name, fd_io_strerror( errno ) ));
1701 0 : }
1702 0 : fd_cstr_ncpy( inode->name, partial_name, sizeof(inode->name) );
1703 0 : inode->full_slot = ULONG_MAX;
1704 0 : inode->incr_slot = ULONG_MAX;
1705 0 : return snap_pool_idx;
1706 0 : }
1707 :
1708 : /* snap_start boots the snap production pipeline.
1709 : Returns:
1710 : - 1 if snapshot production was started
1711 : - 0 if system is not ready yet, and start attempt should be retried
1712 : - -1 if attempt was rejected */
1713 :
1714 : static int
1715 : snap_start( fd_snapmk_t * ctx,
1716 : fd_stem_context_t * stem,
1717 0 : fd_replay_snap_start_t const * msg ) {
1718 0 : switch( ctx->state ) {
1719 0 : case SNAPMK_STATE_IDLE:
1720 0 : case SNAPMK_STATE_SLEEP:
1721 0 : break;
1722 0 : case SNAPMK_STATE_STARTUP:
1723 0 : case SNAPMK_STATE_STARTUP_BURST:
1724 0 : return 0; /* not ready yet */
1725 0 : default:
1726 0 : FD_LOG_ERR(( "invariant violation: snapshot creation requested state is %u", ctx->state ));
1727 0 : }
1728 :
1729 0 : fd_bank_t * bank = fd_banks_bank_query( ctx->banks, msg->bank_idx );
1730 0 : FD_TEST( bank );
1731 0 : ctx->bank = bank;
1732 :
1733 0 : int incremental = msg->slot!=msg->base_slot;
1734 0 : if( FD_UNLIKELY( incremental ) ) {
1735 0 : FD_CHECK_CRIT( msg->base_slot!=ULONG_MAX, "incremental snapshot requested without a base full snapshot" );
1736 0 : ctx->base_slot = msg->base_slot;
1737 0 : } else {
1738 0 : ctx->base_slot = ULONG_MAX;
1739 0 : }
1740 0 : ctx->incremental = incremental;
1741 0 : ctx->start_time = fd_log_wallclock();
1742 0 : ctx->accounts_start_time = 0L;
1743 0 : ctx->end_time = 0L;
1744 0 : fd_blake3_hash( ctx->bank->f.lthash.bytes, FD_LTHASH_LEN_BYTES, ctx->snap_hash );
1745 :
1746 : /* wait for accdb root to match published root */
1747 0 : fd_accdb_fork_id_t root_fork_id = bank->accdb_fork_id;
1748 0 : FD_TEST( root_fork_id.val!=USHORT_MAX );
1749 0 : if( FD_UNLIKELY( __atomic_load_n( &ctx->accdb_root_fork->val, __ATOMIC_ACQUIRE )!=root_fork_id.val ) ) {
1750 0 : return 0; /* not ready */
1751 0 : }
1752 0 : ulong root_generation = __atomic_load_n( &ctx->accdb_shfork[ root_fork_id.val ].generation, __ATOMIC_ACQUIRE );
1753 :
1754 : /* wait for accdb to disable compaction */
1755 0 : ulong sync_req = incremental ? FD_ACCDB_SNAPSHOT_SYNC_START_INCR
1756 0 : : FD_ACCDB_SNAPSHOT_SYNC_START_FULL;
1757 0 : ulong sync_ack = snapshot_sync_request( ctx, FD_ACCDB_SNAPSHOT_SYNC_IDLE, sync_req );
1758 0 : if( FD_UNLIKELY( sync_ack==FD_ACCDB_SNAPSHOT_SYNC_FAIL ) ) {
1759 0 : FD_LOG_WARNING(( "cannot create incremental snapshot, too many accounts changed (increase [snapshots.max_incremental_snapshot_accounts])" ));
1760 0 : snapshot_sync_transition( ctx, FD_ACCDB_SNAPSHOT_SYNC_FAIL, FD_ACCDB_SNAPSHOT_SYNC_DONE, FD_ACCDB_SNAPSHOT_SYNC_IDLE );
1761 0 : ctx->end_time = fd_log_wallclock();
1762 0 : ctx->state = SNAPMK_STATE_FAIL;
1763 0 : return -1;
1764 0 : }
1765 0 : if( FD_UNLIKELY( sync_ack!=FD_ACCDB_SNAPSHOT_SYNC_RUNNING ) ) {
1766 0 : FD_LOG_CRIT(( "unexpected accdb snapshot sync state %lu", sync_ack ));
1767 0 : }
1768 :
1769 : /* user might have changed available snapshots */
1770 0 : fd_snap_pool_recover( ctx->snap_dir_fd, ctx->snap_dir, ctx->pool, ctx->snap_max );
1771 :
1772 : /* final name of snap (during compression has a "partial" name) */
1773 0 : char encoded_hash[ FD_BASE58_ENCODED_32_SZ ];
1774 0 : fd_base58_encode_32( ctx->snap_hash, NULL, encoded_hash );
1775 0 : if( FD_UNLIKELY( incremental ) ) {
1776 0 : FD_TEST( fd_cstr_printf_check( ctx->final_name, FD_SNAP_NAME_MAX, NULL,
1777 0 : "incremental-snapshot-%lu-%lu-%s.tar.zst", ctx->base_slot, ctx->bank->f.slot, encoded_hash ) );
1778 0 : } else {
1779 0 : FD_TEST( fd_cstr_printf_check( ctx->final_name, FD_SNAP_NAME_MAX, NULL,
1780 0 : "snapshot-%lu-%s.tar.zst", ctx->bank->f.slot, encoded_hash ) );
1781 0 : }
1782 :
1783 0 : uint snap_idx = snap_pool_acquire( ctx, stem );
1784 0 : if( FD_UNLIKELY( snap_idx==UINT_MAX ) ) {
1785 0 : snapshot_sync_transition( ctx, FD_ACCDB_SNAPSHOT_SYNC_RUNNING, FD_ACCDB_SNAPSHOT_SYNC_DONE, FD_ACCDB_SNAPSHOT_SYNC_IDLE );
1786 0 : return 0; /* not ready */
1787 0 : }
1788 0 : ctx->snap_idx = snap_idx;
1789 0 : ctx->snap_fd = FD_SNAP_FD( ctx->snap_idx );
1790 :
1791 0 : snapmk_msg_alloc( ctx )->started = (fd_snapmk_msg_started_t) {
1792 0 : .slot = ctx->bank->f.slot,
1793 0 : .base_slot = ctx->base_slot,
1794 0 : .pool_idx = ctx->snap_idx
1795 0 : };
1796 0 : snapmk_msg_publish( ctx, stem, FD_SNAPMK_MSG_STARTED );
1797 :
1798 0 : if( FD_UNLIKELY( ftruncate( ctx->snap_fd, 0L ) ) ) {
1799 0 : FD_LOG_ERR(( "ftruncate(%s) failed: %i-%s", ctx->pool[ ctx->snap_idx ].name, errno, fd_io_strerror( errno ) ));
1800 0 : }
1801 0 : ctx->pool_sz[ ctx->snap_idx ] = 0UL;
1802 0 : if( FD_UNLIKELY( lseek( ctx->snap_fd, 0L, SEEK_SET )<0L ) ) {
1803 0 : FD_LOG_ERR(( "lseek(%s) failed: %i-%s", ctx->pool[ ctx->snap_idx ].name, errno, fd_io_strerror( errno ) ));
1804 0 : }
1805 :
1806 0 : atomic_store_explicit( ctx->file_off_p, 0UL, memory_order_relaxed );
1807 0 : atomic_store_explicit( ctx->appendvec_slot_ticket_p, ctx->bank->f.slot, memory_order_relaxed );
1808 :
1809 : /* compression buffers */
1810 :
1811 0 : ctx->raw_buf.size = 0UL;
1812 0 : ctx->raw_buf.pos = 0UL;
1813 0 : ctx->comp_buf.pos = 0UL;
1814 0 : ctx->comp_buf.size = COMP_BUF_SZ;
1815 0 : ctx->zstd_data_frame_cnt = 0UL;
1816 0 : ctx->zstd_padding_sz = 0UL;
1817 0 : ctx->uncompressed_sz = 0UL;
1818 0 : ctx->snapshot_compress_ticks0 = ctx->metrics.compress_ticks;
1819 0 : ctx->snapshot_io_blocked_ticks0 = ctx->metrics.io_blocked_ticks;
1820 0 : ulong zst_err = ZSTD_CCtx_reset( ctx->zst, ZSTD_reset_session_only );
1821 0 : if( FD_UNLIKELY( ZSTD_isError( zst_err ) ) ) {
1822 0 : FD_LOG_ERR(( "ZSTD_CCtx_reset failed: %s", ZSTD_getErrorName( zst_err ) ));
1823 0 : }
1824 :
1825 : /* misc */
1826 :
1827 0 : ctx->leader = msg->leader;
1828 0 : fd_ssmanifest_writer_init( ctx->manifest_writer, bank, &ctx->leader, ctx->accdb, root_fork_id, ctx->raw );
1829 :
1830 : /* accdb cache/disk parsers */
1831 :
1832 0 : fd_backup_cache_reset( ctx->acc_cache, root_generation );
1833 0 : *ctx->accparse = (fd_snapmk_accparse_t) {
1834 0 : .idx = ctx->acc_cache->idx, /* reset above, incl root_generation */
1835 0 : .acc_keep = 1U,
1836 0 : .visited_set = ctx->visited_set
1837 0 : };
1838 :
1839 0 : ctx->delta.chain_idx = 0UL;
1840 0 : ctx->delta.ele_idx = UINT_MAX;
1841 0 : ctx->delta.ele_cnt = 0UL;
1842 0 : ctx->delta.ele_total = incremental ? __atomic_load_n( &ctx->accdb_shmem->delta.head, __ATOMIC_RELAXED ) : 0UL;
1843 :
1844 0 : visited_set_null( ctx->visited_set );
1845 :
1846 0 : ctx->state = SNAPMK_STATE_START;
1847 0 : ctx->zp_ready = 0UL;
1848 0 : ctx->disk_out_idx = -1;
1849 0 : ctx->disk_batch_pending = 0;
1850 0 : if( FD_UNLIKELY( incremental ) ) {
1851 0 : ctx->metrics.snapshots_started[ FD_METRICS_ENUM_SNAP_TYPE_V_INCREMENTAL_IDX ]++;
1852 0 : ctx->metrics.last_snapshot_slot_started_incremental = ctx->bank->f.slot;
1853 0 : } else {
1854 0 : ctx->metrics.snapshots_started[ FD_METRICS_ENUM_SNAP_TYPE_V_FULL_IDX ]++;
1855 0 : ctx->metrics.last_snapshot_slot_started_full = ctx->bank->f.slot;
1856 0 : }
1857 0 : broadcast_prepare( ctx );
1858 :
1859 0 : if( FD_UNLIKELY( incremental ) ) {
1860 0 : FD_LOG_INFO(( "incremental snapshot creation started (slot %lu, base slot %lu)",
1861 0 : ctx->bank->f.slot, ctx->base_slot ));
1862 0 : } else {
1863 0 : FD_LOG_INFO(( "snapshot creation started (slot %lu)", ctx->bank->f.slot ));
1864 0 : }
1865 0 : return 1;
1866 0 : }
1867 :
1868 : /* snapzp_stamp_shadow tracks the snaprd frag seq corresponding to an
1869 : upcoming snapmk_zp publish. */
1870 :
1871 : static inline void
1872 : snapzp_stamp_shadow( fd_snapmk_t * ctx,
1873 : ulong out_idx,
1874 0 : ulong pub_seq ) {
1875 0 : ctx->rd_shadow[ out_idx ][ pub_seq & (FD_SNAPMK_ZP_DEPTH-1UL) ] = ctx->rd_seq;
1876 0 : }
1877 :
1878 : /* snaprd_frag ingests a new accdb disk data frag from snaprd. */
1879 :
1880 : static int
1881 : snaprd_frag( fd_snapmk_t * ctx,
1882 : fd_stem_context_t * stem,
1883 : ulong seq,
1884 : ulong sig,
1885 : ulong chunk,
1886 : ulong ctl,
1887 0 : ulong tspub ) {
1888 0 : ulong frag_sz = tspub;
1889 0 : FD_CHECK_CRIT( ctx->state==SNAPMK_STATE_ACCDB_DISK, "lifecycle bug" );
1890 0 : FD_CHECK_CRIT( fd_frag_meta_ctl_orig( ctl )==FD_BACKUP_ORIG_DISK_FRAG, "unexpected snaprd frag orig" );
1891 0 : FD_CHECK_CRIT( frag_sz<=FD_BACKUP_RD_MTU && frag_sz<UINT_MAX, "invalid snaprd frag data size" );
1892 0 : FD_CHECK_CRIT( frag_sz || fd_frag_meta_ctl_eom( ctl ), "empty snaprd frag" );
1893 :
1894 0 : fd_snapmk_accparse_t * parse = ctx->accparse;
1895 0 : ctx->rd_seq = seq;
1896 0 : if( FD_LIKELY( !parse->input_active ) ) {
1897 0 : uchar const * data = fd_chunk_to_laddr_const( ctx->rd_in_mem, chunk );
1898 0 : parse->data = data;
1899 0 : parse->data_sz = frag_sz;
1900 0 : parse->src_gaddr = fd_wksp_gaddr_fast( ctx->rd_in_mem, data );
1901 0 : parse->src_off = sig;
1902 0 : parse->frag_base_gaddr = fd_wksp_gaddr_fast( ctx->rd_in_mem, data );
1903 0 : parse->pf_cursor = data;
1904 0 : parse->input_active = 1;
1905 0 : }
1906 :
1907 0 : for(;;) {
1908 : /* (A) Flush a previously-staged batch once an output link frees up.
1909 : A batch is self-contained within one snaprd frag, so it may be
1910 : routed to any ready zp tile (no disk_out_idx pinning). */
1911 0 : if( FD_UNLIKELY( ctx->disk_batch_pending ) ) {
1912 0 : if( FD_UNLIKELY( !ctx->zp_ready ) ) return 1;
1913 0 : ulong out_idx = zp_rr_next( ctx );
1914 0 : ulong out_chunk;
1915 0 : void * payload = zp_alloc( ctx, out_idx, sizeof(fd_backup_disk_batch_msg_t), &out_chunk );
1916 0 : fd_memcpy( payload, ctx->disk_batch, sizeof(fd_backup_disk_batch_msg_t) );
1917 0 : ulong ctl_batch = fd_frag_meta_ctl( FD_BACKUP_ORIG_ACC_DISK_BATCH, 1, 1, 0 );
1918 0 : snapzp_stamp_shadow( ctx, out_idx, stem->seqs[ out_idx ] );
1919 0 : zp_publish( ctx, stem, out_idx, ctx->disk_batch_base_gaddr, out_chunk,
1920 0 : sizeof(fd_backup_disk_batch_msg_t), ctl_batch, 0UL, 0UL );
1921 0 : ctx->disk_batch_pending = 0;
1922 0 : return 1;
1923 0 : }
1924 :
1925 :
1926 : /* (B) Stage a batch of wholly-contained accounts (no straddle).
1927 : Staging consumes the accounts into ctx->disk_batch; (A) flushes
1928 : it on the next iteration once credit is available. */
1929 0 : if( FD_LIKELY( ctx->disk_out_idx < 0 ) ) {
1930 0 : ulong n = fd_snapmk_accparse_publish_batch( parse, ctx->disk_batch );
1931 0 : if( n ) {
1932 0 : ctx->disk_batch_pending = 1;
1933 0 : ctx->disk_batch_base_gaddr = parse->frag_base_gaddr;
1934 0 : continue;
1935 0 : }
1936 0 : }
1937 :
1938 : /* (C) Single-account fallback for straddling / mid-record accounts. */
1939 0 : ulong out_idx = (ulong)ctx->disk_out_idx;
1940 0 : if( FD_LIKELY( out_idx<ctx->zp_cnt ) ) {
1941 0 : if( FD_UNLIKELY( !stem->cr_avail[ out_idx ] ) ) return 1;
1942 0 : } else {
1943 0 : if( FD_UNLIKELY( !ctx->zp_ready ) ) return 1;
1944 0 : out_idx = zp_rr_next( ctx );
1945 0 : }
1946 :
1947 0 : fd_frag_meta_t meta[1];
1948 0 : if( FD_UNLIKELY( !fd_snapmk_accparse_publish( parse, meta ) ) ) {
1949 : /* A discarded straddling record can finish before the end of this
1950 : frag. Retry from (B) while the parser is at the clean boundary,
1951 : instead of consuming the rest of the frag through the scalar
1952 : fallback. */
1953 0 : if( FD_LIKELY( parse->data_sz && !parse->acc_active &&
1954 0 : !parse->meta_sz && !parse->pub_pending ) ) continue;
1955 :
1956 0 : parse->input_active = 0;
1957 : /* A prestaged batch references the current frag's bytes and must
1958 : be drained before this frag is released (publish_batch above
1959 : returns 0 only once the prestage is empty). */
1960 0 : FD_CHECK_ERR( !parse->ps_cnt, "prestaged batch outlived its frag" );
1961 0 : if( FD_UNLIKELY( fd_frag_meta_ctl_eom( ctl ) ) ) {
1962 0 : if( FD_UNLIKELY( parse->meta_sz || parse->acc_active || parse->pub_pending ) ) {
1963 0 : FD_LOG_CRIT(( "snaprd stream ended mid-account record at accdb offset %lu "
1964 0 : "(meta_sz=%u acc_active=%i acc_off=%u acc_sz=%u pub_pending=%i)",
1965 0 : parse->src_off, parse->meta_sz, parse->acc_active,
1966 0 : parse->acc_off, parse->acc_sz, parse->pub_pending ));
1967 0 : }
1968 0 : ctx->disk_out_idx = -1;
1969 0 : barrier_install( ctx, stem );
1970 0 : broadcast_prepare( ctx );
1971 0 : ctx->state = SNAPMK_STATE_ACCDB_DISK_FLUSH;
1972 0 : }
1973 0 : return 0;
1974 0 : }
1975 :
1976 : /* An account may straddle multiple snaprd frags. The first frag of
1977 : an account (som) carries the fd_backup_disk_msg_t header and pins
1978 : the account to out_idx; continuation frags carry data only and
1979 : must go to the same zp tile, until eom unpins it. */
1980 0 : int som = fd_frag_meta_ctl_som( meta->ctl );
1981 0 : int eom = fd_frag_meta_ctl_eom( meta->ctl );
1982 0 : ulong out_chunk = 0UL;
1983 0 : ulong out_sz = 0UL;
1984 0 : if( FD_UNLIKELY( som ) ) {
1985 0 : ctx->disk_out_idx = (int)out_idx;
1986 0 : fd_backup_disk_msg_t * frag = zp_alloc( ctx, out_idx, sizeof(fd_backup_disk_msg_t), &out_chunk );
1987 0 : memcpy( frag->pubkey.uc, parse->meta.pubkey, sizeof(fd_pubkey_t) );
1988 0 : memcpy( frag->owner.uc, parse->meta.owner, sizeof(fd_pubkey_t) );
1989 0 : frag->size = parse->meta.size;
1990 0 : frag->acc_idx = parse->acc_idx;
1991 0 : frag->snap_sz = parse->acc_snap_sz;
1992 0 : frag->data_sz = (uint)meta->tspub;
1993 0 : out_sz = sizeof(fd_backup_disk_msg_t);
1994 0 : }
1995 :
1996 0 : snapzp_stamp_shadow( ctx, out_idx, stem->seqs[ out_idx ] );
1997 0 : zp_publish( ctx, stem, out_idx, meta->sig, out_chunk, out_sz, meta->ctl, meta->tsorig, meta->tspub );
1998 0 : if( FD_UNLIKELY( eom ) ) ctx->disk_out_idx = -1;
1999 0 : return 1;
2000 0 : }
2001 0 : }
2002 :
2003 : /* returnable_frag is called for every input frag. */
2004 :
2005 : static int
2006 : returnable_frag( fd_snapmk_t * ctx,
2007 : ulong in_idx,
2008 : ulong seq,
2009 : ulong sig,
2010 : ulong chunk,
2011 : ulong sz,
2012 : ulong ctl,
2013 : ulong tsorig,
2014 : ulong tspub,
2015 0 : fd_stem_context_t * stem ) {
2016 0 : (void)sz; (void)tsorig;
2017 0 : fd_startup_gate_busy( ctx->startup_gate );
2018 0 : switch( ctx->in_kind[ in_idx ] ) {
2019 0 : case IN_KIND_REPLAY:
2020 0 : switch( sig ) {
2021 0 : case REPLAY_SIG_SNAP_START: {
2022 0 : fd_replay_snap_start_t const * msg = fd_chunk_to_laddr_const( ctx->replay_in_mem, chunk );
2023 0 : int res = snap_start( ctx, stem, msg );
2024 0 : if( res==0 ) return 1; /* not ready yet */
2025 0 : break;
2026 0 : }
2027 0 : default:
2028 0 : FD_LOG_ERR(( "unexpected replay_snapmk message (sig=%lu)", sig ));
2029 0 : }
2030 0 : ctx->replay_in_seq_cons = fd_seq_inc( seq, 1UL );
2031 0 : return 0;
2032 0 : case IN_KIND_SNAPRD:
2033 0 : return snaprd_frag( ctx, stem, seq, sig, chunk, ctl, tspub );
2034 0 : default:
2035 0 : FD_LOG_CRIT(( "unexpected msg from link %lu with sig %lu", in_idx, sig ));
2036 0 : }
2037 0 : }
2038 :
2039 : static void
2040 0 : metrics_write( fd_snapmk_t * ctx ) {
2041 0 : ulong disk_alloc_bytes = 0UL;
2042 0 : for( uint i=0U; i<ctx->snap_max; i++ ) disk_alloc_bytes += ctx->pool_sz[ i ];
2043 0 : if( FD_LIKELY( ctx->snap_idx<ctx->snap_max ) ) {
2044 0 : ulong file_off = atomic_load_explicit( ctx->file_off_p, memory_order_relaxed );
2045 0 : disk_alloc_bytes += fd_ulong_sat_sub( file_off, ctx->pool_sz[ ctx->snap_idx ] );
2046 0 : }
2047 :
2048 0 : FD_MGAUGE_SET ( SNAPMK, JOB_STATE, ctx->state );
2049 0 : FD_MCNT_ENUM_COPY( SNAPMK, SNAPSHOT_STARTED, ctx->metrics.snapshots_started );
2050 0 : FD_MCNT_ENUM_COPY( SNAPMK, SNAPSHOT_FINISHED, ctx->metrics.snapshots_finished );
2051 0 : FD_MGAUGE_SET ( SNAPMK, LAST_STARTED_SLOT_FULL, ctx->metrics.last_snapshot_slot_started_full );
2052 0 : FD_MGAUGE_SET ( SNAPMK, LAST_STARTED_SLOT_INCREMENTAL, ctx->metrics.last_snapshot_slot_started_incremental );
2053 0 : FD_MGAUGE_SET ( SNAPMK, LAST_FINISHED_SLOT_FULL, ctx->metrics.last_snapshot_slot_finished_full );
2054 0 : FD_MGAUGE_SET ( SNAPMK, LAST_FINISHED_SLOT_INCREMENTAL, ctx->metrics.last_snapshot_slot_finished_incremental );
2055 :
2056 0 : FD_MCNT_SET ( SNAPMK, COMPRESS_CONSUMED_BYTES, ctx->metrics.bytes_compressed );
2057 0 : FD_MCNT_SET ( SNAPMK, DISK_WRITTEN_BYTES, ctx->metrics.bytes_written );
2058 :
2059 0 : FD_MGAUGE_SET ( SNAPMK, CACHE_PROGRESS_BYTES, fd_backup_cache_scanned_bytes( ctx->acc_cache ) );
2060 0 : FD_MGAUGE_SET ( SNAPMK, CACHE_EXPECTED_BYTES, fd_backup_cache_total_bytes ( ctx->acc_cache ) );
2061 0 : FD_MGAUGE_SET ( SNAPMK, INCREMENTAL_ACCOUNT_PROGRESS, ctx->delta.ele_cnt );
2062 0 : FD_MGAUGE_SET ( SNAPMK, INCREMENTAL_ACCOUNT_EXPECTED, ctx->delta.ele_total );
2063 :
2064 0 : FD_MCNT_SET ( SNAPMK, IO_BLOCKED_DURATION_NANOS, ctx->metrics.io_blocked_ticks );
2065 0 : FD_MCNT_SET ( SNAPMK, COMPRESS_DURATION_NANOS, ctx->metrics.compress_ticks );
2066 0 : FD_MGAUGE_SET ( SNAPMK, DISK_ALLOCATED_BYTES, disk_alloc_bytes );
2067 :
2068 0 : FD_MGAUGE_SET ( SNAPMK, INCREMENTAL_ACCOUNT_USED, __atomic_load_n( &ctx->accdb_shmem->delta.head, __ATOMIC_RELAXED ) );
2069 0 : FD_MGAUGE_SET ( SNAPMK, INCREMENTAL_ACCOUNT_CAPACITY, ctx->accdb_shmem->delta.ele_max );
2070 0 : }
2071 :
2072 : #define STEM_BURST SNAPMK_STEM_BURST
2073 : #define STEM_LAZY SNAPMK_STEM_LAZY
2074 : #define STEM_CALLBACK_CONTEXT_TYPE fd_snapmk_t
2075 : #define STEM_CALLBACK_CONTEXT_ALIGN alignof(fd_snapmk_t)
2076 0 : #define STEM_CALLBACK_RECV_CREDIT recv_credit
2077 0 : #define STEM_CALLBACK_CHECK_CREDIT check_credit
2078 0 : #define STEM_CALLBACK_AFTER_CREDIT after_credit
2079 0 : #define STEM_CALLBACK_RETURNABLE_FRAG returnable_frag
2080 0 : #define STEM_CALLBACK_METRICS_WRITE metrics_write
2081 : #include "../../disco/stem/fd_stem.c"
2082 :
2083 :
2084 : /* snapmk_run contains a bunch of boilerplate to hijack flow control
2085 : away from stem. */
2086 :
2087 : static void
2088 : snapmk_run( fd_topo_t * topo,
2089 0 : fd_topo_tile_t * tile ) {
2090 0 : fd_snapmk_t * ctx = (fd_snapmk_t *)fd_ulong_align_up( (ulong)fd_topo_obj_laddr( topo, tile->tile_obj_id ), alignof(fd_snapmk_t) );
2091 :
2092 0 : fd_frag_meta_t const * in_mcache[ FD_TOPO_MAX_LINKS ];
2093 0 : ulong * in_fseq [ FD_TOPO_MAX_TILE_IN_LINKS ];
2094 :
2095 0 : ulong polled_in_cnt = 0UL;
2096 0 : for( ulong i=0UL; i<tile->in_cnt; i++ ) {
2097 0 : if( FD_UNLIKELY( !tile->in_link_poll[ i ] ) ) continue;
2098 0 : fd_topo_link_t const * link = &topo->links[ tile->in_link_id[ i ] ];
2099 0 : in_mcache[ polled_in_cnt ] = link->mcache;
2100 0 : FD_TEST( in_mcache[ polled_in_cnt ] );
2101 : /* Redirect links with custom flow control to local shadows. */
2102 0 : if( 0==strcmp( link->name, "snaprd_out" ) ) {
2103 : /* snaprd_out fseq set using custom logic, disable stem fseq updates */
2104 0 : in_fseq[ polled_in_cnt ] = &ctx->rd_fseq_dummy;
2105 0 : } else {
2106 0 : in_fseq[ polled_in_cnt ] = tile->in_link_fseq[ i ];
2107 0 : }
2108 0 : FD_TEST( in_fseq[ polled_in_cnt ] );
2109 0 : polled_in_cnt += 1UL;
2110 0 : }
2111 :
2112 0 : fd_frag_meta_t * out_mcache[ FD_TOPO_MAX_LINKS ];
2113 0 : for( ulong i=0UL; i<tile->out_cnt; i++ ) {
2114 0 : out_mcache[ i ] = topo->links[ tile->out_link_id[ i ] ].mcache;
2115 0 : FD_TEST( out_mcache[ i ] );
2116 0 : }
2117 :
2118 0 : ulong reliable_cons_cnt = 0UL;
2119 0 : ulong cons_out [ FD_TOPO_MAX_LINKS ];
2120 0 : ulong * cons_fseq[ FD_TOPO_MAX_LINKS ];
2121 0 : volatile ulong * cons_slow[ FD_TOPO_MAX_LINKS ];
2122 0 : for( ulong i=0UL; i<topo->tile_cnt; i++ ) {
2123 0 : fd_topo_tile_t * consumer_tile = &topo->tiles[ i ];
2124 0 : ulong polled_in_idx = 0UL;
2125 0 : for( ulong j=0UL; j<consumer_tile->in_cnt; j++ ) {
2126 0 : int is_polled = consumer_tile->in_link_poll[ j ];
2127 0 : for( ulong k=0UL; k<tile->out_cnt; k++ ) {
2128 0 : if( FD_UNLIKELY( consumer_tile->in_link_id[ j ]==tile->out_link_id[ k ] && consumer_tile->in_link_reliable[ j ] ) ) {
2129 0 : cons_out [ reliable_cons_cnt ] = k;
2130 0 : cons_fseq[ reliable_cons_cnt ] = consumer_tile->in_link_fseq[ j ];
2131 0 : FD_TEST( cons_fseq[ reliable_cons_cnt ] );
2132 0 : cons_slow[ reliable_cons_cnt ] = fd_metrics_link_in( consumer_tile->metrics, polled_in_idx ) + FD_METRICS_COUNTER_LINK_BACKPRESSURE_BLAMED_OFF;
2133 0 : reliable_cons_cnt++;
2134 0 : FD_TEST( reliable_cons_cnt<FD_TOPO_MAX_LINKS );
2135 0 : }
2136 0 : }
2137 0 : if( FD_LIKELY( is_polled ) ) polled_in_idx++;
2138 0 : }
2139 0 : }
2140 :
2141 0 : fd_rng_t rng[1];
2142 0 : FD_TEST( fd_rng_join( fd_rng_new( rng, (uint)fd_ulong_hash( (ulong)fd_tickcount() + tile->id ), 0UL ) ) );
2143 :
2144 0 : uchar __attribute__((aligned(FD_STEM_SCRATCH_ALIGN))) stem_scratch[ stem_scratch_footprint( polled_in_cnt, tile->out_cnt, reliable_cons_cnt ) ];
2145 :
2146 0 : fd_stem_sleep_t sleep[ 1 ] = {{ .shmem = NULL }};
2147 0 : if( FD_UNLIKELY( topo->sleep_obj_id!=ULONG_MAX ) ) {
2148 0 : sleep->shmem = fd_sleep_join( fd_topo_obj_laddr( topo, topo->sleep_obj_id ) );
2149 0 : FD_TEST( sleep->shmem );
2150 0 : sleep->tile_id = tile->id;
2151 0 : sleep->waker_fseq = NULL;
2152 0 : sleep->out_link_id = tile->out_link_id;
2153 :
2154 0 : ulong polled_idx = 0UL;
2155 0 : for( ulong i=0UL; i<tile->in_cnt; i++ ) {
2156 0 : if( FD_LIKELY( !tile->in_link_poll[ i ] ) ) continue;
2157 0 : fd_topo_link_t const * link = &topo->links[ tile->in_link_id[ i ] ];
2158 0 : sleep->in_link_id [ polled_idx ] = tile->in_link_id[ i ];
2159 : /* snaprd_out credits go through rd_ack, which rings snaprd itself;
2160 : the stem writes a dummy fseq for it and must not ring */
2161 0 : sleep->in_producer[ polled_idx ] = tile->in_link_reliable[ i ] && strcmp( link->name, "snaprd_out" ) ? fd_topo_find_link_producer( topo, link ) : ULONG_MAX;
2162 0 : polled_idx++;
2163 0 : }
2164 :
2165 0 : ulong pair_cnt = 0UL;
2166 0 : for( ulong i=0UL; i<tile->out_cnt; i++ ) {
2167 0 : sleep->wake_off[ i ] = (ushort)pair_cnt;
2168 0 : pair_cnt += fd_sleep_wake_table( sleep->wake+pair_cnt, topo, tile->out_link_id[ i ] );
2169 0 : }
2170 0 : sleep->wake_off[ tile->out_cnt ] = (ushort)pair_cnt;
2171 0 : }
2172 :
2173 0 : stem_run1( polled_in_cnt, in_mcache, in_fseq,
2174 0 : tile->out_cnt, out_mcache,
2175 0 : reliable_cons_cnt, cons_out, cons_fseq, cons_slow,
2176 0 : SNAPMK_STEM_BURST, SNAPMK_STEM_LAZY,
2177 0 : rng, stem_scratch, ctx, sleep );
2178 0 : }
2179 :
2180 : static ulong
2181 0 : max_event_sz( fd_topo_tile_t const * tile FD_PARAM_UNUSED ) {
2182 0 : return sizeof(fd_event_snapshot_created_t);
2183 0 : }
2184 :
2185 : fd_topo_run_tile_t fd_tile_snapmk = {
2186 : .name = "snapmk",
2187 : .populate_allowed_fds = populate_allowed_fds,
2188 : .populate_allowed_seccomp = populate_allowed_seccomp,
2189 : .scratch_align = scratch_align,
2190 : .scratch_footprint = scratch_footprint,
2191 : .privileged_init = privileged_init,
2192 : .unprivileged_init = unprivileged_init,
2193 : .run = snapmk_run,
2194 : .max_event_sz = max_event_sz,
2195 : .allow_renameat = 1
2196 : };
|
