Line data Source code
1 : #include "fd_snapwm_tile_private.h"
2 : #include "utils/fd_ssctrl.h"
3 : #include "utils/fd_vinyl_io_wd.h"
4 :
5 : #include "../../disco/topo/fd_topo.h"
6 : #include "../../disco/metrics/fd_metrics.h"
7 : #include "../../flamenco/runtime/sysvar/fd_sysvar_slot_history.h"
8 : #include "../../flamenco/runtime/fd_system_ids.h"
9 :
10 : #define NAME "snapwm"
11 :
12 : /* The snapwm tile is a state machine responsible for loading accounts
13 : into vinyl database. It processes pre-assembled bstream pairs
14 : and handles vinyl's meta_map and bstream actual allocation. */
15 :
16 : static inline int
17 0 : should_shutdown( fd_snapwm_tile_t * ctx ) {
18 0 : if( FD_UNLIKELY( ctx->state==FD_SNAPSHOT_STATE_SHUTDOWN ) ) {
19 0 : ulong accounts_dup = ctx->metrics.accounts_ignored + ctx->metrics.accounts_replaced;
20 0 : ulong accounts = ctx->metrics.accounts_loaded - accounts_dup;
21 0 : long elapsed_ns = fd_log_wallclock() - ctx->boot_timestamp;
22 0 : FD_LOG_NOTICE(( "loaded %.1fM accounts (%.1fM dups) from snapshot in %.3f seconds",
23 0 : (double)accounts/1e6,
24 0 : (double)accounts_dup/1e6,
25 0 : (double)elapsed_ns/1e9 ));
26 0 : }
27 0 : return ctx->state==FD_SNAPSHOT_STATE_SHUTDOWN;
28 0 : }
29 :
30 : static ulong
31 0 : scratch_align( void ) {
32 0 : return 512UL;
33 0 : }
34 :
35 : static ulong
36 0 : scratch_footprint( fd_topo_tile_t const * tile ) {
37 0 : (void)tile;
38 0 : ulong l = FD_LAYOUT_INIT;
39 0 : l = FD_LAYOUT_APPEND( l, alignof(fd_snapwm_tile_t), sizeof(fd_snapwm_tile_t) );
40 0 : l = FD_LAYOUT_APPEND( l, fd_vinyl_io_wd_align(), fd_vinyl_io_wd_footprint( tile->snapwm.snapwr_depth ) );
41 0 : l = FD_LAYOUT_APPEND( l, fd_vinyl_io_mm_align(), fd_vinyl_io_mm_footprint( FD_SNAPWM_IO_SPAD_MAX ) );
42 0 : return FD_LAYOUT_FINI( l, scratch_align() );
43 0 : }
44 :
45 : static void
46 0 : metrics_write( fd_snapwm_tile_t * ctx ) {
47 0 : FD_MGAUGE_SET( SNAPWM, STATE, (ulong)ctx->state );
48 0 : FD_MGAUGE_SET( SNAPWM, ACCOUNTS_LOADED, ctx->metrics.accounts_loaded );
49 0 : FD_MGAUGE_SET( SNAPWM, ACCOUNTS_REPLACED, ctx->metrics.accounts_replaced );
50 0 : FD_MGAUGE_SET( SNAPWM, ACCOUNTS_IGNORED, ctx->metrics.accounts_ignored );
51 0 : FD_MGAUGE_SET( SNAPWM, ACCOUNTS_ACTIVE, ctx->vinyl.pair_cnt );
52 0 : }
53 :
54 : static inline void
55 : seq_to_tsorig_tspub( ulong * tsorig,
56 : ulong * tspub,
57 0 : ulong seq ) {
58 0 : *tsorig = ( seq >> 0 ) & ((1UL<<32)-1UL);
59 0 : *tspub = ( seq >> 32 ) & ((1UL<<32)-1UL);
60 0 : }
61 :
62 : static void
63 : transition_malformed( fd_snapwm_tile_t * ctx,
64 0 : fd_stem_context_t * stem ) {
65 0 : if( FD_UNLIKELY( ctx->state==FD_SNAPSHOT_STATE_ERROR ) ) return;
66 0 : ctx->state = FD_SNAPSHOT_STATE_ERROR;
67 0 : fd_stem_publish( stem, ctx->out_ct_idx, FD_SNAPSHOT_MSG_CTRL_ERROR, 0UL, 0UL, 0UL, 0UL, 0UL );
68 0 : }
69 :
70 : /* verify_slot_deltas_with_slot_history verifies the 'SlotHistory'
71 : sysvar account after loading a snapshot. The full database
72 : architecture is only instantiated after snapshot loading, so this
73 : function uses a primitive/cache-free mechanism to query the parts of
74 : the account database that are available.
75 :
76 : Returns 0 if verification passed, -1 if not. */
77 :
78 : static int
79 0 : verify_slot_deltas_with_slot_history( fd_snapwm_tile_t * ctx ) {
80 : /* Do a raw read of the slot history sysvar account from the database.
81 : Requires approx 500kB stack space. */
82 :
83 0 : fd_account_meta_t meta;
84 0 : uchar data[ FD_SYSVAR_SLOT_HISTORY_BINCODE_SZ ];
85 0 : union {
86 0 : uchar buf[ FD_SYSVAR_SLOT_HISTORY_FOOTPRINT ];
87 0 : fd_slot_history_global_t o;
88 0 : } decoded;
89 0 : FD_STATIC_ASSERT( offsetof( __typeof__(decoded), buf)==offsetof( __typeof__(decoded), o ), memory_layout );
90 0 : fd_snapwm_vinyl_read_account( ctx, &fd_sysvar_slot_history_id, &meta, data, sizeof(data) );
91 :
92 0 : if( FD_UNLIKELY( !meta.lamports || !meta.dlen ) ) {
93 0 : FD_LOG_WARNING(( "SlotHistory sysvar account missing or empty" ));
94 0 : return -1;
95 0 : }
96 0 : if( FD_UNLIKELY( meta.dlen > FD_SYSVAR_SLOT_HISTORY_BINCODE_SZ ) ) {
97 0 : FD_LOG_WARNING(( "SlotHistory sysvar account data too large: %u bytes", meta.dlen ));
98 0 : return -1;
99 0 : }
100 0 : if( FD_UNLIKELY( !fd_memeq( meta.owner, fd_sysvar_owner_id.uc, sizeof(fd_pubkey_t) ) ) ) {
101 0 : FD_BASE58_ENCODE_32_BYTES( meta.owner, owner_b58 );
102 0 : FD_LOG_WARNING(( "SlotHistory sysvar owner is invalid: %s != sysvar_owner_id", owner_b58 ));
103 0 : return -1;
104 0 : }
105 :
106 0 : if( FD_UNLIKELY(
107 0 : !fd_bincode_decode_static_global(
108 0 : slot_history,
109 0 : &decoded.o,
110 0 : data,
111 0 : meta.dlen )
112 0 : ) ) {
113 0 : FD_LOG_WARNING(( "SlotHistory sysvar account data is corrupt" ));
114 0 : return -1;
115 0 : }
116 :
117 0 : ulong txncache_entries_len = fd_ulong_load_8( ctx->txncache_entries_len_ptr );
118 0 : if( FD_UNLIKELY( !txncache_entries_len ) ) {
119 0 : FD_LOG_WARNING(( "unexpected txncache entries length %lu. continue...", txncache_entries_len ));
120 0 : }
121 :
122 0 : for( ulong i=0UL; i<txncache_entries_len; i++ ) {
123 0 : fd_sstxncache_entry_t const * entry = &ctx->txncache_entries[i];
124 0 : if( FD_UNLIKELY( fd_sysvar_slot_history_find_slot( &decoded.o, entry->slot )!=FD_SLOT_HISTORY_SLOT_FOUND ) ) {
125 0 : FD_LOG_WARNING(( "slot %lu missing from SlotHistory sysvar account", entry->slot ));
126 0 : return -1;
127 0 : }
128 0 : }
129 0 : return 0;
130 0 : }
131 :
132 : static int
133 : handle_data_frag( fd_snapwm_tile_t * ctx,
134 : ulong chunk,
135 : ulong acc_cnt,
136 0 : fd_stem_context_t * stem ) {
137 :
138 0 : if( FD_UNLIKELY( ctx->state==FD_SNAPSHOT_STATE_FINISHING ) ) {
139 0 : FD_LOG_WARNING(( "received unexpected data frag while in state %s (%lu)",
140 0 : fd_ssctrl_state_str( (ulong)ctx->state ), (ulong)ctx->state ));
141 0 : transition_malformed( ctx, stem );
142 0 : return 0;
143 0 : }
144 0 : if( FD_UNLIKELY( ctx->state==FD_SNAPSHOT_STATE_ERROR ) ) {
145 : /* Ignore all data frags after observing an error in the stream until
146 : we receive fail & init control messages to restart processing. */
147 0 : return 0;
148 0 : }
149 0 : if( FD_UNLIKELY( ctx->state!=FD_SNAPSHOT_STATE_PROCESSING ) ) {
150 0 : FD_LOG_ERR(( "received data frag during invalid state %s (%lu)",
151 0 : fd_ssctrl_state_str( (ulong)ctx->state ), (ulong)ctx->state ));
152 0 : }
153 :
154 0 : FD_TEST( chunk>=ctx->in.chunk0 && chunk<=ctx->in.wmark && acc_cnt<=FD_SNAPWM_PAIR_BATCH_CNT_MAX );
155 :
156 0 : fd_snapwm_vinyl_process_account( ctx, chunk, acc_cnt, stem );
157 :
158 0 : return 0;
159 0 : }
160 :
161 : static void
162 : handle_control_frag( fd_snapwm_tile_t * ctx,
163 : ulong sig,
164 0 : fd_stem_context_t * stem ) {
165 0 : if( ctx->state==FD_SNAPSHOT_STATE_ERROR && sig!=FD_SNAPSHOT_MSG_CTRL_FAIL ) {
166 : /* Control messages move along the snapshot load pipeline. Since
167 : error conditions can be triggered by any tile in the pipeline,
168 : it is possible to be in error state and still receive otherwise
169 : valid messages. Only a fail message can revert this. */
170 0 : return;
171 0 : };
172 :
173 0 : int forward_msg = 1;
174 :
175 0 : switch( sig ) {
176 0 : case FD_SNAPSHOT_MSG_CTRL_INIT_FULL:
177 0 : case FD_SNAPSHOT_MSG_CTRL_INIT_INCR: {
178 0 : FD_TEST( ctx->state==FD_SNAPSHOT_STATE_IDLE );
179 0 : ctx->state = FD_SNAPSHOT_STATE_PROCESSING;
180 0 : ctx->full = sig==FD_SNAPSHOT_MSG_CTRL_INIT_FULL;
181 : /* When lthash verification is disabled, vinyl txn operation can
182 : be used, providing a fast revert mechanism when the incr
183 : snapshot is cancelled. However, when the lthash verification
184 : is enabled, the meta map needs to be updated as the accounts
185 : are processed, in order to detemine which duplicates are new
186 : and which are old. This prevents us from using txn_commit
187 : when lthash is enabled. (In more detail: the need of having
188 : recovery_seq in the pair's phdr info makes an update on the
189 : bstream during txn_commit not feasible, since it would require
190 : recomputing the pair's integrity hashes). For the case when
191 : lthash verification is enabled, there is a recovery mechanism
192 : in place. */
193 0 : if( sig==FD_SNAPSHOT_MSG_CTRL_INIT_INCR && ctx->lthash_disabled ) {
194 0 : fd_snapwm_vinyl_txn_begin( ctx );
195 0 : }
196 0 : fd_snapwm_vinyl_wd_init( ctx );
197 : /* backup bstream seq, independent of whether full or incr. */
198 0 : fd_snapwm_vinyl_recovery_seq_backup( ctx );
199 :
200 : /* There is a way to avoid a lock here: every other writer, in
201 : particular the write tiles that update wr_seq, have already
202 : finished processing all standing writes and are idle. And
203 : even though any read tile may see partial updates as they
204 : occur, this all happens while they are idle waiting for the
205 : init full/incr command. */
206 0 : if( !!ctx->vinyl.admin ) fd_snapwm_vinyl_update_admin( ctx, 0/*do_rwlock*/ );
207 :
208 : /* Rewind metric counters (no-op unless recovering from a fail) */
209 0 : if( sig==FD_SNAPSHOT_MSG_CTRL_INIT_FULL ) {
210 0 : ctx->metrics.accounts_loaded = ctx->metrics.full_accounts_loaded = 0UL;
211 0 : ctx->metrics.accounts_replaced = ctx->metrics.full_accounts_replaced = 0UL;
212 0 : ctx->metrics.accounts_ignored = ctx->metrics.full_accounts_ignored = 0UL;
213 0 : ctx->vinyl.pair_cnt = ctx->vinyl.full_pair_cnt = 0UL;
214 0 : } else {
215 0 : ctx->metrics.accounts_loaded = ctx->metrics.full_accounts_loaded;
216 0 : ctx->metrics.accounts_replaced = ctx->metrics.full_accounts_replaced;
217 0 : ctx->metrics.accounts_ignored = ctx->metrics.full_accounts_ignored;
218 0 : ctx->vinyl.pair_cnt = ctx->vinyl.full_pair_cnt;
219 0 : }
220 0 : break;
221 0 : }
222 :
223 0 : case FD_SNAPSHOT_MSG_CTRL_FINI: {
224 0 : FD_TEST( ctx->state==FD_SNAPSHOT_STATE_PROCESSING );
225 0 : ctx->state = FD_SNAPSHOT_STATE_FINISHING;
226 0 : fd_snapwm_vinyl_wd_fini( ctx );
227 0 : if( ctx->vinyl.txn_active ) {
228 0 : fd_snapwm_vinyl_txn_commit( ctx, stem );
229 0 : }
230 0 : break;
231 0 : }
232 :
233 0 : case FD_SNAPSHOT_MSG_CTRL_NEXT: {
234 0 : FD_TEST( ctx->state==FD_SNAPSHOT_STATE_FINISHING );
235 0 : ctx->state = FD_SNAPSHOT_STATE_IDLE;
236 :
237 : /* Backup metric counters */
238 0 : ctx->metrics.full_accounts_loaded = ctx->metrics.accounts_loaded;
239 0 : ctx->metrics.full_accounts_replaced = ctx->metrics.accounts_replaced;
240 0 : ctx->metrics.full_accounts_ignored = ctx->metrics.accounts_ignored;
241 0 : ctx->vinyl.full_pair_cnt = ctx->vinyl.pair_cnt;
242 :
243 0 : if( FD_UNLIKELY( verify_slot_deltas_with_slot_history( ctx ) ) ) {
244 0 : FD_LOG_WARNING(( "slot deltas verification failed for full snapshot" ));
245 0 : transition_malformed( ctx, stem );
246 0 : forward_msg = 0;
247 0 : break;
248 0 : }
249 0 : break;
250 0 : }
251 :
252 0 : case FD_SNAPSHOT_MSG_CTRL_DONE: {
253 0 : FD_TEST( ctx->state==FD_SNAPSHOT_STATE_FINISHING );
254 0 : ctx->state = FD_SNAPSHOT_STATE_IDLE;
255 :
256 0 : if( FD_UNLIKELY( verify_slot_deltas_with_slot_history( ctx ) ) ) {
257 0 : if( ctx->full ) FD_LOG_WARNING(( "slot deltas verification failed for full snapshot" ));
258 0 : else FD_LOG_WARNING(( "slot deltas verification failed for incremental snapshot" ));
259 0 : transition_malformed( ctx, stem );
260 0 : forward_msg = 0;
261 0 : break;
262 0 : }
263 0 : break;
264 0 : }
265 :
266 0 : case FD_SNAPSHOT_MSG_CTRL_ERROR: {
267 0 : FD_TEST( ctx->state!=FD_SNAPSHOT_STATE_SHUTDOWN );
268 0 : ctx->state = FD_SNAPSHOT_STATE_ERROR;
269 0 : break;
270 0 : }
271 :
272 0 : case FD_SNAPSHOT_MSG_CTRL_FAIL: {
273 0 : FD_TEST( ctx->state!=FD_SNAPSHOT_STATE_SHUTDOWN );
274 0 : ctx->state = FD_SNAPSHOT_STATE_IDLE;
275 0 : fd_snapwm_vinyl_wd_fini( ctx );
276 0 : if( ctx->full ) {
277 0 : fd_snapwm_vinyl_revert_full( ctx );
278 0 : } else {
279 0 : if( ctx->vinyl.txn_active ) {
280 0 : fd_snapwm_vinyl_txn_cancel( ctx );
281 0 : } else {
282 0 : fd_snapwm_vinyl_revert_incr( ctx );
283 0 : }
284 0 : }
285 0 : break;
286 0 : }
287 :
288 0 : case FD_SNAPSHOT_MSG_CTRL_SHUTDOWN: {
289 0 : FD_TEST( ctx->state==FD_SNAPSHOT_STATE_IDLE );
290 0 : ctx->state = FD_SNAPSHOT_STATE_SHUTDOWN;
291 0 : fd_snapwm_vinyl_shutdown( ctx );
292 0 : break;
293 0 : }
294 :
295 0 : default: {
296 0 : FD_LOG_ERR(( "unexpected control frag %s (%lu) in state %s (%lu)",
297 0 : fd_ssctrl_msg_ctrl_str( sig ), sig,
298 0 : fd_ssctrl_state_str( (ulong)ctx->state ), (ulong)ctx->state ));
299 0 : break;
300 0 : }
301 0 : }
302 :
303 0 : if( FD_LIKELY( forward_msg ) ) {
304 : /* Encode wr_seq into tsorig and tspub. Downstream will make use
305 : of wr_seq when needed. */
306 0 : ulong tsorig = 0UL;
307 0 : ulong tspub = 0UL;
308 0 : if( !ctx->lthash_disabled ) {
309 0 : seq_to_tsorig_tspub( &tsorig, &tspub, ((fd_vinyl_io_wd_t const *)ctx->vinyl.io_wd)->wr_seq );
310 0 : }
311 : /* Forward the control message down the pipeline */
312 0 : fd_stem_publish( stem, ctx->out_ct_idx, sig, 0UL, 0UL, 0UL, tsorig, tspub );
313 0 : }
314 0 : }
315 :
316 : static inline void
317 : handle_expected_hash_message( fd_snapwm_tile_t * ctx,
318 : ulong chunk,
319 : ulong sz,
320 0 : fd_stem_context_t * stem ) {
321 0 : if( FD_UNLIKELY( ctx->lthash_disabled ) ) return;
322 0 : if( FD_UNLIKELY( sz!=sizeof(fd_lthash_value_t) ) ) {
323 0 : FD_LOG_ERR(( "unexpected msg sz %lu for sig FD_SNAPSHOT_HASH_MSG_EXPECTED", sz ));
324 0 : return;
325 0 : }
326 :
327 0 : uchar * src = fd_chunk_to_laddr( ctx->in.wksp, chunk );
328 0 : uchar * dst = fd_chunk_to_laddr( ctx->hash_out.mem, ctx->hash_out.chunk );
329 0 : memcpy( dst, src, sz );
330 0 : fd_stem_publish( stem, ctx->out_ct_idx, FD_SNAPSHOT_HASH_MSG_EXPECTED, ctx->hash_out.chunk, sz, 0UL, 0UL, 0UL );
331 0 : ctx->hash_out.chunk = fd_dcache_compact_next( ctx->hash_out.chunk, sz, ctx->hash_out.chunk0, ctx->hash_out.wmark );
332 0 : }
333 :
334 : static inline void
335 : handle_expected_capitalization_message( fd_snapwm_tile_t * ctx,
336 : ulong chunk,
337 : ulong sz,
338 0 : fd_stem_context_t * stem ) {
339 0 : if( FD_UNLIKELY( sz!=sizeof(fd_ssctrl_capitalization_t) ) ) {
340 0 : FD_LOG_ERR(( "unexpected msg sz %lu for sig FD_SNAPSHOT_MSG_EXP_CAPITALIZATION", sz ));
341 0 : return;
342 0 : }
343 :
344 0 : fd_ssctrl_capitalization_t const * src = fd_chunk_to_laddr_const( ctx->in.wksp, chunk );
345 0 : fd_ssctrl_capitalization_t * dst = fd_chunk_to_laddr( ctx->hash_out.mem, ctx->hash_out.chunk );
346 0 : memcpy( dst, src, sz );
347 0 : fd_stem_publish( stem, ctx->out_ct_idx, FD_SNAPSHOT_MSG_EXP_CAPITALIZATION, ctx->hash_out.chunk, sz, 0UL, 0UL, 0UL );
348 0 : ctx->hash_out.chunk = fd_dcache_compact_next( ctx->hash_out.chunk, sz, ctx->hash_out.chunk0, ctx->hash_out.wmark );
349 0 : }
350 :
351 : static inline int
352 : returnable_frag( fd_snapwm_tile_t * ctx,
353 : ulong in_idx FD_PARAM_UNUSED,
354 : ulong seq FD_PARAM_UNUSED,
355 : ulong sig,
356 : ulong chunk,
357 : ulong sz,
358 : ulong ctl FD_PARAM_UNUSED,
359 : ulong tsorig FD_PARAM_UNUSED,
360 : ulong tspub FD_PARAM_UNUSED,
361 0 : fd_stem_context_t * stem ) {
362 0 : FD_TEST( ctx->state!=FD_SNAPSHOT_STATE_SHUTDOWN );
363 :
364 0 : int ret = 0;
365 0 : if( FD_LIKELY( sig==FD_SNAPSHOT_MSG_DATA ) ) ret = handle_data_frag( ctx, chunk, sz/*acc_cnt*/, stem );
366 0 : else if( FD_UNLIKELY( sig==FD_SNAPSHOT_HASH_MSG_EXPECTED ) ) handle_expected_hash_message( ctx, chunk, sz, stem );
367 0 : else if( FD_UNLIKELY( sig==FD_SNAPSHOT_MSG_EXP_CAPITALIZATION ) ) handle_expected_capitalization_message( ctx, chunk, sz, stem );
368 0 : else handle_control_frag( ctx, sig, stem );
369 :
370 0 : return ret;
371 0 : }
372 :
373 : static ulong
374 : populate_allowed_fds( fd_topo_t const * topo FD_PARAM_UNUSED,
375 : fd_topo_tile_t const * tile FD_PARAM_UNUSED,
376 : ulong out_fds_cnt,
377 0 : int * out_fds ) {
378 0 : if( FD_UNLIKELY( out_fds_cnt<2UL ) ) FD_LOG_ERR(( "invalid out_fds_cnt %lu", out_fds_cnt ));
379 :
380 0 : ulong out_cnt = 0;
381 0 : out_fds[ out_cnt++ ] = 2UL; /* stderr */
382 0 : if( FD_LIKELY( -1!=fd_log_private_logfile_fd() ) ) {
383 0 : out_fds[ out_cnt++ ] = fd_log_private_logfile_fd(); /* logfile */
384 0 : }
385 :
386 0 : return out_cnt;
387 0 : }
388 :
389 : static ulong
390 : populate_allowed_seccomp( fd_topo_t const * topo,
391 : fd_topo_tile_t const * tile,
392 : ulong out_cnt,
393 0 : struct sock_filter * out ) {
394 0 : (void)topo; (void)tile;
395 0 : return fd_snapwm_vinyl_seccomp( out_cnt, out );
396 0 : }
397 :
398 : static void
399 : privileged_init( fd_topo_t * topo,
400 0 : fd_topo_tile_t * tile ) {
401 0 : fd_snapwm_tile_t * ctx = fd_topo_obj_laddr( topo, tile->tile_obj_id );
402 0 : memset( ctx, 0, sizeof(fd_snapwm_tile_t) );
403 0 : FD_TEST( fd_rng_secure( &ctx->seed, 8UL ) );
404 :
405 0 : fd_snapwm_vinyl_privileged_init( ctx, topo, tile );
406 0 : }
407 :
408 : static inline fd_snapwm_out_link_t
409 : out1( fd_topo_t const * topo,
410 : fd_topo_tile_t const * tile,
411 0 : char const * name ) {
412 0 : ulong idx = fd_topo_find_tile_out_link( topo, tile, name, 0UL );
413 :
414 0 : fd_topo_link_t const * out_link = &topo->links[ tile->out_link_id[ idx ] ];
415 :
416 0 : if( FD_UNLIKELY( idx==ULONG_MAX ) ) return (fd_snapwm_out_link_t){0};
417 :
418 0 : ulong mtu = topo->links[ tile->out_link_id[ idx ] ].mtu;
419 0 : if( FD_UNLIKELY( mtu==0UL ) ) return (fd_snapwm_out_link_t){0};
420 :
421 0 : void * mem = topo->workspaces[ topo->objs[ topo->links[ tile->out_link_id[ idx ] ].dcache_obj_id ].wksp_id ].wksp;
422 0 : ulong chunk0 = fd_dcache_compact_chunk0( mem, topo->links[ tile->out_link_id[ idx ] ].dcache );
423 0 : ulong wmark = fd_dcache_compact_wmark ( mem, topo->links[ tile->out_link_id[ idx ] ].dcache, mtu );
424 0 : ulong depth = out_link->depth;
425 :
426 : /* This only works when there is a single consumer for the given
427 : output link. */
428 0 : ulong const * consumer_fseq = NULL;
429 0 : ulong consumer_cnt = 0UL;
430 :
431 0 : for( ulong tile_idx=0UL; tile_idx<topo->tile_cnt; tile_idx++ ) {
432 0 : fd_topo_tile_t const * consumer_tile = &topo->tiles[ tile_idx ];
433 0 : for( ulong i=0UL; i<consumer_tile->in_cnt; i++ ) {
434 0 : if( consumer_tile->in_link_id[ i ]==out_link->id ) {
435 0 : consumer_fseq = consumer_tile->in_link_fseq[ i ];
436 0 : consumer_cnt++;
437 0 : }
438 0 : }
439 0 : }
440 0 : FD_TEST( consumer_cnt==1UL );
441 0 : FD_TEST( !!consumer_fseq );
442 :
443 0 : return (fd_snapwm_out_link_t){ .idx = idx, .mem = mem, .chunk0 = chunk0, .wmark = wmark, .chunk = chunk0,
444 0 : .mtu = mtu, .depth = depth, .consumer_fseq = consumer_fseq };
445 0 : }
446 :
447 : FD_FN_UNUSED static void
448 : unprivileged_init( fd_topo_t * topo,
449 0 : fd_topo_tile_t * tile ) {
450 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
451 :
452 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
453 0 : fd_snapwm_tile_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_snapwm_tile_t), sizeof(fd_snapwm_tile_t) );
454 0 : void * _io_wd = FD_SCRATCH_ALLOC_APPEND( l, fd_vinyl_io_wd_align(), fd_vinyl_io_wd_footprint( tile->snapwm.snapwr_depth ) );
455 0 : void * _io_mm = FD_SCRATCH_ALLOC_APPEND( l, fd_vinyl_io_mm_align(), fd_vinyl_io_mm_footprint( FD_SNAPWM_IO_SPAD_MAX ) );
456 :
457 0 : ctx->full = 1;
458 0 : ctx->state = FD_SNAPSHOT_STATE_IDLE;
459 0 : ctx->lthash_disabled = tile->snapwm.lthash_disabled;
460 :
461 0 : ctx->boot_timestamp = fd_log_wallclock();
462 :
463 0 : fd_memset( &ctx->metrics, 0, sizeof(ctx->metrics) );
464 :
465 0 : if( FD_UNLIKELY( tile->kind_id ) ) FD_LOG_ERR(( "There can only be one `" NAME "` tile" ));
466 0 : if( FD_UNLIKELY( tile->in_cnt!=2UL ) ) FD_LOG_ERR(( "tile `" NAME "` has %lu ins, expected 2", tile->in_cnt ));
467 0 : if( FD_UNLIKELY( tile->out_cnt!=2UL ) ) FD_LOG_ERR(( "tile `" NAME "` has %lu outs, expected 2", tile->out_cnt ));
468 :
469 0 : ulong out_link_ct_idx = fd_topo_find_tile_out_link( topo, tile, "snapwm_ct", 0UL );
470 0 : if( out_link_ct_idx==ULONG_MAX ) out_link_ct_idx = fd_topo_find_tile_out_link( topo, tile, "snapwm_lv", 0UL );
471 0 : if( FD_UNLIKELY( out_link_ct_idx==ULONG_MAX ) ) FD_LOG_ERR(( "tile `" NAME "` missing required out link `snapwm_ct` or `snapwm_lv`" ));
472 0 : fd_topo_link_t * snapwm_out_link = &topo->links[ tile->out_link_id[ out_link_ct_idx ] ];
473 0 : ctx->out_ct_idx = out_link_ct_idx;
474 :
475 0 : if( 0==strcmp( snapwm_out_link->name, "snapwm_lv" ) ) {
476 0 : ctx->hash_out = out1( topo, tile, "snapwm_lv" );
477 0 : FD_TEST( ctx->hash_out.mtu==FD_SNAPWM_DUP_META_BATCH_SZ );
478 0 : }
479 :
480 0 : for( ulong i=0UL; i<tile->in_cnt; i++ ) {
481 0 : fd_topo_link_t const * in_link = &topo->links[ tile->in_link_id[ i ] ];
482 0 : if( 0==strcmp( in_link->name, "snapin_wm" ) ) {
483 0 : fd_topo_wksp_t const * in_wksp = &topo->workspaces[ topo->objs[ in_link->dcache_obj_id ].wksp_id ];
484 0 : ctx->in.wksp = in_wksp->wksp;
485 0 : ctx->in.chunk0 = fd_dcache_compact_chunk0( ctx->in.wksp, in_link->dcache );
486 0 : ctx->in.wmark = fd_dcache_compact_wmark( ctx->in.wksp, in_link->dcache, in_link->mtu );
487 0 : ctx->in.mtu = in_link->mtu;
488 0 : ctx->in.pos = 0UL;
489 0 : } else if( 0==strcmp( in_link->name, "snapin_txn" ) ) {
490 : /* snapwm needs all txn_cache data in order to verify the slot
491 : deltas with the slot history. To make this possible, snapin
492 : uses the dcache of the snapin_txn link as the scratch memory.
493 : The app field of the dcache is used to communicate the
494 : txncache_entries_len value. */
495 0 : fd_topo_wksp_t * in_wksp = &topo->workspaces[ topo->objs[ in_link->dcache_obj_id ].wksp_id ];
496 0 : ulong chunk0 = fd_dcache_compact_chunk0( in_wksp->wksp, in_link->dcache );
497 0 : ctx->txncache_entries = fd_chunk_to_laddr( in_wksp->wksp, chunk0 );
498 0 : ctx->txncache_entries_len_ptr = (ulong const *)fd_dcache_app_laddr_const( in_link->dcache );
499 0 : } else {
500 0 : FD_LOG_ERR(( "tile `" NAME "` unrecognized in link %s", in_link->name ));
501 0 : }
502 0 : }
503 0 : FD_TEST( !!ctx->in.wksp );
504 0 : FD_TEST( !!ctx->txncache_entries );
505 :
506 0 : fd_snapwm_vinyl_unprivileged_init( ctx, topo, tile, _io_mm, _io_wd );
507 0 : }
508 :
509 : /* Control fragments can result in one extra publish to forward the
510 : message down the pipeline, in addition to the result / malformed
511 : message. It can send one duplicate account message as well.
512 : When fd_snapwm_vinyl_txn_commit is invoked, the latter will handle
513 : fseq checks internally, since the amount of messages it needs to
514 : send far exceed the STEM_BURST. */
515 0 : #define STEM_BURST 3UL
516 :
517 0 : #define STEM_LAZY 1000L
518 :
519 0 : #define STEM_CALLBACK_CONTEXT_TYPE fd_snapwm_tile_t
520 0 : #define STEM_CALLBACK_CONTEXT_ALIGN alignof(fd_snapwm_tile_t)
521 :
522 : #define STEM_CALLBACK_SHOULD_SHUTDOWN should_shutdown
523 0 : #define STEM_CALLBACK_METRICS_WRITE metrics_write
524 0 : #define STEM_CALLBACK_RETURNABLE_FRAG returnable_frag
525 :
526 : #include "../../disco/stem/fd_stem.c"
527 :
528 : fd_topo_run_tile_t fd_tile_snapwm = {
529 : .name = NAME,
530 : .populate_allowed_fds = populate_allowed_fds,
531 : .populate_allowed_seccomp = populate_allowed_seccomp,
532 : .scratch_align = scratch_align,
533 : .scratch_footprint = scratch_footprint,
534 : .privileged_init = privileged_init,
535 : .unprivileged_init = unprivileged_init,
536 : .run = stem_run,
537 : };
538 :
539 : #undef NAME
|
