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