Line data Source code
1 : #include "fd_txncache.h"
2 : #include "fd_txncache_private.h"
3 : #include "../../util/log/fd_log.h"
4 :
5 : struct blockcache {
6 : fd_txncache_blockcache_shmem_t * shmem;
7 :
8 : uint * heads; /* The hash table for the blockhash. Each entry is a pointer to the head of a linked list of
9 : transactions that reference this blockhash. As we add transactions to the bucket, the head
10 : pointer is updated to the new item, and the new item is pointed to the previous head. */
11 : ushort * pages; /* A list of the txnpages containing the transactions for this blockcache. */
12 :
13 : descends_set_t * descends; /* Each fork can descend from other forks in the txncache, and this bit vector contains one
14 : value for each fork in the txncache. If this fork descends from some other fork F, then
15 : the bit at index F in descends[] is set. */
16 : };
17 :
18 : typedef struct blockcache blockcache_t;
19 :
20 : struct fd_txncache_private {
21 : fd_txncache_shmem_t * shmem;
22 :
23 : fd_txncache_blockcache_shmem_t * blockcache_shmem_pool;
24 : blockcache_t * blockcache_pool;
25 : blockhash_map_t * blockhash_map;
26 :
27 : ushort * txnpages_free; /* The index in the txnpages array that is free, for each of the free pages. */
28 :
29 : fd_txncache_txnpage_t * txnpages; /* The actual storage for the transactions. The blockcache points to these
30 : pages when storing transactions. Transaction are grouped into pages of
31 : size 16384 to make certain allocation and deallocation operations faster
32 : (just the pages are acquired/released, rather than each txn). */
33 :
34 : ushort * scratch_pages;
35 : uint * scratch_heads;
36 : fd_txncache_txnpage_t * scratch_txnpage;
37 : };
38 :
39 : FD_FN_CONST ulong
40 279 : fd_txncache_align( void ) {
41 279 : return FD_TXNCACHE_ALIGN;
42 279 : }
43 :
44 : FD_FN_CONST ulong
45 132 : fd_txncache_footprint( ulong max_live_slots ) {
46 132 : ulong max_active_slots = FD_TXNCACHE_MAX_BLOCKHASH_DISTANCE+max_live_slots;
47 :
48 132 : ulong l;
49 132 : l = FD_LAYOUT_INIT;
50 132 : l = FD_LAYOUT_APPEND( l, FD_TXNCACHE_SHMEM_ALIGN, sizeof(fd_txncache_t) );
51 132 : l = FD_LAYOUT_APPEND( l, alignof(blockcache_t), max_active_slots*sizeof(blockcache_t) );
52 132 : return FD_LAYOUT_FINI( l, FD_TXNCACHE_ALIGN );
53 132 : }
54 :
55 : void *
56 : fd_txncache_new( void * ljoin,
57 72 : fd_txncache_shmem_t * shmem ) {
58 72 : if( FD_UNLIKELY( !ljoin ) ) {
59 0 : FD_LOG_WARNING(( "NULL ljoin" ));
60 0 : return NULL;
61 0 : }
62 :
63 72 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)ljoin, fd_txncache_align() ) ) ) {
64 0 : FD_LOG_WARNING(( "misaligned ljoin" ));
65 0 : return NULL;
66 0 : }
67 :
68 72 : ulong max_active_slots = shmem->active_slots_max;
69 72 : ulong blockhash_map_chains = fd_ulong_pow2_up( 2UL*shmem->active_slots_max );
70 72 : ulong bucket_cnt = shmem->bucket_cnt;
71 :
72 : /* Page counts come from the shmem header rather than being
73 : re-derived, so the layout walk below cannot desync from the one in
74 : fd_txncache_shmem_new. */
75 72 : ushort _max_txnpages = shmem->max_txnpages;
76 72 : ushort _max_txnpages_per_blockhash = shmem->txnpages_per_blockhash_max;
77 :
78 72 : ulong _descends_footprint = descends_set_footprint( max_active_slots );
79 72 : if( FD_UNLIKELY( !_descends_footprint ) ) {
80 0 : FD_LOG_WARNING(( "invalid max_active_slots" ));
81 0 : return NULL;
82 0 : }
83 :
84 72 : FD_SCRATCH_ALLOC_INIT( l, shmem );
85 72 : fd_txncache_shmem_t * tc = FD_SCRATCH_ALLOC_APPEND( l, FD_TXNCACHE_SHMEM_ALIGN, sizeof(fd_txncache_shmem_t) );
86 72 : void * _blockhash_map = FD_SCRATCH_ALLOC_APPEND( l, blockhash_map_align(), blockhash_map_footprint( blockhash_map_chains ) );
87 72 : void * _blockcache_pool = FD_SCRATCH_ALLOC_APPEND( l, blockcache_pool_align(), blockcache_pool_footprint( max_active_slots ) );
88 72 : void * _blockcache_pages = FD_SCRATCH_ALLOC_APPEND( l, alignof(ushort), max_active_slots*_max_txnpages_per_blockhash*sizeof(ushort) );
89 72 : void * _blockcache_heads = FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), max_active_slots*bucket_cnt*sizeof(uint) );
90 72 : void * _blockcache_descends = FD_SCRATCH_ALLOC_APPEND( l, descends_set_align(), max_active_slots*_descends_footprint );
91 72 : void * _txnpages_free = FD_SCRATCH_ALLOC_APPEND( l, alignof(ushort), _max_txnpages*sizeof(ushort) );
92 72 : void * _txnpages = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_txncache_txnpage_t), _max_txnpages*sizeof(fd_txncache_txnpage_t) );
93 72 : void * _scratch_pages = FD_SCRATCH_ALLOC_APPEND( l, alignof(ushort), _max_txnpages_per_blockhash*sizeof(ushort) );
94 72 : void * _scratch_heads = FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), bucket_cnt*sizeof(uint) );
95 72 : void * _scratch_txnpage = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_txncache_txnpage_t), sizeof(fd_txncache_txnpage_t) );
96 :
97 72 : FD_SCRATCH_ALLOC_INIT( l2, ljoin );
98 72 : fd_txncache_t * ltc = FD_SCRATCH_ALLOC_APPEND( l2, FD_TXNCACHE_ALIGN, sizeof(fd_txncache_t) );
99 72 : void * _local_blockcache_pool = FD_SCRATCH_ALLOC_APPEND( l2, alignof(blockcache_t), max_active_slots*sizeof(blockcache_t) );
100 :
101 72 : ltc->shmem = tc;
102 :
103 72 : ltc->blockcache_pool = (blockcache_t*)_local_blockcache_pool;
104 72 : ltc->blockcache_shmem_pool = blockcache_pool_join( _blockcache_pool );
105 :
106 12060 : for( ulong i=0UL; i<shmem->active_slots_max; i++ ) {
107 11988 : ltc->blockcache_pool[ i ].pages = (ushort *)_blockcache_pages + i*_max_txnpages_per_blockhash;
108 11988 : ltc->blockcache_pool[ i ].heads = (uint *)_blockcache_heads + i*bucket_cnt;
109 11988 : ltc->blockcache_pool[ i ].descends = descends_set_join( (uchar *)_blockcache_descends + i*_descends_footprint );
110 11988 : ltc->blockcache_pool[ i ].shmem = ltc->blockcache_shmem_pool + i;
111 11988 : FD_TEST( ltc->blockcache_pool[ i ].shmem );
112 11988 : }
113 :
114 72 : FD_TEST( ltc->blockcache_shmem_pool );
115 :
116 72 : ltc->blockhash_map = blockhash_map_join( _blockhash_map );
117 72 : FD_TEST( ltc->blockhash_map );
118 :
119 72 : ltc->txnpages_free = (ushort *)_txnpages_free;
120 72 : ltc->txnpages = (fd_txncache_txnpage_t *)_txnpages;
121 :
122 72 : ltc->scratch_pages = _scratch_pages;
123 72 : ltc->scratch_heads = _scratch_heads;
124 72 : ltc->scratch_txnpage = _scratch_txnpage;
125 :
126 72 : return (void *)ltc;
127 72 : }
128 :
129 : fd_txncache_t *
130 72 : fd_txncache_join( void * ljoin ) {
131 72 : if( FD_UNLIKELY( !ljoin ) ) {
132 0 : FD_LOG_WARNING(( "NULL ljoin" ));
133 0 : return NULL;
134 0 : }
135 :
136 72 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)ljoin, fd_txncache_align() ) ) ) {
137 0 : FD_LOG_WARNING(( "misaligned ljoin" ));
138 0 : return NULL;
139 0 : }
140 :
141 72 : fd_txncache_t * tc = (fd_txncache_t *)ljoin;
142 :
143 72 : return tc;
144 72 : }
145 :
146 : void
147 3945 : fd_txncache_reset( fd_txncache_t * tc ) {
148 3945 : fd_rwlock_write( tc->shmem->lock );
149 :
150 3945 : tc->shmem->root_cnt = 0UL;
151 3945 : root_slist_remove_all( tc->shmem->root_ll, tc->blockcache_shmem_pool );
152 :
153 3945 : tc->shmem->txnpages_free_cnt = tc->shmem->max_txnpages;
154 690804 : for( ushort i=0; i<tc->shmem->max_txnpages; i++ ) tc->txnpages_free[ i ] = i;
155 :
156 3945 : blockcache_pool_reset( tc->blockcache_shmem_pool );
157 3945 : blockhash_map_reset( tc->blockhash_map );
158 :
159 3945 : fd_rwlock_unwrite( tc->shmem->lock );
160 3945 : }
161 :
162 : void *
163 : fd_txncache_snapin_scratch( fd_txncache_t * tc,
164 6 : ulong * out_sz ) {
165 6 : FD_TEST_ERR( tc->shmem->txnpages_free_cnt==tc->shmem->max_txnpages );
166 6 : *out_sz = (ulong)tc->shmem->max_txnpages*sizeof(fd_txncache_txnpage_t);
167 6 : return tc->txnpages;
168 6 : }
169 :
170 : FD_FN_PURE static inline ulong
171 : fd_txncache_bucket( fd_txncache_t const * tc,
172 663 : uchar const * txnhash ) {
173 663 : return fd_ulong_hash( FD_LOAD( ulong, txnhash )^tc->shmem->seed )%tc->shmem->bucket_cnt;
174 663 : }
175 :
176 : static fd_txncache_txnpage_t *
177 : fd_txncache_ensure_txnpage( fd_txncache_t * tc,
178 273 : blockcache_t * blockcache ) {
179 273 : ushort page_cnt = blockcache->shmem->pages_cnt;
180 273 : if( FD_UNLIKELY( page_cnt>tc->shmem->txnpages_per_blockhash_max ) ) return NULL;
181 :
182 273 : if( FD_LIKELY( page_cnt ) ) {
183 138 : ushort txnpage_idx = blockcache->pages[ page_cnt-1 ];
184 138 : ushort txnpage_free = tc->txnpages[ txnpage_idx ].free;
185 138 : if( FD_LIKELY( txnpage_free ) ) return &tc->txnpages[ txnpage_idx ];
186 138 : }
187 :
188 135 : if( FD_UNLIKELY( page_cnt==tc->shmem->txnpages_per_blockhash_max ) ) return NULL;
189 135 : if( FD_LIKELY( FD_ATOMIC_CAS( &blockcache->pages[ page_cnt ], (ushort)USHORT_MAX, (ushort)(USHORT_MAX-1UL) )==(ushort)USHORT_MAX ) ) {
190 135 : ulong txnpages_free_cnt = tc->shmem->txnpages_free_cnt;
191 135 : for(;;) {
192 135 : if( FD_UNLIKELY( !txnpages_free_cnt ) ) {
193 0 : blockcache->pages[ page_cnt ] = (ushort)USHORT_MAX;
194 0 : FD_COMPILER_MFENCE();
195 0 : return NULL;
196 0 : }
197 135 : ulong old_txnpages_free_cnt = FD_ATOMIC_CAS( &tc->shmem->txnpages_free_cnt, (ushort)txnpages_free_cnt, (ushort)(txnpages_free_cnt-1UL) );
198 135 : if( FD_LIKELY( old_txnpages_free_cnt==txnpages_free_cnt ) ) break;
199 0 : txnpages_free_cnt = old_txnpages_free_cnt;
200 0 : FD_SPIN_PAUSE();
201 0 : }
202 :
203 135 : ushort txnpage_idx = tc->txnpages_free[ txnpages_free_cnt-1UL ];
204 135 : fd_txncache_txnpage_t * txnpage = &tc->txnpages[ txnpage_idx ];
205 135 : txnpage->free = FD_TXNCACHE_TXNS_PER_PAGE;
206 135 : FD_COMPILER_MFENCE();
207 135 : blockcache->pages[ page_cnt ] = txnpage_idx;
208 135 : FD_COMPILER_MFENCE();
209 135 : blockcache->shmem->pages_cnt = (ushort)(page_cnt+1);
210 135 : return txnpage;
211 135 : } else {
212 0 : ushort txnpage_idx = blockcache->pages[ page_cnt ];
213 0 : while( FD_UNLIKELY( txnpage_idx==(ushort)(USHORT_MAX-1UL) ) ) {
214 0 : txnpage_idx = blockcache->pages[ page_cnt ];
215 0 : FD_SPIN_PAUSE();
216 0 : }
217 0 : if( FD_UNLIKELY( txnpage_idx==(ushort)USHORT_MAX ) ) return NULL;
218 0 : return &tc->txnpages[ txnpage_idx ];
219 0 : }
220 135 : }
221 :
222 : static int
223 : fd_txncache_insert_txn( fd_txncache_t * tc,
224 : blockcache_t * blockcache,
225 : fd_txncache_txnpage_t * txnpage,
226 : fd_txncache_fork_id_t fork_id,
227 273 : uchar const * txnhash ) {
228 273 : ulong txnpage_idx = (ulong)(txnpage - tc->txnpages);
229 :
230 273 : for(;;) {
231 273 : ushort txnpage_free = txnpage->free;
232 273 : if( FD_UNLIKELY( !txnpage_free ) ) return 0;
233 273 : if( FD_UNLIKELY( FD_ATOMIC_CAS( &txnpage->free, txnpage_free, txnpage_free-1UL )!=txnpage_free ) ) {
234 0 : FD_SPIN_PAUSE();
235 0 : continue;
236 0 : }
237 :
238 273 : ulong txn_idx = FD_TXNCACHE_TXNS_PER_PAGE-txnpage_free;
239 273 : ulong txnhash_offset = blockcache->shmem->txnhash_offset;
240 273 : memcpy( txnpage->txns[ txn_idx ]->txnhash, txnhash+txnhash_offset, 20UL );
241 273 : txnpage->txns[ txn_idx ]->fork_id = fork_id;
242 273 : txnpage->txns[ txn_idx ]->generation = tc->blockcache_pool[ fork_id.val ].shmem->generation;
243 273 : FD_COMPILER_MFENCE();
244 :
245 273 : ulong txn_bucket = fd_txncache_bucket( tc, txnhash+txnhash_offset );
246 273 : for(;;) {
247 273 : uint head = blockcache->heads[ txn_bucket ];
248 273 : txnpage->txns[ txn_idx ]->blockcache_next = head;
249 273 : FD_COMPILER_MFENCE();
250 273 : if( FD_LIKELY( FD_ATOMIC_CAS( &blockcache->heads[ txn_bucket ], head, (uint)(FD_TXNCACHE_TXNS_PER_PAGE*txnpage_idx+txn_idx) )==head ) ) break;
251 0 : FD_SPIN_PAUSE();
252 0 : }
253 :
254 273 : return 1;
255 273 : }
256 273 : }
257 :
258 : fd_txncache_fork_id_t
259 : fd_txncache_attach_child( fd_txncache_t * tc,
260 8094 : fd_txncache_fork_id_t parent_fork_id ) {
261 8094 : fd_rwlock_write( tc->shmem->lock );
262 :
263 8094 : FD_TEST( blockcache_pool_free( tc->blockcache_shmem_pool ) );
264 8094 : ulong idx = blockcache_pool_idx_acquire( tc->blockcache_shmem_pool );
265 :
266 8094 : blockcache_t * fork = &tc->blockcache_pool[ idx ];
267 8094 : fd_txncache_fork_id_t fork_id = { .val = (ushort)idx };
268 :
269 8094 : fork->shmem->generation = tc->shmem->blockcache_generation++;
270 8094 : fork->shmem->child_id = (fd_txncache_fork_id_t){ .val = USHORT_MAX };
271 :
272 8094 : if( FD_LIKELY( parent_fork_id.val==USHORT_MAX ) ) {
273 3951 : FD_TEST( blockcache_pool_free( tc->blockcache_shmem_pool )==blockcache_pool_max( tc->blockcache_shmem_pool )-1UL );
274 3951 : fork->shmem->parent_id = (fd_txncache_fork_id_t){ .val = USHORT_MAX };
275 3951 : fork->shmem->sibling_id = (fd_txncache_fork_id_t){ .val = USHORT_MAX };
276 :
277 3951 : descends_set_null( fork->descends );
278 3951 : root_slist_ele_push_tail( tc->shmem->root_ll, fork->shmem, tc->blockcache_shmem_pool );
279 4143 : } else {
280 4143 : blockcache_t * parent = &tc->blockcache_pool[ parent_fork_id.val ];
281 : /* We might be tempted to freeze the parent here, and it's valid to
282 : do this ordinarily, but not when loading from a snapshot, when
283 : we need to load many transactions into a root parent chain at
284 : once. */
285 4143 : fork->shmem->sibling_id = parent->shmem->child_id;
286 4143 : fork->shmem->parent_id = parent_fork_id;
287 4143 : parent->shmem->child_id = fork_id;
288 :
289 4143 : descends_set_copy( fork->descends, parent->descends );
290 4143 : descends_set_insert( fork->descends, parent_fork_id.val );
291 4143 : }
292 :
293 8094 : fork->shmem->txnhash_offset = 0UL;
294 8094 : fork->shmem->frozen = 0;
295 8094 : memset( fork->heads, 0xFF, tc->shmem->bucket_cnt*sizeof(uint) );
296 8094 : fork->shmem->pages_cnt = 0;
297 8094 : memset( fork->pages, 0xFF, tc->shmem->txnpages_per_blockhash_max*sizeof(fork->pages[ 0 ]) );
298 :
299 8094 : fd_rwlock_unwrite( tc->shmem->lock );
300 8094 : return fork_id;
301 8094 : }
302 :
303 : void
304 : fd_txncache_attach_blockhash( fd_txncache_t * tc,
305 : fd_txncache_fork_id_t fork_id,
306 9 : uchar const * blockhash ) {
307 9 : fd_rwlock_write( tc->shmem->lock );
308 :
309 9 : blockcache_t * fork = &tc->blockcache_pool[ fork_id.val ];
310 9 : FD_TEST( !fork->shmem->frozen );
311 9 : fork->shmem->frozen = 1;
312 :
313 9 : memcpy( fork->shmem->blockhash.uc, blockhash, 32UL );
314 :
315 9 : blockhash_map_ele_insert( tc->blockhash_map, fork->shmem, tc->blockcache_shmem_pool );
316 :
317 9 : fd_rwlock_unwrite( tc->shmem->lock );
318 9 : }
319 :
320 : void
321 : fd_txncache_finalize_fork( fd_txncache_t * tc,
322 : fd_txncache_fork_id_t fork_id,
323 : ulong txnhash_offset,
324 4350 : uchar const * blockhash ) {
325 4350 : fd_rwlock_write( tc->shmem->lock );
326 :
327 4350 : blockcache_t * fork = &tc->blockcache_pool[ fork_id.val ];
328 4350 : FD_TEST( fork->shmem->frozen<=1 );
329 4350 : FD_TEST( fork->shmem->frozen>=0 );
330 4350 : fork->shmem->txnhash_offset = txnhash_offset;
331 :
332 4350 : memcpy( fork->shmem->blockhash.uc, blockhash, 32UL );
333 :
334 4350 : if( FD_LIKELY( !fork->shmem->frozen ) ) blockhash_map_ele_insert( tc->blockhash_map, fork->shmem, tc->blockcache_shmem_pool );
335 4350 : fork->shmem->frozen = 2;
336 :
337 4350 : fd_rwlock_unwrite( tc->shmem->lock );
338 4350 : }
339 :
340 : static inline void
341 : remove_blockcache( fd_txncache_t * tc,
342 0 : blockcache_t * blockcache ) {
343 0 : FD_TEST( blockcache->shmem->frozen>=0 );
344 0 : memcpy( tc->txnpages_free+tc->shmem->txnpages_free_cnt, blockcache->pages, blockcache->shmem->pages_cnt*sizeof(tc->txnpages_free[ 0 ]) );
345 0 : tc->shmem->txnpages_free_cnt = (ushort)(tc->shmem->txnpages_free_cnt+blockcache->shmem->pages_cnt);
346 :
347 0 : ulong idx = blockcache_pool_idx( tc->blockcache_shmem_pool, blockcache->shmem );
348 0 : for( ulong i=0UL; i<tc->shmem->active_slots_max; i++ ) descends_set_remove( tc->blockcache_pool[ i ].descends, idx );
349 :
350 0 : if( FD_LIKELY( blockcache->shmem->frozen ) ) blockhash_map_ele_remove_fast( tc->blockhash_map, blockcache->shmem, tc->blockcache_shmem_pool );
351 0 : blockcache->shmem->frozen = -1;
352 0 : blockcache_pool_ele_release( tc->blockcache_shmem_pool, blockcache->shmem );
353 0 : }
354 :
355 : static inline void
356 : remove_children( fd_txncache_t * tc,
357 : blockcache_t const * fork,
358 99 : blockcache_t const * except ) {
359 99 : fd_txncache_fork_id_t sibling_idx = fork->shmem->child_id;
360 198 : while( sibling_idx.val!=USHORT_MAX ) {
361 99 : blockcache_t * sibling = &tc->blockcache_pool[ sibling_idx.val ];
362 :
363 99 : sibling_idx = sibling->shmem->sibling_id;
364 99 : if( FD_UNLIKELY( sibling==except ) ) continue;
365 :
366 0 : remove_children( tc, sibling, except );
367 0 : remove_blockcache( tc, sibling );
368 0 : }
369 99 : }
370 :
371 : void
372 : fd_txncache_cancel_fork( fd_txncache_t * tc,
373 0 : fd_txncache_fork_id_t fork_id ) {
374 0 : fd_rwlock_write( tc->shmem->lock );
375 0 : blockcache_t * fork = &tc->blockcache_pool[ fork_id.val ];
376 0 : FD_TEST( fork->shmem->parent_id.val!=USHORT_MAX );
377 :
378 : /* The soon-to-be-pruned subtree must be unrooted. */
379 0 : fd_txncache_blockcache_shmem_t const * latest_root = root_slist_ele_peek_tail_const( tc->shmem->root_ll, tc->blockcache_shmem_pool );
380 0 : FD_TEST( latest_root );
381 0 : FD_TEST( descends_set_test( fork->descends, blockcache_pool_idx( tc->blockcache_shmem_pool, latest_root ) ) );
382 :
383 0 : remove_children( tc, fork, NULL );
384 0 : remove_blockcache( tc, fork );
385 0 : ushort * fork_id_p = &(tc->blockcache_pool[ fork->shmem->parent_id.val ].shmem->child_id.val);
386 0 : while( *fork_id_p!=fork_id.val ) {
387 0 : fork_id_p = &(tc->blockcache_pool[ *fork_id_p ].shmem->sibling_id.val);
388 0 : }
389 0 : *fork_id_p = fork->shmem->sibling_id.val;
390 0 : fd_rwlock_unwrite( tc->shmem->lock );
391 0 : }
392 :
393 : void
394 : fd_txncache_advance_root( fd_txncache_t * tc,
395 99 : fd_txncache_fork_id_t fork_id ) {
396 99 : fd_rwlock_write( tc->shmem->lock );
397 :
398 99 : blockcache_t * fork = &tc->blockcache_pool[ fork_id.val ];
399 99 : FD_TEST( fork->shmem->parent_id.val!=USHORT_MAX );
400 :
401 99 : blockcache_t * parent_fork = &tc->blockcache_pool[ fork->shmem->parent_id.val ];
402 99 : if( FD_UNLIKELY( root_slist_ele_peek_tail( tc->shmem->root_ll, tc->blockcache_shmem_pool )!=parent_fork->shmem ) ) {
403 0 : FD_BASE58_ENCODE_32_BYTES( parent_fork->shmem->blockhash.uc, parent_blockhash_b58 );
404 0 : FD_BASE58_ENCODE_32_BYTES( fork->shmem->blockhash.uc, fork_blockhash_b58 );
405 0 : FD_BASE58_ENCODE_32_BYTES( root_slist_ele_peek_tail( tc->shmem->root_ll, tc->blockcache_shmem_pool )->blockhash.uc, root_blockhash_b58 );
406 0 : FD_LOG_CRIT(( "advancing root from %s to %s but that is not valid, last root is %s",
407 0 : parent_blockhash_b58,
408 0 : fork_blockhash_b58,
409 0 : root_blockhash_b58 ));
410 0 : }
411 :
412 99 : FD_BASE58_ENCODE_32_BYTES( parent_fork->shmem->blockhash.uc, parent_blockhash_b58 );
413 99 : FD_BASE58_ENCODE_32_BYTES( fork->shmem->blockhash.uc, fork_blockhash_b58 );
414 99 : FD_LOG_DEBUG(( "advancing root from %s to %s",
415 99 : parent_blockhash_b58,
416 99 : fork_blockhash_b58 ));
417 :
418 : /* When a fork is rooted, any competing forks can be immediately
419 : removed as they will not be needed again. This includes child
420 : forks of the pruned siblings as well. */
421 99 : remove_children( tc, parent_fork, fork );
422 99 : parent_fork->shmem->child_id = fork_id;
423 99 : fork->shmem->sibling_id = (fd_txncache_fork_id_t){ .val = USHORT_MAX };
424 :
425 : /* Now, the earliest known rooted fork can likely be removed since its
426 : blockhashes cannot be referenced anymore (they are older than 151
427 : blockhashes away). */
428 99 : tc->shmem->root_cnt++;
429 99 : root_slist_ele_push_tail( tc->shmem->root_ll, fork->shmem, tc->blockcache_shmem_pool );
430 99 : if( FD_LIKELY( tc->shmem->root_cnt>FD_TXNCACHE_MAX_BLOCKHASH_DISTANCE ) ) {
431 0 : fd_txncache_blockcache_shmem_t * old_root_shmem = root_slist_ele_pop_head( tc->shmem->root_ll, tc->blockcache_shmem_pool );
432 0 : FD_TEST( old_root_shmem );
433 0 : blockcache_t * old_root = &tc->blockcache_pool[ blockcache_pool_idx( tc->blockcache_shmem_pool, old_root_shmem ) ];
434 :
435 0 : root_slist_ele_peek_head( tc->shmem->root_ll, tc->blockcache_shmem_pool )->parent_id.val = USHORT_MAX;
436 :
437 0 : remove_blockcache( tc, old_root );
438 0 : tc->shmem->root_cnt--;
439 0 : }
440 :
441 99 : fd_rwlock_unwrite( tc->shmem->lock );
442 99 : }
443 :
444 : static inline blockcache_t *
445 : blockhash_on_fork( fd_txncache_t * tc,
446 : blockcache_t const * fork,
447 663 : uchar const * blockhash ) {
448 663 : fd_txncache_blockcache_shmem_t const * candidate = blockhash_map_ele_query_const( tc->blockhash_map, fd_type_pun_const( blockhash ), NULL, tc->blockcache_shmem_pool );
449 663 : if( FD_UNLIKELY( !candidate ) ) return NULL;
450 :
451 663 : while( candidate ) {
452 663 : ulong candidate_idx = blockcache_pool_idx( tc->blockcache_shmem_pool, candidate );
453 663 : if( FD_LIKELY( descends_set_test( fork->descends, candidate_idx ) ) ) return &tc->blockcache_pool[ candidate_idx ];
454 0 : candidate = blockhash_map_ele_next_const( candidate, NULL, tc->blockcache_shmem_pool );
455 0 : }
456 0 : return NULL;
457 663 : }
458 :
459 : static void
460 : purge_stale_on_blockcache( fd_txncache_t * tc,
461 0 : blockcache_t * blockcache ) {
462 0 : FD_TEST( blockcache->shmem->frozen>=0 );
463 0 : memset( tc->scratch_heads, 0xFF, tc->shmem->bucket_cnt*sizeof(tc->scratch_heads[ 0 ]) );
464 0 : memset( tc->scratch_pages, 0xFF, tc->shmem->txnpages_per_blockhash_max*sizeof(tc->scratch_pages[ 0 ]) );
465 0 : ushort scratch_pages_cnt = 0;
466 0 : ushort scratch_txnpage_idx = USHORT_MAX;
467 0 : tc->scratch_txnpage->free = 0;
468 0 : for( ulong i=0UL; i<blockcache->shmem->pages_cnt; i++ ) {
469 0 : ushort curr_txnpage_idx = blockcache->pages[ blockcache->shmem->pages_cnt-i-1UL ];
470 0 : ulong curr_txn_cnt = FD_TXNCACHE_TXNS_PER_PAGE-tc->txnpages[ curr_txnpage_idx ].free;
471 0 : for( ulong j=0UL; j<curr_txn_cnt; j++ ) {
472 0 : fd_txncache_single_txn_t * curr_txn = tc->txnpages[ curr_txnpage_idx ].txns[ curr_txn_cnt-j-1UL ];
473 0 : blockcache_t const * txn_fork = &tc->blockcache_pool[ curr_txn->fork_id.val ];
474 0 : if( FD_LIKELY( txn_fork->shmem->frozen>=0 && txn_fork->shmem->generation==curr_txn->generation ) ) {
475 : /* Valid transaction. Keep. */
476 0 : if( FD_UNLIKELY( !tc->scratch_txnpage->free ) ) {
477 0 : FD_TEST( scratch_txnpage_idx!=curr_txnpage_idx );
478 0 : if( FD_LIKELY( scratch_txnpage_idx!=USHORT_MAX ) ) {
479 0 : fd_txncache_txnpage_t * txnpage = &tc->txnpages[ scratch_txnpage_idx ];
480 0 : memcpy( txnpage, tc->scratch_txnpage, sizeof(*txnpage) );
481 0 : }
482 0 : scratch_txnpage_idx = curr_txnpage_idx;
483 0 : tc->scratch_txnpage->free = FD_TXNCACHE_TXNS_PER_PAGE;
484 0 : tc->scratch_pages[ scratch_pages_cnt ] = scratch_txnpage_idx;
485 0 : scratch_pages_cnt++;
486 0 : }
487 0 : ulong txn_idx = FD_TXNCACHE_TXNS_PER_PAGE-tc->scratch_txnpage->free;
488 0 : memcpy( tc->scratch_txnpage->txns[ txn_idx ], curr_txn, sizeof(*curr_txn) );
489 0 : ulong txn_bucket = fd_txncache_bucket( tc, curr_txn->txnhash );
490 0 : uint head = tc->scratch_heads[ txn_bucket ];
491 0 : tc->scratch_txnpage->txns[ txn_idx ]->blockcache_next = head;
492 0 : ulong txn_gidx = FD_TXNCACHE_TXNS_PER_PAGE*scratch_txnpage_idx+txn_idx;
493 0 : FD_TEST( txn_gidx<UINT_MAX );
494 0 : tc->scratch_heads[ txn_bucket ] = (uint)txn_gidx;
495 0 : tc->scratch_txnpage->free--;
496 0 : } else {
497 : /* Stale transaction. Drop. */
498 0 : continue;
499 0 : }
500 0 : }
501 0 : if( FD_UNLIKELY( curr_txnpage_idx!=scratch_txnpage_idx ) ) {
502 : /* The txnpage is not being used for compaction, free it up. */
503 0 : tc->txnpages_free[ tc->shmem->txnpages_free_cnt ] = curr_txnpage_idx;
504 0 : tc->shmem->txnpages_free_cnt++;
505 0 : }
506 0 : }
507 0 : if( FD_LIKELY( scratch_txnpage_idx!=USHORT_MAX ) ) {
508 0 : fd_txncache_txnpage_t * txnpage = &tc->txnpages[ scratch_txnpage_idx ];
509 0 : memcpy( txnpage, tc->scratch_txnpage, sizeof(*txnpage) );
510 0 : }
511 0 : blockcache->shmem->pages_cnt = scratch_pages_cnt;
512 0 : memcpy( blockcache->pages, tc->scratch_pages, tc->shmem->txnpages_per_blockhash_max*sizeof(blockcache->pages[0]) );
513 0 : memcpy( blockcache->heads, tc->scratch_heads, tc->shmem->bucket_cnt*sizeof(blockcache->heads[0]) );
514 0 : }
515 :
516 : static void
517 : purge_stale_on_fork( fd_txncache_t * tc,
518 0 : blockcache_t * fork ) {
519 0 : purge_stale_on_blockcache( tc, fork );
520 :
521 0 : fd_txncache_fork_id_t sibling_idx = fork->shmem->child_id;
522 0 : while( sibling_idx.val!=USHORT_MAX ) {
523 0 : blockcache_t * sibling = &tc->blockcache_pool[ sibling_idx.val ];
524 0 : purge_stale_on_fork( tc, sibling );
525 0 : sibling_idx = sibling->shmem->sibling_id;
526 0 : }
527 0 : }
528 :
529 : static void
530 0 : purge_stale( fd_txncache_t * tc ) {
531 0 : fd_txncache_blockcache_shmem_t * root_shmem = root_slist_ele_peek_head( tc->shmem->root_ll, tc->blockcache_shmem_pool );
532 0 : FD_TEST( root_shmem );
533 0 : blockcache_t * root = &tc->blockcache_pool[ blockcache_pool_idx( tc->blockcache_shmem_pool, root_shmem ) ];
534 0 : ushort free_before = tc->shmem->txnpages_free_cnt;
535 : /* One might think that an optimization here is to stop the descent on
536 : the latest rooted blockcache. There could be no pruned minority
537 : forks from that point on. As a result, there could be no stale
538 : transactions in any blockcache descending from that.
539 : Unfortunately, frontier eviction means that any blockcache in the
540 : fork tree can have stale transactions. */
541 0 : purge_stale_on_fork( tc, root );
542 0 : FD_LOG_WARNING(( "purge_stale: txnpages_free %hu -> %hu", free_before, tc->shmem->txnpages_free_cnt ));
543 0 : }
544 :
545 : void
546 : fd_txncache_insert( fd_txncache_t * tc,
547 : fd_txncache_fork_id_t fork_id,
548 : uchar const * blockhash,
549 273 : uchar const * txnhash ) {
550 273 : fd_rwlock_read( tc->shmem->lock );
551 :
552 273 : blockcache_t const * fork = &tc->blockcache_pool[ fork_id.val ];
553 273 : FD_TEST( fork->shmem->frozen<=1 );
554 273 : FD_TEST( fork->shmem->frozen>=0 );
555 273 : blockcache_t * blockcache = blockhash_on_fork( tc, fork, blockhash );
556 273 : FD_TEST( blockcache );
557 :
558 273 : for(;;) {
559 273 : fd_txncache_txnpage_t * txnpage = fd_txncache_ensure_txnpage( tc, blockcache );
560 273 : if( FD_UNLIKELY( !txnpage ) ) {
561 : /* Because of sizing invariants when creating the structure, it is
562 : not typically possible to fill it, unless there are stale
563 : transactions from minority forks that were purged floating
564 : around, in which case we can purge them here and try again. */
565 0 : fd_rwlock_unread( tc->shmem->lock );
566 0 : fd_rwlock_write( tc->shmem->lock );
567 0 : if( FD_LIKELY( !fd_txncache_ensure_txnpage( tc, blockcache ) ) ) purge_stale( tc );
568 0 : fd_rwlock_unwrite( tc->shmem->lock );
569 0 : fd_rwlock_read( tc->shmem->lock );
570 0 : continue;
571 0 : }
572 :
573 273 : int success = fd_txncache_insert_txn( tc, blockcache, txnpage, fork_id, txnhash );
574 273 : if( FD_LIKELY( success ) ) break;
575 :
576 0 : FD_SPIN_PAUSE();
577 0 : }
578 :
579 273 : fd_rwlock_unread( tc->shmem->lock );
580 273 : }
581 :
582 : int
583 : fd_txncache_query( fd_txncache_t * tc,
584 : fd_txncache_fork_id_t fork_id,
585 : uchar const * blockhash,
586 390 : uchar const * txnhash ) {
587 390 : fd_rwlock_read( tc->shmem->lock );
588 :
589 390 : blockcache_t const * fork = &tc->blockcache_pool[ fork_id.val ];
590 390 : FD_TEST( fork->shmem->frozen>=0 );
591 390 : blockcache_t const * blockcache = blockhash_on_fork( tc, fork, blockhash );
592 390 : FD_TEST( blockcache );
593 390 : FD_TEST( blockcache->shmem->frozen==2 );
594 :
595 390 : int found = 0;
596 :
597 390 : ulong txnhash_offset = blockcache->shmem->txnhash_offset;
598 390 : ulong head_hash = fd_txncache_bucket( tc, txnhash+txnhash_offset );
599 405 : for( uint head=blockcache->heads[ head_hash ]; head!=UINT_MAX; head=tc->txnpages[ head/FD_TXNCACHE_TXNS_PER_PAGE ].txns[ head%FD_TXNCACHE_TXNS_PER_PAGE ]->blockcache_next ) {
600 18 : fd_txncache_single_txn_t * txn = tc->txnpages[ head/FD_TXNCACHE_TXNS_PER_PAGE ].txns[ head%FD_TXNCACHE_TXNS_PER_PAGE ];
601 :
602 18 : blockcache_t const * txn_fork = &tc->blockcache_pool[ txn->fork_id.val ];
603 18 : int descends = (txn->fork_id.val==fork_id.val || descends_set_test( fork->descends, txn->fork_id.val )) && txn_fork->shmem->frozen>=0 && txn_fork->shmem->generation==txn->generation;
604 18 : if( FD_LIKELY( descends && !memcmp( txnhash+txnhash_offset, txn->txnhash, 20UL ) ) ) {
605 3 : found = 1;
606 3 : break;
607 3 : }
608 18 : }
609 :
610 390 : fd_rwlock_unread( tc->shmem->lock );
611 390 : return found;
612 390 : }
|
