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 0 : #define STATE_HEADER 1
7 0 : #define STATE_SLOT 2
8 0 : #define STATE_GROUP 3
9 0 : #define STATE_ENTRIES 4
10 0 : #define STATE_DONE 5
11 :
12 : /* Wire sizes */
13 :
14 0 : #define HEADER_SZ (8UL) /* slot_deltas_len */
15 0 : #define SLOT_DELTA_SZ (8UL+1UL+8UL) /* slot, is_root, status_len */
16 0 : #define GROUP_HDR_SZ (32UL+8UL+8UL) /* blockhash, txnhash_offset, txn_cnt */
17 0 : #define TXN_SZ (sizeof(fd_txnhash_t)+4UL) /* txnhash, result */
18 :
19 0 : #define FD_TXNCACHE_WRITER_ARENA_MIN (64UL<<20)
20 :
21 0 : #define WALK_COUNT 0
22 0 : #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 0 : group_key( ulong slot_i, ulong blockhash_i ) {
30 0 : return slot_i*FD_TXNCACHE_WRITER_MAX_BLOCKHASHES+blockhash_i;
31 0 : }
32 :
33 : FD_FN_CONST ulong
34 0 : fd_txncache_writer_arena_align( void ) {
35 0 : return 64UL;
36 0 : }
37 :
38 : FD_FN_CONST ulong
39 0 : 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 0 : return fd_ulong_max( FD_TXNCACHE_WRITER_ARENA_MIN, 2UL*2UL*max_txn_per_slot*sizeof(fd_txnhash_t) );
42 0 : }
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 0 : fd_txncache_single_txn_t const * txn ) {
52 0 : ulong desc_i = writer->fork_id_to_blockhash_i[ txn->fork_id.val ];
53 0 : if( FD_UNLIKELY( desc_i==USHORT_MAX ) ) return NULL;
54 0 : fd_txncache_writer_blockhash_desc_t const * desc = &writer->blockhash_descs[ desc_i ];
55 0 : return txn->generation==desc->generation ? desc : NULL;
56 0 : }
57 :
58 : static void
59 : writer_slots_init( fd_txncache_writer_t * writer,
60 : uchar const * slot_history,
61 0 : 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 0 : fd_slot_history_view_t view[1];
66 0 : 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 0 : if( FD_UNLIKELY( view->bits_len!=FD_SLOT_HISTORY_MAX_ENTRIES ||
70 0 : view->blocks_len*64UL!=FD_SLOT_HISTORY_MAX_ENTRIES ||
71 0 : 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 0 : ulong scan_cnt = fd_ulong_min( writer->snapshot_slot+1UL, FD_SLOT_HISTORY_MAX_ENTRIES );
80 0 : ulong cnt = 0UL;
81 0 : for( ulong i=0UL; i<scan_cnt && cnt<FD_TXNCACHE_WRITER_MAX_SLOT_DELTAS; i++ ) {
82 0 : ulong slot = writer->snapshot_slot-i;
83 0 : if( slot!=writer->snapshot_slot && fd_sysvar_slot_history_find_slot( view, slot )!=FD_SLOT_HISTORY_SLOT_FOUND ) continue;
84 0 : writer->slots[ FD_TXNCACHE_WRITER_MAX_SLOT_DELTAS-1UL-cnt ] = slot; /* fill from the back to get ascending order */
85 0 : cnt++;
86 0 : }
87 0 : FD_TEST( cnt ); /* i==0 always counts */
88 :
89 0 : memmove( writer->slots, writer->slots+FD_TXNCACHE_WRITER_MAX_SLOT_DELTAS-cnt, cnt*sizeof(ulong) );
90 0 : writer->slot_cnt = cnt;
91 0 : }
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 0 : writer_map_execution_slots( fd_txncache_writer_t * writer ) {
101 0 : ulong blockhash_cnt = writer->blockhash_cnt;
102 0 : ulong slot_cnt = writer->slot_cnt;
103 0 : 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 0 : memset( writer->fork_id_to_blockhash_i, 0xFF, sizeof(writer->fork_id_to_blockhash_i) );
108 0 : for( ulong blockhash_i=0UL; blockhash_i<blockhash_cnt; blockhash_i++ ) {
109 0 : fd_txncache_writer_blockhash_desc_t * desc = &writer->blockhash_descs[ blockhash_i ];
110 0 : desc->slot_i = USHORT_MAX;
111 0 : writer->fork_id_to_blockhash_i[ desc->blockcache_idx ] = (ushort)blockhash_i;
112 0 : }
113 :
114 0 : ulong mapped_cnt = fd_ulong_min( blockhash_cnt, slot_cnt );
115 0 : ulong blockhash_i0 = blockhash_cnt-mapped_cnt;
116 0 : ulong slot_i0 = slot_cnt -mapped_cnt;
117 0 : for( ulong pair_i=0UL; pair_i<mapped_cnt; pair_i++ ) {
118 0 : ulong blockhash_i = blockhash_i0+pair_i;
119 0 : ulong slot_i = slot_i0 +pair_i;
120 0 : writer->blockhash_descs[ blockhash_i ].slot_i = (ushort)slot_i;
121 0 : }
122 0 : }
123 :
124 : static void
125 : writer_walk_blockhash( fd_txncache_writer_t * writer,
126 : ulong blockhash_idx,
127 0 : int mode ) {
128 :
129 0 : fd_txncache_blockcache_iter_t iter[1];
130 0 : for( fd_txncache_blockcache_iter_init( iter, writer->tc, writer->blockhash_descs[ blockhash_idx ].blockcache_idx );
131 0 : !fd_txncache_blockcache_iter_done( iter );
132 0 : fd_txncache_blockcache_iter_next( iter ) ) {
133 0 : fd_txncache_single_txn_t const * txn = fd_txncache_blockcache_iter_ele( iter );
134 :
135 0 : fd_txncache_writer_blockhash_desc_t const * exec_desc = writer_exec_desc( writer, txn );
136 0 : if( FD_UNLIKELY( !exec_desc ) ) continue;
137 :
138 0 : ushort slot_i = exec_desc->slot_i;
139 0 : 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 0 : ulong key = group_key( slot_i, blockhash_idx );
143 0 : if( mode==WALK_COUNT ) {
144 0 : writer->entry_cnt_by_key[ key ]++;
145 0 : continue;
146 0 : }
147 :
148 0 : ulong group_i = writer->group_i_by_key[ key ];
149 0 : if( group_i<writer->group_i || group_i>=writer->batch_hi ) continue;
150 0 : fd_txncache_writer_group_t * group = &writer->groups[ group_i ];
151 0 : 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 0 : memcpy( writer->arena[ group->arena_entry_off+group->filled_entry_cnt ], txn->txnhash, sizeof(writer->arena[0]) );
155 0 : group->filled_entry_cnt++;
156 0 : }
157 0 : fd_txncache_blockcache_iter_fini( iter );
158 0 : }
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 0 : writer_groups_init( fd_txncache_writer_t * writer ) {
167 0 : memset( writer->entry_cnt_by_key, 0, sizeof(writer->entry_cnt_by_key) );
168 0 : for( ulong blockhash_i=0UL; blockhash_i<writer->blockhash_cnt; blockhash_i++ ) {
169 0 : writer_walk_blockhash( writer, blockhash_i, WALK_COUNT );
170 0 : }
171 :
172 0 : ulong group_cnt = 0UL;
173 0 : ulong entry_cnt = 0UL;
174 0 : for( ulong slot_i=0UL; slot_i<writer->slot_cnt; slot_i++ ) {
175 0 : writer->group_cnt_by_slot_i[ slot_i ] = 0UL;
176 0 : for( ulong blockhash_i=0UL; blockhash_i<writer->blockhash_cnt; blockhash_i++ ) {
177 0 : ulong key = group_key( slot_i, blockhash_i );
178 0 : uint group_entry_cnt = writer->entry_cnt_by_key[ key ];
179 0 : if( !group_entry_cnt ) continue;
180 0 : fd_txncache_writer_group_t * group = &writer->groups[ group_cnt ];
181 0 : group->slot_i = (ushort)slot_i;
182 0 : group->blockhash_i = (ushort)blockhash_i;
183 0 : group->entry_cnt = group_entry_cnt;
184 0 : group->arena_entry_off = 0U;
185 0 : group->filled_entry_cnt = 0U;
186 0 : writer->group_i_by_key[ key ] = (uint)group_cnt;
187 0 : writer->group_cnt_by_slot_i[ slot_i ]++;
188 0 : group_cnt++;
189 0 : entry_cnt += group_entry_cnt;
190 0 : }
191 0 : }
192 0 : writer->group_cnt = group_cnt;
193 0 : writer->entry_cnt = entry_cnt;
194 0 : }
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 0 : writer_fill_batch( fd_txncache_writer_t * writer ) {
201 0 : ulong group_i0 = writer->group_i;
202 0 : fd_txncache_writer_group_t * first = &writer->groups[ group_i0 ];
203 0 : 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 0 : uchar blockhash_in_batch[ FD_TXNCACHE_WRITER_MAX_BLOCKHASHES ] = {0};
209 0 : ulong arena_entry_off = 0UL;
210 0 : ulong batch_hi = group_i0;
211 0 : while( batch_hi<writer->group_cnt && arena_entry_off+(ulong)writer->groups[ batch_hi ].entry_cnt<=writer->arena_entry_cnt_max ) {
212 0 : fd_txncache_writer_group_t * group = &writer->groups[ batch_hi ];
213 0 : group->arena_entry_off = (uint)arena_entry_off;
214 0 : group->filled_entry_cnt = 0U;
215 0 : blockhash_in_batch[ group->blockhash_i ] = 1;
216 0 : arena_entry_off += group->entry_cnt;
217 0 : batch_hi++;
218 0 : }
219 0 : writer->batch_hi = batch_hi;
220 :
221 0 : for( ulong blockhash_i=0UL; blockhash_i<writer->blockhash_cnt; blockhash_i++ ) {
222 0 : if( blockhash_in_batch[ blockhash_i ] ) writer_walk_blockhash( writer, blockhash_i, WALK_FILL );
223 0 : }
224 :
225 0 : for( ulong group_i=group_i0; group_i<batch_hi; group_i++ ) {
226 0 : fd_txncache_writer_group_t const * group = &writer->groups[ group_i ];
227 0 : 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 0 : }
231 0 : }
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 0 : ulong arena_sz ) {
241 0 : 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 0 : ulong arena_entry_cnt_max = arena_sz/sizeof(fd_txnhash_t);
246 0 : FD_CHECK_CRIT( arena_entry_cnt_max<=UINT_MAX, "txncache writer arena has too many entries" ); /* arena_entry_off is uint */
247 :
248 0 : writer->state = STATE_HEADER;
249 0 : writer->tc = tc;
250 0 : writer->snapshot_slot = snapshot_slot;
251 0 : writer->arena = arena;
252 0 : writer->arena_entry_cnt_max = arena_entry_cnt_max;
253 0 : writer->batch_hi = 0UL;
254 0 : writer->slot_i = 0UL;
255 0 : writer->group_i = 0UL;
256 0 : writer->entry_i = 0UL;
257 :
258 0 : writer_slots_init( writer, slot_history, slot_history_sz );
259 :
260 : /* Snapshot the root list. The snapshot root is the newest root. */
261 :
262 0 : fd_txncache_root_t roots[ FD_TXNCACHE_MAX_ROOTS ];
263 0 : ulong root_cnt = fd_txncache_snapshot_roots( tc, roots );
264 0 : for( ulong i=0UL; i<root_cnt; i++ ) {
265 0 : fd_txncache_writer_blockhash_desc_t * blockhash_desc = &writer->blockhash_descs[ i ];
266 0 : blockhash_desc->blockcache_idx = roots[ i ].blockcache_idx;
267 0 : blockhash_desc->generation = roots[ i ].generation;
268 0 : blockhash_desc->txnhash_offset = roots[ i ].txnhash_offset;
269 0 : memcpy( blockhash_desc->blockhash, roots[ i ].blockhash, 32UL );
270 0 : }
271 0 : writer->blockhash_cnt = root_cnt;
272 :
273 0 : writer_map_execution_slots( writer );
274 0 : writer_groups_init( writer );
275 :
276 0 : return writer;
277 0 : }
278 :
279 : FD_FN_PURE ulong
280 0 : fd_txncache_writer_serialized_sz( fd_txncache_writer_t const * writer ) {
281 0 : return HEADER_SZ+writer->slot_cnt*SLOT_DELTA_SZ+writer->group_cnt*GROUP_HDR_SZ+writer->entry_cnt*TXN_SZ;
282 0 : }
283 :
284 : ulong
285 : fd_txncache_writer_serialize( fd_txncache_writer_t * writer,
286 : uchar * out_buf,
287 0 : ulong buf_sz ) {
288 0 : 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 0 : uchar * p = out_buf;
292 0 : uchar * p1 = out_buf+buf_sz;
293 :
294 0 : for(;;) {
295 0 : switch( writer->state ) {
296 :
297 0 : case STATE_HEADER: {
298 0 : if( FD_UNLIKELY( p+HEADER_SZ>p1 ) ) goto done;
299 0 : FD_STORE( ulong, p, writer->slot_cnt ); p += 8UL; /* slot_deltas_len */
300 0 : writer->slot_i = 0UL;
301 0 : writer->group_i = 0UL;
302 0 : writer->state = writer->slot_cnt ? STATE_SLOT : STATE_DONE;
303 0 : break;
304 0 : }
305 :
306 0 : case STATE_SLOT: {
307 0 : if( FD_UNLIKELY( p+SLOT_DELTA_SZ>p1 ) ) goto done;
308 0 : ulong slot_i = writer->slot_i;
309 0 : FD_STORE( ulong, p, writer->slots[ slot_i ] ); p += 8UL; /* slot */
310 0 : FD_STORE( uchar, p, 1 ); p += 1UL; /* is_root */
311 0 : FD_STORE( ulong, p, writer->group_cnt_by_slot_i[ slot_i ] ); p += 8UL; /* status_len */
312 0 : if( writer->group_cnt_by_slot_i[ slot_i ] ) {
313 0 : writer->state = STATE_GROUP;
314 0 : } else {
315 0 : writer->slot_i++;
316 0 : if( writer->slot_i>=writer->slot_cnt ) writer->state = STATE_DONE;
317 0 : }
318 0 : break;
319 0 : }
320 :
321 0 : case STATE_GROUP: {
322 0 : fd_txncache_writer_group_t const * group = &writer->groups[ writer->group_i ];
323 0 : FD_TEST( group->slot_i==writer->slot_i );
324 0 : if( FD_UNLIKELY( writer->group_i>=writer->batch_hi ) ) writer_fill_batch( writer );
325 0 : if( FD_UNLIKELY( p+GROUP_HDR_SZ>p1 ) ) goto done;
326 0 : fd_txncache_writer_blockhash_desc_t const * blockhash_desc = &writer->blockhash_descs[ group->blockhash_i ];
327 0 : memcpy( p, blockhash_desc->blockhash, 32UL ); p += 32UL;
328 0 : FD_STORE( ulong, p, blockhash_desc->txnhash_offset ); p += 8UL;
329 0 : FD_STORE( ulong, p, (ulong)group->entry_cnt ); p += 8UL;
330 0 : writer->entry_i = 0UL;
331 0 : writer->state = STATE_ENTRIES;
332 0 : break;
333 0 : }
334 :
335 0 : case STATE_ENTRIES: {
336 0 : fd_txncache_writer_group_t const * group = &writer->groups[ writer->group_i ];
337 0 : while( writer->entry_i<group->entry_cnt ) {
338 0 : if( FD_UNLIKELY( p+TXN_SZ>p1 ) ) goto done;
339 0 : memcpy( p, writer->arena[ group->arena_entry_off+writer->entry_i ], sizeof(writer->arena[0]) ); p += sizeof(writer->arena[0]);
340 0 : FD_STORE( uint, p, 0U ); p += 4UL; /* result = Ok */
341 0 : writer->entry_i++;
342 0 : }
343 0 : writer->group_i++;
344 0 : if( writer->group_i<writer->group_cnt && writer->groups[ writer->group_i ].slot_i==writer->slot_i ) {
345 0 : writer->state = STATE_GROUP;
346 0 : } else {
347 0 : writer->slot_i++;
348 0 : writer->state = writer->slot_i<writer->slot_cnt ? STATE_SLOT : STATE_DONE;
349 0 : }
350 0 : break;
351 0 : }
352 :
353 0 : case STATE_DONE:
354 0 : goto done;
355 :
356 0 : default:
357 0 : FD_LOG_CRIT(( "invalid state reached (%u)", writer->state ));
358 0 : }
359 0 : }
360 :
361 0 : done:
362 0 : return (ulong)( p-out_buf );
363 0 : }
|
