Line data Source code
1 : #include "fd_txncache_writer.h"
2 : #include "../../flamenco/runtime/fd_txncache_private.h"
3 : #include "../../flamenco/runtime/sysvar/fd_sysvar_slot_history.h"
4 : #include "../../util/fd_util.h"
5 : #include "../../util/racesan/fd_racesan_target.h"
6 :
7 0 : #define STATE_HEADER 1
8 0 : #define STATE_SLOT 2
9 0 : #define STATE_GROUP 3
10 0 : #define STATE_ENTRIES 4
11 0 : #define STATE_DONE 5
12 :
13 : /* Wire sizes */
14 :
15 0 : #define HEADER_SZ (8UL) /* slot_deltas_len */
16 0 : #define SLOT_DELTA_SZ (8UL+1UL+8UL) /* slot, is_root, status_len */
17 0 : #define GROUP_HDR_SZ (32UL+8UL+8UL) /* blockhash, txnhash_offset, txn_cnt */
18 0 : #define TXN_SZ (sizeof(fd_txnhash_t)+4UL) /* txnhash, result */
19 :
20 0 : #define FD_TXNCACHE_WRITER_ARENA_MIN (64UL<<20)
21 :
22 0 : #define WALK_COUNT 0
23 0 : #define WALK_FILL 1
24 :
25 0 : #define WALK_ANOMALY_NONE 0
26 0 : #define WALK_ANOMALY_CHAIN_TXN_IDX_OOB 1
27 0 : #define WALK_ANOMALY_CYCLE 2
28 0 : #define WALK_ANOMALY_CAPTURED_FORK_NO_EXEC_SLOT 3
29 0 : #define WALK_ANOMALY_GROUP_COUNT_EXCEEDED 4
30 :
31 : FD_STATIC_ASSERT( FD_TXNCACHE_WRITER_CHECK_INTERVAL, check_interval_nonzero );
32 :
33 : FD_STATIC_ASSERT( FD_TXNCACHE_WRITER_MAX_SLOT_DELTAS<=USHORT_MAX, group_slot_i );
34 : FD_STATIC_ASSERT( FD_TXNCACHE_WRITER_MAX_BLOCKHASHES<=USHORT_MAX, group_blockhash_i );
35 :
36 : static inline ulong
37 0 : group_key( ulong slot_i, ulong blockhash_i ) {
38 0 : return slot_i*FD_TXNCACHE_WRITER_MAX_BLOCKHASHES+blockhash_i;
39 0 : }
40 :
41 : FD_FN_CONST ulong
42 0 : fd_txncache_writer_arena_align( void ) {
43 0 : return 64UL;
44 0 : }
45 :
46 : FD_FN_CONST ulong
47 0 : fd_txncache_writer_arena_sz( ulong max_txn_per_slot ) {
48 : /* Write at least 2 slots at once, and have at least 64 MiB. */
49 0 : return fd_ulong_max( FD_TXNCACHE_WRITER_ARENA_MIN, 2UL*2UL*max_txn_per_slot*sizeof(fd_txnhash_t) );
50 0 : }
51 :
52 : /* Returns the captured rooted descriptor of the fork the txn executed
53 : on, or NULL if that fork is not in the captured rooted set, either
54 : because it is unrooted or because it is a cancelled fork whose pool
55 : slot has since been reused. */
56 :
57 : static inline fd_txncache_writer_blockhash_desc_t const *
58 : writer_exec_desc( fd_txncache_writer_t const * writer,
59 0 : fd_txncache_single_txn_t const * txn ) {
60 0 : ulong desc_i = writer->fork_id_to_blockhash_i[ txn->fork_id.val ];
61 0 : if( FD_UNLIKELY( desc_i==USHORT_MAX ) ) return NULL;
62 0 : fd_txncache_writer_blockhash_desc_t const * desc = &writer->blockhash_descs[ desc_i ];
63 0 : return txn->generation==desc->generation ? desc : NULL;
64 0 : }
65 :
66 : static void
67 : writer_slots_init( fd_txncache_writer_t * writer,
68 : uchar const * slot_history,
69 0 : ulong slot_history_sz ) {
70 : /* Agave's verify_slot_history rejects a snapshot whose SlotHistory is
71 : not exactly MAX_ENTRIES bits ending at the bank's slot. */
72 :
73 0 : fd_slot_history_view_t view[1];
74 0 : if( FD_UNLIKELY( !fd_sysvar_slot_history_view( view, slot_history, slot_history_sz ) ) ) {
75 0 : FD_LOG_CRIT(( "malformed SlotHistory sysvar at slot %lu", writer->snapshot_slot ));
76 0 : }
77 0 : if( FD_UNLIKELY( view->bits_len!=FD_SLOT_HISTORY_MAX_ENTRIES ||
78 0 : view->blocks_len*64UL!=FD_SLOT_HISTORY_MAX_ENTRIES ||
79 0 : view->next_slot>writer->snapshot_slot+1UL ) ) {
80 0 : FD_LOG_CRIT(( "malformed SlotHistory sysvar at slot %lu (bits_len %lu blocks_len %lu next_slot %lu)",
81 0 : writer->snapshot_slot, view->bits_len, view->blocks_len, view->next_slot ));
82 0 : }
83 :
84 : /* The SlotHistory in the bank's sysvar cache is one slot stale: it
85 : is missing the bit for the snapshot slot itself, so the loop below
86 : forces that slot in. */
87 :
88 0 : ulong scan_cnt = fd_ulong_min( writer->snapshot_slot+1UL, FD_SLOT_HISTORY_MAX_ENTRIES );
89 0 : ulong cnt = 0UL;
90 0 : for( ulong i=0UL; i<scan_cnt && cnt<FD_TXNCACHE_WRITER_MAX_SLOT_DELTAS; i++ ) {
91 0 : ulong slot = writer->snapshot_slot-i;
92 0 : if( slot!=writer->snapshot_slot && fd_sysvar_slot_history_find_slot( view, slot )!=FD_SLOT_HISTORY_SLOT_FOUND ) continue;
93 0 : writer->slots[ FD_TXNCACHE_WRITER_MAX_SLOT_DELTAS-1UL-cnt ] = slot; /* fill from the back to get ascending order */
94 0 : cnt++;
95 0 : }
96 0 : FD_TEST( cnt ); /* i==0 always counts */
97 :
98 0 : memmove( writer->slots, writer->slots+FD_TXNCACHE_WRITER_MAX_SLOT_DELTAS-cnt, cnt*sizeof(ulong) );
99 0 : writer->slot_cnt = cnt;
100 0 : }
101 :
102 : /* Pairs rooted blockcache descriptors with execution slots by rank from
103 : the newest end. Every rooted block registered exactly one blockhash
104 : and set exactly one SlotHistory bit, and both sequences end at the
105 : snapshot slot. The initial genesis descriptor may precede the named
106 : execution slots; it remains usable as a referenced blockhash but has
107 : no execution slot mapping. */
108 : static int
109 0 : writer_map_execution_slots( fd_txncache_writer_t * writer ) {
110 0 : ulong blockhash_cnt = writer->blockhash_cnt;
111 0 : ulong slot_cnt = writer->slot_cnt;
112 0 : if( FD_UNLIKELY( blockhash_cnt>slot_cnt+1UL ) ) {
113 0 : FD_LOG_WARNING(( "txncache has %lu rooted blockhash descriptors but SlotHistory names only %lu execution slots", blockhash_cnt, slot_cnt ));
114 0 : return 0;
115 0 : }
116 :
117 0 : memset( writer->fork_id_to_blockhash_i, 0xFF, sizeof(writer->fork_id_to_blockhash_i) );
118 0 : for( ulong blockhash_i=0UL; blockhash_i<blockhash_cnt; blockhash_i++ ) {
119 0 : fd_txncache_writer_blockhash_desc_t * desc = &writer->blockhash_descs[ blockhash_i ];
120 0 : desc->slot_i = USHORT_MAX;
121 0 : writer->fork_id_to_blockhash_i[ desc->blockcache_idx ] = (ushort)blockhash_i;
122 0 : }
123 :
124 0 : ulong mapped_cnt = fd_ulong_min( blockhash_cnt, slot_cnt );
125 0 : ulong blockhash_i0 = blockhash_cnt-mapped_cnt;
126 0 : ulong slot_i0 = slot_cnt -mapped_cnt;
127 0 : for( ulong pair_i=0UL; pair_i<mapped_cnt; pair_i++ ) {
128 0 : ulong blockhash_i = blockhash_i0+pair_i;
129 0 : ulong slot_i = slot_i0 +pair_i;
130 0 : writer->blockhash_descs[ blockhash_i ].slot_i = (ushort)slot_i;
131 0 : }
132 0 : return 1;
133 0 : }
134 :
135 : /* Walks every hash chain of a rooted blockcache without taking the
136 : txncache lock. For each transaction that matches a captured rooted
137 : descriptor, the function either counts the transaction into its
138 : (execution slot, blockhash) group, or copies its hash into the
139 : group's arena range. */
140 :
141 : static void
142 : writer_walk_blockhash( fd_txncache_writer_t * writer,
143 : ulong blockhash_i,
144 0 : int mode ) {
145 0 : fd_txncache_t const * tc = writer->tc;
146 0 : fd_txncache_writer_blockhash_desc_t const * blockhash_desc = &writer->blockhash_descs[ blockhash_i ];
147 :
148 : /* Loop invariant variables are hoisted because the compiler fences
149 : would otherwise force the address chains to be reloaded every
150 : bucket. */
151 0 : fd_txncache_shmem_t const * shmem = tc->shmem;
152 0 : uint const * heads = tc->blockcache_pool[ blockhash_desc->blockcache_idx ].heads;
153 0 : fd_txncache_txnpage_t const * txnpages = tc->txnpages;
154 0 : ulong bucket_cnt = shmem->bucket_cnt;
155 0 : ulong txn_cap = shmem->max_txnpages*FD_TXNCACHE_TXNS_PER_PAGE;
156 0 : ulong root_gen_init = writer->root_gen;
157 0 : fd_txnhash_t * arena = writer->arena;
158 0 : ulong group_lo = writer->group_i;
159 0 : ulong group_hi = writer->batch_hi;
160 :
161 0 : uint bucket_entry_cnt[ FD_TXNCACHE_WRITER_MAX_SLOT_DELTAS ] = {0};
162 0 : ushort touched_slot_i[ FD_TXNCACHE_WRITER_MAX_SLOT_DELTAS ];
163 :
164 0 : for( ulong bucket=0UL; bucket<bucket_cnt; bucket++ ) {
165 0 : for(;;) {
166 0 : ulong gen0 = __atomic_load_n( &shmem->mutation_gen, __ATOMIC_ACQUIRE );
167 0 : if( FD_UNLIKELY( gen0&1UL ) ) {
168 0 : fd_racesan_hook( "txncache_writer:mutation_in_progress" );
169 0 : FD_SPIN_PAUSE();
170 0 : continue;
171 0 : }
172 :
173 : /* root_gen only changes inside a mutation bracket, so we check it
174 : once up front. If mutation_gen checks out in the end, that
175 : implies root_gen is also good. */
176 0 : if( FD_UNLIKELY( __atomic_load_n( &shmem->root_gen, __ATOMIC_RELAXED )!=root_gen_init ) ) {
177 0 : FD_LOG_CRIT(( "txncache root advanced while snapshot of slot %lu was in progress", writer->snapshot_slot ));
178 0 : }
179 :
180 0 : ulong touched_cnt = 0UL;
181 0 : ulong steps = 0UL;
182 0 : int anomaly = WALK_ANOMALY_NONE;
183 0 : uint anomaly_txn_idx = UINT_MAX;
184 0 : fd_txncache_writer_blockhash_desc_t const * anomaly_desc = NULL;
185 :
186 0 : fd_racesan_hook( "txncache_writer:bucket_started" );
187 :
188 0 : for( uint head=__atomic_load_n( &heads[ bucket ], __ATOMIC_ACQUIRE ); head!=UINT_MAX; ) {
189 0 : if( FD_UNLIKELY( (ulong)head>=txn_cap ) ) {
190 0 : anomaly = WALK_ANOMALY_CHAIN_TXN_IDX_OOB;
191 0 : anomaly_txn_idx = head;
192 0 : break;
193 0 : }
194 0 : fd_txncache_single_txn_t const * txn = txnpages[ head/FD_TXNCACHE_TXNS_PER_PAGE ].txns[ head%FD_TXNCACHE_TXNS_PER_PAGE ];
195 : /* Pigeonhole principle. If we visited more than the max number
196 : of entries, then at least one entry has been visited twice,
197 : meaning a potential cycle. */
198 0 : if( FD_UNLIKELY( ++steps>txn_cap ) ) {
199 : /* Record the anomaly rather than fatally crash immediately.
200 : A concurrent compaction for example could legitimately send
201 : us into a loop. We should simply retry in that case. */
202 0 : anomaly = WALK_ANOMALY_CYCLE;
203 0 : break;
204 0 : }
205 :
206 0 : fd_txncache_writer_blockhash_desc_t const * exec_desc = writer_exec_desc( writer, txn );
207 0 : if( FD_LIKELY( exec_desc ) ) {
208 0 : ushort slot_i = exec_desc->slot_i;
209 0 : if( FD_UNLIKELY( slot_i==USHORT_MAX ) ) {
210 0 : anomaly = WALK_ANOMALY_CAPTURED_FORK_NO_EXEC_SLOT;
211 0 : anomaly_desc = exec_desc;
212 0 : break;
213 0 : }
214 0 : ulong key = group_key( slot_i, blockhash_i );
215 0 : if( mode==WALK_COUNT ) {
216 0 : if( FD_UNLIKELY( !bucket_entry_cnt[ slot_i ] ) ) touched_slot_i[ touched_cnt++ ] = slot_i;
217 0 : bucket_entry_cnt[ slot_i ]++;
218 0 : } else {
219 0 : ulong group_i = writer->group_i_by_key[ key ];
220 0 : if( group_i>=group_lo && group_i<group_hi ) {
221 0 : fd_txncache_writer_group_t * group = &writer->groups[ group_i ];
222 0 : uint entry_i = group->filled_entry_cnt+bucket_entry_cnt[ slot_i ];
223 0 : if( FD_UNLIKELY( entry_i<group->filled_entry_cnt || entry_i>=group->entry_cnt ) ) {
224 0 : anomaly = WALK_ANOMALY_GROUP_COUNT_EXCEEDED;
225 0 : break;
226 0 : }
227 0 : if( FD_UNLIKELY( !bucket_entry_cnt[ slot_i ] ) ) touched_slot_i[ touched_cnt++ ] = slot_i;
228 :
229 : /* Speculative. These txnhash values are considered
230 : committed when entry_cnt is bumped from local
231 : accumulators, after the final generation check. */
232 0 : memcpy( arena[ group->arena_entry_off+entry_i ], txn->txnhash, sizeof(arena[0]) );
233 0 : bucket_entry_cnt[ slot_i ]++;
234 0 : fd_racesan_hook( "txncache_writer:txnhash_staged" );
235 0 : }
236 0 : }
237 0 : }
238 :
239 0 : if( FD_UNLIKELY( !(steps%FD_TXNCACHE_WRITER_CHECK_INTERVAL) ) ) {
240 0 : FD_HW_MFENCE_LD();
241 0 : if( __atomic_load_n( &shmem->mutation_gen, __ATOMIC_RELAXED )!=gen0 ) { break; }
242 0 : }
243 :
244 : /* Volatile: head is bounds checked and then used to index the
245 : pool, so it must be loaded exactly once. */
246 0 : head = FD_VOLATILE_CONST( txn->blockcache_next );
247 0 : }
248 :
249 0 : if( FD_UNLIKELY( anomaly!=WALK_ANOMALY_NONE ) ) { fd_racesan_hook( "txncache_writer:anomaly_detected" ); }
250 :
251 : /* Drain reads before we check generation number. */
252 0 : FD_HW_MFENCE_LD();
253 0 : ulong gen1 = __atomic_load_n( &shmem->mutation_gen, __ATOMIC_RELAXED );
254 :
255 0 : if( FD_UNLIKELY( gen1!=gen0 ) ) {
256 0 : for( ulong i=0UL; i<touched_cnt; i++ ) bucket_entry_cnt[ touched_slot_i[ i ] ] = 0U;
257 0 : FD_SPIN_PAUSE();
258 0 : continue;
259 0 : }
260 :
261 : /* At this point, we have an anomaly without any generation number
262 : change that would explain the anomaly. Fatal. */
263 0 : switch( anomaly ) {
264 0 : case WALK_ANOMALY_NONE:
265 0 : break;
266 0 : case WALK_ANOMALY_CHAIN_TXN_IDX_OOB:
267 0 : FD_LOG_CRIT(( "txncache chain of bucket %lu of blockcache %lu has out-of-bounds transaction index %u (capacity %lu)", bucket, blockhash_desc->blockcache_idx, anomaly_txn_idx, txn_cap ));
268 0 : case WALK_ANOMALY_CYCLE:
269 0 : FD_LOG_CRIT(( "txncache chain walk of bucket %lu of blockcache %lu exceeded the pool's %lu transactions: cycle", bucket, blockhash_desc->blockcache_idx, txn_cap ));
270 0 : case WALK_ANOMALY_CAPTURED_FORK_NO_EXEC_SLOT:
271 0 : FD_LOG_CRIT(( "txncache transaction on captured fork %lu (descriptor %ld) has no execution slot", anomaly_desc->blockcache_idx, (long)(anomaly_desc-writer->blockhash_descs) ));
272 0 : case WALK_ANOMALY_GROUP_COUNT_EXCEEDED:
273 0 : FD_LOG_CRIT(( "txncache group in blockcache %lu exceeds its counted transaction count while snapshot of slot %lu is in progress", blockhash_desc->blockcache_idx, writer->snapshot_slot ));
274 0 : default:
275 0 : FD_LOG_CRIT(( "invalid txncache walk anomaly %d", anomaly ));
276 0 : }
277 :
278 : /* Yay! This was a successful bucket walk. Aggregate local
279 : accumulators into writer. Then break out of the retry loop to
280 : start the next bucket. */
281 0 : for( ulong i=0UL; i<touched_cnt; i++ ) {
282 0 : ulong slot_i = touched_slot_i[ i ];
283 0 : ulong key = group_key( slot_i, blockhash_i );
284 0 : uint delta = bucket_entry_cnt[ slot_i ];
285 0 : if( mode==WALK_COUNT ) writer->entry_cnt_by_key[ key ] += delta;
286 0 : else writer->groups[ writer->group_i_by_key[ key ] ].filled_entry_cnt += delta;
287 0 : bucket_entry_cnt[ slot_i ] = 0U;
288 0 : }
289 0 : break;
290 0 : }
291 0 : }
292 0 : }
293 :
294 : /* Walks every rooted blockcache once to count the transactions of each
295 : group, then lays the non-empty groups out in wire order: slot delta
296 : ascending, then wire group in blockhash descriptor order (oldest
297 : rooted blockcache first). */
298 :
299 : static void
300 0 : writer_groups_init( fd_txncache_writer_t * writer ) {
301 0 : memset( writer->entry_cnt_by_key, 0, sizeof(writer->entry_cnt_by_key) );
302 0 : for( ulong blockhash_i=0UL; blockhash_i<writer->blockhash_cnt; blockhash_i++ ) {
303 0 : writer_walk_blockhash( writer, blockhash_i, WALK_COUNT );
304 0 : }
305 :
306 0 : ulong group_cnt = 0UL;
307 0 : ulong entry_cnt = 0UL;
308 0 : for( ulong slot_i=0UL; slot_i<writer->slot_cnt; slot_i++ ) {
309 0 : writer->group_cnt_by_slot_i[ slot_i ] = 0UL;
310 0 : for( ulong blockhash_i=0UL; blockhash_i<writer->blockhash_cnt; blockhash_i++ ) {
311 0 : ulong key = group_key( slot_i, blockhash_i );
312 0 : uint group_entry_cnt = writer->entry_cnt_by_key[ key ];
313 0 : if( !group_entry_cnt ) continue;
314 0 : fd_txncache_writer_group_t * group = &writer->groups[ group_cnt ];
315 0 : group->slot_i = (ushort)slot_i;
316 0 : group->blockhash_i = (ushort)blockhash_i;
317 0 : group->entry_cnt = group_entry_cnt;
318 0 : group->arena_entry_off = 0U;
319 0 : group->filled_entry_cnt = 0U;
320 0 : writer->group_i_by_key[ key ] = (uint)group_cnt;
321 0 : writer->group_cnt_by_slot_i[ slot_i ]++;
322 0 : group_cnt++;
323 0 : entry_cnt += group_entry_cnt;
324 0 : }
325 0 : }
326 0 : writer->group_cnt = group_cnt;
327 0 : writer->entry_cnt = entry_cnt;
328 0 : }
329 :
330 : /* Writes groups [group_i,batch_hi) into the arena, with batch_hi as far
331 : as the arena reaches. Walks each rooted blockcache that has a group
332 : in the batch once. */
333 : static void
334 0 : writer_fill_batch( fd_txncache_writer_t * writer ) {
335 0 : ulong group_i0 = writer->group_i;
336 0 : fd_txncache_writer_group_t * first = &writer->groups[ group_i0 ];
337 0 : if( FD_UNLIKELY( (ulong)first->entry_cnt>writer->arena_entry_cnt_max ) ) {
338 0 : FD_BASE58_ENCODE_32_BYTES( writer->blockhash_descs[ first->blockhash_i ].blockhash, blockhash_b58 );
339 0 : FD_LOG_CRIT(( "status cache group of slot %lu under blockhash %s holds %u transactions, exceeding the %lu entry arena", writer->slots[ first->slot_i ], blockhash_b58, first->entry_cnt, writer->arena_entry_cnt_max ));
340 0 : }
341 :
342 0 : uchar blockhash_in_batch[ FD_TXNCACHE_WRITER_MAX_BLOCKHASHES ] = {0};
343 0 : ulong arena_entry_off = 0UL;
344 0 : ulong batch_hi = group_i0;
345 0 : while( batch_hi<writer->group_cnt && arena_entry_off+(ulong)writer->groups[ batch_hi ].entry_cnt<=writer->arena_entry_cnt_max ) {
346 0 : fd_txncache_writer_group_t * group = &writer->groups[ batch_hi ];
347 0 : group->arena_entry_off = (uint)arena_entry_off;
348 0 : group->filled_entry_cnt = 0U;
349 0 : blockhash_in_batch[ group->blockhash_i ] = 1;
350 0 : arena_entry_off += group->entry_cnt;
351 0 : batch_hi++;
352 0 : }
353 0 : writer->batch_hi = batch_hi;
354 :
355 0 : for( ulong blockhash_i=0UL; blockhash_i<writer->blockhash_cnt; blockhash_i++ ) {
356 0 : if( blockhash_in_batch[ blockhash_i ] ) writer_walk_blockhash( writer, blockhash_i, WALK_FILL );
357 0 : }
358 :
359 0 : for( ulong group_i=group_i0; group_i<batch_hi; group_i++ ) {
360 0 : fd_txncache_writer_group_t const * group = &writer->groups[ group_i ];
361 0 : if( FD_UNLIKELY( group->filled_entry_cnt!=group->entry_cnt ) ) {
362 0 : FD_LOG_CRIT(( "txncache group %u of slot %lu has %u transactions, expected %u while snapshot of slot %lu is in progress", group->blockhash_i, writer->slots[ group->slot_i ], group->filled_entry_cnt, group->entry_cnt, writer->snapshot_slot ));
363 0 : }
364 0 : }
365 0 : }
366 :
367 : fd_txncache_writer_t *
368 : fd_txncache_writer_init( fd_txncache_writer_t * writer,
369 : fd_txncache_t * tc,
370 : fd_txncache_fork_id_t fork_id,
371 : ulong snapshot_slot,
372 : uchar const * slot_history,
373 : ulong slot_history_sz,
374 : void * arena,
375 0 : ulong arena_sz ) {
376 0 : fd_rwlock_t * lock = tc->shmem->lock;
377 :
378 0 : if( FD_UNLIKELY( !arena || !fd_ulong_is_aligned( (ulong)arena, fd_txncache_writer_arena_align() ) || arena_sz<2UL*2UL*sizeof(fd_txnhash_t) ) ) {
379 0 : FD_LOG_CRIT(( "invalid txncache writer arena (%p, %lu bytes)", arena, arena_sz ));
380 0 : }
381 :
382 0 : ulong arena_entry_cnt_max = arena_sz/sizeof(fd_txnhash_t);
383 0 : FD_CHECK_CRIT( arena_entry_cnt_max<=UINT_MAX, "txncache writer arena has too many entries" ); /* arena_entry_off is uint */
384 :
385 0 : writer->state = STATE_HEADER;
386 0 : writer->tc = tc;
387 0 : writer->snapshot_slot = snapshot_slot;
388 0 : writer->arena = arena;
389 0 : writer->arena_entry_cnt_max = arena_entry_cnt_max;
390 0 : writer->batch_hi = 0UL;
391 0 : writer->slot_i = 0UL;
392 0 : writer->group_i = 0UL;
393 0 : writer->entry_i = 0UL;
394 :
395 0 : writer_slots_init( writer, slot_history, slot_history_sz );
396 :
397 : /* Snapshot the root list. The snapshot root must be the newest root. */
398 :
399 0 : fd_rwlock_read( lock );
400 0 : if( FD_UNLIKELY( fork_id.val>=tc->shmem->active_slots_max ||
401 0 : root_slist_is_empty( tc->shmem->root_ll, tc->blockcache_shmem_pool ) ||
402 0 : root_slist_idx_peek_tail( tc->shmem->root_ll, tc->blockcache_shmem_pool )!=fork_id.val ||
403 0 : tc->blockcache_shmem_pool[ fork_id.val ].frozen!=2 ) ) {
404 0 : fd_rwlock_unread( lock );
405 0 : return NULL;
406 0 : }
407 0 : writer->root_gen = __atomic_load_n( &tc->shmem->root_gen, __ATOMIC_RELAXED );
408 :
409 0 : ulong blockhash_cnt = 0UL;
410 0 : for( ulong it = root_slist_iter_init( tc->shmem->root_ll, tc->blockcache_shmem_pool );
411 0 : !root_slist_iter_done( it, tc->shmem->root_ll, tc->blockcache_shmem_pool );
412 0 : it = root_slist_iter_next( it, tc->shmem->root_ll, tc->blockcache_shmem_pool ) ) {
413 0 : ulong blockcache_idx = root_slist_iter_idx( it, tc->shmem->root_ll, tc->blockcache_shmem_pool );
414 0 : if( FD_UNLIKELY( blockhash_cnt>=FD_TXNCACHE_WRITER_MAX_BLOCKHASHES ) ) {
415 0 : FD_LOG_CRIT(( "txncache has more than %lu roots", FD_TXNCACHE_WRITER_MAX_BLOCKHASHES ));
416 0 : }
417 0 : fd_txncache_blockcache_shmem_t const * blockcache_shmem = &tc->blockcache_shmem_pool[ blockcache_idx ];
418 0 : fd_txncache_writer_blockhash_desc_t * blockhash_desc = &writer->blockhash_descs[ blockhash_cnt ];
419 0 : blockhash_desc->blockcache_idx = blockcache_idx;
420 0 : blockhash_desc->generation = blockcache_shmem->generation;
421 0 : blockhash_desc->txnhash_offset = blockcache_shmem->txnhash_offset;
422 0 : memcpy( blockhash_desc->blockhash, blockcache_shmem->blockhash.uc, 32UL );
423 0 : blockhash_cnt++;
424 0 : }
425 0 : fd_rwlock_unread( lock );
426 0 : writer->blockhash_cnt = blockhash_cnt;
427 :
428 0 : if( FD_UNLIKELY( !writer_map_execution_slots( writer ) ) ) return NULL;
429 0 : writer_groups_init( writer );
430 :
431 0 : return writer;
432 0 : }
433 :
434 : FD_FN_PURE ulong
435 0 : fd_txncache_writer_serialized_sz( fd_txncache_writer_t const * writer ) {
436 0 : return HEADER_SZ+writer->slot_cnt*SLOT_DELTA_SZ+writer->group_cnt*GROUP_HDR_SZ+writer->entry_cnt*TXN_SZ;
437 0 : }
438 :
439 : ulong
440 : fd_txncache_writer_serialize( fd_txncache_writer_t * writer,
441 : uchar * out_buf,
442 0 : ulong buf_sz ) {
443 0 : if( FD_UNLIKELY( buf_sz<FD_TXNCACHE_WRITER_BUF_MIN ) ) {
444 0 : FD_LOG_CRIT(( "buffer too small (%lu bytes, need at least %lu)", buf_sz, FD_TXNCACHE_WRITER_BUF_MIN ));
445 0 : }
446 0 : uchar * p = out_buf;
447 0 : uchar * p1 = out_buf+buf_sz;
448 :
449 0 : for(;;) {
450 0 : switch( writer->state ) {
451 :
452 0 : case STATE_HEADER: {
453 0 : if( FD_UNLIKELY( p+HEADER_SZ>p1 ) ) goto done;
454 0 : FD_STORE( ulong, p, writer->slot_cnt ); p += 8UL; /* slot_deltas_len */
455 0 : writer->slot_i = 0UL;
456 0 : writer->group_i = 0UL;
457 0 : writer->state = writer->slot_cnt ? STATE_SLOT : STATE_DONE;
458 0 : break;
459 0 : }
460 :
461 0 : case STATE_SLOT: {
462 0 : if( FD_UNLIKELY( p+SLOT_DELTA_SZ>p1 ) ) goto done;
463 0 : ulong slot_i = writer->slot_i;
464 0 : FD_STORE( ulong, p, writer->slots[ slot_i ] ); p += 8UL; /* slot */
465 0 : FD_STORE( uchar, p, 1 ); p += 1UL; /* is_root */
466 0 : FD_STORE( ulong, p, writer->group_cnt_by_slot_i[ slot_i ] ); p += 8UL; /* status_len */
467 0 : if( writer->group_cnt_by_slot_i[ slot_i ] ) {
468 0 : writer->state = STATE_GROUP;
469 0 : } else {
470 0 : writer->slot_i++;
471 0 : if( writer->slot_i>=writer->slot_cnt ) writer->state = STATE_DONE;
472 0 : }
473 0 : break;
474 0 : }
475 :
476 0 : case STATE_GROUP: {
477 0 : fd_txncache_writer_group_t const * group = &writer->groups[ writer->group_i ];
478 0 : FD_TEST( group->slot_i==writer->slot_i );
479 0 : if( FD_UNLIKELY( writer->group_i>=writer->batch_hi ) ) writer_fill_batch( writer );
480 0 : if( FD_UNLIKELY( p+GROUP_HDR_SZ>p1 ) ) goto done;
481 0 : fd_txncache_writer_blockhash_desc_t const * blockhash_desc = &writer->blockhash_descs[ group->blockhash_i ];
482 0 : memcpy( p, blockhash_desc->blockhash, 32UL ); p += 32UL;
483 0 : FD_STORE( ulong, p, blockhash_desc->txnhash_offset ); p += 8UL;
484 0 : FD_STORE( ulong, p, (ulong)group->entry_cnt ); p += 8UL;
485 0 : writer->entry_i = 0UL;
486 0 : writer->state = STATE_ENTRIES;
487 0 : break;
488 0 : }
489 :
490 0 : case STATE_ENTRIES: {
491 0 : fd_txncache_writer_group_t const * group = &writer->groups[ writer->group_i ];
492 0 : while( writer->entry_i<group->entry_cnt ) {
493 0 : if( FD_UNLIKELY( p+TXN_SZ>p1 ) ) goto done;
494 0 : memcpy( p, writer->arena[ group->arena_entry_off+writer->entry_i ], sizeof(writer->arena[0]) ); p += sizeof(writer->arena[0]);
495 0 : FD_STORE( uint, p, 0U ); p += 4UL; /* result = Ok */
496 0 : writer->entry_i++;
497 0 : }
498 0 : writer->group_i++;
499 0 : if( writer->group_i<writer->group_cnt && writer->groups[ writer->group_i ].slot_i==writer->slot_i ) {
500 0 : writer->state = STATE_GROUP;
501 0 : } else {
502 0 : writer->slot_i++;
503 0 : writer->state = writer->slot_i<writer->slot_cnt ? STATE_SLOT : STATE_DONE;
504 0 : }
505 0 : break;
506 0 : }
507 :
508 0 : case STATE_DONE:
509 0 : goto done;
510 :
511 0 : default:
512 0 : FD_LOG_CRIT(( "invalid state reached (%u)", writer->state ));
513 0 : }
514 0 : }
515 :
516 0 : done:
517 0 : return (ulong)( p-out_buf );
518 0 : }
|
