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