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