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 :
11 : /* FD_SNAPMK_PF_LEAD is how far ahead of the record cursor the parser
12 : prefetches disk bytes. */
13 :
14 9 : #define FD_SNAPMK_PF_LEAD 4096UL
15 :
16 : /* fd_snapmk_accparse_t does streaming zero-copy parsing of accdb
17 : partitions. Ingests a stream of disk data (arbitrarily fragmented)
18 : and produces an account-aligned fragment stream. */
19 :
20 : struct fd_snapmk_accparse {
21 :
22 : /* current input frag */
23 : uchar const * data;
24 : ulong data_sz;
25 : ulong src_gaddr;
26 : ulong src_off; /* accdb file offset of data */
27 : ulong frag_base_gaddr; /* src_gaddr at start of current frag */
28 : uchar const * pf_cursor; /* prefetch high-water mark within current frag */
29 : int input_active;
30 :
31 : /* record being parsed */
32 : uint meta_sz; /* header bytes buffered so far, if torn */
33 : int acc_active;
34 : uint acc_off; /* account data bytes consumed so far */
35 : uint acc_sz; /* account data byte count */
36 : uint acc_snap_sz; /* account byte count in snapshot format */
37 : uint acc_idx; /* index entry, UINT_MAX if not in the snapshot */
38 : uint acc_keep;
39 : ulong acc_file_off; /* accdb file offset of the record header */
40 :
41 : /* output frag staged by fd_snapmk_accparse_publish, drained on its
42 : next call (the rest of the frag is derived from the fields above,
43 : which hold still until the record completes) */
44 : ulong pub_gaddr;
45 : uint pub_sz;
46 : int pub_pending;
47 : int pub_som;
48 : int pub_eom;
49 :
50 : fd_backup_accidx_t idx;
51 : visited_set_t * visited_set;
52 :
53 : /* defrag buffer for torn disk partition data */
54 : union __attribute__((packed)) {
55 : uchar buf[ sizeof(fd_accdb_disk_meta_t) ];
56 : fd_accdb_disk_meta_t meta;
57 : };
58 :
59 : /* Whole records parsed out of the current frag, awaiting index
60 : lookup. Staged one batch ahead of the batch being resolved so the
61 : chain head prefetches issued here land during that walk. */
62 : uint ps_cnt;
63 : ulong ps_chain_idx[ FD_BACKUP_DISK_PARA ];
64 : uint ps_frag_off [ FD_BACKUP_DISK_PARA ];
65 : ulong ps_file_off [ FD_BACKUP_DISK_PARA ];
66 : };
67 :
68 : typedef struct fd_snapmk_accparse fd_snapmk_accparse_t;
69 :
70 : FD_PROTOTYPES_BEGIN
71 :
72 : /* Implementation */
73 :
74 : /* fd_snapmk_accparse_lookup resolves cnt parsed disk records against
75 : the accdb index. chain_idx[ i ] is the acc_map chain that record i's
76 : address hashes to, file_off[ i ] its accdb file offset. Sets
77 : acc_idx[ i ] to the index entry to copy into the snapshot, or
78 : UINT_MAX if the record should be skipped.
79 :
80 : A record is claimed by the index entry pointing at its exact file
81 : offset. accdb file offsets are unique. The record is skipped if
82 : no entry points at it (a superseded version still on disk), if the
83 : entry is unrooted or a tombstone, or if the account was already
84 : emitted from cache or from another partition. */
85 :
86 : static inline void
87 : fd_snapmk_accparse_lookup( fd_snapmk_accparse_t * parse,
88 : ulong const * chain_idx,
89 : ulong const * file_off,
90 : uint * acc_idx,
91 435 : ulong cnt ) {
92 435 : fd_backup_accidx_t * idx = &parse->idx;
93 435 : fd_accdb_accmeta_t const * acc_pool = idx->acc_pool;
94 :
95 435 : uint cur[ FD_BACKUP_DISK_PARA ]; /* chain node lane i sits on */
96 :
97 435 : FD_COMPILER_MFENCE();
98 435 : FD_VOLATILE( *idx->epoch_slot ) = FD_VOLATILE_CONST( *idx->epoch );
99 435 : FD_HW_MFENCE();
100 :
101 : /* Chain heads were prefetched when this batch was prestaged, a full
102 : batch resolve ago. */
103 1635 : for( ulong i=0UL; i<cnt; i++ ) {
104 1200 : acc_idx[ i ] = UINT_MAX;
105 1200 : cur [ i ] = FD_VOLATILE_CONST( idx->acc_map[ chain_idx[ i ] ] );
106 1200 : }
107 :
108 435 : int any;
109 714 : do {
110 714 : any = 0;
111 2955 : for( ulong i=0UL; i<cnt; i++ ) {
112 2241 : uint ele = cur[ i ];
113 2241 : if( !fd_backup_accidx_valid( idx, ele ) ) continue; /* lane retired */
114 :
115 1998 : fd_accdb_accmeta_t const * m = &acc_pool[ ele ];
116 1998 : uint next = FD_VOLATILE_CONST( m->map.next );
117 1998 : uint gen = FD_VOLATILE_CONST( m->key.generation );
118 1998 : ulong off = FD_VOLATILE_CONST( m->offset_fork );
119 1998 : ulong lam = FD_VOLATILE_CONST( m->lamports );
120 :
121 1998 : int hit = fd_backup_accidx_rooted( idx, gen, lam )
122 1998 : & ( ( off & FD_ACCDB_OFF_MASK )==file_off[ i ] );
123 1998 : fd_uint_store_if( hit, &acc_idx[ i ], ele );
124 :
125 : /* Resolving hit and next in the same pass keeps the next node
126 : prefetch to the lanes that will actually use it. */
127 1998 : int cont = (!hit) & fd_backup_accidx_valid( idx, next );
128 1998 : cur[ i ] = fd_uint_if( cont, next, UINT_MAX );
129 1998 : any |= cont;
130 1998 : __builtin_prefetch( &acc_pool[ next & (uint)(-cont) ], 0, 3 );
131 1998 : }
132 714 : } while( any );
133 :
134 435 : FD_COMPILER_MFENCE();
135 435 : FD_VOLATILE( *idx->epoch_slot ) = ULONG_MAX;
136 :
137 : /* Claim each account for this snapshot, dropping the records that
138 : lost the claim to an earlier sighting. */
139 1635 : for( ulong i=0UL; i<cnt; i++ ) {
140 1200 : uint ele = acc_idx[ i ];
141 1200 : if( FD_UNLIKELY( ele==UINT_MAX ) ) continue;
142 756 : if( FD_UNLIKELY( fd_backup_visited_test_and_set( parse->visited_set, (ulong)ele ) ) ) {
143 246 : acc_idx[ i ] = UINT_MAX;
144 246 : }
145 756 : }
146 435 : }
147 :
148 : /* fd_snapmk_accparse_keep returns 1 if the record whose header is in
149 : parse->meta should be copied into the snapshot being produced, 0
150 : otherwise, and sets parse->acc_idx to its index entry. This is the
151 : fallback for records that straddle an input frag; whole records go
152 : through fd_snapmk_accparse_publish_batch below. */
153 :
154 : static inline int
155 12 : fd_snapmk_accparse_keep( fd_snapmk_accparse_t * parse ) {
156 12 : if( FD_UNLIKELY( parse->meta.generation>parse->idx.root_generation ) ) {
157 3 : parse->acc_idx = UINT_MAX;
158 3 : return 0;
159 3 : }
160 9 : ulong chain_idx = fd_backup_accidx_chain( &parse->idx, parse->meta.pubkey );
161 9 : fd_snapmk_accparse_lookup( parse, &chain_idx, &parse->acc_file_off, &parse->acc_idx, 1UL );
162 9 : return parse->acc_idx!=UINT_MAX;
163 12 : }
164 :
165 : /* fd_snapmk_accparse_prestage consumes up to FD_BACKUP_DISK_PARA whole
166 : records from the current frag into parse->ps_*, and prefetches the
167 : index lines their lookup will need. Does nothing unless the prestage
168 : is empty and the parser sits on a clean record boundary. */
169 :
170 : static inline void
171 6 : fd_snapmk_accparse_prestage( fd_snapmk_accparse_t * parse ) {
172 6 : if( FD_UNLIKELY( parse->ps_cnt ) ) return;
173 6 : if( FD_UNLIKELY( parse->pub_pending || parse->acc_active || parse->meta_sz ) ) return;
174 :
175 6 : ulong const meta_sz = sizeof(fd_accdb_disk_meta_t);
176 :
177 6 : fd_backup_accidx_t const * idx = &parse->idx;
178 6 : uint const * acc_map = idx->acc_map;
179 6 : fd_accdb_accmeta_t const * acc_pool = idx->acc_pool;
180 :
181 6 : ulong * chain_idx = parse->ps_chain_idx;
182 6 : ulong n = 0UL;
183 :
184 15 : while( n<FD_BACKUP_DISK_PARA ) {
185 15 : if( parse->data_sz < meta_sz ) break; /* partial meta straddles frag end */
186 :
187 : /* Prefetch a fixed window ahead of the record cursor. pf_cursor is
188 : a per-frag high-water mark, so each line is prefetched once no
189 : matter how the records fall across the window. */
190 9 : uchar const * pf_lim = parse->data + fd_ulong_min( parse->data_sz, FD_SNAPMK_PF_LEAD );
191 9 : uchar const * pf = fd_ptr_if( parse->pf_cursor>parse->data, parse->pf_cursor, parse->data );
192 21 : for( ; pf<pf_lim; pf+=64UL ) __builtin_prefetch( pf, 0, 2 );
193 9 : parse->pf_cursor = pf_lim;
194 :
195 9 : fd_accdb_disk_meta_t const * dm = (fd_accdb_disk_meta_t const *)parse->data;
196 9 : ulong data_len = (ulong)FD_ACCDB_SIZE_DATA( dm->size );
197 9 : ulong rec = meta_sz + data_len;
198 9 : if( parse->data_sz < rec ) break; /* account data straddles frag end */
199 :
200 9 : if( FD_LIKELY( dm->generation <= idx->root_generation ) ) {
201 6 : chain_idx[ n ] = fd_backup_accidx_chain( idx, dm->pubkey );
202 6 : __builtin_prefetch( &acc_map[ chain_idx[ n ] ], 0, 3 );
203 6 : parse->ps_frag_off[ n ] = (uint)( parse->src_gaddr - parse->frag_base_gaddr );
204 6 : parse->ps_file_off[ n ] = parse->src_off;
205 6 : n++;
206 6 : }
207 :
208 9 : parse->data += rec;
209 9 : parse->data_sz -= rec;
210 9 : parse->src_gaddr += rec;
211 9 : parse->src_off += rec;
212 9 : }
213 :
214 : /* Head pass: acc_map lines were prefetched during the record walk
215 : above; prefetch the acc_pool[head] lines here. */
216 12 : for( ulong i=0UL; i<n; i++ ) {
217 6 : uint head = FD_VOLATILE_CONST( acc_map[ chain_idx[ i ] ] );
218 6 : int live = fd_backup_accidx_valid( idx, head );
219 6 : __builtin_prefetch( &acc_pool[ head & (uint)(-live) ], 0, 3 );
220 6 : }
221 :
222 6 : parse->ps_cnt = (uint)n;
223 6 : }
224 :
225 : /* fd_snapmk_accparse_publish_batch fills batch with the next run of
226 : whole records from the current frag and returns how many. Returns 0
227 : once the frag holds no more whole records, leaving any straddling
228 : remainder to fd_snapmk_accparse_publish. Frag offsets in the batch
229 : are relative to parse->frag_base_gaddr. */
230 :
231 : static inline ulong
232 : fd_snapmk_accparse_publish_batch( fd_snapmk_accparse_t * parse,
233 3 : fd_backup_disk_batch_msg_t * batch ) {
234 3 : fd_snapmk_accparse_prestage( parse );
235 :
236 3 : ulong n = (ulong)parse->ps_cnt;
237 3 : if( !n ) return 0UL;
238 :
239 : /* Move batch N out of the prestage buffer, then stage N+1 before
240 : resolving N so N+1's prefetches overlap N's chain walk. */
241 3 : ulong chain_idx[ FD_BACKUP_DISK_PARA ];
242 3 : ulong file_off [ FD_BACKUP_DISK_PARA ];
243 3 : memcpy( chain_idx, parse->ps_chain_idx, n*sizeof(ulong) );
244 3 : memcpy( file_off, parse->ps_file_off, n*sizeof(ulong) );
245 3 : memcpy( batch->frag_off, parse->ps_frag_off, n*sizeof(uint) );
246 3 : parse->ps_cnt = 0U;
247 :
248 3 : fd_snapmk_accparse_prestage( parse );
249 :
250 3 : fd_snapmk_accparse_lookup( parse, chain_idx, file_off, batch->acc_idx, n );
251 381 : for( ulong i=n; i<FD_BACKUP_DISK_PARA; i++ ) batch->acc_idx[ i ] = UINT_MAX;
252 :
253 3 : return n;
254 3 : }
255 :
256 : FD_PROTOTYPES_END
257 :
258 : #endif /* HEADER_fd_src_discof_backup_fd_backup_disk_h */
|
