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