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