Line data Source code
1 : #ifndef HEADER_fd_src_discof_backup_fd_backup_disk_h
2 : #define HEADER_fd_src_discof_backup_fd_backup_disk_h
3 :
4 : /* fd_backup_disk.h does streaming parsing of accdb disk partitions.
5 : Publishes discovered accounts (by disk offset) onto mcache/dcache. */
6 :
7 : #include "fd_backup.h"
8 : #include "fd_backup_accidx.h"
9 : #include "fd_backup_shmem.h"
10 : #include "../../tango/fd_tango_base.h"
11 :
12 : /* FD_SNAPMK_PF_LEAD is how far ahead of the record cursor the parser
13 : prefetches disk bytes. */
14 :
15 0 : #define FD_SNAPMK_PF_LEAD 4096UL
16 :
17 : /* fd_snapmk_accparse_t does streaming zero-copy parsing of accdb
18 : partitions. Ingests a stream of disk data (arbitrarily fragmented)
19 : and produces an account-aligned fragment stream. */
20 :
21 : struct fd_snapmk_accparse {
22 :
23 : /* current input frag */
24 : uchar const * data;
25 : ulong data_sz;
26 : ulong src_gaddr;
27 : ulong src_off; /* accdb file offset of data */
28 : ulong frag_base_gaddr; /* src_gaddr at start of current frag */
29 : uchar const * pf_cursor; /* prefetch high-water mark within current frag */
30 : int input_active;
31 :
32 : /* record being parsed */
33 : uint meta_sz; /* header bytes buffered so far, if torn */
34 : int acc_active;
35 : uint acc_off; /* account data bytes consumed so far */
36 : uint acc_sz; /* account data byte count */
37 : uint acc_snap_sz; /* account byte count in snapshot format */
38 : uint acc_idx; /* index entry, UINT_MAX if not in the snapshot */
39 : uint acc_keep;
40 : ulong acc_file_off; /* accdb file offset of the record header */
41 :
42 : /* output frag staged by fd_snapmk_accparse_publish, drained on its
43 : next call (the rest of the frag is derived from the fields above,
44 : which hold still until the record completes) */
45 : ulong pub_gaddr;
46 : uint pub_sz;
47 : int pub_pending;
48 : int pub_som;
49 : int pub_eom;
50 :
51 : fd_backup_accidx_t idx;
52 : visited_set_t * visited_set;
53 :
54 : /* defrag buffer for torn disk partition data */
55 : union __attribute__((packed)) {
56 : uchar buf[ sizeof(fd_accdb_disk_meta_t) ];
57 : fd_accdb_disk_meta_t meta;
58 : };
59 :
60 : /* Whole records parsed out of the current frag, awaiting index
61 : lookup. Staged one batch ahead of the batch being resolved so the
62 : chain head prefetches issued here land during that walk. */
63 : uint ps_cnt;
64 : ulong ps_chain_idx[ FD_BACKUP_DISK_PARA ];
65 : uint ps_frag_off [ FD_BACKUP_DISK_PARA ];
66 : ulong ps_file_off [ FD_BACKUP_DISK_PARA ];
67 : };
68 :
69 : typedef struct fd_snapmk_accparse fd_snapmk_accparse_t;
70 :
71 : FD_PROTOTYPES_BEGIN
72 :
73 : /* Implementation */
74 :
75 : /* fd_snapmk_accparse_lookup resolves cnt parsed disk records against
76 : the accdb index. chain_idx[ i ] is the acc_map chain that record i's
77 : address hashes to, file_off[ i ] its accdb file offset. Sets
78 : acc_idx[ i ] to the index entry to copy into the snapshot, or
79 : UINT_MAX if the record should be skipped.
80 :
81 : A record is claimed by the index entry pointing at its exact file
82 : offset. accdb file offsets are unique. The record is skipped if
83 : no entry points at it (a superseded version still on disk), if the
84 : entry is unrooted or a tombstone, or if the account was already
85 : emitted from cache or from another partition. */
86 :
87 : static inline void
88 : fd_snapmk_accparse_lookup( fd_snapmk_accparse_t * parse,
89 : ulong const * chain_idx,
90 : ulong const * file_off,
91 : uint * acc_idx,
92 0 : ulong cnt ) {
93 0 : fd_backup_accidx_t * idx = &parse->idx;
94 0 : fd_accdb_accmeta_t const * acc_pool = idx->acc_pool;
95 :
96 0 : uint cur[ FD_BACKUP_DISK_PARA ]; /* chain node lane i sits on */
97 :
98 0 : FD_COMPILER_MFENCE();
99 0 : FD_VOLATILE( *idx->epoch_slot ) = FD_VOLATILE_CONST( *idx->epoch );
100 0 : FD_HW_MFENCE();
101 :
102 : /* Chain heads were prefetched when this batch was prestaged, a full
103 : batch resolve ago. */
104 0 : for( ulong i=0UL; i<cnt; i++ ) {
105 0 : acc_idx[ i ] = UINT_MAX;
106 0 : cur [ i ] = FD_VOLATILE_CONST( idx->acc_map[ chain_idx[ i ] ] );
107 0 : }
108 :
109 0 : int any;
110 0 : do {
111 0 : any = 0;
112 0 : for( ulong i=0UL; i<cnt; i++ ) {
113 0 : uint ele = cur[ i ];
114 0 : if( !fd_backup_accidx_valid( idx, ele ) ) continue; /* lane retired */
115 :
116 0 : fd_accdb_accmeta_t const * m = &acc_pool[ ele ];
117 0 : uint next = FD_VOLATILE_CONST( m->map.next );
118 0 : uint gen = FD_VOLATILE_CONST( m->key.generation );
119 0 : ulong off = FD_VOLATILE_CONST( m->offset_fork );
120 0 : ulong lam = FD_VOLATILE_CONST( m->lamports );
121 :
122 0 : int hit = fd_backup_accidx_rooted( idx, gen, lam )
123 0 : & ( ( off & FD_ACCDB_OFF_MASK )==file_off[ i ] );
124 0 : fd_uint_store_if( hit, &acc_idx[ i ], ele );
125 :
126 : /* Resolving hit and next in the same pass keeps the next node
127 : prefetch to the lanes that will actually use it. */
128 0 : int cont = (!hit) & fd_backup_accidx_valid( idx, next );
129 0 : cur[ i ] = fd_uint_if( cont, next, UINT_MAX );
130 0 : any |= cont;
131 0 : __builtin_prefetch( &acc_pool[ next & (uint)(-cont) ], 0, 3 );
132 0 : }
133 0 : } while( any );
134 :
135 0 : FD_COMPILER_MFENCE();
136 0 : FD_VOLATILE( *idx->epoch_slot ) = ULONG_MAX;
137 :
138 : /* Claim each account for this snapshot, dropping the records that
139 : lost the claim to an earlier sighting. */
140 0 : for( ulong i=0UL; i<cnt; i++ ) {
141 0 : uint ele = acc_idx[ i ];
142 0 : if( FD_UNLIKELY( ele==UINT_MAX ) ) continue;
143 0 : if( FD_UNLIKELY( fd_backup_visited_test_and_set( parse->visited_set, (ulong)ele ) ) ) {
144 0 : acc_idx[ i ] = UINT_MAX;
145 0 : }
146 0 : }
147 0 : }
148 :
149 : /* fd_snapmk_accparse_keep returns 1 if the record whose header is in
150 : parse->meta should be copied into the snapshot being produced, 0
151 : otherwise, and sets parse->acc_idx to its index entry. This is the
152 : fallback for records that straddle an input frag; whole records go
153 : through fd_snapmk_accparse_publish_batch below. */
154 :
155 : static inline int
156 0 : fd_snapmk_accparse_keep( fd_snapmk_accparse_t * parse ) {
157 0 : if( FD_UNLIKELY( parse->meta.generation>parse->idx.root_generation ) ) {
158 0 : parse->acc_idx = UINT_MAX;
159 0 : return 0;
160 0 : }
161 0 : ulong chain_idx = fd_backup_accidx_chain( &parse->idx, parse->meta.pubkey );
162 0 : fd_snapmk_accparse_lookup( parse, &chain_idx, &parse->acc_file_off, &parse->acc_idx, 1UL );
163 0 : return parse->acc_idx!=UINT_MAX;
164 0 : }
165 :
166 : /* fd_snapmk_accparse_publish produces an account-aligned frag from
167 : accumulated source data. It returns meta if a frag was produced and
168 : NULL otherwise. In particular, after discarding a complete record it
169 : returns NULL at the clean record boundary so the caller can retry the
170 : whole-record batch path on the remaining input. */
171 :
172 : static inline fd_frag_meta_t *
173 : fd_snapmk_accparse_publish( fd_snapmk_accparse_t * parse,
174 0 : fd_frag_meta_t * meta ) {
175 0 : for(;;) {
176 0 : if( FD_UNLIKELY( parse->pub_pending ) ) {
177 0 : meta->sig = parse->pub_gaddr;
178 0 : meta->chunk = parse->acc_idx;
179 0 : meta->sz = 0;
180 0 : meta->ctl = (ushort)fd_frag_meta_ctl( FD_BACKUP_ORIG_ACC_DISK, parse->pub_som, parse->pub_eom, 0 );
181 0 : meta->tsorig = 0U;
182 0 : meta->tspub = (uint)parse->pub_sz;
183 0 : parse->pub_pending = 0;
184 0 : return meta;
185 0 : }
186 :
187 0 : if( FD_UNLIKELY( !parse->data_sz ) ) return NULL;
188 :
189 0 : if( FD_UNLIKELY( !parse->acc_active ) ) {
190 0 : if( FD_UNLIKELY( !parse->meta_sz ) ) {
191 0 : parse->acc_file_off = parse->src_off;
192 0 : parse->acc_snap_sz = 0U;
193 0 : parse->acc_idx = UINT_MAX;
194 0 : parse->acc_keep = 1U;
195 0 : }
196 :
197 0 : ulong meta_rem = sizeof(fd_accdb_disk_meta_t) - (ulong)parse->meta_sz;
198 0 : ulong take = fd_ulong_min( meta_rem, parse->data_sz );
199 0 : fd_memcpy( parse->buf + parse->meta_sz, parse->data, take );
200 0 : parse->meta_sz += (uint)take;
201 0 : parse->data += take;
202 0 : parse->data_sz -= take;
203 0 : parse->src_gaddr += take;
204 0 : parse->src_off += take;
205 :
206 0 : if( FD_UNLIKELY( parse->meta_sz < sizeof(fd_accdb_disk_meta_t) ) ) continue;
207 :
208 0 : ulong data_sz = (ulong)FD_ACCDB_SIZE_DATA( parse->meta.size );
209 0 : ulong snap_sz = sizeof(snap_acc_hdr_t) + fd_ulong_align_up( data_sz, 8UL );
210 0 : if( FD_UNLIKELY( data_sz>UINT_MAX ) ) {
211 0 : FD_LOG_CRIT(( "accdb disk account data too large (%lu bytes)", data_sz ));
212 0 : }
213 0 : if( FD_UNLIKELY( snap_sz>UINT_MAX ) ) {
214 0 : FD_LOG_CRIT(( "snapshot account record too large (%lu bytes)", snap_sz ));
215 0 : }
216 :
217 0 : parse->acc_active = 1;
218 0 : parse->acc_off = 0U;
219 0 : parse->acc_sz = (uint)data_sz;
220 0 : parse->acc_snap_sz = (uint)snap_sz;
221 0 : parse->meta_sz = 0U;
222 0 : parse->acc_keep = (uint)fd_snapmk_accparse_keep( parse );
223 :
224 0 : if( FD_UNLIKELY( !parse->acc_sz ) ) {
225 0 : if( FD_LIKELY( parse->acc_keep ) ) {
226 0 : parse->pub_gaddr = 0UL;
227 0 : parse->pub_sz = 0U;
228 0 : parse->pub_som = 1;
229 0 : parse->pub_eom = 1;
230 0 : parse->pub_pending = 1;
231 0 : }
232 0 : parse->acc_active = 0;
233 0 : parse->acc_off = 0U;
234 0 : parse->acc_sz = 0U;
235 0 : if( FD_UNLIKELY( !parse->acc_keep ) ) return NULL;
236 0 : continue;
237 0 : }
238 :
239 0 : continue;
240 0 : }
241 :
242 0 : ulong acc_rem = (ulong)parse->acc_sz - (ulong)parse->acc_off;
243 0 : ulong take = fd_ulong_min( acc_rem, parse->data_sz );
244 0 : if( FD_UNLIKELY( !take ) ) return NULL;
245 :
246 0 : if( FD_UNLIKELY( !parse->acc_keep ) ) {
247 0 : parse->acc_off += (uint)take;
248 0 : parse->data += take;
249 0 : parse->data_sz -= take;
250 0 : parse->src_gaddr += take;
251 0 : parse->src_off += take;
252 0 : if( FD_UNLIKELY( parse->acc_off==parse->acc_sz ) ) {
253 0 : parse->acc_active = 0;
254 0 : parse->acc_off = 0U;
255 0 : parse->acc_sz = 0U;
256 0 : parse->acc_keep = 1U;
257 0 : return NULL;
258 0 : }
259 0 : continue;
260 0 : }
261 :
262 0 : uint old_acc_off = parse->acc_off;
263 0 : parse->pub_gaddr = parse->src_gaddr;
264 0 : parse->pub_sz = (uint)take;
265 0 : parse->pub_som = !old_acc_off;
266 0 : parse->pub_eom = ( old_acc_off + take )==parse->acc_sz;
267 0 : parse->pub_pending = 1;
268 :
269 0 : parse->acc_off += (uint)take;
270 0 : parse->data += take;
271 0 : parse->data_sz -= take;
272 0 : parse->src_gaddr += take;
273 0 : parse->src_off += take;
274 :
275 0 : if( FD_UNLIKELY( parse->pub_eom ) ) {
276 0 : parse->acc_active = 0;
277 0 : parse->acc_sz = 0U;
278 0 : parse->acc_off = 0U;
279 0 : }
280 0 : }
281 0 : }
282 :
283 : /* fd_snapmk_accparse_prestage consumes up to FD_BACKUP_DISK_PARA whole
284 : records from the current frag into parse->ps_*, and prefetches the
285 : index lines their lookup will need. Does nothing unless the prestage
286 : is empty and the parser sits on a clean record boundary. */
287 :
288 : static inline void
289 0 : fd_snapmk_accparse_prestage( fd_snapmk_accparse_t * parse ) {
290 0 : if( FD_UNLIKELY( parse->ps_cnt ) ) return;
291 0 : if( FD_UNLIKELY( parse->pub_pending || parse->acc_active || parse->meta_sz ) ) return;
292 :
293 0 : ulong const meta_sz = sizeof(fd_accdb_disk_meta_t);
294 :
295 0 : fd_backup_accidx_t const * idx = &parse->idx;
296 0 : uint const * acc_map = idx->acc_map;
297 0 : fd_accdb_accmeta_t const * acc_pool = idx->acc_pool;
298 :
299 0 : ulong * chain_idx = parse->ps_chain_idx;
300 0 : ulong n = 0UL;
301 :
302 0 : while( n<FD_BACKUP_DISK_PARA ) {
303 0 : if( parse->data_sz < meta_sz ) break; /* partial meta straddles frag end */
304 :
305 : /* Prefetch a fixed window ahead of the record cursor. pf_cursor is
306 : a per-frag high-water mark, so each line is prefetched once no
307 : matter how the records fall across the window. */
308 0 : uchar const * pf_lim = parse->data + fd_ulong_min( parse->data_sz, FD_SNAPMK_PF_LEAD );
309 0 : uchar const * pf = fd_ptr_if( parse->pf_cursor>parse->data, parse->pf_cursor, parse->data );
310 0 : for( ; pf<pf_lim; pf+=64UL ) __builtin_prefetch( pf, 0, 2 );
311 0 : parse->pf_cursor = pf_lim;
312 :
313 0 : fd_accdb_disk_meta_t const * dm = (fd_accdb_disk_meta_t const *)parse->data;
314 0 : ulong data_len = (ulong)FD_ACCDB_SIZE_DATA( dm->size );
315 0 : ulong rec = meta_sz + data_len;
316 0 : if( parse->data_sz < rec ) break; /* account data straddles frag end */
317 :
318 0 : if( FD_LIKELY( dm->generation <= idx->root_generation ) ) {
319 0 : chain_idx[ n ] = fd_backup_accidx_chain( idx, dm->pubkey );
320 0 : __builtin_prefetch( &acc_map[ chain_idx[ n ] ], 0, 3 );
321 0 : parse->ps_frag_off[ n ] = (uint)( parse->src_gaddr - parse->frag_base_gaddr );
322 0 : parse->ps_file_off[ n ] = parse->src_off;
323 0 : n++;
324 0 : }
325 :
326 0 : parse->data += rec;
327 0 : parse->data_sz -= rec;
328 0 : parse->src_gaddr += rec;
329 0 : parse->src_off += rec;
330 0 : }
331 :
332 : /* Head pass: acc_map lines were prefetched during the record walk
333 : above; prefetch the acc_pool[head] lines here. */
334 0 : for( ulong i=0UL; i<n; i++ ) {
335 0 : uint head = FD_VOLATILE_CONST( acc_map[ chain_idx[ i ] ] );
336 0 : int live = fd_backup_accidx_valid( idx, head );
337 0 : __builtin_prefetch( &acc_pool[ head & (uint)(-live) ], 0, 3 );
338 0 : }
339 :
340 0 : parse->ps_cnt = (uint)n;
341 0 : }
342 :
343 : /* fd_snapmk_accparse_publish_batch fills batch with the next run of
344 : whole records from the current frag and returns how many. Returns 0
345 : once the frag holds no more whole records, leaving any straddling
346 : remainder to fd_snapmk_accparse_publish. Frag offsets in the batch
347 : are relative to parse->frag_base_gaddr. */
348 :
349 : static inline ulong
350 : fd_snapmk_accparse_publish_batch( fd_snapmk_accparse_t * parse,
351 0 : fd_backup_disk_batch_msg_t * batch ) {
352 0 : fd_snapmk_accparse_prestage( parse );
353 :
354 0 : ulong n = (ulong)parse->ps_cnt;
355 0 : if( !n ) return 0UL;
356 :
357 : /* Move batch N out of the prestage buffer, then stage N+1 before
358 : resolving N so N+1's prefetches overlap N's chain walk. */
359 0 : ulong chain_idx[ FD_BACKUP_DISK_PARA ];
360 0 : ulong file_off [ FD_BACKUP_DISK_PARA ];
361 0 : memcpy( chain_idx, parse->ps_chain_idx, n*sizeof(ulong) );
362 0 : memcpy( file_off, parse->ps_file_off, n*sizeof(ulong) );
363 0 : memcpy( batch->frag_off, parse->ps_frag_off, n*sizeof(uint) );
364 0 : parse->ps_cnt = 0U;
365 :
366 0 : fd_snapmk_accparse_prestage( parse );
367 :
368 0 : fd_snapmk_accparse_lookup( parse, chain_idx, file_off, batch->acc_idx, n );
369 0 : for( ulong i=n; i<FD_BACKUP_DISK_PARA; i++ ) batch->acc_idx[ i ] = UINT_MAX;
370 :
371 0 : return n;
372 0 : }
373 :
374 : FD_PROTOTYPES_END
375 :
376 : #endif /* HEADER_fd_src_discof_backup_fd_backup_disk_h */
|
