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 :
6 183926 : #define STATE_HEADER 1
7 207573 : #define STATE_SLOT 2
8 198790 : #define STATE_GROUP 3
9 199168 : #define STATE_ENTRIES 4
10 467139 : #define STATE_DONE 5
11 :
12 : /* Wire sizes */
13 :
14 92002 : #define HEADER_SZ (8UL) /* slot_deltas_len */
15 92002 : #define SLOT_DELTA_SZ (8UL+1UL+8UL) /* slot, is_root, status_len */
16 92002 : #define GROUP_HDR_SZ (32UL+8UL+8UL) /* blockhash, txnhash_offset, txn_cnt */
17 92002 : #define TXN_SZ (sizeof(fd_txnhash_t)+4UL) /* txnhash, result */
18 :
19 30 : #define FD_TXNCACHE_WRITER_ARENA_MIN (64UL<<20)
20 :
21 832946 : #define WALK_COUNT 0
22 99323 : #define WALK_FILL 1
23 :
24 : FD_STATIC_ASSERT( FD_TXNCACHE_WRITER_MAX_SLOT_DELTAS<=USHORT_MAX, group_slot_i );
25 : FD_STATIC_ASSERT( FD_TXNCACHE_WRITER_MAX_BLOCKHASHES<=USHORT_MAX, group_blockhash_i );
26 : FD_STATIC_ASSERT( FD_TXNCACHE_MAX_ROOTS<=FD_TXNCACHE_WRITER_MAX_BLOCKHASHES, roots );
27 :
28 : static inline ulong
29 1164182 : group_key( ulong slot_i, ulong blockhash_i ) {
30 1164182 : return slot_i*FD_TXNCACHE_WRITER_MAX_BLOCKHASHES+blockhash_i;
31 1164182 : }
32 :
33 : FD_FN_CONST ulong
34 92065 : fd_txncache_writer_arena_align( void ) {
35 92065 : return 64UL;
36 92065 : }
37 :
38 : FD_FN_CONST ulong
39 30 : fd_txncache_writer_arena_sz( ulong max_txn_per_slot ) {
40 : /* Write at least 2 slots at once, and have at least 64 MiB. */
41 30 : return fd_ulong_max( FD_TXNCACHE_WRITER_ARENA_MIN, 2UL*2UL*max_txn_per_slot*sizeof(fd_txnhash_t) );
42 30 : }
43 :
44 : /* Returns the captured rooted descriptor of the fork the txn executed
45 : on, or NULL if that fork is not in the captured rooted set, either
46 : because it is unrooted or because it is a cancelled fork whose pool
47 : slot has since been reused. */
48 :
49 : static inline fd_txncache_writer_blockhash_desc_t const *
50 : writer_exec_desc( fd_txncache_writer_t const * writer,
51 1408582457 : fd_txncache_single_txn_t const * txn ) {
52 1408582457 : ulong desc_i = writer->fork_id_to_blockhash_i[ txn->fork_id.val ];
53 1408582457 : if( FD_UNLIKELY( desc_i==USHORT_MAX ) ) return NULL;
54 640796 : fd_txncache_writer_blockhash_desc_t const * desc = &writer->blockhash_descs[ desc_i ];
55 640796 : return txn->generation==desc->generation ? desc : NULL;
56 1408582457 : }
57 :
58 : static void
59 : writer_slots_init( fd_txncache_writer_t * writer,
60 : uchar const * slot_history,
61 91963 : ulong slot_history_sz ) {
62 : /* Agave's verify_slot_history rejects a snapshot whose SlotHistory is
63 : not exactly MAX_ENTRIES bits ending at the bank's slot. */
64 :
65 91963 : fd_slot_history_view_t view[1];
66 91963 : if( FD_UNLIKELY( !fd_sysvar_slot_history_view( view, slot_history, slot_history_sz ) ) ) {
67 0 : FD_LOG_CRIT(( "malformed SlotHistory sysvar at slot %lu", writer->snapshot_slot ));
68 0 : }
69 91963 : if( FD_UNLIKELY( view->bits_len!=FD_SLOT_HISTORY_MAX_ENTRIES ||
70 91963 : view->blocks_len*64UL!=FD_SLOT_HISTORY_MAX_ENTRIES ||
71 91963 : view->next_slot>writer->snapshot_slot+1UL ) ) {
72 0 : FD_LOG_CRIT(( "malformed SlotHistory sysvar at slot %lu (bits_len %lu blocks_len %lu next_slot %lu)",
73 0 : writer->snapshot_slot, view->bits_len, view->blocks_len, view->next_slot ));
74 0 : }
75 :
76 : /* The SlotHistory may be one slot stale (missing the bit for the
77 : snapshot slot itself), so the loop below forces that slot in. */
78 :
79 91963 : ulong scan_cnt = fd_ulong_min( writer->snapshot_slot+1UL, FD_SLOT_HISTORY_MAX_ENTRIES );
80 91963 : ulong cnt = 0UL;
81 6690364 : for( ulong i=0UL; i<scan_cnt && cnt<FD_TXNCACHE_WRITER_MAX_SLOT_DELTAS; i++ ) {
82 6598401 : ulong slot = writer->snapshot_slot-i;
83 6598401 : if( slot!=writer->snapshot_slot && fd_sysvar_slot_history_find_slot( view, slot )!=FD_SLOT_HISTORY_SLOT_FOUND ) continue;
84 108268 : writer->slots[ FD_TXNCACHE_WRITER_MAX_SLOT_DELTAS-1UL-cnt ] = slot; /* fill from the back to get ascending order */
85 108268 : cnt++;
86 108268 : }
87 91963 : FD_TEST( cnt ); /* i==0 always counts */
88 :
89 91963 : memmove( writer->slots, writer->slots+FD_TXNCACHE_WRITER_MAX_SLOT_DELTAS-cnt, cnt*sizeof(ulong) );
90 91963 : writer->slot_cnt = cnt;
91 91963 : }
92 :
93 : /* Pairs rooted blockcache descriptors with execution slots by rank from
94 : the newest end. Every rooted block registered exactly one blockhash
95 : and set exactly one SlotHistory bit, and both sequences end at the
96 : snapshot slot. The initial genesis descriptor may precede the named
97 : execution slots; it remains usable as a referenced blockhash but has
98 : no execution slot mapping. */
99 : static void
100 91963 : writer_map_execution_slots( fd_txncache_writer_t * writer ) {
101 91963 : ulong blockhash_cnt = writer->blockhash_cnt;
102 91963 : ulong slot_cnt = writer->slot_cnt;
103 91963 : if( FD_UNLIKELY( blockhash_cnt>slot_cnt+1UL ) ) {
104 0 : FD_LOG_CRIT(( "txncache has %lu rooted blockhash descriptors but SlotHistory names only %lu execution slots", blockhash_cnt, slot_cnt ));
105 0 : }
106 :
107 91963 : memset( writer->fork_id_to_blockhash_i, 0xFF, sizeof(writer->fork_id_to_blockhash_i) );
108 284113 : for( ulong blockhash_i=0UL; blockhash_i<blockhash_cnt; blockhash_i++ ) {
109 192150 : fd_txncache_writer_blockhash_desc_t * desc = &writer->blockhash_descs[ blockhash_i ];
110 192150 : desc->slot_i = USHORT_MAX;
111 192150 : writer->fork_id_to_blockhash_i[ desc->blockcache_idx ] = (ushort)blockhash_i;
112 192150 : }
113 :
114 91963 : ulong mapped_cnt = fd_ulong_min( blockhash_cnt, slot_cnt );
115 91963 : ulong blockhash_i0 = blockhash_cnt-mapped_cnt;
116 91963 : ulong slot_i0 = slot_cnt -mapped_cnt;
117 195791 : for( ulong pair_i=0UL; pair_i<mapped_cnt; pair_i++ ) {
118 103828 : ulong blockhash_i = blockhash_i0+pair_i;
119 103828 : ulong slot_i = slot_i0 +pair_i;
120 103828 : writer->blockhash_descs[ blockhash_i ].slot_i = (ushort)slot_i;
121 103828 : }
122 91963 : }
123 :
124 : static void
125 : writer_walk_blockhash( fd_txncache_writer_t * writer,
126 : ulong blockhash_idx,
127 291473 : int mode ) {
128 :
129 291473 : fd_txncache_blockcache_iter_t iter[1];
130 291473 : for( fd_txncache_blockcache_iter_init( iter, writer->tc, writer->blockhash_descs[ blockhash_idx ].blockcache_idx );
131 1408873930 : !fd_txncache_blockcache_iter_done( iter );
132 1408582457 : fd_txncache_blockcache_iter_next( iter ) ) {
133 1408582457 : fd_txncache_single_txn_t const * txn = fd_txncache_blockcache_iter_ele( iter );
134 :
135 1408582457 : fd_txncache_writer_blockhash_desc_t const * exec_desc = writer_exec_desc( writer, txn );
136 1408582457 : if( FD_UNLIKELY( !exec_desc ) ) continue;
137 :
138 640796 : ushort slot_i = exec_desc->slot_i;
139 640796 : if( FD_UNLIKELY( slot_i==USHORT_MAX ) ) {
140 0 : FD_LOG_CRIT(( "txncache transaction on captured fork %lu (descriptor %ld) has no execution slot", exec_desc->blockcache_idx, (long)(exec_desc-writer->blockhash_descs) ));
141 0 : }
142 640796 : ulong key = group_key( slot_i, blockhash_idx );
143 640796 : if( mode==WALK_COUNT ) {
144 305998 : writer->entry_cnt_by_key[ key ]++;
145 305998 : continue;
146 305998 : }
147 :
148 334798 : ulong group_i = writer->group_i_by_key[ key ];
149 334798 : if( group_i<writer->group_i || group_i>=writer->batch_hi ) continue;
150 305998 : fd_txncache_writer_group_t * group = &writer->groups[ group_i ];
151 305998 : if( FD_UNLIKELY( group->filled_entry_cnt>=group->entry_cnt ) ) {
152 0 : FD_LOG_CRIT(( "txncache group in blockcache %lu exceeds its counted transaction count while snapshot of slot %lu is in progress", writer->blockhash_descs[ blockhash_idx ].blockcache_idx, writer->snapshot_slot ));
153 0 : }
154 305998 : memcpy( writer->arena[ group->arena_entry_off+group->filled_entry_cnt ], txn->txnhash, sizeof(writer->arena[0]) );
155 305998 : group->filled_entry_cnt++;
156 305998 : }
157 291473 : fd_txncache_blockcache_iter_fini( iter );
158 291473 : }
159 :
160 : /* Walks every rooted blockcache once to count the transactions of each
161 : group, then lays the non-empty groups out in wire order: slot delta
162 : ascending, then wire group in blockhash descriptor order (oldest
163 : rooted blockcache first). */
164 :
165 : static void
166 91963 : writer_groups_init( fd_txncache_writer_t * writer ) {
167 91963 : memset( writer->entry_cnt_by_key, 0, sizeof(writer->entry_cnt_by_key) );
168 284113 : for( ulong blockhash_i=0UL; blockhash_i<writer->blockhash_cnt; blockhash_i++ ) {
169 192150 : writer_walk_blockhash( writer, blockhash_i, WALK_COUNT );
170 192150 : }
171 :
172 91963 : ulong group_cnt = 0UL;
173 91963 : ulong entry_cnt = 0UL;
174 200231 : for( ulong slot_i=0UL; slot_i<writer->slot_cnt; slot_i++ ) {
175 108268 : writer->group_cnt_by_slot_i[ slot_i ] = 0UL;
176 631654 : for( ulong blockhash_i=0UL; blockhash_i<writer->blockhash_cnt; blockhash_i++ ) {
177 523386 : ulong key = group_key( slot_i, blockhash_i );
178 523386 : uint group_entry_cnt = writer->entry_cnt_by_key[ key ];
179 523386 : if( !group_entry_cnt ) continue;
180 99395 : fd_txncache_writer_group_t * group = &writer->groups[ group_cnt ];
181 99395 : group->slot_i = (ushort)slot_i;
182 99395 : group->blockhash_i = (ushort)blockhash_i;
183 99395 : group->entry_cnt = group_entry_cnt;
184 99395 : group->arena_entry_off = 0U;
185 99395 : group->filled_entry_cnt = 0U;
186 99395 : writer->group_i_by_key[ key ] = (uint)group_cnt;
187 99395 : writer->group_cnt_by_slot_i[ slot_i ]++;
188 99395 : group_cnt++;
189 99395 : entry_cnt += group_entry_cnt;
190 99395 : }
191 108268 : }
192 91963 : writer->group_cnt = group_cnt;
193 91963 : writer->entry_cnt = entry_cnt;
194 91963 : }
195 :
196 : /* Writes groups [group_i,batch_hi) into the arena, with batch_hi as far
197 : as the arena reaches. Walks each rooted blockcache that has a group
198 : in the batch once. */
199 : static void
200 92065 : writer_fill_batch( fd_txncache_writer_t * writer ) {
201 92065 : ulong group_i0 = writer->group_i;
202 92065 : fd_txncache_writer_group_t * first = &writer->groups[ group_i0 ];
203 92065 : if( FD_UNLIKELY( (ulong)first->entry_cnt>writer->arena_entry_cnt_max ) ) {
204 0 : FD_BASE58_ENCODE_32_BYTES( writer->blockhash_descs[ first->blockhash_i ].blockhash, blockhash_b58 );
205 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 ));
206 0 : }
207 :
208 92065 : uchar blockhash_in_batch[ FD_TXNCACHE_WRITER_MAX_BLOCKHASHES ] = {0};
209 92065 : ulong arena_entry_off = 0UL;
210 92065 : ulong batch_hi = group_i0;
211 191460 : while( batch_hi<writer->group_cnt && arena_entry_off+(ulong)writer->groups[ batch_hi ].entry_cnt<=writer->arena_entry_cnt_max ) {
212 99395 : fd_txncache_writer_group_t * group = &writer->groups[ batch_hi ];
213 99395 : group->arena_entry_off = (uint)arena_entry_off;
214 99395 : group->filled_entry_cnt = 0U;
215 99395 : blockhash_in_batch[ group->blockhash_i ] = 1;
216 99395 : arena_entry_off += group->entry_cnt;
217 99395 : batch_hi++;
218 99395 : }
219 92065 : writer->batch_hi = batch_hi;
220 :
221 285073 : for( ulong blockhash_i=0UL; blockhash_i<writer->blockhash_cnt; blockhash_i++ ) {
222 193008 : if( blockhash_in_batch[ blockhash_i ] ) writer_walk_blockhash( writer, blockhash_i, WALK_FILL );
223 193008 : }
224 :
225 191460 : for( ulong group_i=group_i0; group_i<batch_hi; group_i++ ) {
226 99395 : fd_txncache_writer_group_t const * group = &writer->groups[ group_i ];
227 99395 : if( FD_UNLIKELY( group->filled_entry_cnt!=group->entry_cnt ) ) {
228 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 ));
229 0 : }
230 99395 : }
231 92065 : }
232 :
233 : fd_txncache_writer_t *
234 : fd_txncache_writer_init( fd_txncache_writer_t * writer,
235 : fd_txncache_t * tc,
236 : ulong snapshot_slot,
237 : uchar const * slot_history,
238 : ulong slot_history_sz,
239 : void * arena,
240 91963 : ulong arena_sz ) {
241 91963 : if( FD_UNLIKELY( !arena || !fd_ulong_is_aligned( (ulong)arena, fd_txncache_writer_arena_align() ) || arena_sz<2UL*2UL*sizeof(fd_txnhash_t) ) ) {
242 0 : FD_LOG_CRIT(( "invalid txncache writer arena (%p, %lu bytes)", arena, arena_sz ));
243 0 : }
244 :
245 91963 : ulong arena_entry_cnt_max = arena_sz/sizeof(fd_txnhash_t);
246 91963 : FD_CHECK_CRIT( arena_entry_cnt_max<=UINT_MAX, "txncache writer arena has too many entries" ); /* arena_entry_off is uint */
247 :
248 91963 : writer->state = STATE_HEADER;
249 91963 : writer->tc = tc;
250 91963 : writer->snapshot_slot = snapshot_slot;
251 91963 : writer->arena = arena;
252 91963 : writer->arena_entry_cnt_max = arena_entry_cnt_max;
253 91963 : writer->batch_hi = 0UL;
254 91963 : writer->slot_i = 0UL;
255 91963 : writer->group_i = 0UL;
256 91963 : writer->entry_i = 0UL;
257 :
258 91963 : writer_slots_init( writer, slot_history, slot_history_sz );
259 :
260 : /* Snapshot the root list. The snapshot root is the newest root. */
261 :
262 91963 : fd_txncache_root_t roots[ FD_TXNCACHE_MAX_ROOTS ];
263 91963 : ulong root_cnt = fd_txncache_snapshot_roots( tc, roots );
264 284113 : for( ulong i=0UL; i<root_cnt; i++ ) {
265 192150 : fd_txncache_writer_blockhash_desc_t * blockhash_desc = &writer->blockhash_descs[ i ];
266 192150 : blockhash_desc->blockcache_idx = roots[ i ].blockcache_idx;
267 192150 : blockhash_desc->generation = roots[ i ].generation;
268 192150 : blockhash_desc->txnhash_offset = roots[ i ].txnhash_offset;
269 192150 : memcpy( blockhash_desc->blockhash, roots[ i ].blockhash, 32UL );
270 192150 : }
271 91963 : writer->blockhash_cnt = root_cnt;
272 :
273 91963 : writer_map_execution_slots( writer );
274 91963 : writer_groups_init( writer );
275 :
276 91963 : return writer;
277 91963 : }
278 :
279 : FD_FN_PURE ulong
280 92002 : fd_txncache_writer_serialized_sz( fd_txncache_writer_t const * writer ) {
281 92002 : return HEADER_SZ+writer->slot_cnt*SLOT_DELTA_SZ+writer->group_cnt*GROUP_HDR_SZ+writer->entry_cnt*TXN_SZ;
282 92002 : }
283 :
284 : ulong
285 : fd_txncache_writer_serialize( fd_txncache_writer_t * writer,
286 : uchar * out_buf,
287 184316 : ulong buf_sz ) {
288 184316 : if( FD_UNLIKELY( buf_sz<FD_TXNCACHE_WRITER_BUF_MIN ) ) {
289 0 : FD_LOG_CRIT(( "buffer too small (%lu bytes, need at least %lu)", buf_sz, FD_TXNCACHE_WRITER_BUF_MIN ));
290 0 : }
291 184316 : uchar * p = out_buf;
292 184316 : uchar * p1 = out_buf+buf_sz;
293 :
294 583337 : for(;;) {
295 583337 : switch( writer->state ) {
296 :
297 91963 : case STATE_HEADER: {
298 91963 : if( FD_UNLIKELY( p+HEADER_SZ>p1 ) ) goto done;
299 91963 : FD_STORE( ulong, p, writer->slot_cnt ); p += 8UL; /* slot_deltas_len */
300 91963 : writer->slot_i = 0UL;
301 91963 : writer->group_i = 0UL;
302 91963 : writer->state = writer->slot_cnt ? STATE_SLOT : STATE_DONE;
303 91963 : break;
304 91963 : }
305 :
306 108280 : case STATE_SLOT: {
307 108280 : if( FD_UNLIKELY( p+SLOT_DELTA_SZ>p1 ) ) goto done;
308 108268 : ulong slot_i = writer->slot_i;
309 108268 : FD_STORE( ulong, p, writer->slots[ slot_i ] ); p += 8UL; /* slot */
310 108268 : FD_STORE( uchar, p, 1 ); p += 1UL; /* is_root */
311 108268 : FD_STORE( ulong, p, writer->group_cnt_by_slot_i[ slot_i ] ); p += 8UL; /* status_len */
312 108268 : if( writer->group_cnt_by_slot_i[ slot_i ] ) {
313 99287 : writer->state = STATE_GROUP;
314 99287 : } else {
315 8981 : writer->slot_i++;
316 8981 : if( writer->slot_i>=writer->slot_cnt ) writer->state = STATE_DONE;
317 8981 : }
318 108268 : break;
319 108280 : }
320 :
321 99395 : case STATE_GROUP: {
322 99395 : fd_txncache_writer_group_t const * group = &writer->groups[ writer->group_i ];
323 99395 : FD_TEST( group->slot_i==writer->slot_i );
324 99395 : if( FD_UNLIKELY( writer->group_i>=writer->batch_hi ) ) writer_fill_batch( writer );
325 99395 : if( FD_UNLIKELY( p+GROUP_HDR_SZ>p1 ) ) goto done;
326 99395 : fd_txncache_writer_blockhash_desc_t const * blockhash_desc = &writer->blockhash_descs[ group->blockhash_i ];
327 99395 : memcpy( p, blockhash_desc->blockhash, 32UL ); p += 32UL;
328 99395 : FD_STORE( ulong, p, blockhash_desc->txnhash_offset ); p += 8UL;
329 99395 : FD_STORE( ulong, p, (ulong)group->entry_cnt ); p += 8UL;
330 99395 : writer->entry_i = 0UL;
331 99395 : writer->state = STATE_ENTRIES;
332 99395 : break;
333 99395 : }
334 :
335 99773 : case STATE_ENTRIES: {
336 99773 : fd_txncache_writer_group_t const * group = &writer->groups[ writer->group_i ];
337 405771 : while( writer->entry_i<group->entry_cnt ) {
338 306376 : if( FD_UNLIKELY( p+TXN_SZ>p1 ) ) goto done;
339 305998 : memcpy( p, writer->arena[ group->arena_entry_off+writer->entry_i ], sizeof(writer->arena[0]) ); p += sizeof(writer->arena[0]);
340 305998 : FD_STORE( uint, p, 0U ); p += 4UL; /* result = Ok */
341 305998 : writer->entry_i++;
342 305998 : }
343 99395 : writer->group_i++;
344 99395 : if( writer->group_i<writer->group_cnt && writer->groups[ writer->group_i ].slot_i==writer->slot_i ) {
345 108 : writer->state = STATE_GROUP;
346 99287 : } else {
347 99287 : writer->slot_i++;
348 99287 : writer->state = writer->slot_i<writer->slot_cnt ? STATE_SLOT : STATE_DONE;
349 99287 : }
350 99395 : break;
351 99773 : }
352 :
353 183926 : case STATE_DONE:
354 183926 : goto done;
355 :
356 0 : default:
357 0 : FD_LOG_CRIT(( "invalid state reached (%u)", writer->state ));
358 583337 : }
359 583337 : }
360 :
361 184316 : done:
362 184316 : return (ulong)( p-out_buf );
363 184316 : }
|
