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