Line data Source code
1 : #include "fd_snapshot_loader.h"
2 : #include "fd_snapshot_http.h"
3 : #include "fd_snapshot_restore_private.h"
4 : #include "../runtime/fd_acc_mgr.h"
5 : #include "../runtime/context/fd_exec_slot_ctx.h"
6 : #include "../../ballet/zstd/fd_zstd.h"
7 : #include "../../flamenco/types/fd_types.h"
8 :
9 : #include <assert.h>
10 : #include <errno.h>
11 : #include <fcntl.h>
12 : #include <netdb.h>
13 : #include <regex.h>
14 : #include <stddef.h>
15 : #include <stdio.h>
16 : #include <stdlib.h>
17 : #include <unistd.h>
18 : #include <sys/random.h>
19 :
20 : /* Snapshot restore ***************************************************/
21 :
22 0 : #define OSTREAM_BUFSZ (32768UL)
23 :
24 : struct fd_snapshot_dumper {
25 : fd_alloc_t * alloc;
26 : fd_funk_t funk[1];
27 :
28 : fd_exec_slot_ctx_t * slot_ctx;
29 :
30 : int snapshot_fd;
31 :
32 : fd_snapshot_loader_t * loader;
33 : fd_snapshot_restore_t * restore;
34 :
35 : int csv_fd;
36 : fd_io_buffered_ostream_t csv_out;
37 : uchar csv_buf[ OSTREAM_BUFSZ ];
38 :
39 : int want_accounts;
40 : int has_fail;
41 : };
42 :
43 : typedef struct fd_snapshot_dumper fd_snapshot_dumper_t;
44 :
45 : static fd_snapshot_dumper_t *
46 0 : fd_snapshot_dumper_new( void * mem ) {
47 0 : fd_snapshot_dumper_t * dumper = mem;
48 0 : *dumper = (fd_snapshot_dumper_t) {
49 0 : .snapshot_fd = -1,
50 0 : .csv_fd = -1
51 0 : };
52 0 : return dumper;
53 0 : }
54 :
55 : static void *
56 0 : fd_snapshot_dumper_delete( fd_snapshot_dumper_t * dumper ) {
57 :
58 0 : if( dumper->loader ) {
59 0 : fd_snapshot_loader_delete( dumper->loader );
60 0 : dumper->loader = NULL;
61 0 : }
62 :
63 0 : if( dumper->restore ) {
64 0 : fd_snapshot_restore_delete( dumper->restore );
65 0 : dumper->restore = NULL;
66 0 : }
67 :
68 0 : if( dumper->slot_ctx ) {
69 0 : fd_exec_slot_ctx_delete( fd_exec_slot_ctx_leave( dumper->slot_ctx ) );
70 0 : dumper->slot_ctx = NULL;
71 0 : }
72 :
73 0 : if( dumper->funk->shmem ) {
74 0 : void * shfunk = NULL;
75 0 : fd_funk_leave( dumper->funk, &shfunk );
76 0 : fd_wksp_free_laddr( fd_funk_delete( shfunk ) );
77 0 : }
78 :
79 0 : if( dumper->alloc ) {
80 0 : fd_wksp_free_laddr( fd_alloc_delete( fd_alloc_leave( dumper->alloc ) ) );
81 0 : dumper->alloc = NULL;
82 0 : }
83 :
84 0 : if( dumper->csv_fd>=0 ) {
85 0 : fd_io_buffered_ostream_fini( &dumper->csv_out );
86 0 : if( FD_UNLIKELY( 0!=close( dumper->csv_fd ) ) )
87 0 : FD_LOG_WARNING(( "close(%d) failed (%d-%s)", dumper->csv_fd, errno, fd_io_strerror( errno ) ));
88 0 : dumper->csv_fd = -1;
89 0 : }
90 :
91 0 : fd_memset( dumper, 0, sizeof(fd_snapshot_dumper_t) );
92 0 : return dumper;
93 0 : }
94 :
95 : /* fd_snapshot_dumper_record processes a newly encountered account
96 : record. */
97 :
98 : union fd_snapshot_csv_rec {
99 : char line[ 180 ];
100 : struct __attribute__((packed)) {
101 : char acct_addr[ FD_BASE58_ENCODED_32_LEN ];
102 : char comma1;
103 : char owner_addr[ FD_BASE58_ENCODED_32_LEN ];
104 : char comma2;
105 : char hash[ FD_BASE58_ENCODED_32_LEN ];
106 : char comma3;
107 : char slot[ 14 ]; /* enough for 10000 years at 400ms slot time */
108 : char comma4;
109 : char size[ 8 ]; /* can represent [0,10<<20) */
110 : char comma5;
111 : char lamports[ 20 ]; /* can represent [0,1<<64) */
112 : char newline;
113 : };
114 : };
115 :
116 : typedef union fd_snapshot_csv_rec fd_snapshot_csv_rec_t;
117 :
118 : static void
119 : fd_snapshot_dumper_record( fd_snapshot_dumper_t * d,
120 : fd_funk_rec_t const * rec,
121 0 : fd_wksp_t * wksp ) {
122 :
123 0 : uchar const * rec_val = fd_funk_val_const( rec, wksp );
124 0 : fd_account_meta_t const * meta = (fd_account_meta_t const *)rec_val;
125 : //uchar const * data = rec_val + meta->hlen;
126 :
127 0 : if( d->csv_fd>=0 ) {
128 0 : fd_snapshot_csv_rec_t csv_rec;
129 0 : fd_memset( &csv_rec, ' ', sizeof(csv_rec) );
130 :
131 0 : ulong b58sz;
132 0 : fd_base58_encode_32( rec->pair.key->uc, &b58sz, csv_rec.acct_addr );
133 0 : csv_rec.line[ offsetof(fd_snapshot_csv_rec_t,acct_addr)+b58sz ] = ' ';
134 0 : csv_rec.comma1 = ',';
135 :
136 0 : fd_base58_encode_32( meta->info.owner, &b58sz, csv_rec.owner_addr );
137 0 : csv_rec.line[ offsetof(fd_snapshot_csv_rec_t,owner_addr)+b58sz ] = ' ';
138 0 : csv_rec.comma2 = ',';
139 :
140 0 : fd_base58_encode_32( meta->hash, &b58sz, csv_rec.hash );
141 0 : csv_rec.line[ offsetof(fd_snapshot_csv_rec_t,hash)+b58sz ] = ' ';
142 0 : csv_rec.comma3 = ',';
143 :
144 0 : fd_cstr_append_ulong_as_text( csv_rec.slot, ' ', '\0', meta->dlen, 15 );
145 0 : csv_rec.comma4 = ',';
146 :
147 0 : fd_cstr_append_ulong_as_text( csv_rec.size, ' ', '\0', meta->dlen, 8 );
148 0 : csv_rec.comma5 = ',';
149 :
150 0 : fd_cstr_append_ulong_as_text( csv_rec.lamports, ' ', '\0', meta->info.lamports, 20 );
151 0 : csv_rec.newline = '\n';
152 :
153 0 : fd_io_buffered_ostream_write( &d->csv_out, csv_rec.line, sizeof(csv_rec.line) );
154 0 : }
155 0 : }
156 :
157 : /* fd_snapshot_dumper_release visits any newly appeared accounts and
158 : removes their records from the database. */
159 :
160 : static int
161 0 : fd_snapshot_dumper_release( fd_snapshot_dumper_t * d ) {
162 :
163 0 : fd_funk_txn_t * funk_txn = d->restore->funk_txn;
164 0 : fd_funk_txn_xid_t txn_xid = funk_txn->xid;
165 0 : fd_funk_t * funk = d->funk;
166 0 : fd_wksp_t * wksp = fd_funk_wksp( funk );
167 :
168 : /* Dump all the records */
169 0 : for( fd_funk_rec_t const * rec = fd_funk_txn_first_rec( funk, funk_txn );
170 0 : rec;
171 0 : rec = fd_funk_txn_next_rec( funk, rec ) ) {
172 0 : if( FD_UNLIKELY( !fd_funk_key_is_acc( rec->pair.key ) ) ) continue;
173 0 : fd_snapshot_dumper_record( d, rec, wksp );
174 0 : }
175 :
176 : /* In order to save heap space, evict all the accounts we just
177 : visited. We can do this because we know we'll never read them
178 : again. */
179 :
180 0 : if( FD_UNLIKELY( fd_funk_txn_cancel( funk, funk_txn, 1 )!=1UL ) )
181 0 : FD_LOG_ERR(( "Failed to cancel funk txn" )); /* unreachable */
182 :
183 0 : funk_txn = fd_funk_txn_prepare( funk, NULL, &txn_xid, 1 );
184 0 : if( FD_UNLIKELY( !funk_txn ) )
185 0 : FD_LOG_ERR(( "Failed to prepare funk txn" )); /* unreachable */
186 :
187 0 : d->restore->funk_txn = funk_txn;
188 0 : return 0;
189 0 : }
190 :
191 : /* fd_snapshot_dumper_advance polls the tar reader for data and handles
192 : any newly appeared accounts. */
193 :
194 : static int
195 0 : fd_snapshot_dumper_advance( fd_snapshot_dumper_t * dumper ) {
196 :
197 0 : int advance_err = fd_snapshot_loader_advance( dumper->loader );
198 0 : if( FD_UNLIKELY( advance_err ) ) {
199 0 : if( advance_err==MANIFEST_DONE ) return 0;
200 0 : if( advance_err>0 ) FD_LOG_WARNING(( "fd_snapshot_loader_advance() failed (%d)", advance_err ));
201 0 : return advance_err;
202 0 : }
203 :
204 0 : int collect_err = fd_snapshot_dumper_release( dumper );
205 0 : if( FD_UNLIKELY( collect_err ) ) return collect_err;
206 :
207 0 : return 0;
208 0 : }
209 :
210 : /* fd_snapshot_dump_args_t contains the command-line arguments for the
211 : dump command. */
212 :
213 : struct fd_snapshot_dump_args {
214 : char const * _page_sz;
215 : ulong page_cnt;
216 : ulong near_cpu;
217 : ulong zstd_window_sz;
218 : char * snapshot;
219 : char const * csv_path;
220 : int csv_hdr;
221 : ushort http_redirs;
222 : };
223 :
224 : typedef struct fd_snapshot_dump_args fd_snapshot_dump_args_t;
225 :
226 : static int
227 : do_dump( fd_snapshot_dumper_t * d,
228 : fd_snapshot_dump_args_t * args,
229 : fd_wksp_t * wksp,
230 0 : fd_spad_t * spad ) {
231 :
232 : /* Resolve snapshot source */
233 :
234 0 : fd_snapshot_src_t src[1];
235 0 : src->snapshot_dir = NULL;
236 0 : if( FD_UNLIKELY( !fd_snapshot_src_parse_type_unknown( src, args->snapshot ) ) )
237 0 : return EXIT_FAILURE;
238 :
239 : /* Create a heap */
240 :
241 0 : ulong const fd_alloc_tag = 41UL;
242 0 : d->alloc = fd_alloc_join( fd_alloc_new( fd_wksp_alloc_laddr( wksp, fd_alloc_align(), fd_alloc_footprint(), fd_alloc_tag ), fd_alloc_tag ), 0UL );
243 0 : if( FD_UNLIKELY( !d->alloc ) ) { FD_LOG_WARNING(( "fd_alloc_join() failed" )); return EXIT_FAILURE; }
244 :
245 0 : fd_wksp_usage_t wksp_usage[1] = {0};
246 0 : fd_wksp_usage( wksp, NULL, 0UL, wksp_usage );
247 :
248 0 : if( args->csv_path ) {
249 0 : d->csv_fd = open( args->csv_path, O_WRONLY|O_CREAT|O_TRUNC, 0644 );
250 0 : if( FD_UNLIKELY( d->csv_fd<0 ) ) { FD_LOG_WARNING(( "open(%s) failed (%d-%s)", args->csv_path, errno, fd_io_strerror( errno ) )); return EXIT_FAILURE; }
251 0 : fd_io_buffered_ostream_init( &d->csv_out, d->csv_fd, d->csv_buf, OSTREAM_BUFSZ );
252 0 : }
253 :
254 : /* Create loader */
255 :
256 0 : d->loader = fd_snapshot_loader_new( fd_spad_alloc( spad, fd_snapshot_loader_align(), fd_snapshot_loader_footprint( args->zstd_window_sz ) ), args->zstd_window_sz );
257 0 : if( FD_UNLIKELY( !d->loader ) ) { FD_LOG_WARNING(( "Failed to create fd_snapshot_loader_t" )); return EXIT_FAILURE; }
258 :
259 : /* Create a high-quality hash seed for fd_funk */
260 :
261 0 : ulong funk_seed;
262 0 : if( FD_UNLIKELY( sizeof(ulong)!=getrandom( &funk_seed, sizeof(ulong), 0 ) ) )
263 0 : { FD_LOG_WARNING(( "getrandom() failed (%d-%s)", errno, fd_io_strerror( errno ) )); return EXIT_FAILURE; }
264 :
265 : /* Create a funk database */
266 :
267 0 : ulong const txn_max = 16UL; /* we really only need 1 */
268 0 : uint const rec_max = 1024UL; /* we evict records as we go */
269 :
270 0 : ulong funk_tag = 42UL;
271 0 : int funk_ok = !!fd_funk_join( d->funk, fd_funk_new( fd_wksp_alloc_laddr( wksp, fd_funk_align(), fd_funk_footprint(txn_max, rec_max), funk_tag ), funk_tag, funk_seed, txn_max, rec_max ) );
272 0 : if( FD_UNLIKELY( !funk_ok ) ) { FD_LOG_WARNING(( "Failed to create fd_funk_t" )); return EXIT_FAILURE; }
273 :
274 : /* Create a new processing context */
275 :
276 0 : d->slot_ctx = fd_exec_slot_ctx_join( fd_exec_slot_ctx_new( fd_spad_alloc( spad, FD_EXEC_SLOT_CTX_ALIGN, FD_EXEC_SLOT_CTX_FOOTPRINT ) ) );
277 0 : if( FD_UNLIKELY( !d->slot_ctx ) ) { FD_LOG_WARNING(( "Failed to create fd_exec_slot_ctx_t" )); return EXIT_FAILURE; }
278 :
279 : /* funk_txn is destroyed automatically when deleting fd_funk_t. */
280 :
281 0 : fd_funk_txn_xid_t funk_txn_xid = { .ul = { 1UL } };
282 0 : fd_funk_txn_t * funk_txn = fd_funk_txn_prepare( d->funk, NULL, &funk_txn_xid, 1 );
283 0 : d->slot_ctx->funk_txn = funk_txn;
284 :
285 0 : void * restore_mem = fd_spad_alloc( spad, fd_snapshot_restore_align(), fd_snapshot_restore_footprint() );
286 0 : if( FD_UNLIKELY( !restore_mem ) ) FD_LOG_ERR(( "Failed to allocate restore buffer" )); /* unreachable */
287 :
288 0 : d->restore = fd_snapshot_restore_new( restore_mem, d->funk, funk_txn, spad, d, NULL, NULL );
289 0 : if( FD_UNLIKELY( !d->restore ) ) { FD_LOG_WARNING(( "Failed to create fd_snapshot_restore_t" )); return EXIT_FAILURE; }
290 :
291 : /* Set up the snapshot loader */
292 :
293 0 : if( FD_UNLIKELY( !fd_snapshot_loader_init( d->loader, d->restore, src, 0UL, 0 ) ) ) {
294 0 : FD_LOG_WARNING(( "fd_snapshot_loader_init failed" ));
295 0 : return EXIT_FAILURE;
296 0 : }
297 :
298 0 : d->want_accounts = (!!args->csv_path);
299 :
300 0 : if( FD_UNLIKELY( !d->want_accounts ) ) {
301 0 : FD_LOG_NOTICE(( "Nothing to do, exiting." ));
302 0 : return EXIT_SUCCESS;
303 0 : }
304 :
305 0 : if( (d->csv_fd>=0) & (args->csv_hdr) ) {
306 0 : fd_snapshot_csv_rec_t csv_rec;
307 0 : memset( &csv_rec, ' ', sizeof(fd_snapshot_csv_rec_t) );
308 0 : memcpy( csv_rec.acct_addr, "address", strlen( "address" ) );
309 0 : memcpy( csv_rec.owner_addr, "owner", strlen( "owner" ) );
310 0 : memcpy( csv_rec.hash, "hash", strlen( "hash" ) );
311 0 : memcpy( csv_rec.slot, "slot", strlen( "slot" ) );
312 0 : memcpy( csv_rec.size, "size", strlen( "size" ) );
313 0 : memcpy( csv_rec.lamports, "lamports", strlen( "lamports" ) );
314 0 : csv_rec.comma1 = ',';
315 0 : csv_rec.comma2 = ',';
316 0 : csv_rec.comma3 = ',';
317 0 : csv_rec.comma4 = ',';
318 0 : csv_rec.comma5 = ',';
319 0 : csv_rec.newline = '\n';
320 :
321 0 : if( FD_UNLIKELY( write( d->csv_fd, csv_rec.line, sizeof(fd_snapshot_csv_rec_t) )
322 0 : != sizeof(fd_snapshot_csv_rec_t) ) ) {
323 0 : FD_LOG_WARNING(( "Failed to write CSV header (%d-%s)", errno, fd_io_strerror( errno ) ));
324 0 : d->has_fail = 1;
325 0 : return EXIT_FAILURE;
326 0 : }
327 0 : }
328 :
329 0 : for(;;) {
330 0 : int err = fd_snapshot_dumper_advance( d );
331 0 : if( err==0 ) { /* ok */ }
332 0 : else if( err<0 ) { /* EOF */ break; }
333 0 : else { return EXIT_FAILURE; }
334 :
335 0 : if( FD_UNLIKELY( !d->want_accounts ) )
336 0 : break;
337 0 : }
338 :
339 0 : return d->has_fail ? EXIT_FAILURE : EXIT_SUCCESS;
340 0 : }
341 :
342 : int
343 : cmd_dump( int argc,
344 0 : char ** argv ) {
345 :
346 0 : fd_snapshot_dump_args_t args[1] = {{0}};
347 0 : args->_page_sz = fd_env_strip_cmdline_cstr ( &argc, &argv, "--page-sz", NULL, "gigantic" );
348 0 : args->page_cnt = fd_env_strip_cmdline_ulong ( &argc, &argv, "--page-cnt", NULL, 5UL );
349 0 : args->near_cpu = fd_env_strip_cmdline_ulong ( &argc, &argv, "--near-cpu", NULL, fd_log_cpu_id() );
350 0 : args->zstd_window_sz = fd_env_strip_cmdline_ulong ( &argc, &argv, "--zstd-window-sz", NULL, 33554432UL );
351 0 : args->snapshot = (char *)fd_env_strip_cmdline_cstr ( &argc, &argv, "--snapshot", NULL, NULL );
352 0 : args->csv_path = fd_env_strip_cmdline_cstr ( &argc, &argv, "--csv", NULL, NULL );
353 0 : args->csv_hdr = fd_env_strip_cmdline_int ( &argc, &argv, "--csv-hdr", NULL, 1 );
354 0 : args->http_redirs = (ushort)fd_env_strip_cmdline_ushort( &argc, &argv, "--http-redirs", NULL, 5 );
355 :
356 0 : if( FD_UNLIKELY( argc!=1 ) )
357 0 : FD_LOG_ERR(( "Unexpected command-line arguments" ));
358 0 : if( FD_UNLIKELY( !args->snapshot ) )
359 0 : FD_LOG_ERR(( "Missing --snapshot argument" ));
360 :
361 0 : FD_LOG_NOTICE(( "Creating workspace (--page-cnt %lu, --page-sz %s)", args->page_cnt, args->_page_sz ));
362 :
363 : /* With workspace */
364 :
365 0 : fd_wksp_t * wksp = fd_wksp_new_anonymous( fd_cstr_to_shmem_page_sz( args->_page_sz ), args->page_cnt, args->near_cpu, "wksp", 0UL );
366 0 : if( FD_UNLIKELY( !wksp ) ) FD_LOG_ERR(( "fd_wksp_new_anonymous() failed" ));
367 :
368 : /* With spad */
369 :
370 0 : ulong mem_max = args->zstd_window_sz + (1UL<<32); /* manifest plus 4 GiB headroom */
371 0 : uchar * mem = fd_wksp_alloc_laddr( wksp, FD_SPAD_ALIGN, FD_SPAD_FOOTPRINT( mem_max ), 1UL );
372 0 : fd_spad_t * spad = fd_spad_join( fd_spad_new( mem, mem_max ) );
373 0 : if( FD_UNLIKELY( !spad ) ) {
374 0 : FD_LOG_ERR(( "Failed to allocate spad" ));
375 0 : }
376 0 : fd_spad_push( spad );
377 :
378 : /* With dump context */
379 :
380 0 : fd_snapshot_dumper_t _dumper[1];
381 0 : fd_snapshot_dumper_t * dumper = fd_snapshot_dumper_new( _dumper );
382 :
383 0 : int rc = do_dump( dumper, args, wksp, spad );
384 0 : FD_LOG_INFO(( "Done. Cleaning up." ));
385 :
386 0 : fd_snapshot_dumper_delete( dumper );
387 :
388 0 : fd_spad_pop( spad );
389 0 : void * spad_mem = fd_spad_delete( fd_spad_leave( spad ) );
390 0 : fd_wksp_free_laddr( spad_mem );
391 :
392 0 : fd_wksp_delete_anonymous( wksp );
393 0 : return rc;
394 0 : }
395 :
396 : FD_IMPORT_CSTR( _help, "src/flamenco/snapshot/fd_snapshot_help.txt" );
397 :
398 : __attribute__((noreturn)) static int
399 0 : usage( int code ) {
400 0 : fwrite( _help, 1, _help_sz, stderr );
401 0 : fflush( stderr );
402 0 : exit( code );
403 0 : }
404 :
405 : int
406 : main( int argc,
407 : char ** argv ) {
408 : fd_boot( &argc, &argv );
409 :
410 : if( argc==1 ) return usage(1);
411 : if( 0==strcmp( argv[1], "help" ) ) return usage(0);
412 : for( int i=1; i<argc; i++ )
413 : if( 0==strcmp( argv[i], "--help" ) )
414 : return usage(0);
415 :
416 : argc--; argv++;
417 : char const * cmd = argv[0];
418 :
419 : if( 0==strcmp( cmd, "dump" ) ) {
420 : return cmd_dump( argc, argv );
421 : } else {
422 : fprintf( stderr, "Unknown command: %s\n", cmd );
423 : return usage(1);
424 : }
425 :
426 : fd_halt();
427 : return 0;
428 : }
|
