Line data Source code
1 : /* The snapzp tile gathers account data, compresses it, and writes it to
2 : disk.
3 :
4 : It has two modes:
5 : - cache: resolve account pointers from in-memory DB cache (fall back
6 : to disk on overrun)
7 : - disk: recover accounts from streaming disk reads (from snaprd)
8 :
9 : Internally, snapzp does streaming compression from DRAM into tile
10 : scratch memory. Whenever compression scratch buffer fills, it is
11 : prepended on-the-fly with a matching compressed TAR header. Finally,
12 : everything is written to disk. */
13 :
14 : #define _GNU_SOURCE
15 : #define ZSTD_STATIC_LINKING_ONLY
16 : #include <zstd.h>
17 : #include "fd_backup.h"
18 : #include "fd_backup_cache.h"
19 : #include "fd_backup_shmem.h"
20 : #include "../../disco/metrics/fd_metrics.h"
21 : #include "../../disco/stem/fd_stem.h"
22 : #include "../../disco/topo/fd_topo.h"
23 :
24 : #include <time.h> /* CLOCK_REALTIME */
25 : #include <linux/futex.h>
26 : #include "generated/fd_snapzp_tile_seccomp.h"
27 : #include "../../tango/fseq/fd_fseq.h"
28 : #include <fcntl.h>
29 : #include <unistd.h>
30 : #include <errno.h>
31 :
32 : /* Compression buffer params */
33 0 : #define RAW_BUF_SZ (32UL<<20) /* FIXME make this configurable */
34 : #define COMP_BOUND ZSTD_COMPRESSBOUND( RAW_BUF_SZ )
35 0 : #define COMP_HEAD 522 /* 10 byte Zstandard uncompressed header + 512 byte plaintext tar header */
36 0 : #define COMP_BUF_SZ FD_ULONG_ALIGN_UP( COMP_HEAD+COMP_BOUND+8UL, 4096UL )
37 :
38 : struct fd_snapzp {
39 : fd_backup_cache_t acc_cache[1];
40 :
41 : fd_backup_overrun_t * overrun;
42 : fd_backup_stats_t * stats;
43 : fd_accdb_t * accdb;
44 : fd_accdb_fork_id_t fork_id;
45 :
46 : /* compression buffer */
47 : ZSTD_CCtx * zst;
48 : ulong zst_in_rec; /* ZSTD_CStreamInSize() */
49 : uchar * raw;
50 : ZSTD_inBuffer raw_buf;
51 : ZSTD_outBuffer comp_buf;
52 :
53 : ulong idle_cnt;
54 :
55 : ulong kind_id; /* index of this tile kind */
56 : ulong frame_cnt; /* frames written by this tile */
57 : ulong snapshot_slot;
58 : ulong slot_lo; /* lowest appendvec slot this archive may use */
59 : ulong snapshot_account_cnt;
60 : ulong snapshot_account_sz;
61 : ulong snapshot_tombstone_cnt;
62 : ulong snapshot_cached_account_cnt;
63 : ulong snapshot_disk_account_cnt;
64 : ulong snapshot_zstd_padding_sz;
65 : ulong snapshot_uncompressed_sz;
66 : ulong snapshot_compress_ticks0;
67 : ulong snapshot_io_blocked_ticks0;
68 :
69 : /* input link */
70 : void * snapmk_zp_mem;
71 : ulong snapmk_zp_chunk0;
72 : ulong snapmk_zp_wmark;
73 : void * snaprd_mem;
74 : ulong snaprd_data0;
75 : ulong snaprd_data1;
76 :
77 : /* snapshot files (all O_DIRECT write-only seekable) */
78 : int snap_fd; /* current snap being written */
79 : ulong snap_fd_cnt; /* open snap fd count (see FD_SNAP_DIO_FD) */
80 :
81 : /* bump allocator for file offsets used to coordinate snapzp take
82 : turns to write to a file. 512 byte aligned for direct I/O. */
83 : ulong volatile * file_off;
84 :
85 : /* countdown allocator for appendvec slots, shared the same way */
86 : ulong volatile * appendvec_slot_ticket;
87 :
88 : struct {
89 : int active;
90 : fd_pubkey_t pubkey;
91 : fd_pubkey_t owner;
92 : uint size;
93 : uint acc_idx;
94 : ulong data_rem;
95 : ulong data_pad;
96 : } disk;
97 :
98 : struct {
99 : ulong accounts_compressed;
100 : ulong cache_read_torn;
101 : ulong bytes_compressed;
102 : ulong bytes_written;
103 : ulong io_blocked_ticks;
104 : ulong compress_ticks;
105 : } metrics;
106 :
107 : __attribute__((aligned(4096))) uchar raw_buf1 [ RAW_BUF_SZ ];
108 : __attribute__((aligned(4096))) uchar comp_buf1[ COMP_BUF_SZ ];
109 : };
110 : typedef struct fd_snapzp fd_snapzp_t;
111 :
112 :
113 : FD_FN_CONST static inline ulong
114 0 : scratch_align( void ) {
115 0 : return FD_SHMEM_HUGE_PAGE_SZ;
116 0 : }
117 :
118 : FD_FN_PURE static inline ulong
119 0 : scratch_footprint( fd_topo_tile_t const * tile ) {
120 0 : (void)tile;
121 0 : ulong l = FD_LAYOUT_INIT;
122 0 : l = FD_LAYOUT_APPEND( l, alignof(fd_snapzp_t), sizeof(fd_snapzp_t) );
123 0 : l = FD_LAYOUT_APPEND( l, fd_accdb_align(), fd_accdb_footprint( tile->snapzp.max_live_slots, 0 ) );
124 0 : l = FD_LAYOUT_APPEND( l, 32UL, ZSTD_estimateCStreamSize( FD_BACKUP_ZSTD_LEVEL ) );
125 0 : return FD_LAYOUT_FINI( l, FD_SHMEM_HUGE_PAGE_SZ );
126 0 : }
127 :
128 : static void
129 : privileged_init( fd_topo_t const * topo,
130 0 : fd_topo_tile_t const * tile ) {
131 0 : (void)topo;
132 0 : ulong snap_fd_max = tile->snapzp.snap_fd_cnt;
133 0 : FD_CHECK_ERR( snap_fd_max>0UL && snap_fd_max<=FD_SNAP_MAX,
134 0 : "invalid snap_fd_max" );
135 0 : for( uint i=0U; i<snap_fd_max; i++ ) {
136 0 : if( FD_UNLIKELY( -1==fcntl( FD_SNAP_DIO_FD( i ), F_GETFD ) ) )
137 0 : FD_LOG_ERR(( "fcntl(snapshot pool fd %d) failed (%i-%s), was the snapshot pool initialized on boot?",
138 0 : FD_SNAP_DIO_FD( i ), errno, fd_io_strerror( errno ) ));
139 0 : }
140 0 : }
141 :
142 : static void
143 : unprivileged_init( fd_topo_t const * topo,
144 0 : fd_topo_tile_t const * tile ) {
145 0 : FD_CHECK_ERR( tile->kind_id < SNAPZP_TILE_MAX, "too many snapzp tiles" );
146 :
147 0 : FD_SCRATCH_ALLOC_INIT( l, fd_topo_obj_laddr( topo, tile->tile_obj_id ) );
148 0 : fd_snapzp_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_snapzp_t), sizeof(fd_snapzp_t) );
149 0 : void * _accdb = FD_SCRATCH_ALLOC_APPEND( l, fd_accdb_align(), fd_accdb_footprint( tile->snapzp.max_live_slots, 0 ) );
150 0 : void * _zstd = FD_SCRATCH_ALLOC_APPEND( l, 32UL, ZSTD_estimateCStreamSize( FD_BACKUP_ZSTD_LEVEL ) );
151 0 : FD_SCRATCH_ALLOC_FINI( l, scratch_align() );
152 :
153 0 : memset( ctx, 0, sizeof(fd_snapzp_t) ); /* 64 MiB-ish memset */
154 0 : ctx->snap_fd_cnt = tile->snapzp.snap_fd_cnt;
155 0 : ctx->snap_fd = -1;
156 :
157 0 : ctx->zst = ZSTD_initStaticCStream( _zstd, ZSTD_estimateCStreamSize( FD_BACKUP_ZSTD_LEVEL ) );
158 0 : FD_TEST( ctx->zst );
159 0 : ulong zst_err;
160 0 : zst_err = ZSTD_CCtx_setParameter( ctx->zst, ZSTD_c_compressionLevel, FD_BACKUP_ZSTD_LEVEL );
161 0 : if( FD_UNLIKELY( ZSTD_isError( zst_err ) ) ) {
162 0 : FD_LOG_ERR(( "ZSTD_CCtx_setParameter(ZSTD_c_compressionLevel) failed: %s", ZSTD_getErrorName( zst_err ) ));
163 0 : }
164 0 : zst_err = ZSTD_CCtx_setParameter( ctx->zst, ZSTD_c_stableInBuffer, 1 );
165 0 : if( FD_UNLIKELY( ZSTD_isError( zst_err ) ) ) {
166 0 : FD_LOG_ERR(( "ZSTD_CCtx_setParameter(ZSTD_c_stableInBuffer=1) failed: %s", ZSTD_getErrorName( zst_err ) ));
167 0 : }
168 0 : zst_err = ZSTD_CCtx_setParameter( ctx->zst, ZSTD_c_stableOutBuffer, 1 );
169 0 : if( FD_UNLIKELY( ZSTD_isError( zst_err ) ) ) {
170 0 : FD_LOG_ERR(( "ZSTD_CCtx_setParameter(ZSTD_c_stableOutBuffer=1) failed: %s", ZSTD_getErrorName( zst_err ) ));
171 0 : }
172 0 : zst_err = ZSTD_CCtx_setParameter( ctx->zst, ZSTD_c_srcSizeHint, (int)RAW_BUF_SZ );
173 0 : if( FD_UNLIKELY( ZSTD_isError( zst_err ) ) ) {
174 0 : FD_LOG_ERR(( "ZSTD_CCtx_setParameter(ZSTD_c_srcSizeHint) failed: %s", ZSTD_getErrorName( zst_err ) ));
175 0 : }
176 0 : ctx->zst_in_rec = ZSTD_CStreamInSize();
177 0 : ctx->raw = ctx->raw_buf1;
178 0 : ctx->raw_buf = (ZSTD_inBuffer ){ .src = ctx->raw_buf1, .size = 0UL };
179 0 : ctx->comp_buf = (ZSTD_outBuffer){ .dst = ctx->comp_buf1+COMP_HEAD, .size = COMP_BUF_SZ-COMP_HEAD };
180 :
181 0 : for( ulong i=0UL; i<tile->in_cnt; i++ ) {
182 0 : fd_topo_link_t const * link = &topo->links[ tile->in_link_id[ i ] ];
183 0 : if( 0==strcmp( link->name, "snapmk_zp" ) ) {
184 0 : ctx->snapmk_zp_mem = topo->workspaces[ topo->objs[ link->dcache_obj_id ].wksp_id ].wksp;
185 0 : ctx->snapmk_zp_chunk0 = fd_dcache_compact_chunk0( ctx->snapmk_zp_mem, link->dcache );
186 0 : ctx->snapmk_zp_wmark = fd_dcache_compact_wmark ( ctx->snapmk_zp_mem, link->dcache, link->mtu );
187 0 : }
188 0 : }
189 0 : FD_TEST( ctx->snapmk_zp_mem );
190 :
191 : /* discover snaprd_out link */
192 0 : ulong snaprd_wksp_id = fd_topo_find_wksp( topo, "snaprd_out" );
193 0 : FD_CHECK_ERR( snaprd_wksp_id!=ULONG_MAX, "snapzp tile not joined to snaprd_out wksp" );
194 0 : ulong snaprd_link_id = fd_topo_find_link( topo, "snaprd_out", 0UL );
195 0 : FD_CHECK_ERR( snaprd_link_id!=ULONG_MAX, "snaprd_out:0: link not found" );
196 0 : ulong snaprd_dcache_id = topo->links[ snaprd_link_id ].dcache_obj_id;
197 0 : FD_CHECK_ERR( snaprd_dcache_id!=ULONG_MAX, "snaprd_out:0: dcache_obj_id not found" );
198 0 : FD_CHECK_ERR( topo->objs[ snaprd_dcache_id ].wksp_id==snaprd_wksp_id, "snaprd_out:0: dcache_obj_id not in snaprd_out wksp" );
199 0 : uchar const * dcache = topo->links[ snaprd_link_id ].dcache;
200 0 : FD_CHECK_ERR( dcache, "snaprd_out:0: dcache pointer is NULL" );
201 0 : ctx->snaprd_data0 = (ulong)dcache;
202 0 : ctx->snaprd_data1 = (ulong)dcache + fd_dcache_data_sz( dcache );
203 :
204 0 : ctx->snaprd_mem = topo->workspaces[ snaprd_wksp_id ].wksp;
205 0 : FD_TEST( ctx->snaprd_mem );
206 :
207 0 : ctx->kind_id = tile->kind_id;
208 0 : ctx->frame_cnt = 0UL;
209 0 : memset( &ctx->disk, 0, sizeof(ctx->disk) );
210 :
211 0 : void * _accdb_shmem = fd_topo_obj_laddr( topo, tile->snapzp.accdb_obj_id );
212 0 : fd_accdb_shmem_t * accdb_shmem_ro = fd_accdb_shmem_join( _accdb_shmem );
213 0 : FD_TEST( accdb_shmem_ro );
214 0 : ulong * epoch_fseq = fd_fseq_join( fd_topo_obj_laddr( topo, tile->snapzp.accdb_epoch_obj_id ) );
215 0 : FD_TEST( epoch_fseq );
216 0 : ctx->accdb = fd_accdb_join_readonly( _accdb, accdb_shmem_ro, epoch_fseq, FD_ACCDB_FD_RO );
217 0 : FD_TEST( ctx->accdb );
218 0 : FD_TEST( fd_backup_cache_join( ctx->acc_cache, accdb_shmem_ro, epoch_fseq ) );
219 0 : ctx->overrun = fd_backup_overrun( fd_topo_obj_laddr( topo, tile->snapzp.visited_set_obj_id ) );
220 0 : ctx->stats = fd_backup_stats ( fd_topo_obj_laddr( topo, tile->snapzp.visited_set_obj_id ) );
221 0 : FD_TEST( ctx->overrun );
222 0 : FD_TEST( ctx->stats );
223 0 : FD_STATIC_ASSERT( SNAPZP_TILE_MAX==FD_BACKUP_STATS_MAX, backup_stats_max );
224 :
225 0 : ulong * zp_fseq = fd_fseq_join( fd_topo_obj_laddr( topo, tile->snapzp.zp_fseq_id ) ); FD_TEST( zp_fseq );
226 0 : ulong * zp_app = fd_fseq_app_laddr( zp_fseq );
227 0 : ctx->file_off = &zp_app[ 0 ];
228 0 : ctx->appendvec_slot_ticket = &zp_app[ 1 ];
229 0 : }
230 :
231 : static ulong
232 : populate_allowed_fds( fd_topo_t const * topo,
233 : fd_topo_tile_t const * tile,
234 : ulong out_fds_cnt,
235 0 : int * out_fds ) {
236 0 : (void)topo;
237 0 : ulong snap_fd_cnt = tile->snapzp.snap_fd_cnt;
238 0 : FD_CHECK_ERR( out_fds_cnt>=3UL+snap_fd_cnt, "out_fds[] too small" );
239 0 : ulong out_cnt = 0UL;
240 0 : out_fds[ out_cnt++ ] = 2; /* stderr */
241 0 : if( FD_LIKELY( -1!=fd_log_private_logfile_fd() ) )
242 0 : out_fds[ out_cnt++ ] = fd_log_private_logfile_fd(); /* logfile */
243 0 : out_fds[ out_cnt++ ] = FD_ACCDB_FD_RO;
244 0 : for( uint i=0U; i<snap_fd_cnt; i++ )
245 0 : out_fds[ out_cnt++ ] = FD_SNAP_DIO_FD( i );
246 0 : return out_cnt;
247 0 : }
248 :
249 : static ulong
250 : populate_allowed_seccomp( fd_topo_t const * topo,
251 : fd_topo_tile_t const * tile,
252 : ulong out_cnt,
253 0 : struct sock_filter * out ) {
254 0 : (void)topo;
255 0 : ulong snap_fd_cnt = tile->snapzp.snap_fd_cnt;
256 0 : populate_sock_filter_policy_fd_snapzp_tile(
257 0 : out_cnt, out,
258 0 : (uint)fd_log_private_logfile_fd(),
259 0 : (uint)FD_SNAP_DIO_FD( 0 ),
260 0 : (uint)FD_SNAP_DIO_FD( snap_fd_cnt-1U ),
261 0 : (uint)FD_ACCDB_FD_RO );
262 0 : return sock_filter_policy_fd_snapzp_tile_instr_cnt;
263 0 : }
264 :
265 : static void
266 : before_credit( fd_snapzp_t * ctx,
267 : fd_stem_context_t * stem,
268 0 : int * charge_busy ) {
269 0 : (void)charge_busy;
270 0 : if( FD_LIKELY( ctx->snap_fd>=0 ) ) {
271 : /* Don't sleep while snapshot production is active
272 : FIXME this is quite wasteful */
273 0 : ctx->idle_cnt = 0UL;
274 0 : return;
275 0 : }
276 0 : if( FD_UNLIKELY( ctx->idle_cnt++ > 16384UL && !stem->sleep ) ) {
277 0 : fd_log_sleep( (long)1e6 );
278 0 : }
279 0 : }
280 :
281 : /* msg_start is called on all snapzp tiles before snapshot production
282 : starts. */
283 :
284 : static void
285 : msg_start( fd_snapzp_t * ctx,
286 0 : fd_backup_start_msg_t const * frag ) {
287 0 : FD_CHECK_CRIT( frag->snap_idx < ctx->snap_fd_cnt, "invalid snapshot pool slot" );
288 0 : ctx->snap_fd = FD_SNAP_DIO_FD( frag->snap_idx );
289 0 : ctx->fork_id = (fd_accdb_fork_id_t){ .val = frag->fork_id };
290 0 : ctx->snapshot_slot = frag->slot;
291 0 : ctx->slot_lo = frag->slot_lo;
292 :
293 0 : ulong zst_err = ZSTD_CCtx_reset( ctx->zst, ZSTD_reset_session_only );
294 0 : if( FD_UNLIKELY( ZSTD_isError( zst_err ) ) ) {
295 0 : FD_LOG_ERR(( "ZSTD_CCtx_reset failed: %s", ZSTD_getErrorName( zst_err ) ));
296 0 : }
297 0 : ctx->frame_cnt = 0UL;
298 0 : ctx->snapshot_account_cnt = 0UL;
299 0 : ctx->snapshot_account_sz = 0UL;
300 0 : ctx->snapshot_tombstone_cnt = 0UL;
301 0 : ctx->snapshot_cached_account_cnt = 0UL;
302 0 : ctx->snapshot_disk_account_cnt = 0UL;
303 0 : ctx->snapshot_zstd_padding_sz = 0UL;
304 0 : ctx->snapshot_uncompressed_sz = 0UL;
305 0 : ctx->snapshot_compress_ticks0 = ctx->metrics.compress_ticks;
306 0 : ctx->snapshot_io_blocked_ticks0 = ctx->metrics.io_blocked_ticks;
307 0 : fd_backup_worker_stats_t * stats = &ctx->stats->worker[ ctx->kind_id ];
308 0 : __atomic_store_n( &stats->account_cnt, 0UL, __ATOMIC_RELAXED );
309 0 : __atomic_store_n( &stats->account_sz, 0UL, __ATOMIC_RELAXED );
310 0 : __atomic_store_n( &stats->tombstone_cnt, 0UL, __ATOMIC_RELAXED );
311 0 : __atomic_store_n( &stats->cached_account_cnt, 0UL, __ATOMIC_RELAXED );
312 0 : __atomic_store_n( &stats->disk_account_cnt, 0UL, __ATOMIC_RELAXED );
313 0 : __atomic_store_n( &stats->zstd_data_frame_cnt, 0UL, __ATOMIC_RELAXED );
314 0 : __atomic_store_n( &stats->zstd_padding_sz, 0UL, __ATOMIC_RELAXED );
315 0 : __atomic_store_n( &stats->compress_ticks, 0UL, __ATOMIC_RELAXED );
316 0 : __atomic_store_n( &stats->io_blocked_ticks, 0UL, __ATOMIC_RELAXED );
317 0 : __atomic_store_n( &stats->uncompressed_sz, 0UL, __ATOMIC_RELEASE );
318 0 : memset( &ctx->disk, 0, sizeof(ctx->disk) );
319 0 : ctx->raw_buf.pos = 0UL;
320 0 : ctx->raw_buf.size = 0UL;
321 0 : ctx->comp_buf.pos = 0UL;
322 0 : ctx->comp_buf.size = COMP_BUF_SZ-COMP_HEAD;
323 0 : }
324 :
325 : /* zip_work does opportunistic Zstandard compression work in memory.
326 : This function is tuned for a good tradeoff between latency
327 : (contiguous busy cycles) and throughput (dispatch overhead of libzstd
328 : function calls). */
329 :
330 : static void
331 0 : zip_work( fd_snapzp_t * ctx ) {
332 0 : if( FD_LIKELY( ctx->raw_buf.size - ctx->raw_buf.pos < ctx->zst_in_rec ) ) return;
333 0 : ulong raw_pos = ctx->raw_buf.pos;
334 0 : long t0 = fd_tickcount();
335 0 : ulong ret = ZSTD_compressStream2( ctx->zst, &ctx->comp_buf, &ctx->raw_buf, ZSTD_e_continue );
336 0 : long t1 = fd_tickcount();
337 0 : if( FD_UNLIKELY( ZSTD_isError( ret ) ) ) {
338 0 : FD_LOG_ERR(( "ZSTD_compressStream2(ZSTD_e_continue) failed: %s", ZSTD_getErrorName( ret ) ));
339 0 : }
340 0 : ctx->metrics.bytes_compressed += ctx->raw_buf.pos - raw_pos;
341 0 : ctx->metrics.compress_ticks += (ulong)( t1-t0 );
342 0 : }
343 :
344 : /* zip_flush ends the current Zstandard compression frame and does a
345 : blocking direct I/O write. Both uncompressed and compressed streams
346 : are padded up to 512 byte alignment to meet TAR and direct I/O
347 : requirements respectively. */
348 :
349 : static void
350 0 : zip_flush( fd_snapzp_t * ctx ) {
351 0 : FD_CHECK_CRIT( !ctx->disk.active, "attempted to flush with active defrag op" );
352 :
353 : /* Nothing buffered */
354 0 : if( FD_UNLIKELY( !ctx->raw_buf.size ) ) return;
355 :
356 : /* Align input frame by 512 bytes (TAR file format) */
357 0 : ulong content_usz = ctx->raw_buf.size;
358 0 : ulong content_asz = fd_ulong_align_up( content_usz, 512UL );
359 0 : if( content_asz > content_usz ) {
360 0 : FD_TEST( content_asz <= RAW_BUF_SZ );
361 0 : fd_memset( ctx->raw + content_usz, 0, content_asz - content_usz );
362 0 : ctx->raw_buf.size = content_asz;
363 0 : }
364 0 : ctx->snapshot_uncompressed_sz += sizeof(fd_tar_meta_t) + content_asz;
365 :
366 : /* Finish content compression frame */
367 0 : ulong raw_pos = ctx->raw_buf.pos;
368 0 : long t0 = fd_tickcount();
369 0 : ulong ret = ZSTD_compressStream2( ctx->zst, &ctx->comp_buf, &ctx->raw_buf, ZSTD_e_end );
370 0 : long t1 = fd_tickcount();
371 0 : if( FD_UNLIKELY( ZSTD_isError( ret ) ) ) {
372 0 : FD_LOG_ERR(( "ZSTD_compressStream2(ZSTD_e_end) failed: %s", ZSTD_getErrorName( ret ) ));
373 0 : }
374 0 : if( FD_UNLIKELY( ret!=0UL ) ) {
375 0 : FD_LOG_ERR(( "ZSTD_compressStream2(ZSTD_e_end) did not finish frame" ));
376 0 : }
377 0 : ctx->metrics.bytes_compressed += ctx->raw_buf.pos - raw_pos;
378 0 : ctx->metrics.compress_ticks += (ulong)( t1-t0 );
379 0 : FD_TEST( ctx->raw_buf.pos == ctx->raw_buf.size );
380 0 : ctx->raw_buf.pos = 0UL;
381 0 : ctx->raw_buf.size = 0UL;
382 :
383 : /* One appendvec per frame. If, for some reason, we run out of slots
384 : to use, start incrementing the file index. */
385 0 : ulong appendvec_slot = __atomic_fetch_sub( ctx->appendvec_slot_ticket, 1UL, __ATOMIC_RELAXED );
386 0 : ulong appendvec_id = 0UL;
387 0 : if( FD_UNLIKELY( appendvec_slot<ctx->slot_lo || appendvec_slot>ctx->snapshot_slot ) ) { /* out of slots (> means the counter wrapped below 0) */
388 0 : appendvec_id = ctx->slot_lo - appendvec_slot; /* modular: 1, 2, ... */
389 0 : appendvec_slot = ctx->slot_lo;
390 0 : }
391 :
392 : /* Prepend compression frame with a TAR header
393 : (Zstandard frame with a 512 byte uncompressed block) */
394 0 : uchar * comp_head = (uchar *)ctx->comp_buf.dst - COMP_HEAD;
395 0 : memcpy( comp_head, (uchar[]){0x28,0xB5,0x2F,0xFD,0x60,0x00,0x01,0x01,0x10,0x00}, 10 );
396 0 : fd_tar_meta_t meta; fd_backup_tar_file_hdr( &meta, content_usz );
397 :
398 : /* Generate a unique file name */
399 0 : ctx->frame_cnt++;
400 0 : do {
401 0 : char * p = fd_cstr_init( meta.name );
402 0 : p = fd_cstr_append_cstr( p, "accounts/" );
403 0 : p = fd_cstr_append_ulong_as_text( p, 0, 0, appendvec_slot, fd_ulong_base10_dig_cnt( appendvec_slot ) );
404 0 : p = fd_cstr_append_char( p, '.' );
405 0 : p = fd_cstr_append_ulong_as_text( p, 0, 0, appendvec_id, fd_ulong_base10_dig_cnt( appendvec_id ) );
406 0 : fd_cstr_fini( p );
407 0 : } while(0);
408 0 : fd_tar_meta_set_chksum( &meta );
409 0 : memcpy( comp_head+10, &meta, sizeof(fd_tar_meta_t) );
410 :
411 : /* Align to block size with a skippable frame */
412 0 : ulong comp_usz = COMP_HEAD + ctx->comp_buf.pos;
413 0 : ulong comp_asz = fd_ulong_align_up( comp_usz, 4096UL );
414 0 : ulong pad_sz = comp_asz - comp_usz;
415 0 : if( FD_UNLIKELY( pad_sz>0UL && pad_sz<8UL ) ) {
416 0 : comp_asz += 4096UL;
417 0 : pad_sz += 4096UL;
418 0 : }
419 0 : FD_TEST( comp_asz <= COMP_BUF_SZ );
420 0 : if( FD_LIKELY( pad_sz>0UL ) ) {
421 0 : uchar * tail = (uchar *)ctx->comp_buf.dst + ctx->comp_buf.pos;
422 0 : FD_STORE( uint, tail, ZSTD_MAGIC_SKIPPABLE_START );
423 0 : FD_STORE( uint, tail+4, (uint)( pad_sz-8 ) );
424 0 : fd_memset( tail+8, 0, pad_sz-8 );
425 0 : ctx->snapshot_zstd_padding_sz += pad_sz;
426 0 : }
427 :
428 : /* Allocate file range to write into */
429 0 : ulong off = __atomic_fetch_add( ctx->file_off, comp_asz, __ATOMIC_RELAXED );
430 0 : FD_TEST( fd_ulong_is_aligned( off, 4096UL ) );
431 0 : FD_TEST( fd_ulong_is_aligned( comp_asz, 4096UL ) );
432 0 : t0 = fd_tickcount();
433 0 : long write_sz = pwrite( ctx->snap_fd, comp_head, comp_asz, (long)off );
434 0 : t1 = fd_tickcount();
435 0 : if( FD_UNLIKELY( write_sz!=(long)comp_asz ) ) {
436 0 : FD_LOG_ERR(( "pwrite failed: %i-%s", errno, fd_io_strerror( errno ) ));
437 0 : }
438 0 : ctx->metrics.bytes_written += comp_asz;
439 0 : ctx->metrics.io_blocked_ticks += (ulong)( t1-t0 );
440 :
441 : /* Free compressed buffer */
442 0 : ctx->comp_buf.pos = 0UL;
443 0 : }
444 :
445 : /* accmeta_await_evict waits until the account data belonging to the
446 : given index entry is evicted from cache to disk. This is used to
447 : fallback to disk reads when a cache read is torn by eviction. */
448 :
449 : static void
450 : accmeta_await_evict( fd_snapzp_t * ctx,
451 0 : uint acc_idx ) {
452 0 : fd_backup_accidx_t * idx = &ctx->acc_cache->idx;
453 0 : FD_CHECK_CRIT( fd_backup_accidx_valid( idx, acc_idx ), "invalid account index" );
454 :
455 0 : fd_accdb_accmeta_t const * acc = &idx->acc_pool[ acc_idx ];
456 0 : for(;;) {
457 0 : FD_COMPILER_MFENCE();
458 0 : FD_VOLATILE( *idx->epoch_slot ) = FD_VOLATILE_CONST( *idx->epoch );
459 0 : FD_HW_MFENCE();
460 :
461 0 : ulong off_packed = FD_VOLATILE_CONST( acc->offset_fork );
462 0 : int done = ( off_packed & FD_ACCDB_OFF_MASK )!=FD_ACCDB_OFF_INVAL;
463 :
464 0 : FD_COMPILER_MFENCE();
465 0 : FD_VOLATILE( *idx->epoch_slot ) = ULONG_MAX;
466 :
467 0 : if( FD_LIKELY( done ) ) break;
468 0 : FD_SPIN_PAUSE(); /* FIXME yield to OS instead? */
469 0 : }
470 :
471 : /* tell snapmk that we were overrun */
472 0 : fd_backup_overrun_push( ctx->overrun, acc_idx );
473 0 : ctx->metrics.cache_read_torn++;
474 :
475 0 : FD_COMPILER_MFENCE();
476 0 : }
477 :
478 : /* msg_acc_cache instructs snapzp to gather, compress, and write out a
479 : bunch of accounts in cache. Note that accdb cache entries are not
480 : pinned. Account cache data can be evicted during read, leading to
481 : use-after-free, which this function gracefully recovers from. */
482 :
483 : static void
484 : msg_acc_cache( fd_snapzp_t * ctx,
485 0 : fd_backup_cache_msg_t const * batch ) {
486 0 : FD_CHECK_CRIT( ctx->snap_fd>=0, "invalid snapshot file descriptor" );
487 :
488 0 : fd_backup_accidx_t * idx = &ctx->acc_cache->idx;
489 0 : int in_epoch = 0;
490 :
491 0 : ZSTD_inBuffer * buf = &ctx->raw_buf;
492 0 : for( ulong i=0UL; i<FD_BACKUP_CACHE_PARA; i++ ) {
493 0 : fd_pubkey_t const * pubkey = &batch->pubkey [ i ];
494 0 : uint acc_idx = batch->acc_idx[ i ];
495 0 : if( acc_idx==UINT_MAX ) continue;
496 :
497 0 : if( FD_UNLIKELY( !in_epoch ) ) {
498 0 : FD_COMPILER_MFENCE();
499 0 : FD_VOLATILE( *idx->epoch_slot ) = FD_VOLATILE_CONST( *idx->epoch );
500 : /* a lock prefix is an expensive CPU pipeline hazard, therefore
501 : we hold onto our epoch for multiple accounts */
502 0 : FD_HW_MFENCE();
503 0 : in_epoch = 1;
504 0 : }
505 :
506 0 : ulong prev_sz = buf->size;
507 0 : int err = fd_backup_cache_read( ctx->acc_cache, pubkey, acc_idx, ctx->raw, &buf->size, RAW_BUF_SZ );
508 0 : if( FD_UNLIKELY( err==FD_BACKUP_CACHE_ERR_SPACE ) ) {
509 0 : FD_COMPILER_MFENCE();
510 0 : FD_VOLATILE( *idx->epoch_slot ) = ULONG_MAX;
511 0 : zip_flush( ctx );
512 0 : FD_COMPILER_MFENCE();
513 0 : FD_VOLATILE( *idx->epoch_slot ) = FD_VOLATILE_CONST( *idx->epoch );
514 0 : FD_HW_MFENCE();
515 0 : prev_sz = buf->size;
516 0 : err = fd_backup_cache_read( ctx->acc_cache, pubkey, acc_idx, ctx->raw, &buf->size, RAW_BUF_SZ );
517 0 : FD_CHECK_ERR( err!=FD_BACKUP_CACHE_ERR_SPACE, "Zstandard buffer too small" );
518 0 : }
519 :
520 0 : if( FD_UNLIKELY( err==FD_BACKUP_CACHE_ERR_MISS ) ) {
521 0 : FD_COMPILER_MFENCE();
522 0 : FD_VOLATILE( *idx->epoch_slot ) = ULONG_MAX;
523 0 : in_epoch = 0;
524 0 : accmeta_await_evict( ctx, acc_idx );
525 0 : continue;
526 0 : }
527 0 : FD_CHECK_ERR( err==FD_BACKUP_CACHE_SUCCESS, "unexpected cache error code" );
528 0 : ctx->snapshot_account_cnt++;
529 0 : ctx->snapshot_account_sz += buf->size - prev_sz;
530 0 : ctx->snapshot_cached_account_cnt++;
531 0 : ctx->metrics.accounts_compressed++;
532 0 : }
533 :
534 0 : if( FD_LIKELY( in_epoch ) ) {
535 0 : FD_COMPILER_MFENCE();
536 0 : FD_VOLATILE( *idx->epoch_slot ) = ULONG_MAX;
537 0 : }
538 0 : }
539 :
540 : static void
541 : msg_acc_delta( fd_snapzp_t * ctx,
542 0 : fd_backup_delta_msg_t const * batch ) {
543 0 : FD_CHECK_CRIT( ctx->snap_fd>=0, "invalid snapshot file descriptor" );
544 0 : FD_CHECK_CRIT( !ctx->disk.active, "received account delta while already processing a disk account" );
545 0 : FD_CHECK_CRIT( batch->cnt<=FD_BACKUP_CACHE_PARA, "invalid delta account batch" );
546 :
547 0 : ulong const rec_max = sizeof(snap_acc_hdr_t) + FD_RUNTIME_ACC_SZ_MAX;
548 0 : FD_STATIC_ASSERT( sizeof(snap_acc_hdr_t)+FD_RUNTIME_ACC_SZ_MAX<=RAW_BUF_SZ, raw_buf_too_small );
549 :
550 0 : for( ulong i=0UL; i<(ulong)batch->cnt; i++ ) {
551 0 : if( FD_UNLIKELY( ctx->raw_buf.size + rec_max > RAW_BUF_SZ ) ) zip_flush( ctx );
552 :
553 0 : ulong start = ctx->raw_buf.size;
554 0 : snap_acc_hdr_t * hdr = (snap_acc_hdr_t *)( ctx->raw + start );
555 0 : memset( hdr, 0, sizeof(snap_acc_hdr_t) );
556 0 : hdr->pubkey = batch->pubkey[ i ];
557 :
558 0 : ulong lamports = 0UL;
559 0 : ulong data_len = 0UL;
560 0 : int executable = 0;
561 0 : int source = fd_accdb_read_one_nocache( ctx->accdb, ctx->fork_id, batch->pubkey[ i ].uc,
562 0 : &lamports, &executable, hdr->owner.uc,
563 0 : ctx->raw + start + sizeof(snap_acc_hdr_t), &data_len );
564 0 : FD_CHECK_CRIT( data_len<=FD_RUNTIME_ACC_SZ_MAX, "accdb returned oversized account" );
565 0 : hdr->lamports = lamports;
566 0 : hdr->executable = (uchar)!!executable;
567 0 : hdr->data_len = data_len;
568 :
569 0 : ulong data_pad = fd_ulong_align_up( data_len, 8UL ) - data_len;
570 0 : if( data_pad ) fd_memset( ctx->raw + start + sizeof(snap_acc_hdr_t) + data_len, 0, data_pad );
571 0 : ctx->raw_buf.size = start + sizeof(snap_acc_hdr_t) + data_len + data_pad;
572 0 : ctx->snapshot_account_cnt++;
573 0 : ctx->snapshot_account_sz += sizeof(snap_acc_hdr_t) + data_len + data_pad;
574 0 : ctx->snapshot_tombstone_cnt += source==FD_ACCDB_READ_ONE_NOCACHE_MISS;
575 0 : ctx->snapshot_cached_account_cnt += source==FD_ACCDB_READ_ONE_NOCACHE_CACHE;
576 0 : ctx->snapshot_disk_account_cnt += source==FD_ACCDB_READ_ONE_NOCACHE_DISK;
577 0 : ctx->metrics.accounts_compressed++;
578 0 : }
579 0 : }
580 :
581 : /* accmeta_disk validates that an index entry (at acc_idx with given
582 : pubkey) is an on-disk rooted account.
583 : (Safe while compaction and rooting are disabled, as on-disk rooted
584 : accounts are pinned in the index and on-disk under these conditions.
585 :
586 : Returns a pointer to the index entry if a matching record is found,
587 : NULL otherwise. (Never returns NULL under correct usage.) */
588 :
589 : static fd_accdb_accmeta_t const *
590 : accmeta_disk( fd_snapzp_t * ctx,
591 : fd_pubkey_t const * pubkey,
592 : uint size,
593 0 : uint acc_idx ) {
594 0 : fd_backup_accidx_t const * idx = &ctx->acc_cache->idx;
595 0 : if( FD_UNLIKELY( !fd_backup_accidx_valid( idx, acc_idx ) ) ) return NULL;
596 :
597 0 : fd_accdb_accmeta_t const * acc = &idx->acc_pool[ acc_idx ];
598 0 : uint es = FD_VOLATILE_CONST( acc->executable_size );
599 0 : if( FD_UNLIKELY( FD_ACCDB_SIZE_DATA( es )!=FD_ACCDB_SIZE_DATA( size ) ) ) return NULL;
600 0 : if( FD_UNLIKELY( memcmp( acc->key.pubkey, pubkey->uc, sizeof(fd_pubkey_t) ) ) ) return NULL;
601 0 : return acc;
602 0 : }
603 :
604 : /* msg_acc_disk instructs snapzp to compress an account fragment that
605 : was read from the accdb disk file. (slow path)
606 :
607 : The first fragment of an account has an associated backup_disk_msg
608 : descriptor (chunk serves as ptr, sz is the descriptor sz, i.e.
609 : sizeof(fd_backup_disk_msg_t)). The first frag is indicated by the
610 : ctl.som flag.
611 :
612 : The remaining fragments do not have a descriptor. The last frag is
613 : indicated by the ctl.eom frag.
614 :
615 : The account data fragment size is in tspub.
616 :
617 : msg_acc_disk_start is a helper function that starts a disk account
618 : defrag operation. Returns the number of bytes consumed (always the
619 : frag data size). */
620 :
621 : static ulong
622 : msg_acc_disk_start( fd_snapzp_t * ctx,
623 0 : fd_backup_disk_msg_t const * frag ) {
624 0 : FD_CHECK_CRIT( ctx->snap_fd>=0, "invalid snapshot file descriptor" );
625 0 : FD_CHECK_CRIT( !ctx->disk.active, "received account SOM while already processing a disk account" );
626 :
627 0 : ulong data_len = (ulong)FD_ACCDB_SIZE_DATA( frag->size );
628 0 : ulong rec_sz = sizeof(snap_acc_hdr_t) + fd_ulong_align_up( data_len, 8UL );
629 0 : FD_CHECK_CRIT( rec_sz<=RAW_BUF_SZ, "oversize snapshot account record" );
630 0 : FD_CHECK_CRIT( frag->snap_sz==rec_sz, "disk account snapshot size mismatch" );
631 0 : if( FD_UNLIKELY( ctx->raw_buf.size + rec_sz > RAW_BUF_SZ ) ) {
632 0 : zip_flush( ctx );
633 0 : }
634 :
635 0 : memset( &ctx->disk, 0, sizeof(ctx->disk) );
636 0 : ctx->disk.active = 1;
637 0 : ctx->disk.pubkey = frag->pubkey;
638 0 : ctx->disk.owner = frag->owner;
639 0 : ctx->disk.size = frag->size;
640 0 : ctx->disk.acc_idx = frag->acc_idx;
641 :
642 0 : fd_accdb_accmeta_t const * accmeta = accmeta_disk( ctx, &ctx->disk.pubkey, ctx->disk.size, ctx->disk.acc_idx );
643 0 : FD_CHECK_CRIT( accmeta, "bug in snapshot producer: rooted account disappeared from index" );
644 :
645 0 : snap_acc_hdr_t * hdr = (snap_acc_hdr_t *)( ctx->raw + ctx->raw_buf.size );
646 0 : memset( hdr, 0, sizeof(snap_acc_hdr_t) );
647 0 : hdr->pubkey = ctx->disk.pubkey;
648 0 : hdr->owner = ctx->disk.owner;
649 0 : hdr->lamports = FD_VOLATILE_CONST( accmeta->lamports );
650 0 : hdr->executable = !!FD_ACCDB_SIZE_EXEC( FD_VOLATILE_CONST( accmeta->executable_size ) );
651 0 : hdr->data_len = data_len;
652 :
653 0 : ctx->raw_buf.size += sizeof(snap_acc_hdr_t);
654 0 : ctx->disk.data_rem = data_len;
655 0 : ctx->disk.data_pad = fd_ulong_align_up( data_len, 8UL ) - data_len;
656 0 : return (ulong)frag->data_sz;
657 0 : }
658 :
659 : static void
660 : msg_acc_disk( fd_snapzp_t * ctx,
661 : ulong seq, /* unused */
662 : ulong sig, /* data pointer */
663 : ulong chunk, /* descriptor pointer */
664 : ulong sz, /* descriptor size */
665 : ulong ctl, /* holds orig, som, eom */
666 : ulong tsorig, /* unused */
667 0 : ulong tspub ) { /* data size */
668 0 : (void)seq; (void)tsorig;
669 0 : int som = fd_frag_meta_ctl_som( ctl );
670 0 : int eom = fd_frag_meta_ctl_eom( ctl );
671 :
672 : /* begin defrag operation */
673 0 : ulong frag_sz = tspub;
674 0 : if( FD_UNLIKELY( som ) ) {
675 0 : FD_CHECK_CRIT( chunk>=ctx->snapmk_zp_chunk0 && chunk<=ctx->snapmk_zp_wmark && sz==sizeof(fd_backup_disk_msg_t),
676 0 : "input frag bounds check failed: fd_backup_disk_msg_t(som=1)" );
677 0 : fd_backup_disk_msg_t const * frag = fd_chunk_to_laddr_const( ctx->snapmk_zp_mem, chunk );
678 0 : frag_sz = msg_acc_disk_start( ctx, frag );
679 0 : if( FD_UNLIKELY( tspub!=frag_sz ) ) {
680 0 : FD_LOG_CRIT(( "invalid accdb disk frag stream: size mismatch (%lu != %lu)", tspub, frag_sz ));
681 0 : }
682 0 : } else {
683 0 : FD_CHECK_CRIT( !sz, "invalid accdb disk frag stream: non-SOM frag must not have a descriptor" );
684 0 : }
685 0 : FD_CHECK_CRIT( !!ctx->disk.active, "invalid accdb disk frag stream: non-SOM frag seen but no active defrag op" );
686 :
687 : /* locate data pointer */
688 0 : uchar const * frag = NULL;
689 0 : if( FD_LIKELY( ctx->disk.data_rem ) ) {
690 0 : frag = fd_wksp_laddr_fast( ctx->snaprd_mem, sig );
691 0 : }
692 :
693 : /* defrag copy */
694 0 : ulong take = fd_ulong_min( ctx->disk.data_rem, frag_sz );
695 0 : if( FD_LIKELY( take ) ) {
696 0 : FD_CHECK_CRIT( ctx->raw_buf.size + take <= RAW_BUF_SZ,
697 0 : "internal bounds check failed" );
698 0 : FD_CHECK_CRIT( (ulong)frag >= ctx->snaprd_data0 &&
699 0 : (ulong)frag + frag_sz <= ctx->snaprd_data1,
700 0 : "snaprd bounds check failed" );
701 0 : fd_memcpy( ctx->raw + ctx->raw_buf.size, frag, take );
702 0 : ctx->raw_buf.size += take;
703 0 : ctx->disk.data_rem -= take;
704 0 : frag_sz -= take;
705 0 : }
706 0 : FD_CHECK_CRIT( !frag_sz, "invalid accdb disk frag stream: frag spans multiple accounts" );
707 0 : FD_CHECK_CRIT( !( !ctx->disk.data_rem && !eom ), "invalid accdb disk frag stream: non-EOM frag seen but data already complete" );
708 :
709 : /* finish defrag operation */
710 0 : if( eom ) {
711 0 : FD_CHECK_CRIT( !ctx->disk.data_rem, "invalid accdb disk frag stream: EOM frag seen but defrag not complete" );
712 0 : if( ctx->disk.data_pad ) {
713 0 : FD_TEST( ctx->raw_buf.size + ctx->disk.data_pad <= RAW_BUF_SZ );
714 0 : fd_memset( ctx->raw + ctx->raw_buf.size, 0, ctx->disk.data_pad );
715 0 : ctx->raw_buf.size += ctx->disk.data_pad;
716 0 : }
717 0 : ctx->snapshot_account_cnt++;
718 0 : ctx->snapshot_account_sz += sizeof(snap_acc_hdr_t) + fd_ulong_align_up( (ulong)FD_ACCDB_SIZE_DATA( ctx->disk.size ), 8UL );
719 0 : ctx->snapshot_disk_account_cnt++;
720 0 : ctx->metrics.accounts_compressed++;
721 0 : memset( &ctx->disk, 0, sizeof(ctx->disk) );
722 0 : }
723 0 : }
724 :
725 : /* msg_acc_disk instructions snapzp to compress a batch of unfragmented
726 : accounts that were read from the accdb disk file. (fast path) */
727 :
728 : static void
729 : msg_acc_disk_batch( fd_snapzp_t * ctx,
730 : fd_backup_disk_batch_msg_t const * batch,
731 0 : ulong sig ) {
732 0 : FD_CHECK_CRIT( ctx->snap_fd>=0, "invalid snapshot file descriptor" );
733 0 : FD_CHECK_CRIT( !ctx->disk.active, "received account batch while already processing a disk account" );
734 :
735 0 : fd_backup_accidx_t const * idx = &ctx->acc_cache->idx;
736 0 : fd_accdb_accmeta_t const * acc_pool = idx->acc_pool;
737 :
738 : /* MLP gather of account index entries */
739 0 : static fd_accdb_accmeta_t const dead = {0};
740 0 : fd_accdb_accmeta_t const * gather[ FD_BACKUP_DISK_PARA ];
741 0 : for( ulong i=0UL; i<FD_BACKUP_DISK_PARA; i++ ) {
742 0 : uint ai = batch->acc_idx[ i ];
743 0 : gather[ i ] = fd_backup_accidx_valid( idx, ai ) ? &acc_pool[ ai ] : &dead;
744 0 : }
745 0 : ulong lamports[ FD_BACKUP_DISK_PARA ];
746 0 : uint exec_sz [ FD_BACKUP_DISK_PARA ];
747 0 : for( ulong i=0UL; i<FD_BACKUP_DISK_PARA; i++ ) {
748 0 : lamports[ i ] = FD_VOLATILE_CONST( gather[ i ]->lamports );
749 0 : exec_sz [ i ] = FD_VOLATILE_CONST( gather[ i ]->executable_size );
750 0 : }
751 :
752 : /* Bounds check batch (TOCTOU susceptible, but this is an accepted
753 : risk, as snapmk->snapzp is a trusted link) */
754 0 : uchar const * base = fd_wksp_laddr_fast( ctx->snaprd_mem, sig );
755 0 : for( ulong i=0UL; i<FD_BACKUP_DISK_PARA; i++ ) {
756 0 : FD_CHECK_CRIT( ((ulong)base + batch->frag_off[ i ] >= ctx->snaprd_data0) &
757 0 : ((ulong)base + batch->frag_off[ i ] + sizeof(fd_accdb_disk_meta_t) <= (ulong)ctx->snaprd_data1),
758 0 : "account data bounds check fail" );
759 0 : }
760 :
761 : /* Sequential scan of disk data in frag */
762 0 : for( ulong i=0UL; i<FD_BACKUP_DISK_PARA; i++ ) {
763 0 : uint acc_idx = batch->acc_idx[ i ];
764 0 : if( acc_idx==UINT_MAX ) continue;
765 0 : FD_CHECK_CRIT( fd_backup_accidx_valid( idx, acc_idx ), "account index bounds check fail" );
766 :
767 0 : fd_accdb_disk_meta_t const * dm = (fd_accdb_disk_meta_t const *)( base + batch->frag_off[ i ] );
768 0 : FD_CHECK_CRIT( (ulong)(dm+1) <= (ulong)ctx->snaprd_data1, "account data bounds check fail" );
769 :
770 0 : ulong data_len = (ulong)FD_ACCDB_SIZE_DATA( dm->size );
771 0 : FD_CHECK_CRIT( (ulong)(dm+1)+data_len <= (ulong)ctx->snaprd_data1, "account data bounds check fail" );
772 :
773 : /* validate that disk data matches index */
774 0 : FD_CHECK_CRIT( FD_ACCDB_SIZE_DATA( exec_sz[ i ] )==data_len, "account query corruption detected" );
775 0 : FD_CHECK_CRIT( !memcmp( gather[ i ]->key.pubkey, dm->pubkey, sizeof(fd_pubkey_t) ), "account query corruption detected" );
776 :
777 0 : ulong rec_sz = sizeof(snap_acc_hdr_t) + fd_ulong_align_up( data_len, 8UL );
778 0 : ulong data_pad = fd_ulong_align_up( data_len, 8UL ) - data_len;
779 0 : FD_CHECK_CRIT( rec_sz<=RAW_BUF_SZ, "oversize snapshot account record" );
780 0 : if( FD_UNLIKELY( ctx->raw_buf.size + rec_sz > RAW_BUF_SZ ) ) {
781 0 : zip_flush( ctx );
782 0 : }
783 :
784 0 : snap_acc_hdr_t * hdr = (snap_acc_hdr_t *)( ctx->raw + ctx->raw_buf.size );
785 0 : memset( hdr, 0, sizeof(snap_acc_hdr_t) );
786 0 : memcpy( hdr->pubkey.uc, dm->pubkey, sizeof(fd_pubkey_t) );
787 0 : memcpy( hdr->owner.uc, dm->owner, sizeof(fd_pubkey_t) );
788 0 : hdr->lamports = lamports[ i ];
789 0 : hdr->executable = !!FD_ACCDB_SIZE_EXEC( exec_sz[ i ] );
790 0 : hdr->data_len = data_len;
791 0 : ctx->raw_buf.size += sizeof(snap_acc_hdr_t);
792 :
793 0 : if( FD_LIKELY( data_len ) ) {
794 0 : uchar const * data = base + batch->frag_off[ i ] + sizeof(fd_accdb_disk_meta_t);
795 0 : fd_memcpy( ctx->raw + ctx->raw_buf.size, data, data_len );
796 0 : ctx->raw_buf.size += data_len;
797 0 : }
798 0 : if( data_pad ) {
799 0 : fd_memset( ctx->raw + ctx->raw_buf.size, 0, data_pad );
800 0 : ctx->raw_buf.size += data_pad;
801 0 : }
802 0 : ctx->snapshot_account_cnt++;
803 0 : ctx->snapshot_account_sz += rec_sz;
804 0 : ctx->snapshot_disk_account_cnt++;
805 0 : ctx->metrics.accounts_compressed++;
806 0 : }
807 0 : }
808 :
809 : /* msg_done is called on all snapzp tiles once all account data has been
810 : compressed. This causes snapzp to enter sleep state. */
811 :
812 : static void
813 0 : msg_done( fd_snapzp_t * ctx ) {
814 : /* Leave file descriptors open, but release local reference */
815 0 : ctx->snap_fd = -1;
816 0 : }
817 :
818 : /* returnable_frag is called for every message received from snapmk. */
819 :
820 : static int
821 : returnable_frag( fd_snapzp_t * ctx,
822 : ulong in_idx,
823 : ulong seq,
824 : ulong sig,
825 : ulong chunk,
826 : ulong sz,
827 : ulong ctl,
828 : ulong tsorig,
829 : ulong tspub,
830 0 : fd_stem_context_t * stem ) {
831 0 : (void)in_idx; (void)stem;
832 0 : ctx->idle_cnt = 0UL;
833 0 : ulong orig = fd_frag_meta_ctl_orig( ctl );
834 :
835 0 : # define MSG_TRANSLATE( msg_type ) __extension__({ \
836 0 : FD_CHECK_CRIT( chunk >= ctx->snapmk_zp_chunk0 && chunk <= ctx->snapmk_zp_wmark && sz==sizeof(msg_type), "input frag bounds check failed: " #msg_type ); \
837 0 : (msg_type const *)fd_chunk_to_laddr_const( ctx->snapmk_zp_mem, chunk ); \
838 0 : })
839 :
840 0 : switch( orig ) {
841 0 : case FD_BACKUP_ORIG_START:
842 0 : msg_start( ctx, MSG_TRANSLATE( fd_backup_start_msg_t ) );
843 0 : break;
844 0 : case FD_BACKUP_ORIG_ACC_CACHE:
845 0 : msg_acc_cache( ctx, MSG_TRANSLATE( fd_backup_cache_msg_t ) );
846 0 : zip_work( ctx );
847 0 : break;
848 0 : case FD_BACKUP_ORIG_ACC_DELTA:
849 0 : msg_acc_delta( ctx, MSG_TRANSLATE( fd_backup_delta_msg_t ) );
850 0 : zip_work( ctx );
851 0 : break;
852 0 : case FD_BACKUP_ORIG_ACC_DISK:
853 0 : msg_acc_disk( ctx, seq, sig, chunk, sz, ctl, tsorig, tspub );
854 0 : zip_work( ctx );
855 0 : break;
856 0 : case FD_BACKUP_ORIG_ACC_DISK_BATCH:
857 0 : msg_acc_disk_batch( ctx, MSG_TRANSLATE( fd_backup_disk_batch_msg_t ), sig );
858 0 : zip_work( ctx );
859 0 : break;
860 0 : case FD_BACKUP_ORIG_FLUSH:
861 0 : zip_flush( ctx );
862 0 : fd_backup_worker_stats_t * stats = &ctx->stats->worker[ ctx->kind_id ];
863 0 : __atomic_store_n( &stats->account_cnt, ctx->snapshot_account_cnt, __ATOMIC_RELAXED );
864 0 : __atomic_store_n( &stats->account_sz, ctx->snapshot_account_sz, __ATOMIC_RELAXED );
865 0 : __atomic_store_n( &stats->tombstone_cnt, ctx->snapshot_tombstone_cnt, __ATOMIC_RELAXED );
866 0 : __atomic_store_n( &stats->cached_account_cnt, ctx->snapshot_cached_account_cnt, __ATOMIC_RELAXED );
867 0 : __atomic_store_n( &stats->disk_account_cnt, ctx->snapshot_disk_account_cnt, __ATOMIC_RELAXED );
868 0 : __atomic_store_n( &stats->zstd_data_frame_cnt, 2UL*ctx->frame_cnt, __ATOMIC_RELAXED );
869 0 : __atomic_store_n( &stats->zstd_padding_sz, ctx->snapshot_zstd_padding_sz, __ATOMIC_RELAXED );
870 0 : __atomic_store_n( &stats->compress_ticks, ctx->metrics.compress_ticks-ctx->snapshot_compress_ticks0, __ATOMIC_RELAXED );
871 0 : __atomic_store_n( &stats->io_blocked_ticks, ctx->metrics.io_blocked_ticks-ctx->snapshot_io_blocked_ticks0, __ATOMIC_RELAXED );
872 0 : __atomic_store_n( &stats->uncompressed_sz, ctx->snapshot_uncompressed_sz, __ATOMIC_RELEASE );
873 0 : break;
874 0 : case FD_BACKUP_ORIG_DONE:
875 0 : msg_done( ctx );
876 0 : break;
877 0 : default:
878 0 : FD_LOG_CRIT(( "unknown backup instruction (orig=%lu, seq=%lu)", orig, seq ));
879 0 : }
880 :
881 0 : return 0;
882 0 : }
883 :
884 : static void
885 0 : metrics_write( fd_snapzp_t * ctx ) {
886 0 : FD_MCNT_SET( SNAPZP, ACCOUNT_COMPRESSED, ctx->metrics.accounts_compressed );
887 0 : FD_MCNT_SET( SNAPZP, COMPRESS_CONSUMED_BYTES, ctx->metrics.bytes_compressed );
888 0 : FD_MCNT_SET( SNAPZP, DISK_WRITTEN_BYTES, ctx->metrics.bytes_written );
889 0 : FD_MCNT_SET( SNAPZP, IO_BLOCKED_DURATION_NANOS, ctx->metrics.io_blocked_ticks );
890 0 : FD_MCNT_SET( SNAPZP, COMPRESS_DURATION_NANOS, ctx->metrics.compress_ticks );
891 0 : FD_MCNT_SET( SNAPZP, CACHE_READ_TORN, ctx->metrics.cache_read_torn );
892 0 : }
893 :
894 0 : #define STEM_BURST 1UL
895 0 : #define STEM_LAZY 9400UL
896 0 : #define STEM_CALLBACK_CONTEXT_TYPE fd_snapzp_t
897 0 : #define STEM_CALLBACK_CONTEXT_ALIGN alignof(fd_snapzp_t)
898 0 : #define STEM_CALLBACK_BEFORE_CREDIT before_credit
899 0 : #define STEM_CALLBACK_RETURNABLE_FRAG returnable_frag
900 0 : #define STEM_CALLBACK_METRICS_WRITE metrics_write
901 : #include "../../disco/stem/fd_stem.c"
902 :
903 : #ifndef FD_TILE_TEST
904 : fd_topo_run_tile_t fd_tile_snapzp = {
905 : .name = "snapzp",
906 : .populate_allowed_fds = populate_allowed_fds,
907 : .populate_allowed_seccomp = populate_allowed_seccomp,
908 : .scratch_align = scratch_align,
909 : .scratch_footprint = scratch_footprint,
910 : .privileged_init = privileged_init,
911 : .unprivileged_init = unprivileged_init,
912 : .run = stem_run,
913 : };
914 : #endif
|
