Line data Source code
1 : #include <stdio.h>
2 : #include <stdlib.h>
3 : #include <stddef.h>
4 : #include <alloca.h>
5 : #include <unistd.h>
6 : #include <sys/types.h>
7 : #include <sys/stat.h>
8 : #include <fcntl.h>
9 : #include <errno.h>
10 : #include <strings.h>
11 : #include "../../choreo/fd_choreo.h"
12 : #include "../../disco/fd_disco.h"
13 : #include "../../util/fd_util.h"
14 : #include "../../flamenco/fd_flamenco.h"
15 : #include "../../flamenco/nanopb/pb_decode.h"
16 : #include "../../flamenco/runtime/fd_hashes.h"
17 : #include "../../funk/fd_funk_filemap.h"
18 : #include "../../flamenco/types/fd_types.h"
19 : #include "../../flamenco/runtime/fd_runtime.h"
20 : #include "../../flamenco/runtime/fd_borrowed_account.h"
21 : #include "../../flamenco/runtime/fd_rocksdb.h"
22 : #include "../../flamenco/runtime/fd_txncache.h"
23 : #include "../../flamenco/rewards/fd_rewards.h"
24 : #include "../../ballet/base58/fd_base58.h"
25 : #include "../../flamenco/types/fd_solana_block.pb.h"
26 : #include "../../flamenco/runtime/context/fd_capture_ctx.h"
27 : #include "../../flamenco/runtime/fd_blockstore.h"
28 : #include "../../flamenco/runtime/program/fd_builtin_programs.h"
29 : #include "../../flamenco/shredcap/fd_shredcap.h"
30 : #include "../../flamenco/runtime/program/fd_bpf_program_util.h"
31 : #include "../../flamenco/snapshot/fd_snapshot.h"
32 : #include "../../flamenco/snapshot/fd_snapshot_create.h"
33 :
34 : struct fd_ledger_args {
35 : fd_wksp_t * wksp; /* wksp for blockstore */
36 : fd_wksp_t * funk_wksp; /* wksp for funk */
37 : fd_wksp_t * status_cache_wksp; /* wksp for status cache. */
38 : fd_blockstore_t blockstore_ljoin;
39 : fd_blockstore_t * blockstore; /* blockstore for replay */
40 : fd_funk_t * funk; /* handle to funk */
41 : fd_alloc_t * alloc; /* handle to alloc */
42 : char const * cmd; /* user passed command to fd_ledger */
43 : ulong start_slot; /* start slot for offline replay */
44 : ulong end_slot; /* end slot for offline replay */
45 : uint hashseed; /* hashseed */
46 : char const * checkpt; /* wksp checkpoint */
47 : char const * checkpt_funk; /* wksp checkpoint for a funk wksp */
48 : char const * checkpt_status_cache; /* status cache checkpoint */
49 : char const * restore; /* wksp restore */
50 : char const * restore_funk; /* wksp restore for a funk wksp */
51 : char const * allocator; /* allocator used during replay (libc/wksp) */
52 : ulong shred_max; /* maximum number of shreds*/
53 : ulong slot_history_max; /* number of slots stored by blockstore*/
54 : ulong txns_max; /* txns_max*/
55 : ulong index_max; /* size of funk index (same as rec max) */
56 : char const * funk_file; /* path to funk backing store */
57 : ulong funk_page_cnt;
58 : fd_funk_close_file_args_t funk_close_args;
59 : char const * snapshot; /* path to agave snapshot */
60 : char const * incremental; /* path to agave incremental snapshot */
61 : char const * genesis; /* path to agave genesis */
62 : char const * mini_db_dir; /* path to minifed rocksdb that's to be created */
63 : int copy_txn_status; /* determine if txns should be copied to the blockstore during minify/replay */
64 : int funk_only; /* determine if only funk should be ingested */
65 : char const * shredcap; /* path to replay using shredcap instead of rocksdb */
66 : int abort_on_mismatch; /* determine if execution should abort on mismatch*/
67 : char const * capture_fpath; /* solcap: path for solcap file to be created */
68 : int capture_txns; /* solcap: determine if transaction results should be captured for solcap*/
69 : char const * checkpt_path; /* path to dump funk wksp checkpoints during execution*/
70 : ulong checkpt_freq; /* how often funk wksp checkpoints will be dumped (defaults to never) */
71 : int checkpt_mismatch; /* determine if a funk wksp checkpoint should be dumped on a mismatch*/
72 :
73 : int dump_insn_to_pb; /* instruction dumping: should insns be dumped */
74 : int dump_txn_to_pb; /* txn dumping: should txns be dumped */
75 : int dump_block_to_pb; /* block dumping: should blocks be dumped */
76 : ulong dump_proto_start_slot; /* instruction / txn dumping: what slot to start dumping*/
77 : char const * dump_proto_sig_filter; /* instruction / txn dumping: specify txn sig to dump at */
78 : char const * dump_proto_output_dir; /* instruction / txn dumping: output directory for protobuf messages */
79 :
80 : int verify_funk; /* verify funk before execution starts */
81 : uint verify_acc_hash; /* verify account hash from the snapshot */
82 : uint check_acc_hash; /* check account hash by reconstructing with data */
83 : ulong trash_hash; /* trash hash to be used for negative cases*/
84 : ulong vote_acct_max; /* max number of vote accounts */
85 : char const * rocksdb_list[32]; /* max number of rocksdb dirs that can be passed in */
86 : ulong rocksdb_list_slot[32]; /* start slot for each rocksdb dir that's passed in assuming there are mulitple */
87 : ulong rocksdb_list_cnt; /* number of rocksdb dirs passed in */
88 : char * rocksdb_list_strdup;
89 : uint cluster_version[3]; /* What version of solana is the genesis block? */
90 : char const * one_off_features[32]; /* List of one off feature pubkeys to enable for execution agnostic of cluster version */
91 : uint one_off_features_cnt; /* Number of one off features */
92 : char * one_off_features_strdup;
93 : ulong snapshot_freq; /* How often a snapshot should be produced */
94 : ulong incremental_freq; /* How often an incremental snapshot should be produced */
95 : char const * snapshot_dir; /* Directory to create a snapshot in */
96 : ulong snapshot_tcnt; /* Number of threads to use for snapshot creation */
97 : double allowed_mem_delta; /* Percent of memory in the blockstore wksp that can be
98 : used and not freed between the start of end of execution.
99 : If the difference in usage exceeds this value, error out. */
100 :
101 : /* These values are setup and maintained before replay */
102 : fd_capture_ctx_t * capture_ctx; /* capture_ctx is used in runtime_replay for various debugging tasks */
103 : fd_exec_slot_ctx_t * slot_ctx; /* slot_ctx */
104 : fd_exec_epoch_ctx_t * epoch_ctx; /* epoch_ctx */
105 : fd_tpool_t * tpool; /* thread pool for execution */
106 : uchar tpool_mem[FD_TPOOL_FOOTPRINT( FD_TILE_MAX )] __attribute__( ( aligned( FD_TPOOL_ALIGN ) ) );
107 :
108 : fd_spad_t * exec_spads[ 128UL ]; /* bump allocators that are eventually assigned to each txn_ctx */
109 : ulong exec_spad_cnt; /* number of bump allocators, bounded by number of threads */
110 : fd_spad_t * runtime_spad; /* bump allocator used for runtime scoped allocations */
111 : fd_tpool_t * snapshot_tpool; /* thread pool for snapshot creation */
112 : uchar tpool_mem_snapshot[FD_TPOOL_FOOTPRINT( FD_TILE_MAX )] __attribute__( ( aligned( FD_TPOOL_ALIGN ) ) );
113 : fd_tpool_t * snapshot_bg_tpool; /* thread pool for snapshot creation */
114 : uchar tpool_mem_snapshot_bg[FD_TPOOL_FOOTPRINT( FD_TILE_MAX )] __attribute__( ( aligned( FD_TPOOL_ALIGN ) ) );
115 : ulong last_snapshot_slot; /* last snapshot slot */
116 : fd_hash_t last_snapshot_hash; /* last snapshot hash */
117 : ulong last_snapshot_cap; /* last snapshot account capitalization */
118 : int is_snapshotting; /* determine if a snapshot is being created */
119 : int snapshot_mismatch; /* determine if a snapshot should be created on a mismatch */
120 : ulong thread_mem_bound; /* how much spad is allocated by a tpool thread. The default
121 : value is the full runtime bound. If a value of 0 is passed
122 : in, then a reduced bound will be used. */
123 : ulong runtime_mem_bound; /* how much to allocate for a runtime-scoped spad */
124 :
125 : fd_valloc_t valloc; /* wksp valloc that should NOT be used for runtime allocations */
126 :
127 : char const * lthash;
128 : };
129 : typedef struct fd_ledger_args fd_ledger_args_t;
130 :
131 : /* Allocations ****************************************************************/
132 :
133 : static void
134 0 : init_spads( fd_ledger_args_t * args, int has_tpool ) {
135 :
136 0 : FD_LOG_NOTICE(( "setting up exec_spads" ));
137 :
138 : /* Allocate memory for the account mem space. In live execution, each of
139 : the spad allocations should be tied to its respective execution thread.
140 : In the future, the spad should be allocated from its tiles' workspace.
141 : It is important that the exec_spads are only allocated on startup for
142 : performance reasons to avoid dynamic allocation in the critical path. */
143 :
144 0 : if( has_tpool ) {
145 0 : args->exec_spad_cnt = fd_tpool_worker_cnt( args->tpool );
146 0 : for( ulong i=0UL; i<fd_tpool_worker_cnt( args->tpool ); i++ ) {
147 0 : ulong total_mem_sz = args->thread_mem_bound;
148 0 : uchar * mem = fd_wksp_alloc_laddr( args->wksp, FD_SPAD_ALIGN, FD_SPAD_FOOTPRINT( total_mem_sz ), 999UL );
149 0 : fd_spad_t * spad = fd_spad_join( fd_spad_new( mem, total_mem_sz ) );
150 0 : if( FD_UNLIKELY( !spad ) ) {
151 0 : FD_LOG_ERR(( "failed to allocate spad" ));
152 0 : }
153 0 : args->exec_spads[ i ] = spad;
154 0 : }
155 0 : }
156 :
157 : /* Similiarly, we need an allocator that is scoped to the runtime and not
158 : transaction execution. This is responsible for managing the memory at the
159 : epoch boundary and at the start/end of a slot's execution. */
160 :
161 : /* FIXME: This bound should be calculated based on various data structures
162 : that exist at the slot_ctx/epoch_ctx. It should encapsulate all allocations
163 : that happen outside of transaction execution. */
164 :
165 0 : uchar * mem = fd_wksp_alloc_laddr( args->wksp, FD_SPAD_ALIGN, FD_SPAD_FOOTPRINT( args->runtime_mem_bound ), 999UL );
166 0 : fd_spad_t * spad = fd_spad_join( fd_spad_new( mem, args->runtime_mem_bound ) );
167 0 : if( FD_UNLIKELY( !spad ) ) {
168 0 : FD_LOG_ERR(( "Failed to allocate runtime spad" ));
169 0 : }
170 0 : args->runtime_spad = spad;
171 :
172 0 : }
173 :
174 : /* Snapshot *******************************************************************/
175 :
176 : static void
177 : fd_create_snapshot_task( void FD_PARAM_UNUSED *tpool,
178 : ulong t0, ulong t1,
179 : void *args FD_PARAM_UNUSED,
180 : void *reduce FD_PARAM_UNUSED, ulong stride FD_PARAM_UNUSED,
181 : ulong l0 FD_PARAM_UNUSED, ulong l1 FD_PARAM_UNUSED,
182 : ulong m0 FD_PARAM_UNUSED, ulong m1 FD_PARAM_UNUSED,
183 0 : ulong n0 FD_PARAM_UNUSED, ulong n1 FD_PARAM_UNUSED ) {
184 :
185 0 : fd_snapshot_ctx_t * snapshot_ctx = (fd_snapshot_ctx_t *)t0;
186 0 : fd_ledger_args_t * ledger_args = (fd_ledger_args_t *)t1;
187 :
188 0 : char tmp_dir_buf[ FD_SNAPSHOT_DIR_MAX ];
189 0 : int err = snprintf( tmp_dir_buf, FD_SNAPSHOT_DIR_MAX, "%s/%s",
190 0 : snapshot_ctx->out_dir,
191 0 : snapshot_ctx->is_incremental ? FD_SNAPSHOT_TMP_INCR_ARCHIVE : FD_SNAPSHOT_TMP_ARCHIVE );
192 0 : if( FD_UNLIKELY( err<0 ) ) {
193 0 : FD_LOG_WARNING(( "Failed to format directory string" ));
194 0 : return;
195 0 : }
196 :
197 0 : char zstd_dir_buf[ FD_SNAPSHOT_DIR_MAX ];
198 0 : err = snprintf( zstd_dir_buf, FD_SNAPSHOT_DIR_MAX, "%s/%s",
199 0 : snapshot_ctx->out_dir,
200 0 : snapshot_ctx->is_incremental ? FD_SNAPSHOT_TMP_INCR_ARCHIVE_ZSTD : FD_SNAPSHOT_TMP_FULL_ARCHIVE_ZSTD );
201 0 : if( FD_UNLIKELY( err<0 ) ) {
202 0 : FD_LOG_WARNING(( "Failed to format directory string" ));
203 0 : return;
204 0 : }
205 :
206 : /* Create and open the relevant files for snapshots. */
207 :
208 0 : snapshot_ctx->tmp_fd = open( tmp_dir_buf, O_CREAT | O_RDWR | O_TRUNC, 0644 );
209 0 : if( FD_UNLIKELY( snapshot_ctx->tmp_fd==-1 ) ) {
210 0 : FD_LOG_WARNING(( "Failed to open and create tarball for file=%s (%i-%s)", tmp_dir_buf, errno, fd_io_strerror( errno ) ));
211 0 : return;
212 0 : }
213 :
214 0 : snapshot_ctx->snapshot_fd = open( zstd_dir_buf, O_RDWR | O_CREAT | O_TRUNC, 0644 );
215 0 : if( FD_UNLIKELY( snapshot_ctx->snapshot_fd==-1 ) ) {
216 0 : FD_LOG_WARNING(( "Failed to open the snapshot file (%i-%s)", errno, fd_io_strerror( errno ) ));
217 0 : return;
218 0 : }
219 :
220 0 : FD_LOG_WARNING(( "Starting snapshot creation at slot=%lu", snapshot_ctx->slot ));
221 :
222 0 : fd_snapshot_create_new_snapshot( snapshot_ctx,
223 0 : &ledger_args->last_snapshot_hash,
224 0 : &ledger_args->last_snapshot_cap );
225 :
226 0 : FD_LOG_NOTICE(( "Successfully produced a snapshot at directory=%s", ledger_args->snapshot_dir ));
227 :
228 0 : ledger_args->slot_ctx->epoch_ctx->constipate_root = 0;
229 0 : ledger_args->is_snapshotting = 0;
230 :
231 0 : err = close( snapshot_ctx->tmp_fd );
232 0 : if( FD_UNLIKELY( err ) ) {
233 0 : FD_LOG_ERR(( "failed to close tmp_fd" ));
234 0 : }
235 :
236 0 : err = close( snapshot_ctx->snapshot_fd );
237 0 : if( FD_UNLIKELY( err ) ) {
238 0 : FD_LOG_ERR(( "failed to close snapshot_fd" ));
239 0 : }
240 :
241 0 : }
242 :
243 : /* Runtime Replay *************************************************************/
244 : static int
245 0 : init_tpool( fd_ledger_args_t * ledger_args ) {
246 :
247 0 : ulong snapshot_tcnt = ledger_args->snapshot_tcnt;
248 :
249 0 : ulong tcnt = fd_tile_cnt() - snapshot_tcnt;
250 0 : fd_tpool_t * tpool = NULL;
251 :
252 0 : ulong start_idx = 1UL;
253 0 : if( tcnt>=1UL ) {
254 0 : tpool = fd_tpool_init( ledger_args->tpool_mem, tcnt );
255 0 : if( tpool == NULL ) {
256 0 : FD_LOG_ERR(( "failed to create thread pool" ));
257 0 : }
258 0 : for( ulong i=1UL; i<tcnt; ++i ) {
259 0 : if( fd_tpool_worker_push( tpool, start_idx++, NULL, 0UL ) == NULL ) {
260 0 : FD_LOG_ERR(( "failed to launch worker" ));
261 0 : }
262 0 : else {
263 0 : FD_LOG_INFO(( "launched worker %lu", start_idx - 1UL ));
264 0 : }
265 0 : }
266 0 : }
267 :
268 0 : ledger_args->tpool = tpool;
269 :
270 : /* Setup a background thread for the snapshot service as well as a tpool
271 : used for snapshot hashing. */
272 :
273 0 : if( !snapshot_tcnt ) {
274 0 : FD_LOG_NOTICE(( "No snapshot threads requested" ));
275 0 : return 0;
276 0 : }
277 :
278 0 : else if( snapshot_tcnt==1UL ) {
279 0 : FD_LOG_ERR(( "This is an invalid value for the number of threads to use for snapshot creation" ));
280 0 : }
281 :
282 0 : fd_tpool_t * snapshot_bg_tpool = fd_tpool_init( ledger_args->tpool_mem_snapshot_bg, snapshot_tcnt );
283 0 : if( FD_UNLIKELY( !fd_tpool_worker_push( snapshot_bg_tpool, start_idx++, NULL, 0UL ) ) ) {
284 0 : FD_LOG_ERR(( "failed to launch worker" ));
285 0 : } else {
286 0 : FD_LOG_NOTICE(( "launched snapshot bg worker %lu", start_idx - 1UL ));
287 0 : }
288 :
289 0 : ledger_args->snapshot_bg_tpool = snapshot_bg_tpool;
290 :
291 :
292 0 : if( snapshot_tcnt==2UL ) {
293 0 : return 0;
294 0 : }
295 :
296 : /* If a snapshot is being created, setup its own tpool. */
297 :
298 0 : fd_tpool_t * snapshot_tpool = fd_tpool_init( ledger_args->tpool_mem_snapshot, snapshot_tcnt - 1UL );
299 0 : for( ulong i=1UL; i<snapshot_tcnt - 1UL; ++i ) {
300 0 : if( FD_UNLIKELY( !fd_tpool_worker_push( snapshot_tpool, start_idx++, NULL, 0UL ) ) ) {
301 0 : FD_LOG_ERR(( "failed to launch worker" ));
302 0 : } else {
303 0 : FD_LOG_NOTICE(( "launched snapshot hash worker %lu", start_idx - 1UL ));
304 0 : }
305 0 : }
306 :
307 0 : ledger_args->snapshot_tpool = snapshot_tpool;
308 :
309 0 : return 0;
310 0 : }
311 :
312 : void
313 0 : args_cleanup( fd_ledger_args_t * ledger_args ) {
314 0 : if( ledger_args->rocksdb_list_strdup ) free( ledger_args->rocksdb_list_strdup );
315 0 : if( ledger_args->one_off_features_strdup ) free( ledger_args->one_off_features_strdup );
316 0 : }
317 :
318 : int
319 0 : runtime_replay( fd_ledger_args_t * ledger_args ) {
320 0 : int ret = 0;
321 :
322 0 : fd_features_restore( ledger_args->slot_ctx, ledger_args->runtime_spad );
323 :
324 0 : fd_runtime_update_leaders( ledger_args->slot_ctx, ledger_args->slot_ctx->slot_bank.slot, ledger_args->runtime_spad );
325 :
326 0 : fd_calculate_epoch_accounts_hash_values( ledger_args->slot_ctx );
327 :
328 0 : long replay_time = -fd_log_wallclock();
329 0 : ulong txn_cnt = 0;
330 0 : ulong slot_cnt = 0;
331 0 : fd_blockstore_t * blockstore = ledger_args->slot_ctx->blockstore;
332 :
333 0 : ulong prev_slot = ledger_args->slot_ctx->slot_bank.slot;
334 0 : ulong start_slot = ledger_args->slot_ctx->slot_bank.slot + 1;
335 :
336 0 : ledger_args->slot_ctx->root_slot = prev_slot;
337 :
338 : /* On demand rocksdb ingest */
339 0 : fd_rocksdb_t rocks_db = {0};
340 0 : fd_rocksdb_root_iter_t iter = {0};
341 0 : fd_slot_meta_t slot_meta = {0};
342 0 : ulong curr_rocksdb_idx = 0UL;
343 :
344 0 : char * err = fd_rocksdb_init( &rocks_db, ledger_args->rocksdb_list[ 0UL ] );
345 0 : if( FD_UNLIKELY( err!=NULL ) ) {
346 0 : FD_LOG_ERR(( "fd_rocksdb_init at path=%s returned error=%s", ledger_args->rocksdb_list[ 0UL ], err ));
347 0 : }
348 0 : fd_rocksdb_root_iter_new( &iter );
349 :
350 0 : int block_found = -1;
351 0 : while ( block_found!=0 && start_slot<=ledger_args->end_slot ) {
352 0 : block_found = fd_rocksdb_root_iter_seek( &iter, &rocks_db, start_slot, &slot_meta, ledger_args->valloc );
353 0 : if ( block_found!=0 ) {
354 0 : start_slot++;
355 0 : }
356 0 : }
357 :
358 0 : if( FD_UNLIKELY( block_found!=0 ) ) {
359 0 : FD_LOG_ERR(( "unable to seek to any slot" ));
360 0 : }
361 :
362 : /* Setup trash_hash */
363 0 : uchar trash_hash_buf[32];
364 0 : memset( trash_hash_buf, 0xFE, sizeof(trash_hash_buf) );
365 :
366 0 : ledger_args->is_snapshotting = 0;
367 :
368 : /* Calculate and store wksp free size before execution. */
369 0 : fd_wksp_usage_t init_usage = {0};
370 0 : fd_wksp_usage( fd_blockstore_wksp( ledger_args->blockstore ), NULL, 0UL, &init_usage );
371 :
372 0 : ulong block_slot = start_slot;
373 0 : uchar aborted = 0U;
374 0 : for( ulong slot = start_slot; slot<=ledger_args->end_slot && !aborted; ++slot ) {
375 :
376 0 : ledger_args->slot_ctx->slot_bank.prev_slot = prev_slot;
377 0 : ledger_args->slot_ctx->slot_bank.slot = slot;
378 :
379 0 : FD_LOG_DEBUG(( "reading slot %lu", slot ));
380 :
381 : /* If we have reached a new block, load one in from rocksdb to the blockstore */
382 0 : bool block_exists = fd_blockstore_shreds_complete( blockstore, slot);
383 0 : if( !block_exists && slot_meta.slot == slot ) {
384 0 : int err = fd_rocksdb_import_block_blockstore( &rocks_db,
385 0 : &slot_meta, blockstore,
386 0 : ledger_args->copy_txn_status,
387 0 : slot == (ledger_args->trash_hash) ? trash_hash_buf : NULL,
388 0 : ledger_args->valloc );
389 0 : if( FD_UNLIKELY( err ) ) {
390 0 : FD_LOG_ERR(( "Failed to import block %lu", start_slot ));
391 0 : }
392 :
393 : /* Remove the previous block from the blockstore */
394 0 : if( FD_LIKELY( block_slot < slot ) ) {
395 : /* Mark the block as successfully processed */
396 :
397 0 : fd_block_map_query_t query[1] = {0};
398 0 : int err = fd_block_map_prepare( blockstore->block_map, &block_slot, NULL, query, FD_MAP_FLAG_BLOCKING );
399 0 : fd_block_info_t * block_info = fd_block_map_query_ele( query );
400 :
401 0 : if( FD_UNLIKELY( err || block_info->slot != block_slot ) ) FD_LOG_ERR(( "failed to prepare block map query" ));
402 :
403 0 : block_info->flags = fd_uchar_clear_bit( block_info->flags, FD_BLOCK_FLAG_REPLAYING );
404 0 : block_info->flags = fd_uchar_set_bit( block_info->flags, FD_BLOCK_FLAG_PROCESSED );
405 :
406 0 : fd_block_map_publish( query );
407 :
408 : /* Remove the old block from the blockstore */
409 : /*for( uint idx = 0; idx <= slot_complete_idx; idx++ ) {
410 : fd_blockstore_shred_remove( blockstore, block_slot, idx );
411 : }*/
412 0 : fd_blockstore_block_allocs_remove( blockstore, block_slot );
413 0 : fd_blockstore_slot_remove( blockstore, block_slot );
414 0 : }
415 : /* Mark the new block as replaying */
416 0 : fd_block_map_query_t query[1] = {0};
417 0 : err = fd_block_map_prepare( blockstore->block_map, &slot, NULL, query, FD_MAP_FLAG_BLOCKING );
418 0 : fd_block_info_t * block_info = fd_block_map_query_ele( query );
419 0 : if( FD_UNLIKELY( err || block_info->slot != slot ) ) FD_LOG_ERR(( "failed to prepare block map query" ));
420 0 : block_info->flags = fd_uchar_set_bit( block_info->flags, FD_BLOCK_FLAG_REPLAYING );
421 0 : fd_block_map_publish( query );
422 :
423 0 : block_slot = slot;
424 0 : }
425 :
426 0 : fd_block_t * blk = fd_blockstore_block_query( blockstore, slot );
427 0 : if( blk == NULL ) {
428 0 : FD_LOG_WARNING(( "failed to read slot %lu", slot ));
429 0 : continue;
430 0 : }
431 :
432 0 : ledger_args->slot_ctx->slot_bank.tick_height = ledger_args->slot_ctx->slot_bank.max_tick_height;
433 0 : if( FD_UNLIKELY( FD_RUNTIME_EXECUTE_SUCCESS != fd_runtime_compute_max_tick_height( ledger_args->epoch_ctx->epoch_bank.ticks_per_slot, slot, &ledger_args->slot_ctx->slot_bank.max_tick_height ) ) ) {
434 0 : FD_LOG_ERR(( "couldn't compute max tick height slot %lu ticks_per_slot %lu", slot, ledger_args->epoch_ctx->epoch_bank.ticks_per_slot ));
435 0 : }
436 :
437 0 : if( ledger_args->slot_ctx->root_slot%ledger_args->snapshot_freq==0UL && !ledger_args->is_snapshotting ) {
438 :
439 0 : ledger_args->is_snapshotting = 1;
440 :
441 0 : ledger_args->last_snapshot_slot = ledger_args->slot_ctx->root_slot;
442 :
443 0 : fd_snapshot_ctx_t snapshot_ctx = {
444 0 : .slot = ledger_args->slot_ctx->root_slot,
445 0 : .out_dir = ledger_args->snapshot_dir,
446 0 : .is_incremental = 0,
447 0 : .funk = ledger_args->slot_ctx->acc_mgr->funk,
448 0 : .status_cache = ledger_args->slot_ctx->status_cache,
449 0 : .tpool = ledger_args->snapshot_tpool,
450 0 : .spad = ledger_args->runtime_spad,
451 0 : .features = &ledger_args->slot_ctx->epoch_ctx->features
452 0 : };
453 :
454 0 : fd_tpool_exec( ledger_args->snapshot_bg_tpool, 1UL, fd_create_snapshot_task, NULL,
455 0 : (ulong)&snapshot_ctx, (ulong)ledger_args, 0UL, NULL,
456 0 : 0UL, 0UL, 0UL, 0UL, 0UL, 0UL, 0UL );
457 :
458 0 : } else if( ledger_args->slot_ctx->root_slot%ledger_args->incremental_freq==0UL && !ledger_args->is_snapshotting && ledger_args->last_snapshot_slot ) {
459 :
460 0 : ledger_args->is_snapshotting = 1;
461 :
462 0 : fd_snapshot_ctx_t snapshot_ctx = {
463 0 : .features = &ledger_args->slot_ctx->epoch_ctx->features,
464 0 : .slot = ledger_args->slot_ctx->root_slot,
465 0 : .out_dir = ledger_args->snapshot_dir,
466 0 : .is_incremental = 1,
467 0 : .spad = ledger_args->runtime_spad,
468 0 : .funk = ledger_args->slot_ctx->acc_mgr->funk,
469 0 : .status_cache = ledger_args->slot_ctx->status_cache,
470 0 : .last_snap_slot = ledger_args->last_snapshot_slot,
471 0 : .tpool = ledger_args->snapshot_tpool,
472 0 : .last_snap_acc_hash = &ledger_args->last_snapshot_hash,
473 0 : .last_snap_capitalization = ledger_args->last_snapshot_cap
474 0 : };
475 :
476 0 : fd_tpool_exec( ledger_args->snapshot_bg_tpool, 1UL, fd_create_snapshot_task, NULL,
477 0 : (ulong)&snapshot_ctx, (ulong)ledger_args, 0UL, NULL,
478 0 : 0UL, 0UL, 0UL, 0UL, 0UL, 0UL, 0UL );
479 0 : }
480 :
481 0 : ulong blk_txn_cnt = 0UL;
482 0 : FD_LOG_NOTICE(( "Used memory in spad before slot=%lu %lu", ledger_args->slot_ctx->slot_bank.slot, ledger_args->runtime_spad->mem_used ));
483 0 : FD_TEST( fd_runtime_block_eval_tpool( ledger_args->slot_ctx,
484 0 : blk,
485 0 : ledger_args->capture_ctx,
486 0 : ledger_args->tpool,
487 0 : 1,
488 0 : &blk_txn_cnt,
489 0 : ledger_args->exec_spads,
490 0 : ledger_args->exec_spad_cnt,
491 0 : ledger_args->runtime_spad ) == FD_RUNTIME_EXECUTE_SUCCESS );
492 0 : txn_cnt += blk_txn_cnt;
493 0 : slot_cnt++;
494 :
495 0 : fd_hash_t expected;
496 0 : int err = fd_blockstore_block_hash_query( blockstore, slot, &expected );
497 0 : if( FD_UNLIKELY( err ) ) FD_LOG_ERR( ( "slot %lu is missing its hash", slot ) );
498 0 : else if( FD_UNLIKELY( 0 != memcmp( ledger_args->slot_ctx->slot_bank.poh.hash, expected.hash, 32UL ) ) ) {
499 0 : char expected_hash[ FD_BASE58_ENCODED_32_SZ ];
500 0 : fd_acct_addr_cstr( expected_hash, expected.hash );
501 0 : char poh_hash[ FD_BASE58_ENCODED_32_SZ ];
502 0 : fd_acct_addr_cstr( poh_hash, ledger_args->slot_ctx->slot_bank.poh.hash );
503 0 : FD_LOG_WARNING(( "PoH hash mismatch! slot=%lu expected=%s, got=%s",
504 0 : slot,
505 0 : expected_hash,
506 0 : poh_hash ));
507 :
508 0 : if( ledger_args->checkpt_mismatch ) {
509 0 : fd_runtime_checkpt( ledger_args->capture_ctx, ledger_args->slot_ctx, ULONG_MAX );
510 0 : }
511 0 : if( ledger_args->snapshot_mismatch ) {
512 0 : fd_snapshot_ctx_t snapshot_ctx = {
513 0 : .slot = ledger_args->slot_ctx->root_slot,
514 0 : .out_dir = ledger_args->snapshot_dir,
515 0 : .is_incremental = 0,
516 0 : .spad = ledger_args->runtime_spad,
517 0 : .funk = ledger_args->slot_ctx->acc_mgr->funk,
518 0 : .status_cache = ledger_args->slot_ctx->status_cache,
519 0 : .tpool = ledger_args->snapshot_tpool
520 0 : };
521 0 : fd_create_snapshot_task( NULL, (ulong)&snapshot_ctx, (ulong)ledger_args, NULL, NULL, 0UL, 0UL, 0UL, 0UL, 0UL, 0UL, 0UL );
522 0 : }
523 0 : if( ledger_args->abort_on_mismatch ) {
524 0 : ret = 1;
525 0 : aborted = 1U;
526 0 : break;
527 0 : }
528 0 : }
529 :
530 0 : err = fd_blockstore_bank_hash_query( blockstore, slot, &expected );
531 0 : if( FD_UNLIKELY( err) ) {
532 0 : FD_LOG_ERR(( "slot %lu is missing its bank hash", slot ));
533 0 : } else if( FD_UNLIKELY( 0 != memcmp( ledger_args->slot_ctx->slot_bank.banks_hash.hash,
534 0 : expected.hash,
535 0 : 32UL ) ) ) {
536 :
537 0 : char expected_hash[ FD_BASE58_ENCODED_32_SZ ];
538 0 : fd_acct_addr_cstr( expected_hash, expected.hash );
539 0 : char bank_hash[ FD_BASE58_ENCODED_32_SZ ];
540 0 : fd_acct_addr_cstr( bank_hash, ledger_args->slot_ctx->slot_bank.banks_hash.hash );
541 :
542 0 : FD_LOG_WARNING(( "Bank hash mismatch! slot=%lu expected=%s, got=%s",
543 0 : slot,
544 0 : expected_hash,
545 0 : bank_hash ));
546 :
547 0 : if( ledger_args->checkpt_mismatch ) {
548 0 : fd_runtime_checkpt( ledger_args->capture_ctx, ledger_args->slot_ctx, ULONG_MAX );
549 0 : }
550 0 : if( ledger_args->snapshot_mismatch ) {
551 0 : fd_snapshot_ctx_t snapshot_ctx = {
552 0 : .slot = ledger_args->slot_ctx->root_slot,
553 0 : .out_dir = ledger_args->snapshot_dir,
554 0 : .is_incremental = 0,
555 0 : .spad = ledger_args->runtime_spad,
556 0 : .funk = ledger_args->slot_ctx->acc_mgr->funk,
557 0 : .status_cache = ledger_args->slot_ctx->status_cache,
558 0 : .tpool = ledger_args->snapshot_tpool
559 0 : };
560 0 : fd_create_snapshot_task( NULL, (ulong)&snapshot_ctx, (ulong)ledger_args, NULL, NULL, 0UL, 0UL, 0UL, 0UL, 0UL, 0UL, 0UL );
561 0 : }
562 0 : if( ledger_args->abort_on_mismatch ) {
563 0 : ret = 1;
564 0 : break;
565 0 : }
566 0 : }
567 :
568 0 : prev_slot = slot;
569 :
570 0 : if( slot<ledger_args->end_slot ) {
571 : /* TODO: This currently doesn't support switching over on slots that occur on a fork */
572 : /* If need to go to next rocksdb, switch over */
573 0 : if( FD_UNLIKELY( ledger_args->rocksdb_list_cnt>1UL &&
574 0 : slot+1UL==ledger_args->rocksdb_list_slot[curr_rocksdb_idx] ) ) {
575 0 : curr_rocksdb_idx++;
576 0 : FD_LOG_WARNING(( "Switching to next rocksdb=%s", ledger_args->rocksdb_list[curr_rocksdb_idx] ));
577 0 : fd_rocksdb_root_iter_destroy( &iter );
578 0 : fd_rocksdb_destroy( &rocks_db );
579 :
580 0 : fd_memset( &rocks_db, 0, sizeof(fd_rocksdb_t) );
581 0 : fd_memset( &iter, 0, sizeof(fd_rocksdb_root_iter_t) );
582 0 : fd_memset( &slot_meta, 0, sizeof(fd_slot_meta_t) );
583 :
584 0 : char * err = fd_rocksdb_init( &rocks_db, ledger_args->rocksdb_list[curr_rocksdb_idx] );
585 0 : if( FD_UNLIKELY( err!=NULL ) ) {
586 0 : FD_LOG_ERR(( "fd_rocksdb_init at path=%s returned error=%s", ledger_args->rocksdb_list[curr_rocksdb_idx], err ));
587 0 : }
588 0 : fd_rocksdb_root_iter_new( &iter );
589 0 : int ret = fd_rocksdb_root_iter_seek( &iter, &rocks_db, slot+1UL, &slot_meta, ledger_args->valloc );
590 0 : if( ret<0 ) {
591 0 : FD_LOG_ERR(( "Failed to seek to slot %lu", slot+1UL ));
592 0 : }
593 0 : } else {
594 : /* Otherwise look for next slot in current rocksdb */
595 0 : int ret = fd_rocksdb_root_iter_next( &iter, &slot_meta, ledger_args->valloc );
596 0 : if( ret<0 ) {
597 0 : ret = fd_rocksdb_get_meta( &rocks_db, slot+1UL, &slot_meta, ledger_args->valloc );
598 0 : if( ret<0 ) {
599 0 : FD_LOG_ERR(( "Failed to get meta for slot %lu", slot+1UL ));
600 0 : }
601 0 : }
602 0 : }
603 0 : }
604 0 : }
605 :
606 : /* Throw an error if the blockstore wksp has a usage which exceeds the allowed
607 : threshold. This likely indicates that a memory leak was introduced. */
608 :
609 0 : fd_wksp_usage_t final_usage = {0};
610 0 : fd_wksp_usage( fd_blockstore_wksp( ledger_args->blockstore ), NULL, 0UL, &final_usage );
611 :
612 0 : ulong mem_delta = fd_ulong_sat_sub( init_usage.free_sz, final_usage.free_sz );
613 0 : double pcnt_mem_delta = (double)mem_delta / (double)init_usage.free_sz;
614 0 : if( pcnt_mem_delta > ledger_args->allowed_mem_delta ) {
615 0 : FD_LOG_ERR(( "Memory usage delta (%4f%%) exceeded allowed limit (%4f%%)", 100UL * pcnt_mem_delta, 100UL * ledger_args->allowed_mem_delta ));
616 0 : } else {
617 0 : FD_LOG_NOTICE(( "Memory usage delta (%4f%%) within allowed limit (%4f%%)", 100UL * pcnt_mem_delta, 100UL * ledger_args->allowed_mem_delta ));
618 0 : }
619 :
620 : #if FD_SPAD_TRACK_USAGE
621 : for( ulong i=0UL; i<ledger_args->exec_spad_cnt; i++ ) {
622 : fd_spad_t * spad = ledger_args->exec_spads[ i ];
623 : double pcnt_mem_wmark = (double)fd_spad_mem_wmark( spad ) / (double)fd_spad_mem_max( spad );
624 : pcnt_mem_wmark *= 100;
625 : FD_LOG_NOTICE(( "spad %2lu mem_wmark %10lu (%6.2f%%) mem_max %10lu", i, fd_spad_mem_wmark( spad ), pcnt_mem_wmark, fd_spad_mem_max( spad ) ));
626 : }
627 : #endif
628 :
629 0 : if( ledger_args->tpool ) {
630 0 : fd_tpool_fini( ledger_args->tpool );
631 0 : }
632 :
633 0 : fd_rocksdb_root_iter_destroy( &iter );
634 0 : fd_rocksdb_destroy( &rocks_db );
635 :
636 0 : replay_time += fd_log_wallclock();
637 0 : double replay_time_s = (double)replay_time * 1e-9;
638 0 : double tps = (double)txn_cnt / replay_time_s;
639 0 : double sec_per_slot = replay_time_s / (double)slot_cnt;
640 0 : FD_LOG_NOTICE((
641 0 : "replay completed - slots: %lu, elapsed: %6.6f s, txns: %lu, tps: %6.6f, sec/slot: %6.6f",
642 0 : slot_cnt,
643 0 : replay_time_s,
644 0 : txn_cnt,
645 0 : tps,
646 0 : sec_per_slot ));
647 :
648 0 : if ( slot_cnt == 0 ) {
649 0 : FD_LOG_ERR(( "No slots replayed" ));
650 0 : }
651 :
652 0 : args_cleanup( ledger_args );
653 :
654 0 : return ret;
655 0 : }
656 :
657 : /***************************** Helpers ****************************************/
658 : static fd_valloc_t
659 0 : allocator_setup( fd_wksp_t * wksp ) {
660 :
661 0 : if( FD_UNLIKELY( !wksp ) ) {
662 0 : FD_LOG_ERR(( "workspace is NULL" ));
663 0 : }
664 :
665 0 : void * alloc_shmem = fd_wksp_alloc_laddr( wksp, fd_alloc_align(), fd_alloc_footprint(), 3UL );
666 0 : if( FD_UNLIKELY( !alloc_shmem ) ) { FD_LOG_ERR( ( "fd_alloc too large for workspace" ) ); }
667 0 : void * alloc_shalloc = fd_alloc_new( alloc_shmem, 3UL );
668 0 : if( FD_UNLIKELY( !alloc_shalloc ) ) { FD_LOG_ERR( ( "fd_alloc_new failed" ) ); }
669 0 : fd_alloc_t * alloc = fd_alloc_join( alloc_shalloc, 3UL );
670 0 : if( FD_UNLIKELY( !alloc ) ) { FD_LOG_ERR( ( "fd_alloc_join failed" ) ); }
671 0 : fd_valloc_t valloc = fd_alloc_virtual( alloc );
672 0 : return valloc;
673 :
674 : /* NOTE: Enable this if leak hunting */
675 : //return fd_backtracing_alloc_virtual( &valloc );
676 :
677 0 : }
678 :
679 : void
680 0 : fd_ledger_main_setup( fd_ledger_args_t * args ) {
681 0 : fd_flamenco_boot( NULL, NULL );
682 0 : fd_funk_t * funk = args->funk;
683 :
684 : /* Setup capture context */
685 0 : int has_solcap = args->capture_fpath && args->capture_fpath[0] != '\0';
686 0 : int has_checkpt = args->checkpt_path && args->checkpt_path[0] != '\0';
687 0 : int has_checkpt_funk = args->checkpt_funk && args->checkpt_funk[0] != '\0';
688 0 : int has_dump_to_protobuf = args->dump_insn_to_pb || args->dump_txn_to_pb || args->dump_block_to_pb;
689 :
690 0 : if( has_solcap || has_checkpt || has_checkpt_funk || has_dump_to_protobuf ) {
691 0 : FILE * capture_file = NULL;
692 :
693 0 : void * capture_ctx_mem = fd_valloc_malloc( args->valloc, FD_CAPTURE_CTX_ALIGN, FD_CAPTURE_CTX_FOOTPRINT );
694 0 : FD_TEST( capture_ctx_mem );
695 0 : fd_memset( capture_ctx_mem, 0, sizeof( fd_capture_ctx_t ) );
696 0 : args->capture_ctx = fd_capture_ctx_new( capture_ctx_mem );
697 :
698 0 : args->capture_ctx->checkpt_freq = ULONG_MAX;
699 :
700 0 : if( has_solcap ) {
701 0 : capture_file = fopen( args->capture_fpath, "w+" );
702 0 : if( FD_UNLIKELY( !capture_file ) ) {
703 0 : FD_LOG_ERR(( "fopen(%s) failed (%d-%s)", args->capture_fpath, errno, strerror( errno ) ));
704 0 : }
705 0 : fd_solcap_writer_init( args->capture_ctx->capture, capture_file );
706 0 : args->capture_ctx->capture_txns = args->capture_txns;
707 0 : } else {
708 0 : args->capture_ctx->capture = NULL;
709 0 : }
710 :
711 0 : if( has_checkpt || has_checkpt_funk ) {
712 0 : args->capture_ctx->checkpt_path = ( has_checkpt ? args->checkpt_path : args->checkpt_funk );
713 0 : args->capture_ctx->checkpt_freq = args->checkpt_freq;
714 0 : }
715 0 : if( has_dump_to_protobuf ) {
716 0 : args->capture_ctx->dump_insn_to_pb = args->dump_insn_to_pb;
717 0 : args->capture_ctx->dump_txn_to_pb = args->dump_txn_to_pb;
718 0 : args->capture_ctx->dump_block_to_pb = args->dump_block_to_pb;
719 0 : args->capture_ctx->dump_proto_sig_filter = args->dump_proto_sig_filter;
720 0 : args->capture_ctx->dump_proto_output_dir = args->dump_proto_output_dir;
721 0 : args->capture_ctx->dump_proto_start_slot = args->dump_proto_start_slot;
722 0 : }
723 0 : }
724 :
725 0 : args->slot_ctx->snapshot_freq = args->snapshot_freq;
726 0 : args->slot_ctx->incremental_freq = args->incremental_freq;
727 0 : args->slot_ctx->last_snapshot_slot = 0UL;
728 0 : args->last_snapshot_slot = 0UL;
729 :
730 0 : args->slot_ctx->runtime_wksp = fd_wksp_containing( args->runtime_spad );
731 0 : FD_TEST( args->slot_ctx->runtime_wksp );
732 :
733 : /* Finish other runtime setup steps */
734 0 : fd_features_restore( args->slot_ctx, args->runtime_spad );
735 0 : fd_runtime_update_leaders( args->slot_ctx, args->slot_ctx->slot_bank.slot, args->runtime_spad );
736 0 : fd_calculate_epoch_accounts_hash_values( args->slot_ctx );
737 :
738 0 : fd_funk_start_write( funk );
739 0 : fd_bpf_scan_and_create_bpf_program_cache_entry_tpool( args->slot_ctx, args->slot_ctx->funk_txn, args->tpool, args->runtime_spad );
740 0 : fd_funk_end_write( funk );
741 :
742 : /* First, load in the sysvars into the sysvar cache. This is required to
743 : make the StakeHistory sysvar available to the rewards calculation. */
744 :
745 0 : fd_runtime_sysvar_cache_load( args->slot_ctx, args->runtime_spad );
746 :
747 : /* After both snapshots have been loaded in, we can determine if we should
748 : start distributing rewards. */
749 :
750 0 : fd_rewards_recalculate_partitioned_rewards( args->slot_ctx,
751 0 : args->tpool,
752 0 : args->exec_spads,
753 0 : args->exec_spad_cnt,
754 0 : args->runtime_spad );
755 :
756 0 : }
757 :
758 : void
759 0 : fd_ledger_main_teardown( fd_ledger_args_t * args ) {
760 : /* Flush solcap file and cleanup */
761 0 : if( args->capture_ctx && args->capture_ctx->capture ) {
762 0 : fd_solcap_writer_flush( args->capture_ctx->capture );
763 0 : fd_solcap_writer_delete( args->capture_ctx->capture );
764 0 : }
765 :
766 0 : fd_exec_epoch_ctx_delete( fd_exec_epoch_ctx_leave( args->epoch_ctx ) );
767 0 : fd_exec_slot_ctx_delete( fd_exec_slot_ctx_leave( args->slot_ctx ) );
768 0 : }
769 :
770 : void
771 : ingest_rocksdb( char const * file,
772 : ulong start_slot,
773 : ulong end_slot,
774 : fd_blockstore_t * blockstore,
775 : int txn_status,
776 : ulong trash_hash,
777 0 : fd_valloc_t valloc ) {
778 :
779 0 : fd_rocksdb_t rocks_db;
780 0 : char * err = fd_rocksdb_init( &rocks_db, file );
781 0 : if( FD_UNLIKELY( err!=NULL ) ) {
782 0 : FD_LOG_ERR(( "fd_rocksdb_init returned %s", err ));
783 0 : }
784 :
785 0 : ulong last_slot = fd_rocksdb_last_slot( &rocks_db, &err );
786 0 : if( FD_UNLIKELY( err!=NULL ) ) {
787 0 : FD_LOG_ERR(( "fd_rocksdb_last_slot returned %s", err ));
788 0 : }
789 :
790 0 : if( last_slot < start_slot ) {
791 0 : FD_LOG_ERR(( "rocksdb blocks are older than snapshot. first=%lu last=%lu wanted=%lu",
792 0 : fd_rocksdb_first_slot(&rocks_db, &err), last_slot, start_slot ));
793 0 : }
794 :
795 0 : FD_LOG_NOTICE(( "ingesting rocksdb from start=%lu to end=%lu", start_slot, end_slot ));
796 :
797 0 : fd_rocksdb_root_iter_t iter = {0};
798 0 : fd_rocksdb_root_iter_new( &iter );
799 :
800 0 : fd_slot_meta_t slot_meta = {0};
801 0 : fd_memset( &slot_meta, 0, sizeof(slot_meta) );
802 :
803 0 : int block_found = -1;
804 0 : while ( block_found!=0 && start_slot<=end_slot ) {
805 0 : block_found = fd_rocksdb_root_iter_seek( &iter, &rocks_db, start_slot, &slot_meta, valloc );
806 0 : if ( block_found!=0 ) {
807 0 : start_slot++;
808 0 : }
809 0 : }
810 0 : if( FD_UNLIKELY( block_found!=0 ) ) {
811 0 : FD_LOG_ERR(( "unable to seek to any slot" ));
812 0 : }
813 :
814 0 : uchar trash_hash_buf[32];
815 0 : memset( trash_hash_buf, 0xFE, sizeof(trash_hash_buf) );
816 :
817 0 : ulong blk_cnt = 0;
818 0 : do {
819 0 : ulong slot = slot_meta.slot;
820 0 : if( slot > end_slot ) {
821 0 : break;
822 0 : }
823 :
824 : /* Read and deshred block from RocksDB */
825 0 : if( blk_cnt % 100 == 0 ) {
826 0 : FD_LOG_WARNING(( "imported %lu blocks", blk_cnt ));
827 0 : }
828 :
829 0 : int err = fd_rocksdb_import_block_blockstore( &rocks_db,
830 0 : &slot_meta,
831 0 : blockstore,
832 0 : txn_status,
833 0 : (slot == trash_hash) ? trash_hash_buf : NULL,
834 0 : valloc );
835 0 : if( FD_UNLIKELY( err ) ) {
836 0 : FD_LOG_ERR(( "fd_rocksdb_get_block failed" ));
837 0 : }
838 :
839 0 : ++blk_cnt;
840 :
841 0 : fd_slot_meta_destroy( &slot_meta );
842 :
843 0 : int ret = fd_rocksdb_root_iter_next( &iter, &slot_meta, valloc );
844 0 : if( ret < 0 ) {
845 : // FD_LOG_WARNING(("Failed for slot %lu", slot + 1));
846 0 : ret = fd_rocksdb_get_meta( &rocks_db, slot + 1, &slot_meta, valloc );
847 0 : if( ret < 0 ) {
848 0 : break;
849 0 : }
850 0 : }
851 : // FD_LOG_ERR(("fd_rocksdb_root_iter_seek returned %d", ret));
852 0 : } while (1);
853 :
854 0 : fd_rocksdb_root_iter_destroy( &iter );
855 0 : fd_rocksdb_destroy( &rocks_db );
856 :
857 0 : FD_LOG_NOTICE(( "ingested %lu blocks", blk_cnt ));
858 0 : }
859 :
860 : void
861 0 : parse_one_off_features( fd_ledger_args_t * args, char const * one_off_features ) {
862 0 : if( !one_off_features ) {
863 0 : FD_LOG_NOTICE(( "No one-off features passed in" ));
864 0 : return;
865 0 : }
866 :
867 0 : char * one_off_features_str = strdup( one_off_features );
868 0 : args->one_off_features_strdup = one_off_features_str;
869 0 : char * token = NULL;
870 0 : token = strtok( one_off_features_str, "," );
871 0 : while( token ) {
872 0 : args->one_off_features[ args->one_off_features_cnt++ ] = token;
873 0 : token = strtok( NULL, "," );
874 0 : }
875 :
876 0 : FD_LOG_NOTICE(( "Found %u one off features to include", args->one_off_features_cnt ));
877 0 : }
878 :
879 : void
880 : parse_rocksdb_list( fd_ledger_args_t * args,
881 : char const * rocksdb_list,
882 0 : char const * rocksdb_start_slots ) {
883 : /* First parse the paths to the different rocksdb */
884 0 : if( FD_UNLIKELY( !rocksdb_list ) ) {
885 0 : FD_LOG_NOTICE(( "No rocksdb list passed in" ));
886 0 : return;
887 0 : }
888 :
889 0 : char * rocksdb_str = strdup( rocksdb_list );
890 0 : args->rocksdb_list_strdup = rocksdb_str;
891 0 : char * token = NULL;
892 0 : token = strtok( rocksdb_str, "," );
893 0 : while( token ) {
894 0 : args->rocksdb_list[ args->rocksdb_list_cnt++ ] = token;
895 0 : token = strtok( NULL, "," );
896 0 : }
897 :
898 : /* Now repeat for the start slots assuming there are multiple */
899 0 : if( rocksdb_start_slots == NULL && args->rocksdb_list_cnt > 1 ) {
900 0 : FD_LOG_ERR(( "Multiple rocksdb dirs passed in but no start slots" ));
901 0 : }
902 0 : ulong index = 0UL;
903 0 : if( rocksdb_start_slots ) {
904 0 : char * rocksdb_start_slot_str = strdup( rocksdb_start_slots );
905 0 : token = NULL;
906 0 : token = strtok( rocksdb_start_slot_str, "," );
907 0 : while( token ) {
908 0 : args->rocksdb_list_slot[ index++ ] = strtoul( token, NULL, 10 );
909 0 : token = strtok( NULL, "," );
910 0 : }
911 0 : }
912 :
913 0 : if( index != args->rocksdb_list_cnt - 1UL ) {
914 0 : FD_LOG_ERR(( "Number of rocksdb dirs passed in doesn't match number of start slots" ));
915 0 : }
916 0 : }
917 :
918 : void
919 0 : init_funk( fd_ledger_args_t * args ) {
920 0 : fd_funk_t * funk;
921 0 : if( args->restore_funk ) {
922 0 : funk = fd_funk_recover_checkpoint( args->funk_file, 1, args->restore_funk, &args->funk_close_args );
923 0 : } else {
924 0 : funk = fd_funk_open_file( args->funk_file, 1, args->hashseed, args->txns_max, args->index_max, args->funk_page_cnt*(1UL<<30), FD_FUNK_OVERWRITE, &args->funk_close_args );
925 0 : }
926 0 : args->funk = funk;
927 0 : args->funk_wksp = fd_funk_wksp( funk );
928 0 : FD_LOG_NOTICE(( "funky at global address 0x%016lx with %lu records", fd_wksp_gaddr_fast( args->funk_wksp, funk ),
929 0 : fd_funk_rec_cnt( fd_funk_rec_map( funk, args->funk_wksp ) ) ));
930 0 : }
931 :
932 : void
933 0 : cleanup_funk( fd_ledger_args_t * args ) {
934 0 : fd_funk_close_file( &args->funk_close_args );
935 0 : }
936 :
937 : void
938 0 : init_blockstore( fd_ledger_args_t * args ) {
939 0 : fd_wksp_tag_query_info_t info;
940 0 : ulong blockstore_tag = FD_BLOCKSTORE_MAGIC;
941 0 : void * shmem;
942 0 : if( fd_wksp_tag_query( args->wksp, &blockstore_tag, 1, &info, 1 ) > 0 ) {
943 0 : shmem = fd_wksp_laddr_fast( args->wksp, info.gaddr_lo );
944 0 : args->blockstore = fd_blockstore_join( &args->blockstore_ljoin, shmem );
945 0 : if( args->blockstore->shmem->magic != FD_BLOCKSTORE_MAGIC ) {
946 0 : FD_LOG_ERR(( "failed to join a blockstore" ));
947 0 : }
948 0 : FD_LOG_NOTICE(( "joined blockstore" ));
949 0 : } else {
950 0 : ulong txn_max = 256UL;
951 0 : shmem = fd_wksp_alloc_laddr( args->wksp, fd_blockstore_align(), fd_blockstore_footprint( args->shred_max, args->slot_history_max, 16, txn_max ), blockstore_tag );
952 0 : if( shmem == NULL ) {
953 0 : FD_LOG_ERR(( "failed to allocate a blockstore" ));
954 0 : }
955 0 : args->blockstore = fd_blockstore_join( &args->blockstore_ljoin, fd_blockstore_new( shmem, 1, args->hashseed, args->shred_max, args->slot_history_max, 16, txn_max ) );
956 0 : if( args->blockstore->shmem->magic != FD_BLOCKSTORE_MAGIC ) {
957 0 : fd_wksp_free_laddr( shmem );
958 0 : FD_LOG_ERR(( "failed to allocate a blockstore" ));
959 0 : }
960 0 : FD_LOG_NOTICE(( "allocating a new blockstore" ));
961 0 : }
962 0 : }
963 :
964 : void
965 0 : checkpt( fd_ledger_args_t * args ) {
966 0 : if( !args->checkpt && !args->checkpt_funk && !args->checkpt_status_cache ) {
967 0 : FD_LOG_WARNING(( "No checkpt argument specified" ));
968 0 : }
969 :
970 0 : if( args->checkpt_funk ) {
971 0 : if( args->funk_wksp == NULL ) {
972 0 : FD_LOG_ERR(( "funk_wksp is NULL" ));
973 0 : }
974 0 : FD_LOG_NOTICE(( "writing funk checkpt %s", args->checkpt_funk ));
975 0 : unlink( args->checkpt_funk );
976 : #ifdef FD_FUNK_WKSP_PROTECT
977 : fd_wksp_mprotect( args->funk_wksp, 0 );
978 : #endif
979 0 : int err = fd_wksp_checkpt( args->funk_wksp, args->checkpt_funk, 0666, 0, NULL );
980 : #ifdef FD_FUNK_WKSP_PROTECT
981 : fd_wksp_mprotect( args->funk_wksp, 1 );
982 : #endif
983 0 : if( err ) {
984 0 : FD_LOG_ERR(( "funk checkpt failed: error %d", err ));
985 0 : }
986 0 : }
987 0 : if( args->checkpt ) {
988 0 : FD_LOG_NOTICE(( "writing %s", args->checkpt ));
989 0 : unlink( args->checkpt );
990 0 : int err = fd_wksp_checkpt( args->wksp, args->checkpt, 0666, 0, NULL );
991 0 : if( err ) {
992 0 : FD_LOG_ERR(( "checkpt failed: error %d", err ));
993 0 : }
994 0 : }
995 0 : if( args->checkpt_status_cache ) {
996 0 : FD_LOG_NOTICE(( "writing %s", args->checkpt_status_cache ));
997 0 : unlink( args->checkpt_status_cache );
998 0 : int err = fd_wksp_checkpt( args->status_cache_wksp, args->checkpt_status_cache, 0666, 0, NULL );
999 0 : if( err ) {
1000 0 : FD_LOG_ERR(( "status cache checkpt failed: error %d", err ));
1001 0 : }
1002 0 : }
1003 0 : }
1004 :
1005 : void
1006 0 : wksp_restore( fd_ledger_args_t * args ) {
1007 0 : if( args->restore != NULL ) {
1008 0 : FD_LOG_NOTICE(( "restoring wksp %s", args->restore ));
1009 0 : fd_wksp_restore( args->wksp, args->restore, args->hashseed );
1010 0 : }
1011 0 : }
1012 :
1013 : /********************* Main Command Functions and Setup ***********************/
1014 : void
1015 0 : minify( fd_ledger_args_t * args ) {
1016 : /* Example commmand:
1017 : fd_ledger --cmd minify --rocksdb <LARGE_ROCKSDB> --minified-rocksdb <MINI_ROCKSDB>
1018 : --start-slot <START_SLOT> --end-slot <END_SLOT> --copy-txn-status 1
1019 : */
1020 0 : if( args->rocksdb_list[ 0UL ] == NULL ) {
1021 0 : FD_LOG_ERR(( "rocksdb path is NULL" ));
1022 0 : }
1023 0 : if( args->mini_db_dir == NULL ) {
1024 0 : FD_LOG_ERR(( "minified rocksdb path is NULL" ));
1025 0 : }
1026 :
1027 0 : args->valloc = allocator_setup( args->wksp );
1028 0 : init_spads( args, 0 );
1029 :
1030 0 : fd_rocksdb_t big_rocksdb;
1031 0 : char * err = fd_rocksdb_init( &big_rocksdb, args->rocksdb_list[ 0UL ] );
1032 0 : if( FD_UNLIKELY( err!=NULL ) ) {
1033 0 : FD_LOG_ERR(( "fd_rocksdb_init at path=%s returned error=%s", args->rocksdb_list[ 0UL ], err ));
1034 0 : }
1035 :
1036 : /* If the directory for the minified rocksdb already exists, error out */
1037 0 : struct stat statbuf;
1038 0 : if( stat( args->mini_db_dir, &statbuf ) == 0 ) {
1039 0 : FD_LOG_ERR(( "path for mini_db_dir=%s already exists", args->mini_db_dir ));
1040 0 : }
1041 :
1042 : /* Create a new smaller rocksdb */
1043 0 : fd_rocksdb_t mini_rocksdb;
1044 0 : fd_rocksdb_new( &mini_rocksdb, args->mini_db_dir );
1045 :
1046 : /* Correctly bound off start and end slot */
1047 0 : ulong first_slot = fd_rocksdb_first_slot( &big_rocksdb, &err );
1048 0 : ulong last_slot = fd_rocksdb_last_slot( &big_rocksdb, &err );
1049 0 : if( args->start_slot < first_slot ) { args->start_slot = first_slot; }
1050 0 : if( args->end_slot > last_slot ) { args->end_slot = last_slot; }
1051 :
1052 0 : FD_LOG_NOTICE(( "copying over rocks db for range [%lu, %lu]", args->start_slot, args->end_slot ));
1053 :
1054 : /* Copy over all slot indexed columns */
1055 0 : for( ulong cf_idx = 1; cf_idx < FD_ROCKSDB_CF_CNT; ++cf_idx ) {
1056 0 : fd_rocksdb_copy_over_slot_indexed_range( &big_rocksdb, &mini_rocksdb, cf_idx,
1057 0 : args->start_slot, args->end_slot );
1058 0 : }
1059 0 : FD_LOG_NOTICE(("copied over all slot indexed columns"));
1060 :
1061 : /* Copy over transactions. This is more complicated because first, a temporary
1062 : blockstore will be populated. This will be used to look up transactions
1063 : which can be quickly queried */
1064 0 : if( args->copy_txn_status ) {
1065 0 : init_blockstore( args );
1066 : /* Ingest block range into blockstore */
1067 0 : ingest_rocksdb( args->rocksdb_list[ 0UL ],
1068 0 : args->start_slot,
1069 0 : args->end_slot,
1070 0 : args->blockstore,
1071 0 : 0,
1072 0 : ULONG_MAX,
1073 0 : args->valloc );
1074 :
1075 0 : fd_rocksdb_copy_over_txn_status_range( &big_rocksdb, &mini_rocksdb, args->blockstore,
1076 0 : args->start_slot, args->end_slot );
1077 0 : FD_LOG_NOTICE(( "copied over all transaction statuses" ));
1078 0 : } else {
1079 0 : FD_LOG_NOTICE(( "skipping copying of transaction statuses" ));
1080 0 : }
1081 :
1082 : /* TODO: Currently, the address signatures column family isn't copied as it
1083 : is indexed on the pubkey. */
1084 :
1085 0 : fd_rocksdb_destroy( &big_rocksdb );
1086 0 : fd_rocksdb_destroy( &mini_rocksdb );
1087 0 : }
1088 :
1089 : void
1090 0 : ingest( fd_ledger_args_t * args ) {
1091 : /* Setup funk, blockstore, epoch_ctx, and slot_ctx */
1092 0 : wksp_restore( args );
1093 0 : init_funk( args );
1094 0 : if( !args->funk_only ) {
1095 0 : init_blockstore( args );
1096 0 : }
1097 :
1098 0 : init_tpool( args );
1099 0 : init_spads( args, 1 );
1100 :
1101 0 : fd_spad_t * spad = args->runtime_spad;
1102 :
1103 0 : fd_funk_t * funk = args->funk;
1104 :
1105 0 : args->valloc = allocator_setup( args->wksp );
1106 0 : uchar * epoch_ctx_mem = fd_spad_alloc( spad, fd_exec_epoch_ctx_align(), fd_exec_epoch_ctx_footprint( args->vote_acct_max ) );
1107 0 : fd_memset( epoch_ctx_mem, 0, fd_exec_epoch_ctx_footprint( args->vote_acct_max ) );
1108 0 : fd_exec_epoch_ctx_t * epoch_ctx = fd_exec_epoch_ctx_join( fd_exec_epoch_ctx_new( epoch_ctx_mem, args->vote_acct_max ) );
1109 :
1110 0 : uchar slot_ctx_mem[FD_EXEC_SLOT_CTX_FOOTPRINT] __attribute__((aligned(FD_EXEC_SLOT_CTX_ALIGN)));
1111 0 : fd_exec_slot_ctx_t * slot_ctx = fd_exec_slot_ctx_join( fd_exec_slot_ctx_new( slot_ctx_mem, spad ) );
1112 0 : slot_ctx->epoch_ctx = epoch_ctx;
1113 0 : args->slot_ctx = slot_ctx;
1114 :
1115 0 : uchar * acc_mgr_mem = fd_spad_alloc( args->runtime_spad, alignof(fd_acc_mgr_t), sizeof(fd_acc_mgr_t) );
1116 0 : slot_ctx->acc_mgr = fd_acc_mgr_new( acc_mgr_mem, funk );
1117 0 : slot_ctx->blockstore = args->blockstore;
1118 :
1119 0 : if( args->status_cache_wksp ) {
1120 0 : void * status_cache_mem = fd_spad_alloc( spad,
1121 0 : fd_txncache_align(),
1122 0 : fd_txncache_footprint( FD_TXNCACHE_DEFAULT_MAX_ROOTED_SLOTS,
1123 0 : FD_TXNCACHE_DEFAULT_MAX_LIVE_SLOTS,
1124 0 : MAX_CACHE_TXNS_PER_SLOT,
1125 0 : FD_TXNCACHE_DEFAULT_MAX_CONSTIPATED_SLOTS ) );
1126 0 : FD_TEST( status_cache_mem );
1127 0 : slot_ctx->status_cache = fd_txncache_join( fd_txncache_new( status_cache_mem,
1128 0 : FD_TXNCACHE_DEFAULT_MAX_ROOTED_SLOTS,
1129 0 : FD_TXNCACHE_DEFAULT_MAX_LIVE_SLOTS,
1130 0 : MAX_CACHE_TXNS_PER_SLOT,
1131 0 : FD_TXNCACHE_DEFAULT_MAX_CONSTIPATED_SLOTS ) );
1132 0 : FD_TEST( slot_ctx->status_cache );
1133 0 : }
1134 :
1135 : /* Load in snapshot(s) */
1136 0 : if( args->snapshot ) {
1137 0 : fd_snapshot_load_all( args->snapshot,
1138 0 : slot_ctx,
1139 0 : NULL,
1140 0 : args->tpool,
1141 0 : args->verify_acc_hash,
1142 0 : args->check_acc_hash ,
1143 0 : FD_SNAPSHOT_TYPE_FULL,
1144 0 : args->exec_spads,
1145 0 : args->exec_spad_cnt,
1146 0 : args->runtime_spad );
1147 0 : FD_LOG_NOTICE(( "imported %lu records from snapshot", fd_funk_rec_cnt( fd_funk_rec_map( funk, fd_funk_wksp( funk ) ) ) ));
1148 0 : }
1149 0 : if( args->incremental ) {
1150 0 : fd_snapshot_load_all( args->incremental,
1151 0 : slot_ctx,
1152 0 : NULL,
1153 0 : args->tpool,
1154 0 : args->verify_acc_hash,
1155 0 : args->check_acc_hash,
1156 0 : FD_SNAPSHOT_TYPE_INCREMENTAL,
1157 0 : args->exec_spads,
1158 0 : args->exec_spad_cnt,
1159 0 : args->runtime_spad );
1160 0 : FD_LOG_NOTICE(( "imported %lu records from incremental snapshot", fd_funk_rec_cnt( fd_funk_rec_map( funk, fd_funk_wksp( funk ) ) ) ));
1161 0 : }
1162 :
1163 0 : if( args->genesis ) {
1164 0 : fd_runtime_read_genesis( slot_ctx, args->genesis, args->snapshot != NULL, NULL, args->tpool, args->runtime_spad );
1165 0 : }
1166 :
1167 0 : if( !args->snapshot && (args->restore_funk != NULL || args->restore != NULL) ) {
1168 0 : fd_runtime_recover_banks( slot_ctx, 0, 1, args->runtime_spad );
1169 0 : }
1170 :
1171 : /* At this point the account state has been ingested into funk. Intake rocksdb */
1172 0 : if( args->start_slot == 0 ) {
1173 0 : args->start_slot = slot_ctx->slot_bank.slot + 1;
1174 0 : }
1175 0 : fd_blockstore_t * blockstore = args->blockstore;
1176 0 : if( blockstore ) {
1177 0 : blockstore->shmem->lps = blockstore->shmem->hcs = blockstore->shmem->wmk = slot_ctx->slot_bank.slot;
1178 0 : }
1179 :
1180 0 : if( args->funk_only ) {
1181 0 : FD_LOG_NOTICE(( "using funk only, skipping blockstore ingest" ));
1182 0 : } else if( args->shredcap ) {
1183 0 : FD_LOG_NOTICE(( "using shredcap" ));
1184 0 : fd_shredcap_populate_blockstore( args->shredcap, blockstore, args->start_slot, args->end_slot );
1185 0 : } else if( args->rocksdb_list[ 0UL ] ) {
1186 0 : if( args->end_slot >= slot_ctx->slot_bank.slot + args->slot_history_max ) {
1187 0 : args->end_slot = slot_ctx->slot_bank.slot + args->slot_history_max - 1;
1188 0 : }
1189 0 : ingest_rocksdb( args->rocksdb_list[ 0UL ], args->start_slot, args->end_slot,
1190 0 : blockstore, args->copy_txn_status, args->trash_hash, args->valloc );
1191 0 : }
1192 :
1193 : /* Verification */
1194 0 : for( fd_feature_id_t const * id = fd_feature_iter_init();
1195 0 : !fd_feature_iter_done( id );
1196 0 : id = fd_feature_iter_next( id ) ) {
1197 0 : ulong activated_at = fd_features_get( &slot_ctx->epoch_ctx->features, id );
1198 0 : if( activated_at ) {
1199 0 : FD_LOG_DEBUG(( "feature %s activated at slot %lu", FD_BASE58_ENC_32_ALLOCA( id->id.key ), activated_at ));
1200 0 : }
1201 0 : }
1202 :
1203 0 : if( args->verify_funk ) {
1204 0 : FD_LOG_NOTICE(( "verifying funky" ));
1205 0 : if( fd_funk_verify( funk ) ) {
1206 0 : FD_LOG_ERR(( "verification failed" ));
1207 0 : }
1208 0 : }
1209 :
1210 0 : checkpt( args );
1211 :
1212 0 : cleanup_funk( args );
1213 0 : }
1214 :
1215 : int
1216 0 : replay( fd_ledger_args_t * args ) {
1217 : /* Allows for ingest and direct replay. This can be done with a full checkpoint
1218 : that contains a blockstore and funk, a checkpoint that just has funk, or directly
1219 : using a rocksdb and snapshot.
1220 :
1221 : On demand block ingest is enabled by default and can be disabled with
1222 : '--on-demand-block-ingest 0'. The number of blocks retained in a blockstore during
1223 : on demand block ingest can be set with '--on-demand-block-history <N slots>'
1224 :
1225 : In order to replay from a checkpoint, use '--checkpoint <path to checkpoint>'.
1226 :
1227 : To use a checkpoint, but to consume blocks on demand use '--funkonly true'.
1228 : This option MUST be used if the checkpoint was generated during a replay with
1229 : on demand block ingest.
1230 :
1231 : For blocks to contain transaction status information use '--txnstatus true'
1232 :
1233 : Example command loading in from on demand checkpoint and replaying with on demand block ingest.
1234 : It creates a checkpoint every 1000 slots.
1235 : fd_ledger --funk-restore <CHECKPOINT_TO_LOAD_IN> --cmd replay --page-cnt 20
1236 : --abort-on-mismatch 1 --tile-cpus 5-21 --allocator wksp
1237 : --rocksdb dump/rocksdb --checkpt-path dump/checkpoint_new
1238 : --checkpt-freq 1000 --funk-only 1 --on-demand-block-ingest 1 --funk-page-cnt 350
1239 :
1240 : Example command directly loading in a rocksdb and snapshot and replaying.
1241 : fd_ledger --reset 1 --cmd replay --rocksdb dump/mainnet-257068890/rocksdb --index-max 5000000
1242 : --end-slot 257068895 --txn-max 100 --page-cnt 16 --verify-acc-hash 1
1243 : --snapshot dump/mainnet-257068890/snapshot-257068890-uRVtagPzKhYorycp4CRtKdWrYPij6iBxCYYXmqRvdSp.tar.zst
1244 : --slot-history 5000 --allocator wksp --tile-cpus 5-21 --funk-page-cnt 16
1245 : */
1246 :
1247 0 : args->valloc = allocator_setup( args->wksp );
1248 :
1249 0 : wksp_restore( args ); /* Restores checkpointed workspace(s) */
1250 :
1251 0 : init_funk( args ); /* Joins or creates funk based on if one exists in the workspace */
1252 0 : init_blockstore( args ); /* Does the same for the blockstore */
1253 :
1254 0 : init_tpool( args ); /* Sets up tpool */
1255 0 : init_spads( args, 1 ); /* Sets up spad */
1256 :
1257 0 : fd_spad_t * spad = args->runtime_spad;
1258 :
1259 0 : FD_SPAD_FRAME_BEGIN( spad ) {
1260 :
1261 :
1262 : /* Setup slot_ctx */
1263 0 : fd_funk_t * funk = args->funk;
1264 :
1265 0 : void * epoch_ctx_mem = fd_spad_alloc( spad, FD_EXEC_EPOCH_CTX_ALIGN, fd_exec_epoch_ctx_footprint( args->vote_acct_max ) );
1266 0 : fd_memset( epoch_ctx_mem, 0, fd_exec_epoch_ctx_footprint( args->vote_acct_max ) );
1267 0 : args->epoch_ctx = fd_exec_epoch_ctx_join( fd_exec_epoch_ctx_new( epoch_ctx_mem, args->vote_acct_max ) );
1268 0 : fd_exec_epoch_ctx_bank_mem_clear( args->epoch_ctx );
1269 :
1270 : /* TODO: This is very hacky, needs to be cleaned up */
1271 0 : args->epoch_ctx->epoch_bank.cluster_version[0] = args->cluster_version[0];
1272 0 : args->epoch_ctx->epoch_bank.cluster_version[1] = args->cluster_version[1];
1273 0 : args->epoch_ctx->epoch_bank.cluster_version[2] = args->cluster_version[2];
1274 :
1275 0 : void * runtime_public_mem = fd_wksp_alloc_laddr( args->wksp, fd_runtime_public_align(), fd_runtime_public_footprint( ), FD_EXEC_EPOCH_CTX_MAGIC );
1276 0 : fd_memset( runtime_public_mem, 0, fd_runtime_public_footprint( ) );
1277 :
1278 0 : args->epoch_ctx->replay_public = fd_runtime_public_join( runtime_public_mem );
1279 :
1280 0 : fd_features_enable_cleaned_up( &args->epoch_ctx->features, args->epoch_ctx->epoch_bank.cluster_version );
1281 0 : fd_features_enable_one_offs( &args->epoch_ctx->features, args->one_off_features, args->one_off_features_cnt, 0UL );
1282 :
1283 : // activate them
1284 0 : fd_memcpy( &args->epoch_ctx->replay_public->features, &args->epoch_ctx->features, sizeof(fd_features_t) );
1285 :
1286 0 : void * slot_ctx_mem = fd_spad_alloc( spad, FD_EXEC_SLOT_CTX_ALIGN, FD_EXEC_SLOT_CTX_FOOTPRINT );
1287 0 : args->slot_ctx = fd_exec_slot_ctx_join( fd_exec_slot_ctx_new( slot_ctx_mem, spad ) );
1288 0 : args->slot_ctx->epoch_ctx = args->epoch_ctx;
1289 0 : uchar * acc_mgr_mem = fd_spad_alloc( args->runtime_spad, alignof(fd_acc_mgr_t), sizeof(fd_acc_mgr_t) );
1290 0 : args->slot_ctx->acc_mgr = fd_acc_mgr_new( acc_mgr_mem, funk );
1291 0 : args->slot_ctx->blockstore = args->blockstore;
1292 :
1293 0 : void * status_cache_mem = fd_spad_alloc( spad,
1294 0 : FD_TXNCACHE_ALIGN,
1295 0 : fd_txncache_footprint( FD_TXNCACHE_DEFAULT_MAX_ROOTED_SLOTS,
1296 0 : FD_TXNCACHE_DEFAULT_MAX_LIVE_SLOTS,
1297 0 : MAX_CACHE_TXNS_PER_SLOT,
1298 0 : FD_TXNCACHE_DEFAULT_MAX_CONSTIPATED_SLOTS) );
1299 0 : args->slot_ctx->status_cache = fd_txncache_join( fd_txncache_new( status_cache_mem,
1300 0 : FD_TXNCACHE_DEFAULT_MAX_ROOTED_SLOTS,
1301 0 : FD_TXNCACHE_DEFAULT_MAX_LIVE_SLOTS,
1302 0 : MAX_CACHE_TXNS_PER_SLOT,
1303 0 : FD_TXNCACHE_DEFAULT_MAX_CONSTIPATED_SLOTS ) );
1304 0 : if( FD_UNLIKELY( !args->slot_ctx->status_cache ) ) {
1305 0 : FD_LOG_ERR(( "Status cache was not allocated" ));
1306 0 : }
1307 :
1308 : /* Check number of records in funk. If rec_cnt == 0, then it can be assumed
1309 : that you need to load in snapshot(s). */
1310 :
1311 0 : ulong rec_cnt = fd_funk_rec_cnt( fd_funk_rec_map( funk, fd_funk_wksp( funk ) ) );
1312 0 : if( !rec_cnt ) {
1313 : /* Load in snapshot(s) */
1314 0 : if( args->snapshot ) {
1315 0 : fd_snapshot_load_all( args->snapshot,
1316 0 : args->slot_ctx,
1317 0 : NULL,
1318 0 : args->tpool,
1319 0 : args->verify_acc_hash,
1320 0 : args->check_acc_hash,
1321 0 : FD_SNAPSHOT_TYPE_FULL,
1322 0 : args->exec_spads,
1323 0 : args->exec_spad_cnt,
1324 0 : args->runtime_spad );
1325 0 : FD_LOG_NOTICE(( "imported %lu records from snapshot", fd_funk_rec_cnt( fd_funk_rec_map( funk, fd_funk_wksp( funk ) ) ) ));
1326 0 : }
1327 0 : if( args->incremental ) {
1328 0 : fd_snapshot_load_all( args->incremental,
1329 0 : args->slot_ctx,
1330 0 : NULL,
1331 0 : args->tpool,
1332 0 : args->verify_acc_hash,
1333 0 : args->check_acc_hash,
1334 0 : FD_SNAPSHOT_TYPE_INCREMENTAL,
1335 0 : args->exec_spads,
1336 0 : args->exec_spad_cnt,
1337 0 : args->runtime_spad );
1338 0 : FD_LOG_NOTICE(( "imported %lu records from snapshot", fd_funk_rec_cnt( fd_funk_rec_map( funk, fd_funk_wksp( funk ) ) ) ));
1339 0 : }
1340 0 : if( args->genesis ) {
1341 0 : fd_runtime_read_genesis( args->slot_ctx, args->genesis, args->snapshot != NULL, NULL, args->tpool, args->runtime_spad );
1342 0 : }
1343 0 : } else {
1344 0 : FD_LOG_NOTICE(( "found funk with %lu records", rec_cnt ));
1345 0 : }
1346 :
1347 0 : FD_LOG_NOTICE(( "Used memory in spad after loading in snapshot %lu", args->runtime_spad->mem_used ));
1348 :
1349 0 : fd_ledger_main_setup( args );
1350 :
1351 0 : fd_blockstore_init( args->blockstore, -1, FD_BLOCKSTORE_ARCHIVE_MIN_SIZE, &args->slot_ctx->slot_bank );
1352 0 : fd_buf_shred_pool_reset( args->blockstore->shred_pool, 0 );
1353 :
1354 0 : FD_LOG_WARNING(( "setup done" ));
1355 :
1356 0 : int ret = runtime_replay( args );
1357 :
1358 0 : fd_ledger_main_teardown( args );
1359 :
1360 0 : cleanup_funk( args );
1361 :
1362 0 : return ret;
1363 :
1364 0 : } FD_SPAD_FRAME_END;
1365 0 : }
1366 :
1367 : /* Parse user arguments and setup shared data structures used across commands */
1368 : int
1369 0 : initial_setup( int argc, char ** argv, fd_ledger_args_t * args ) {
1370 0 : if( FD_UNLIKELY( argc==1 ) ) {
1371 0 : return 1;
1372 0 : }
1373 :
1374 0 : fd_boot( &argc, &argv );
1375 0 : fd_flamenco_boot( &argc, &argv );
1376 :
1377 0 : char const * wksp_name = fd_env_strip_cmdline_cstr ( &argc, &argv, "--wksp-name", NULL, NULL );
1378 0 : ulong funk_page_cnt = fd_env_strip_cmdline_ulong ( &argc, &argv, "--funk-page-cnt", NULL, 5 );
1379 0 : ulong page_cnt = fd_env_strip_cmdline_ulong ( &argc, &argv, "--page-cnt", NULL, 5 );
1380 0 : int reset = fd_env_strip_cmdline_int ( &argc, &argv, "--reset", NULL, 0 );
1381 0 : char const * cmd = fd_env_strip_cmdline_cstr ( &argc, &argv, "--cmd", NULL, NULL );
1382 0 : ulong index_max = fd_env_strip_cmdline_ulong ( &argc, &argv, "--index-max", NULL, 450000000 );
1383 0 : ulong txns_max = fd_env_strip_cmdline_ulong ( &argc, &argv, "--txn-max", NULL, 1000 );
1384 0 : char const * funk_file = fd_env_strip_cmdline_cstr ( &argc, &argv, "--funk-file", NULL, NULL );
1385 0 : int verify_funk = fd_env_strip_cmdline_int ( &argc, &argv, "--verify-funky", NULL, 0 );
1386 0 : char const * snapshot = fd_env_strip_cmdline_cstr ( &argc, &argv, "--snapshot", NULL, NULL );
1387 0 : char const * incremental = fd_env_strip_cmdline_cstr ( &argc, &argv, "--incremental", NULL, NULL );
1388 0 : char const * genesis = fd_env_strip_cmdline_cstr ( &argc, &argv, "--genesis", NULL, NULL );
1389 0 : int copy_txn_status = fd_env_strip_cmdline_int ( &argc, &argv, "--copy-txn-status", NULL, 0 );
1390 0 : ulong slot_history_max = fd_env_strip_cmdline_ulong ( &argc, &argv, "--slot-history", NULL, 100UL );
1391 0 : ulong shred_max = fd_env_strip_cmdline_ulong ( &argc, &argv, "--shred-max", NULL, 1UL << 17 );
1392 0 : ulong start_slot = fd_env_strip_cmdline_ulong ( &argc, &argv, "--start-slot", NULL, 0UL );
1393 0 : ulong end_slot = fd_env_strip_cmdline_ulong ( &argc, &argv, "--end-slot", NULL, ULONG_MAX );
1394 0 : uint verify_acc_hash = fd_env_strip_cmdline_uint ( &argc, &argv, "--verify-acc-hash", NULL, 1 );
1395 0 : uint check_acc_hash = fd_env_strip_cmdline_uint ( &argc, &argv, "--check-acc-hash", NULL, 1 );
1396 0 : char const * restore = fd_env_strip_cmdline_cstr ( &argc, &argv, "--restore", NULL, NULL );
1397 0 : char const * restore_funk = fd_env_strip_cmdline_cstr ( &argc, &argv, "--funk-restore", NULL, NULL );
1398 0 : char const * shredcap = fd_env_strip_cmdline_cstr ( &argc, &argv, "--shred-cap", NULL, NULL );
1399 0 : ulong trash_hash = fd_env_strip_cmdline_ulong ( &argc, &argv, "--trash-hash", NULL, ULONG_MAX );
1400 0 : char const * mini_db_dir = fd_env_strip_cmdline_cstr ( &argc, &argv, "--minified-rocksdb", NULL, NULL );
1401 0 : int funk_only = fd_env_strip_cmdline_int ( &argc, &argv, "--funk-only", NULL, 0 );
1402 0 : char const * checkpt = fd_env_strip_cmdline_cstr ( &argc, &argv, "--checkpt", NULL, NULL );
1403 0 : char const * checkpt_funk = fd_env_strip_cmdline_cstr ( &argc, &argv, "--checkpt-funk", NULL, NULL );
1404 0 : char const * capture_fpath = fd_env_strip_cmdline_cstr ( &argc, &argv, "--capture-solcap", NULL, NULL );
1405 0 : int capture_txns = fd_env_strip_cmdline_int ( &argc, &argv, "--capture-txns", NULL, 1 );
1406 0 : char const * checkpt_path = fd_env_strip_cmdline_cstr ( &argc, &argv, "--checkpt-path", NULL, NULL );
1407 0 : ulong checkpt_freq = fd_env_strip_cmdline_ulong ( &argc, &argv, "--checkpt-freq", NULL, ULONG_MAX );
1408 0 : int checkpt_mismatch = fd_env_strip_cmdline_int ( &argc, &argv, "--checkpt-mismatch", NULL, 0 );
1409 0 : char const * allocator = fd_env_strip_cmdline_cstr ( &argc, &argv, "--allocator", NULL, "wksp" );
1410 0 : int abort_on_mismatch = fd_env_strip_cmdline_int ( &argc, &argv, "--abort-on-mismatch", NULL, 1 );
1411 0 : int dump_insn_to_pb = fd_env_strip_cmdline_int ( &argc, &argv, "--dump-insn-to-pb", NULL, 0 );
1412 0 : int dump_txn_to_pb = fd_env_strip_cmdline_int ( &argc, &argv, "--dump-txn-to-pb", NULL, 0 );
1413 0 : int dump_block_to_pb = fd_env_strip_cmdline_int ( &argc, &argv, "--dump-block-to-pb", NULL, 0 );
1414 0 : ulong dump_proto_start_slot = fd_env_strip_cmdline_ulong ( &argc, &argv, "--dump-proto-start-slot", NULL, 0 );
1415 0 : char const * dump_proto_sig_filter = fd_env_strip_cmdline_cstr ( &argc, &argv, "--dump-proto-sig-filter", NULL, NULL );
1416 0 : char const * dump_proto_output_dir = fd_env_strip_cmdline_cstr ( &argc, &argv, "--dump-proto-output-dir", NULL, NULL );
1417 0 : ulong vote_acct_max = fd_env_strip_cmdline_ulong ( &argc, &argv, "--vote_acct_max", NULL, 2000000UL );
1418 0 : char const * rocksdb_list = fd_env_strip_cmdline_cstr ( &argc, &argv, "--rocksdb", NULL, NULL );
1419 0 : char const * rocksdb_list_starts = fd_env_strip_cmdline_cstr ( &argc, &argv, "--rocksdb-starts", NULL, NULL );
1420 0 : char const * cluster_version = fd_env_strip_cmdline_cstr ( &argc, &argv, "--cluster-version", NULL, "2.0.0" );
1421 0 : char const * checkpt_status_cache = fd_env_strip_cmdline_cstr ( &argc, &argv, "--checkpt-status-cache", NULL, NULL );
1422 0 : char const * one_off_features = fd_env_strip_cmdline_cstr ( &argc, &argv, "--one-off-features", NULL, NULL );
1423 0 : char const * lthash = fd_env_strip_cmdline_cstr ( &argc, &argv, "--lthash", NULL, "false" );
1424 0 : ulong snapshot_freq = fd_env_strip_cmdline_ulong ( &argc, &argv, "--snapshot-freq", NULL, ULONG_MAX );
1425 0 : ulong incremental_freq = fd_env_strip_cmdline_ulong ( &argc, &argv, "--incremental-freq", NULL, ULONG_MAX );
1426 0 : char const * snapshot_dir = fd_env_strip_cmdline_cstr ( &argc, &argv, "--snapshot-dir", NULL, NULL );
1427 0 : ulong snapshot_tcnt = fd_env_strip_cmdline_ulong ( &argc, &argv, "--snapshot-tcnt", NULL, 0UL );
1428 0 : double allowed_mem_delta = fd_env_strip_cmdline_double( &argc, &argv, "--allowed-mem-delta", NULL, 0.1 );
1429 0 : int snapshot_mismatch = fd_env_strip_cmdline_int ( &argc, &argv, "--snapshot-mismatch", NULL, 0 );
1430 0 : ulong thread_mem_bound = fd_env_strip_cmdline_ulong ( &argc, &argv, "--thread-mem-bound", NULL, FD_RUNTIME_TRANSACTION_EXECUTION_FOOTPRINT_DEFAULT );
1431 0 : ulong runtime_mem_bound = fd_env_strip_cmdline_ulong ( &argc, &argv, "--runtime-mem-bound", NULL, FD_RUNTIME_BLOCK_EXECUTION_FOOTPRINT );
1432 :
1433 0 : if( FD_UNLIKELY( !verify_acc_hash ) ) {
1434 : /* We've got full snapshots that contain all 0s for the account
1435 : hash in account meta. Running hash verify allows us to
1436 : populate the hash in account meta with real values. */
1437 0 : FD_LOG_WARNING(( "verify-acc-hash needs to be 1" ));
1438 0 : }
1439 :
1440 : // TODO: Add argument validation. Make sure that we aren't including any arguments that aren't parsed for
1441 :
1442 0 : char hostname[64];
1443 0 : gethostname( hostname, sizeof(hostname) );
1444 0 : ulong hashseed = fd_hash( 0, hostname, strnlen( hostname, sizeof(hostname) ) );
1445 0 : args->hashseed = (uint)hashseed;
1446 :
1447 : /* Setup workspace */
1448 0 : fd_wksp_t * wksp;
1449 0 : if( wksp_name == NULL ) {
1450 0 : FD_LOG_NOTICE(( "--wksp not specified, using an anonymous local workspace" ));
1451 0 : wksp = fd_wksp_new_anonymous( FD_SHMEM_GIGANTIC_PAGE_SZ, page_cnt, 0, "wksp", 0UL );
1452 0 : } else {
1453 0 : fd_shmem_info_t shmem_info[1];
1454 0 : if( FD_UNLIKELY( fd_shmem_info( wksp_name, 0UL, shmem_info ) ) )
1455 0 : FD_LOG_ERR(( "unable to query region \"%s\"\n\tprobably does not exist or bad permissions", wksp_name ));
1456 0 : wksp = fd_wksp_attach( wksp_name );
1457 0 : }
1458 :
1459 0 : if( wksp == NULL ) {
1460 0 : FD_LOG_ERR(( "failed to attach to workspace %s", wksp_name ));
1461 0 : }
1462 0 : if( reset ) {
1463 0 : fd_wksp_reset( wksp, args->hashseed );
1464 0 : }
1465 0 : args->wksp = wksp;
1466 :
1467 0 : if( checkpt_status_cache && checkpt_status_cache[0] != '\0' ) {
1468 0 : FD_LOG_NOTICE(( "Creating status cache wksp" ));
1469 0 : fd_wksp_t * status_cache_wksp = fd_wksp_new_anonymous( FD_SHMEM_GIGANTIC_PAGE_SZ, 23UL, 0, "status_cache_wksp", 0UL );
1470 0 : fd_wksp_reset( status_cache_wksp, args->hashseed );
1471 0 : args->status_cache_wksp = status_cache_wksp;
1472 0 : } else {
1473 0 : args->status_cache_wksp = NULL;
1474 0 : }
1475 :
1476 : /* Setup alloc */
1477 0 : #define FD_ALLOC_TAG (422UL)
1478 0 : void * alloc_shmem = fd_wksp_alloc_laddr( wksp, fd_alloc_align(), fd_alloc_footprint(), FD_ALLOC_TAG );
1479 0 : if( FD_UNLIKELY( !alloc_shmem ) ) { FD_LOG_ERR( ( "fd_alloc too large for workspace" ) ); }
1480 0 : void * alloc_shalloc = fd_alloc_new( alloc_shmem, FD_ALLOC_TAG );
1481 0 : if( FD_UNLIKELY( !alloc_shalloc ) ) { FD_LOG_ERR( ( "fd_alloc_new failed" ) ); }
1482 0 : fd_alloc_t * alloc = fd_alloc_join( alloc_shalloc, FD_ALLOC_TAG );
1483 0 : args->alloc = alloc;
1484 0 : #undef FD_ALLOC_TAG
1485 :
1486 : /* Copy over arguments */
1487 0 : args->cmd = cmd;
1488 0 : args->start_slot = start_slot;
1489 0 : args->end_slot = end_slot;
1490 0 : args->checkpt = checkpt;
1491 0 : args->checkpt_funk = checkpt_funk;
1492 0 : args->shred_max = shred_max;
1493 0 : args->slot_history_max = slot_history_max;
1494 0 : args->txns_max = txns_max;
1495 0 : args->index_max = index_max;
1496 0 : args->funk_page_cnt = funk_page_cnt;
1497 0 : args->funk_file = funk_file;
1498 0 : args->restore = restore;
1499 0 : args->restore_funk = restore_funk;
1500 0 : args->mini_db_dir = mini_db_dir;
1501 0 : args->funk_only = funk_only;
1502 0 : args->copy_txn_status = copy_txn_status;
1503 0 : args->snapshot = snapshot;
1504 0 : args->incremental = incremental;
1505 0 : args->genesis = genesis;
1506 0 : args->shredcap = shredcap;
1507 0 : args->verify_funk = verify_funk;
1508 0 : args->check_acc_hash = check_acc_hash;
1509 0 : args->verify_acc_hash = verify_acc_hash;
1510 0 : args->trash_hash = trash_hash;
1511 0 : args->capture_fpath = capture_fpath;
1512 0 : args->capture_txns = capture_txns;
1513 0 : args->checkpt_path = checkpt_path;
1514 0 : args->checkpt_freq = checkpt_freq;
1515 0 : args->checkpt_mismatch = checkpt_mismatch;
1516 0 : args->allocator = allocator;
1517 0 : args->abort_on_mismatch = abort_on_mismatch;
1518 0 : args->dump_insn_to_pb = dump_insn_to_pb;
1519 0 : args->dump_txn_to_pb = dump_txn_to_pb;
1520 0 : args->dump_block_to_pb = dump_block_to_pb;
1521 0 : args->dump_proto_start_slot = dump_proto_start_slot;
1522 0 : args->dump_proto_sig_filter = dump_proto_sig_filter;
1523 0 : args->dump_proto_output_dir = dump_proto_output_dir;
1524 0 : args->vote_acct_max = vote_acct_max;
1525 0 : args->rocksdb_list_cnt = 0UL;
1526 0 : args->checkpt_status_cache = checkpt_status_cache;
1527 0 : args->one_off_features_cnt = 0UL;
1528 0 : args->snapshot_freq = snapshot_freq;
1529 0 : args->incremental_freq = incremental_freq;
1530 0 : args->snapshot_dir = snapshot_dir;
1531 0 : args->snapshot_tcnt = snapshot_tcnt;
1532 0 : args->allowed_mem_delta = allowed_mem_delta;
1533 0 : args->lthash = lthash;
1534 0 : args->snapshot_mismatch = snapshot_mismatch;
1535 0 : args->thread_mem_bound = thread_mem_bound ? thread_mem_bound : FD_RUNTIME_BORROWED_ACCOUNT_FOOTPRINT;
1536 0 : args->runtime_mem_bound = runtime_mem_bound;
1537 0 : parse_one_off_features( args, one_off_features );
1538 0 : parse_rocksdb_list( args, rocksdb_list, rocksdb_list_starts );
1539 :
1540 0 : if( FD_UNLIKELY( sscanf( cluster_version, "%u.%u.%u", &args->cluster_version[0], &args->cluster_version[1], &args->cluster_version[2] )!=3 ) ) {
1541 0 : FD_LOG_ERR(( "failed to decode cluster version" ));;
1542 0 : }
1543 :
1544 0 : if( args->rocksdb_list_cnt==1UL ) {
1545 0 : FD_LOG_NOTICE(( "rocksdb=%s", args->rocksdb_list[0] ));
1546 0 : } else {
1547 0 : for( ulong i=0UL; i<args->rocksdb_list_cnt; ++i ) {
1548 0 : FD_LOG_NOTICE(( "rocksdb_list[ %lu ]=%s slot=%lu", i, args->rocksdb_list[i], args->rocksdb_list_slot[i-1] ));
1549 0 : }
1550 0 : }
1551 :
1552 0 : return 0;
1553 0 : }
1554 :
1555 : int main( int argc, char ** argv ) {
1556 : /* Declaring this on the stack gets the alignment wrong when using asan */
1557 : fd_ledger_args_t * args = fd_alloca( alignof(fd_ledger_args_t), sizeof(fd_ledger_args_t) );
1558 : memset( args, 0, sizeof(fd_ledger_args_t) );
1559 : initial_setup( argc, argv, args );
1560 :
1561 : /* TODO: Need to implement snapshot minification. */
1562 :
1563 : if( args->cmd == NULL ) {
1564 : FD_LOG_ERR(( "no command specified" ));
1565 : } else if( strcmp( args->cmd, "replay" ) == 0 ) {
1566 : return replay( args );
1567 : } else if( strcmp( args->cmd, "ingest" ) == 0 ) {
1568 : ingest( args );
1569 : } else if( strcmp( args->cmd, "minify" ) == 0 ) {
1570 : minify( args );
1571 : } else {
1572 : FD_LOG_ERR(( "unknown command=%s", args->cmd ));
1573 : }
1574 :
1575 : return 0;
1576 : }
|
