Line data Source code
1 : #include "fd_txncache.h"
2 : #include "fd_txncache_private.h"
3 : #include "../../util/log/fd_log.h"
4 : #include "../../util/racesan/fd_racesan_target.h"
5 :
6 : #include <unistd.h>
7 : #include <errno.h>
8 :
9 : FD_FN_CONST ulong
10 0 : fd_txncache_align( void ) {
11 0 : return FD_TXNCACHE_ALIGN;
12 0 : }
13 :
14 : FD_FN_CONST ulong
15 0 : fd_txncache_footprint( ulong max_live_slots ) {
16 0 : ulong max_active_slots = FD_TXNCACHE_MAX_BLOCKHASH_DISTANCE+max_live_slots;
17 :
18 0 : ulong l;
19 0 : l = FD_LAYOUT_INIT;
20 0 : l = FD_LAYOUT_APPEND( l, FD_TXNCACHE_ALIGN, sizeof(fd_txncache_t) );
21 0 : l = FD_LAYOUT_APPEND( l, alignof(blockcache_t), max_active_slots*sizeof(blockcache_t) );
22 0 : return FD_LAYOUT_FINI( l, FD_TXNCACHE_ALIGN );
23 0 : }
24 :
25 : void *
26 : fd_txncache_new( void * ljoin,
27 : fd_txncache_shmem_t * shmem,
28 0 : int spill_fd ) {
29 0 : if( FD_UNLIKELY( !ljoin ) ) {
30 0 : FD_LOG_WARNING(( "NULL ljoin" ));
31 0 : return NULL;
32 0 : }
33 :
34 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)ljoin, fd_txncache_align() ) ) ) {
35 0 : FD_LOG_WARNING(( "misaligned ljoin" ));
36 0 : return NULL;
37 0 : }
38 :
39 0 : ulong max_active_slots = shmem->active_slots_max;
40 0 : ulong blockhash_map_chains = fd_ulong_pow2_up( 2UL*shmem->active_slots_max );
41 0 : ulong bucket_cnt = shmem->bucket_cnt;
42 :
43 : /* Page counts come from the shmem header rather than being
44 : re-derived, so the layout walk below cannot desync from the one in
45 : fd_txncache_shmem_new. */
46 0 : ulong _txnpages_max = shmem->txnpages_max;
47 :
48 0 : ulong _descends_footprint = descends_set_footprint( max_active_slots );
49 0 : if( FD_UNLIKELY( !_descends_footprint ) ) {
50 0 : FD_LOG_WARNING(( "invalid max_active_slots" ));
51 0 : return NULL;
52 0 : }
53 :
54 0 : FD_SCRATCH_ALLOC_INIT( l, shmem );
55 0 : fd_txncache_shmem_t * tc = FD_SCRATCH_ALLOC_APPEND( l, FD_TXNCACHE_SHMEM_ALIGN, sizeof(fd_txncache_shmem_t) );
56 0 : void * _blockhash_map = FD_SCRATCH_ALLOC_APPEND( l, blockhash_map_align(), blockhash_map_footprint( blockhash_map_chains ) );
57 0 : void * _blockcache_pool = FD_SCRATCH_ALLOC_APPEND( l, blockcache_pool_align(), blockcache_pool_footprint( max_active_slots ) );
58 0 : void * _blockcache_heads = FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), max_active_slots*bucket_cnt*sizeof(uint) );
59 0 : void * _blockcache_descends = FD_SCRATCH_ALLOC_APPEND( l, descends_set_align(), max_active_slots*_descends_footprint );
60 0 : void * _txnpages_free = FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), _txnpages_max*sizeof(uint) );
61 0 : void * _page_meta = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_txncache_page_meta_t), _txnpages_max*sizeof(fd_txncache_page_meta_t) );
62 0 : void * _frame_page = FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), shmem->frames_max*sizeof(uint) );
63 0 : void * _frames_free = FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), shmem->frames_max*sizeof(uint) );
64 0 : void * _txnpages = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_txncache_txnpage_t), shmem->frames_max*sizeof(fd_txncache_txnpage_t) );
65 0 : void * _scratch_heads = FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), bucket_cnt*sizeof(uint) );
66 0 : void * _scratch_txnpage = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_txncache_txnpage_t), sizeof(fd_txncache_txnpage_t) );
67 0 : void * _scratch_input = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_txncache_txnpage_t), sizeof(fd_txncache_txnpage_t) );
68 :
69 0 : FD_SCRATCH_ALLOC_INIT( l2, ljoin );
70 0 : fd_txncache_t * ltc = FD_SCRATCH_ALLOC_APPEND( l2, FD_TXNCACHE_ALIGN, sizeof(fd_txncache_t) );
71 0 : void * _local_blockcache_pool = FD_SCRATCH_ALLOC_APPEND( l2, alignof(blockcache_t), max_active_slots*sizeof(blockcache_t) );
72 :
73 0 : ltc->shmem = tc;
74 :
75 0 : ltc->blockcache_pool = (blockcache_t*)_local_blockcache_pool;
76 0 : ltc->blockcache_shmem_pool = blockcache_pool_join( _blockcache_pool );
77 :
78 0 : for( ulong i=0UL; i<shmem->active_slots_max; i++ ) {
79 0 : ltc->blockcache_pool[ i ].heads = (uint *)_blockcache_heads + i*bucket_cnt;
80 0 : ltc->blockcache_pool[ i ].descends = descends_set_join( (uchar *)_blockcache_descends + i*_descends_footprint );
81 0 : ltc->blockcache_pool[ i ].shmem = ltc->blockcache_shmem_pool + i;
82 0 : FD_TEST( ltc->blockcache_pool[ i ].shmem );
83 0 : }
84 :
85 0 : FD_TEST( ltc->blockcache_shmem_pool );
86 :
87 0 : ltc->blockhash_map = blockhash_map_join( _blockhash_map );
88 0 : FD_TEST( ltc->blockhash_map );
89 :
90 0 : ltc->txnpages_free = _txnpages_free;
91 0 : ltc->page_meta = _page_meta;
92 0 : ltc->frame_page = _frame_page;
93 0 : ltc->frames_free = _frames_free;
94 0 : ltc->txnpages = (fd_txncache_txnpage_t *)_txnpages;
95 0 : ltc->scratch_heads = _scratch_heads;
96 0 : ltc->scratch_txnpage = _scratch_txnpage;
97 0 : ltc->scratch_input = _scratch_input;
98 0 : ltc->spill_fd = spill_fd;
99 :
100 0 : memset( ltc->metrics, 0, sizeof(ltc->metrics) );
101 0 : fd_histf_join( fd_histf_new( ltc->metrics->query_ticks, 1UL, 1UL<<30 ) );
102 0 : fd_histf_join( fd_histf_new( ltc->metrics->insert_ticks, 1UL, 1UL<<30 ) );
103 :
104 0 : return (void *)ltc;
105 0 : }
106 :
107 : fd_txncache_join_metrics_t *
108 0 : fd_txncache_join_metrics( fd_txncache_t * tc ) {
109 0 : return tc->metrics;
110 0 : }
111 :
112 : fd_txncache_t *
113 0 : fd_txncache_join( void * ljoin ) {
114 0 : if( FD_UNLIKELY( !ljoin ) ) {
115 0 : FD_LOG_WARNING(( "NULL ljoin" ));
116 0 : return NULL;
117 0 : }
118 :
119 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)ljoin, fd_txncache_align() ) ) ) {
120 0 : FD_LOG_WARNING(( "misaligned ljoin" ));
121 0 : return NULL;
122 0 : }
123 :
124 0 : fd_txncache_t * tc = (fd_txncache_t *)ljoin;
125 :
126 0 : return tc;
127 0 : }
128 :
129 : /* lock_read and lock_write take the txncache lock, adding the time
130 : spent waiting to the lock wait metrics. Uncontended acquisitions
131 : are not timed. */
132 :
133 : static inline void
134 0 : lock_read( fd_txncache_t * tc ) FD_ACQUIRE_SHARED( tc->shmem->lock ) FD_NO_THREAD_SAFETY_ANALYSIS {
135 0 : if( FD_LIKELY( fd_rwlock_tryread( tc->shmem->lock ) ) ) return;
136 0 : long t0 = fd_tickcount();
137 0 : fd_rwlock_read( tc->shmem->lock );
138 0 : FD_ATOMIC_FETCH_AND_ADD( &tc->shmem->lock_read_wait_ticks, (ulong)(fd_tickcount()-t0) );
139 0 : }
140 :
141 : static inline void
142 0 : lock_write( fd_txncache_t * tc ) FD_ACQUIRE( tc->shmem->lock ) FD_NO_THREAD_SAFETY_ANALYSIS {
143 0 : if( FD_LIKELY( fd_rwlock_trywrite( tc->shmem->lock ) ) ) return;
144 0 : long t0 = fd_tickcount();
145 0 : fd_rwlock_write( tc->shmem->lock );
146 0 : tc->shmem->lock_write_wait_ticks += (ulong)(fd_tickcount()-t0);
147 0 : }
148 :
149 : static void
150 : spill_rw( fd_txncache_t * tc,
151 : ulong off,
152 : uchar * buf,
153 : ulong sz,
154 0 : int write ) {
155 0 : FD_TEST( tc->spill_fd>=0 );
156 0 : while( sz ) {
157 0 : long n = write ? pwrite( tc->spill_fd, buf, sz, (off_t)off )
158 0 : : pread ( tc->spill_fd, buf, sz, (off_t)off );
159 0 : if( FD_UNLIKELY( n<0L && errno==EINTR ) ) continue;
160 0 : if( FD_UNLIKELY( n<0L ) ) FD_LOG_CRIT(( "txncache spill %s failed at offset %lu (%ld, errno %d)", write ? "write" : "read", off, n, errno ));
161 0 : if( FD_UNLIKELY( !n ) ) FD_LOG_CRIT(( "txncache spill %s at offset %lu", write ? "write made no progress" : "read reached unexpected EOF", off ));
162 0 : sz -= (ulong)n;
163 0 : if( FD_UNLIKELY( sz && !fd_ulong_is_aligned( (ulong)n, FD_TXNCACHE_TXNPAGE_ALIGN ) ) ) {
164 0 : FD_LOG_CRIT(( "txncache spill %s returned an unaligned short transfer at offset %lu", write ? "write" : "read", off ));
165 0 : }
166 0 : buf += (ulong)n;
167 0 : off += (ulong)n;
168 0 : }
169 0 : }
170 :
171 : static void
172 : spill_io( fd_txncache_t * tc,
173 : ulong page,
174 : fd_txncache_txnpage_t * data,
175 0 : int write ) {
176 0 : long t0 = fd_tickcount();
177 0 : spill_rw( tc, page*sizeof(*data), (uchar *)data, sizeof(*data), write );
178 0 : if( write ) FD_ATOMIC_FETCH_AND_ADD( &tc->shmem->page_spilled_cnt, 1UL );
179 0 : else FD_ATOMIC_FETCH_AND_ADD( &tc->shmem->page_spill_read_cnt, 1UL );
180 0 : FD_ATOMIC_FETCH_AND_ADD( &tc->shmem->spill_ticks, (ulong)(fd_tickcount()-t0) );
181 0 : }
182 :
183 : static uint
184 : frame_take( fd_txncache_t * tc,
185 0 : uint * victim ) {
186 0 : fd_txncache_shmem_t * shmem = tc->shmem;
187 0 : *victim = UINT_MAX;
188 0 : if( FD_LIKELY( shmem->frames_free_cnt ) ) return tc->frames_free[ --shmem->frames_free_cnt ];
189 0 : for( ulong i=0UL; i<2UL*shmem->frames_max; i++ ) {
190 0 : uint frame = (uint)shmem->evict_next;
191 0 : shmem->evict_next = (shmem->evict_next+1UL)%shmem->frames_max;
192 0 : uint owner = tc->frame_page[ frame ];
193 0 : FD_TEST( owner!=UINT_MAX );
194 0 : fd_txncache_page_meta_t * old = &tc->page_meta[ owner ];
195 0 : if( FD_VOLATILE_CONST( old->io ) ) continue;
196 0 : if( old->ref ) {
197 0 : old->ref = 0U;
198 0 : continue;
199 0 : }
200 0 : old->frame = UINT_MAX;
201 0 : shmem->frame_evicted_cnt++;
202 0 : *victim = owner;
203 0 : return frame;
204 0 : }
205 0 : return UINT_MAX;
206 0 : }
207 :
208 : static inline fd_txncache_txnpage_t *
209 : page_io( fd_txncache_t * tc,
210 : ulong page,
211 0 : int write_lock ) {
212 : /* Fast path: the page is resident. A fetch may publish the frame
213 : concurrently. */
214 0 : fd_txncache_page_meta_t * meta = &tc->page_meta[ page ];
215 0 : uint resident = FD_VOLATILE_CONST( meta->frame );
216 0 : if( FD_LIKELY( resident!=UINT_MAX ) ) {
217 0 : if( FD_UNLIKELY( !meta->ref ) ) meta->ref = 1U;
218 0 : return &tc->txnpages[ resident ];
219 0 : }
220 :
221 : /* Loading changes the frame cache, which needs the write lock. */
222 0 : if( FD_LIKELY( !write_lock ) ) {
223 0 : FD_ATOMIC_FETCH_AND_ADD( &tc->shmem->page_miss_cnt, 1UL );
224 0 : return NULL;
225 0 : }
226 0 : FD_TEST( page<tc->shmem->txnpages_max );
227 :
228 : /* A fetch may be loading or writing back the page. Fetches publish
229 : without the lock, so this does not deadlock. */
230 0 : while( FD_UNLIKELY( FD_VOLATILE_CONST( meta->io ) ) ) FD_SPIN_PAUSE();
231 0 : FD_COMPILER_MFENCE();
232 0 : if( FD_UNLIKELY( meta->frame!=UINT_MAX ) ) {
233 0 : if( FD_UNLIKELY( !meta->ref ) ) meta->ref = 1U;
234 0 : return &tc->txnpages[ meta->frame ];
235 0 : }
236 :
237 0 : uint victim;
238 0 : uint frame;
239 0 : while( FD_UNLIKELY( (frame=frame_take( tc, &victim ))==UINT_MAX ) ) FD_SPIN_PAUSE();
240 0 : if( victim!=UINT_MAX ) {
241 0 : fd_txncache_page_meta_t * old = &tc->page_meta[ victim ];
242 0 : if( old->dirty ) {
243 0 : spill_io( tc, victim, &tc->txnpages[ frame ], 1 );
244 0 : old->dirty = 0U;
245 0 : old->disk_valid = 1;
246 0 : }
247 0 : }
248 :
249 : /* Read the page back, or start it empty if it was never written. */
250 0 : fd_txncache_txnpage_t * data = &tc->txnpages[ frame ];
251 0 : if( meta->disk_valid ) {
252 0 : spill_io( tc, page, data, 0 );
253 0 : } else {
254 0 : data->free = FD_TXNCACHE_TXNS_PER_PAGE;
255 0 : }
256 0 : tc->frame_page[ frame ] = (uint)page;
257 0 : meta->frame = frame;
258 0 : meta->ref = 1U;
259 0 : return data;
260 0 : }
261 :
262 : static void
263 : page_fetch( fd_txncache_t * tc,
264 : ulong page,
265 0 : ulong seq ) {
266 0 : fd_txncache_page_meta_t * meta = &tc->page_meta[ page ];
267 0 : lock_write( tc );
268 :
269 : /* page may have been freed, or is already resident. */
270 0 : if( FD_UNLIKELY( tc->shmem->release_seq!=seq || meta->frame!=UINT_MAX ) ) {
271 0 : fd_rwlock_unwrite( tc->shmem->lock );
272 0 : return;
273 0 : }
274 :
275 : /* Another fetch is moving page. */
276 0 : if( FD_UNLIKELY( meta->io ) ) {
277 0 : fd_rwlock_unwrite( tc->shmem->lock );
278 0 : while( FD_UNLIKELY( FD_VOLATILE_CONST( meta->io ) ) ) FD_SPIN_PAUSE();
279 0 : return;
280 0 : }
281 :
282 : /* Every frame is busy. */
283 0 : uint victim;
284 0 : uint frame = frame_take( tc, &victim );
285 0 : if( FD_UNLIKELY( frame==UINT_MAX ) ) {
286 0 : fd_rwlock_unwrite( tc->shmem->lock );
287 0 : FD_SPIN_PAUSE();
288 0 : return;
289 0 : }
290 :
291 : /* Mark page and the evicted page busy until i/o finishes. */
292 0 : fd_txncache_page_meta_t * old = victim!=UINT_MAX ? &tc->page_meta[ victim ] : NULL;
293 0 : int write = old && old->dirty;
294 0 : if( write ) {
295 0 : old->dirty = 0U;
296 0 : old->io = 1U;
297 0 : }
298 0 : int read = meta->disk_valid;
299 0 : meta->io = 1U;
300 0 : tc->frame_page[ frame ] = (uint)page;
301 0 : fd_rwlock_unwrite( tc->shmem->lock );
302 :
303 : /* Write back the evicted page, then read page, without the lock. */
304 0 : fd_racesan_hook( "txncache:fetch_io" );
305 0 : fd_txncache_txnpage_t * data = &tc->txnpages[ frame ];
306 0 : if( write ) spill_io( tc, victim, data, 1 );
307 0 : if( read ) spill_io( tc, page, data, 0 );
308 0 : else data->free = FD_TXNCACHE_TXNS_PER_PAGE;
309 :
310 : /* Publish without the lock, as write lock holders may be waiting on
311 : io. io is cleared last. */
312 0 : if( write ) {
313 0 : old->disk_valid = 1;
314 0 : FD_HW_MFENCE_ST();
315 0 : FD_VOLATILE( old->io ) = 0U;
316 0 : }
317 0 : meta->ref = 1U;
318 0 : FD_HW_MFENCE_ST();
319 0 : FD_VOLATILE( meta->frame ) = frame;
320 0 : FD_HW_MFENCE_ST();
321 0 : FD_VOLATILE( meta->io ) = 0U;
322 0 : }
323 :
324 : /* A freed logical page must never reload contents from its former owner. */
325 : static void
326 : release_page( fd_txncache_t * tc,
327 0 : ulong page ) {
328 0 : fd_txncache_page_meta_t * meta = &tc->page_meta[ page ];
329 0 : while( FD_UNLIKELY( FD_VOLATILE_CONST( meta->io ) ) ) FD_SPIN_PAUSE();
330 0 : FD_COMPILER_MFENCE();
331 0 : tc->shmem->release_seq++;
332 0 : if( meta->frame!=UINT_MAX ) {
333 0 : tc->frame_page[ meta->frame ] = UINT_MAX;
334 0 : tc->frames_free[ tc->shmem->frames_free_cnt++ ] = meta->frame;
335 0 : meta->frame = UINT_MAX;
336 0 : }
337 0 : meta->dirty = 0U;
338 0 : meta->disk_valid = 0;
339 0 : meta->next = UINT_MAX;
340 0 : tc->txnpages_free[ tc->shmem->txnpages_free_cnt++ ] = (uint)page;
341 0 : }
342 :
343 : void
344 0 : fd_txncache_reset( fd_txncache_t * tc ) {
345 0 : lock_write( tc );
346 :
347 0 : tc->shmem->root_cnt = 0UL;
348 0 : root_slist_remove_all( tc->shmem->root_ll, tc->blockcache_shmem_pool );
349 :
350 0 : tc->shmem->txnpages_free_cnt = tc->shmem->txnpages_max;
351 0 : tc->shmem->frames_free_cnt = tc->shmem->frames_max;
352 0 : tc->shmem->evict_next = 0UL;
353 0 : tc->shmem->release_seq++;
354 0 : for( ulong i=0UL; i<tc->shmem->txnpages_max; i++ ) {
355 0 : while( FD_UNLIKELY( FD_VOLATILE_CONST( tc->page_meta[ i ].io ) ) ) FD_SPIN_PAUSE();
356 0 : }
357 0 : FD_COMPILER_MFENCE();
358 0 : for( ulong i=0UL; i<tc->shmem->txnpages_max; i++ ) {
359 0 : tc->txnpages_free[ i ] = (uint)i;
360 0 : tc->page_meta[ i ].frame = UINT_MAX;
361 0 : tc->page_meta[ i ].dirty = 0U;
362 0 : tc->page_meta[ i ].disk_valid = 0;
363 0 : tc->page_meta[ i ].next = UINT_MAX;
364 0 : }
365 0 : for( ulong i=0UL; i<tc->shmem->frames_max; i++ ) {
366 0 : tc->frame_page[ i ] = UINT_MAX;
367 0 : tc->frames_free[ i ] = (uint)i;
368 0 : }
369 :
370 0 : blockcache_pool_reset( tc->blockcache_shmem_pool );
371 0 : blockhash_map_reset( tc->blockhash_map );
372 :
373 0 : fd_rwlock_unwrite( tc->shmem->lock );
374 0 : }
375 :
376 : void
377 : fd_txncache_metrics_snap( fd_txncache_t const * tc,
378 0 : fd_txncache_metrics_t * out ) {
379 0 : fd_txncache_shmem_t const * shmem = tc->shmem;
380 0 : out->page_miss_cnt = FD_VOLATILE_CONST( shmem->page_miss_cnt );
381 0 : out->page_spilled_cnt = FD_VOLATILE_CONST( shmem->page_spilled_cnt );
382 0 : out->page_spill_read_cnt = FD_VOLATILE_CONST( shmem->page_spill_read_cnt );
383 0 : out->frame_evicted_cnt = FD_VOLATILE_CONST( shmem->frame_evicted_cnt );
384 0 : out->spill_ticks = FD_VOLATILE_CONST( shmem->spill_ticks );
385 0 : out->lock_read_wait_ticks = FD_VOLATILE_CONST( shmem->lock_read_wait_ticks );
386 0 : out->lock_write_wait_ticks = FD_VOLATILE_CONST( shmem->lock_write_wait_ticks );
387 0 : out->page_capacity = shmem->txnpages_max;
388 0 : out->page_used = shmem->txnpages_max-fd_ulong_min( FD_VOLATILE_CONST( shmem->txnpages_free_cnt ), shmem->txnpages_max );
389 0 : out->frame_capacity = shmem->frames_max;
390 0 : out->frame_used = shmem->frames_max-fd_ulong_min( FD_VOLATILE_CONST( shmem->frames_free_cnt ), shmem->frames_max );
391 0 : }
392 :
393 : FD_STATIC_ASSERT( !(sizeof(fd_txncache_txnpage_t)%FD_TXNCACHE_SNAPIN_STAGE_ALIGN), snapin_stage_align );
394 :
395 : void
396 : fd_txncache_snapin_stage( fd_txncache_t * tc,
397 : ulong off,
398 : void * buf,
399 : ulong sz,
400 0 : int write ) {
401 0 : FD_TEST_ERR( tc->shmem->txnpages_free_cnt==tc->shmem->txnpages_max );
402 0 : FD_TEST( fd_ulong_is_aligned( off, FD_TXNCACHE_SNAPIN_STAGE_ALIGN ) );
403 0 : FD_TEST( fd_ulong_is_aligned( sz, FD_TXNCACHE_SNAPIN_STAGE_ALIGN ) );
404 0 : FD_TEST( fd_ulong_is_aligned( (ulong)buf, FD_TXNCACHE_SNAPIN_STAGE_ALIGN ) );
405 :
406 0 : ulong mem_sz = tc->shmem->frames_max*sizeof(fd_txncache_txnpage_t);
407 0 : uchar * p = (uchar *)buf;
408 0 : if( off<mem_sz ) {
409 0 : ulong n = fd_ulong_min( sz, mem_sz-off );
410 0 : uchar * mem = (uchar *)tc->txnpages+off;
411 0 : if( write ) fd_memcpy( mem, p, n );
412 0 : else fd_memcpy( p, mem, n );
413 0 : off += n;
414 0 : p += n;
415 0 : sz -= n;
416 0 : }
417 0 : if( sz ) spill_rw( tc, off-mem_sz, p, sz, write );
418 0 : }
419 :
420 : FD_FN_PURE static inline ulong
421 : fd_txncache_bucket( fd_txncache_t const * tc,
422 0 : uchar const * txnhash ) {
423 0 : return fd_ulong_hash( FD_LOAD( ulong, txnhash )^tc->shmem->seed )%tc->shmem->bucket_cnt;
424 0 : }
425 :
426 : /* Returns a resident page of blockcache with room, with its index in
427 : *out_idx. Without the write lock, returns NULL with the index of a
428 : page of blockcache to fetch in *out_idx, or ULONG_MAX if the caller
429 : must retry under the write lock. */
430 :
431 : static fd_txncache_txnpage_t *
432 : fd_txncache_ensure_txnpage( fd_txncache_t * tc,
433 : blockcache_t * blockcache,
434 : int write,
435 0 : ulong * out_idx ) {
436 0 : fd_txncache_shmem_t * shmem = tc->shmem;
437 0 : for(;;) {
438 : /* Use the newest page if it has room. */
439 0 : uint head = FD_VOLATILE_CONST( blockcache->shmem->page_head );
440 0 : if( FD_LIKELY( head!=UINT_MAX ) ) {
441 0 : fd_txncache_txnpage_t * page = page_io( tc, head, write );
442 0 : *out_idx = head;
443 0 : if( FD_UNLIKELY( !page ) ) return NULL;
444 0 : if( FD_LIKELY( FD_VOLATILE_CONST( page->free ) ) ) return page;
445 0 : }
446 :
447 : /* Need a new page. Try to make sure we can safely take the lock
448 : and that there's enough capacity. */
449 0 : if( !write ) {
450 0 : fd_rwlock_write( shmem->alloc_lock );
451 0 : if( FD_UNLIKELY( FD_VOLATILE_CONST( blockcache->shmem->page_head )!=head ) ) {
452 0 : fd_rwlock_unwrite( shmem->alloc_lock );
453 0 : continue;
454 0 : }
455 0 : }
456 :
457 0 : ulong page_cnt = blockcache->shmem->pages_cnt;
458 0 : if( FD_UNLIKELY( page_cnt>=shmem->txnpages_per_blockhash_max || !shmem->txnpages_free_cnt ) ) {
459 0 : if( !write ) fd_rwlock_unwrite( shmem->alloc_lock );
460 0 : *out_idx = ULONG_MAX;
461 0 : return NULL;
462 0 : }
463 :
464 : /* We have safe and exclusive access to allocating a new page.
465 : Without the write lock and a free frame, the page is allocated
466 : not resident, and fetching it starts it empty. */
467 0 : ulong idx = tc->txnpages_free[ --shmem->txnpages_free_cnt ];
468 0 : fd_txncache_txnpage_t * page = NULL;
469 0 : if( write ) {
470 0 : page = page_io( tc, idx, 1 );
471 0 : } else if( FD_LIKELY( shmem->frames_free_cnt ) ) {
472 0 : uint frame = tc->frames_free[ --shmem->frames_free_cnt ];
473 0 : tc->frame_page[ frame ] = (uint)idx;
474 0 : tc->page_meta[ idx ].frame = frame;
475 0 : page = &tc->txnpages[ frame ];
476 0 : }
477 0 : if( FD_LIKELY( page ) ) page->free = FD_TXNCACHE_TXNS_PER_PAGE;
478 0 : tc->page_meta[ idx ].next = head;
479 0 : blockcache->shmem->pages_cnt = page_cnt+1UL;
480 0 : FD_HW_MFENCE_ST();
481 0 : FD_VOLATILE( blockcache->shmem->page_head ) = (uint)idx;
482 0 : if( !write ) fd_rwlock_unwrite( shmem->alloc_lock );
483 0 : *out_idx = idx;
484 0 : return page;
485 0 : }
486 0 : }
487 :
488 : static inline int
489 : fd_txncache_insert_txn( fd_txncache_t * tc,
490 : blockcache_t * blockcache,
491 : fd_txncache_txnpage_t * txnpage,
492 : ulong txnpage_idx,
493 : fd_txncache_fork_id_t fork_id,
494 0 : uchar const * txnhash ) {
495 0 : for(;;) {
496 0 : ushort txnpage_free = txnpage->free;
497 0 : if( FD_UNLIKELY( !txnpage_free ) ) return 0;
498 0 : if( FD_UNLIKELY( FD_ATOMIC_CAS( &txnpage->free, txnpage_free, txnpage_free-1UL )!=txnpage_free ) ) {
499 0 : FD_SPIN_PAUSE();
500 0 : continue;
501 0 : }
502 :
503 0 : ulong txn_idx = FD_TXNCACHE_TXNS_PER_PAGE-txnpage_free;
504 0 : ulong txnhash_offset = blockcache->shmem->txnhash_offset;
505 0 : memcpy( txnpage->txns[ txn_idx ]->txnhash, txnhash+txnhash_offset, 20UL );
506 0 : txnpage->txns[ txn_idx ]->fork_id = fork_id;
507 0 : txnpage->txns[ txn_idx ]->generation = tc->blockcache_pool[ fork_id.val ].shmem->generation;
508 0 : FD_COMPILER_MFENCE();
509 :
510 0 : ulong txn_bucket = fd_txncache_bucket( tc, txnhash+txnhash_offset );
511 0 : for(;;) {
512 0 : uint head = blockcache->heads[ txn_bucket ];
513 0 : txnpage->txns[ txn_idx ]->blockcache_next = head;
514 0 : FD_COMPILER_MFENCE();
515 0 : if( FD_LIKELY( FD_ATOMIC_CAS( &blockcache->heads[ txn_bucket ], head, (uint)(FD_TXNCACHE_TXNS_PER_PAGE*txnpage_idx+txn_idx) )==head ) ) break;
516 0 : FD_SPIN_PAUSE();
517 0 : }
518 :
519 : /* This races, but we expect x86 stores to be atomic. */
520 0 : if( FD_UNLIKELY( !tc->page_meta[ txnpage_idx ].dirty ) ) tc->page_meta[ txnpage_idx ].dirty = 1U;
521 0 : return 1;
522 0 : }
523 0 : }
524 :
525 : fd_txncache_fork_id_t
526 : fd_txncache_attach_child( fd_txncache_t * tc,
527 0 : fd_txncache_fork_id_t parent_fork_id ) {
528 0 : lock_write( tc );
529 :
530 0 : FD_TEST( blockcache_pool_free( tc->blockcache_shmem_pool ) );
531 0 : ulong idx = blockcache_pool_idx_acquire( tc->blockcache_shmem_pool );
532 :
533 0 : blockcache_t * fork = &tc->blockcache_pool[ idx ];
534 0 : fd_txncache_fork_id_t fork_id = { .val = (ushort)idx };
535 :
536 0 : fork->shmem->generation = tc->shmem->blockcache_generation++;
537 0 : fork->shmem->child_id = (fd_txncache_fork_id_t){ .val = USHORT_MAX };
538 :
539 0 : if( FD_LIKELY( parent_fork_id.val==USHORT_MAX ) ) {
540 0 : FD_TEST( blockcache_pool_free( tc->blockcache_shmem_pool )==blockcache_pool_max( tc->blockcache_shmem_pool )-1UL );
541 0 : fork->shmem->parent_id = (fd_txncache_fork_id_t){ .val = USHORT_MAX };
542 0 : fork->shmem->sibling_id = (fd_txncache_fork_id_t){ .val = USHORT_MAX };
543 :
544 0 : descends_set_null( fork->descends );
545 0 : root_slist_ele_push_tail( tc->shmem->root_ll, fork->shmem, tc->blockcache_shmem_pool );
546 0 : } else {
547 0 : blockcache_t * parent = &tc->blockcache_pool[ parent_fork_id.val ];
548 : /* We might be tempted to freeze the parent here, and it's valid to
549 : do this ordinarily, but not when loading from a snapshot, when
550 : we need to load many transactions into a root parent chain at
551 : once. */
552 0 : fork->shmem->sibling_id = parent->shmem->child_id;
553 0 : fork->shmem->parent_id = parent_fork_id;
554 0 : parent->shmem->child_id = fork_id;
555 :
556 0 : descends_set_copy( fork->descends, parent->descends );
557 0 : descends_set_insert( fork->descends, parent_fork_id.val );
558 0 : }
559 :
560 0 : fork->shmem->txnhash_offset = 0UL;
561 0 : fork->shmem->frozen = 0;
562 0 : memset( fork->heads, 0xFF, tc->shmem->bucket_cnt*sizeof(uint) );
563 0 : fork->shmem->pages_cnt = 0UL;
564 0 : fork->shmem->page_head = UINT_MAX;
565 :
566 0 : fd_rwlock_unwrite( tc->shmem->lock );
567 0 : return fork_id;
568 0 : }
569 :
570 : void
571 : fd_txncache_attach_blockhash( fd_txncache_t * tc,
572 : fd_txncache_fork_id_t fork_id,
573 0 : uchar const * blockhash ) {
574 0 : lock_write( tc );
575 :
576 0 : blockcache_t * fork = &tc->blockcache_pool[ fork_id.val ];
577 0 : FD_TEST( !fork->shmem->frozen );
578 0 : fork->shmem->frozen = 1;
579 :
580 0 : memcpy( fork->shmem->blockhash.uc, blockhash, 32UL );
581 :
582 0 : blockhash_map_ele_insert( tc->blockhash_map, fork->shmem, tc->blockcache_shmem_pool );
583 :
584 0 : fd_rwlock_unwrite( tc->shmem->lock );
585 0 : }
586 :
587 : void
588 : fd_txncache_finalize_fork( fd_txncache_t * tc,
589 : fd_txncache_fork_id_t fork_id,
590 : ulong txnhash_offset,
591 0 : uchar const * blockhash ) {
592 0 : lock_write( tc );
593 :
594 0 : blockcache_t * fork = &tc->blockcache_pool[ fork_id.val ];
595 0 : FD_TEST( fork->shmem->frozen<=1 );
596 0 : FD_TEST( fork->shmem->frozen>=0 );
597 0 : fork->shmem->txnhash_offset = txnhash_offset;
598 :
599 0 : memcpy( fork->shmem->blockhash.uc, blockhash, 32UL );
600 :
601 0 : if( FD_LIKELY( !fork->shmem->frozen ) ) blockhash_map_ele_insert( tc->blockhash_map, fork->shmem, tc->blockcache_shmem_pool );
602 0 : fork->shmem->frozen = 2;
603 :
604 0 : fd_rwlock_unwrite( tc->shmem->lock );
605 0 : }
606 :
607 : static inline void
608 : remove_blockcache( fd_txncache_t * tc,
609 0 : blockcache_t * blockcache ) {
610 0 : FD_TEST( blockcache->shmem->frozen>=0 );
611 0 : for( uint page=blockcache->shmem->page_head; page!=UINT_MAX; ) {
612 0 : uint next = tc->page_meta[ page ].next;
613 0 : release_page( tc, page );
614 0 : page = next;
615 0 : }
616 0 : blockcache->shmem->page_head = UINT_MAX;
617 0 : blockcache->shmem->pages_cnt = 0UL;
618 :
619 0 : ulong idx = blockcache_pool_idx( tc->blockcache_shmem_pool, blockcache->shmem );
620 0 : for( ulong i=0UL; i<tc->shmem->active_slots_max; i++ ) descends_set_remove( tc->blockcache_pool[ i ].descends, idx );
621 :
622 0 : if( FD_LIKELY( blockcache->shmem->frozen ) ) blockhash_map_ele_remove_fast( tc->blockhash_map, blockcache->shmem, tc->blockcache_shmem_pool );
623 0 : blockcache->shmem->frozen = -1;
624 0 : blockcache_pool_ele_release( tc->blockcache_shmem_pool, blockcache->shmem );
625 0 : }
626 :
627 : static inline void
628 : remove_children( fd_txncache_t * tc,
629 : blockcache_t const * fork,
630 0 : blockcache_t const * except ) {
631 0 : fd_txncache_fork_id_t sibling_idx = fork->shmem->child_id;
632 0 : while( sibling_idx.val!=USHORT_MAX ) {
633 0 : blockcache_t * sibling = &tc->blockcache_pool[ sibling_idx.val ];
634 :
635 0 : sibling_idx = sibling->shmem->sibling_id;
636 0 : if( FD_UNLIKELY( sibling==except ) ) continue;
637 :
638 0 : remove_children( tc, sibling, except );
639 0 : remove_blockcache( tc, sibling );
640 0 : }
641 0 : }
642 :
643 : void
644 : fd_txncache_cancel_fork( fd_txncache_t * tc,
645 0 : fd_txncache_fork_id_t fork_id ) {
646 0 : lock_write( tc );
647 0 : blockcache_t * fork = &tc->blockcache_pool[ fork_id.val ];
648 0 : FD_TEST( fork->shmem->parent_id.val!=USHORT_MAX );
649 :
650 : /* The soon-to-be-pruned subtree must be unrooted. */
651 0 : fd_txncache_blockcache_shmem_t const * latest_root = root_slist_ele_peek_tail_const( tc->shmem->root_ll, tc->blockcache_shmem_pool );
652 0 : FD_TEST( latest_root );
653 0 : FD_TEST( descends_set_test( fork->descends, blockcache_pool_idx( tc->blockcache_shmem_pool, latest_root ) ) );
654 :
655 0 : remove_children( tc, fork, NULL );
656 0 : remove_blockcache( tc, fork );
657 0 : ushort * fork_id_p = &(tc->blockcache_pool[ fork->shmem->parent_id.val ].shmem->child_id.val);
658 0 : while( *fork_id_p!=fork_id.val ) {
659 0 : fork_id_p = &(tc->blockcache_pool[ *fork_id_p ].shmem->sibling_id.val);
660 0 : }
661 0 : *fork_id_p = fork->shmem->sibling_id.val;
662 0 : fd_rwlock_unwrite( tc->shmem->lock );
663 0 : }
664 :
665 : void
666 : fd_txncache_advance_root( fd_txncache_t * tc,
667 0 : fd_txncache_fork_id_t fork_id ) {
668 0 : lock_write( tc );
669 :
670 0 : blockcache_t * fork = &tc->blockcache_pool[ fork_id.val ];
671 0 : FD_TEST( fork->shmem->parent_id.val!=USHORT_MAX );
672 :
673 0 : blockcache_t * parent_fork = &tc->blockcache_pool[ fork->shmem->parent_id.val ];
674 0 : if( FD_UNLIKELY( root_slist_ele_peek_tail( tc->shmem->root_ll, tc->blockcache_shmem_pool )!=parent_fork->shmem ) ) {
675 0 : FD_BASE58_ENCODE_32_BYTES( parent_fork->shmem->blockhash.uc, parent_blockhash_b58 );
676 0 : FD_BASE58_ENCODE_32_BYTES( fork->shmem->blockhash.uc, fork_blockhash_b58 );
677 0 : FD_BASE58_ENCODE_32_BYTES( root_slist_ele_peek_tail( tc->shmem->root_ll, tc->blockcache_shmem_pool )->blockhash.uc, root_blockhash_b58 );
678 0 : FD_LOG_CRIT(( "advancing root from %s to %s but that is not valid, last root is %s",
679 0 : parent_blockhash_b58,
680 0 : fork_blockhash_b58,
681 0 : root_blockhash_b58 ));
682 0 : }
683 :
684 0 : FD_BASE58_ENCODE_32_BYTES( parent_fork->shmem->blockhash.uc, parent_blockhash_b58 );
685 0 : FD_BASE58_ENCODE_32_BYTES( fork->shmem->blockhash.uc, fork_blockhash_b58 );
686 0 : FD_LOG_DEBUG(( "advancing root from %s to %s",
687 0 : parent_blockhash_b58,
688 0 : fork_blockhash_b58 ));
689 :
690 : /* When a fork is rooted, any competing forks can be immediately
691 : removed as they will not be needed again. This includes child
692 : forks of the pruned siblings as well. */
693 0 : remove_children( tc, parent_fork, fork );
694 0 : parent_fork->shmem->child_id = fork_id;
695 0 : fork->shmem->sibling_id = (fd_txncache_fork_id_t){ .val = USHORT_MAX };
696 :
697 : /* Now, the earliest known rooted fork can likely be removed since its
698 : blockhashes cannot be referenced anymore (they are older than 151
699 : blockhashes away). */
700 0 : tc->shmem->root_cnt++;
701 0 : root_slist_ele_push_tail( tc->shmem->root_ll, fork->shmem, tc->blockcache_shmem_pool );
702 0 : if( FD_LIKELY( tc->shmem->root_cnt>FD_TXNCACHE_MAX_BLOCKHASH_DISTANCE ) ) {
703 0 : fd_txncache_blockcache_shmem_t * old_root_shmem = root_slist_ele_pop_head( tc->shmem->root_ll, tc->blockcache_shmem_pool );
704 0 : FD_TEST( old_root_shmem );
705 0 : blockcache_t * old_root = &tc->blockcache_pool[ blockcache_pool_idx( tc->blockcache_shmem_pool, old_root_shmem ) ];
706 :
707 0 : root_slist_ele_peek_head( tc->shmem->root_ll, tc->blockcache_shmem_pool )->parent_id.val = USHORT_MAX;
708 :
709 0 : remove_blockcache( tc, old_root );
710 0 : tc->shmem->root_cnt--;
711 0 : }
712 :
713 0 : fd_rwlock_unwrite( tc->shmem->lock );
714 0 : }
715 :
716 : static inline blockcache_t *
717 : blockhash_on_fork( fd_txncache_t * tc,
718 : blockcache_t const * fork,
719 0 : uchar const * blockhash ) {
720 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 );
721 0 : if( FD_UNLIKELY( !candidate ) ) return NULL;
722 :
723 0 : while( candidate ) {
724 0 : ulong candidate_idx = blockcache_pool_idx( tc->blockcache_shmem_pool, candidate );
725 0 : if( FD_LIKELY( descends_set_test( fork->descends, candidate_idx ) ) ) return &tc->blockcache_pool[ candidate_idx ];
726 0 : candidate = blockhash_map_ele_next_const( candidate, NULL, tc->blockcache_shmem_pool );
727 0 : }
728 0 : return NULL;
729 0 : }
730 :
731 : static void
732 : purge_stale_on_blockcache( fd_txncache_t * tc,
733 0 : blockcache_t * blockcache ) {
734 0 : FD_TEST( blockcache->shmem->frozen>=0 );
735 0 : memset( tc->scratch_heads, 0xFF, tc->shmem->bucket_cnt*sizeof(tc->scratch_heads[ 0 ]) );
736 0 : uint scratch_page_head = UINT_MAX;
737 0 : ulong scratch_pages_cnt = 0UL;
738 0 : ulong scratch_txnpage_idx = ULONG_MAX;
739 0 : tc->scratch_txnpage->free = 0;
740 0 : uint next_txnpage_idx = blockcache->shmem->page_head;
741 0 : while( next_txnpage_idx!=UINT_MAX ) {
742 0 : ulong curr_txnpage_idx = next_txnpage_idx;
743 0 : next_txnpage_idx = tc->page_meta[ curr_txnpage_idx ].next;
744 0 : fd_txncache_txnpage_t * input = tc->scratch_input;
745 0 : memcpy( input, page_io( tc, curr_txnpage_idx, 1 ), sizeof(*input) );
746 0 : ulong curr_txn_cnt = FD_TXNCACHE_TXNS_PER_PAGE-input->free;
747 0 : for( ulong j=0UL; j<curr_txn_cnt; j++ ) {
748 0 : fd_txncache_single_txn_t const * curr_txn = input->txns[ curr_txn_cnt-j-1UL ];
749 0 : blockcache_t const * txn_fork = &tc->blockcache_pool[ curr_txn->fork_id.val ];
750 0 : if( FD_LIKELY( txn_fork->shmem->frozen>=0 && txn_fork->shmem->generation==curr_txn->generation ) ) {
751 : /* Valid transaction. Keep. */
752 0 : if( FD_UNLIKELY( !tc->scratch_txnpage->free ) ) {
753 0 : FD_TEST( scratch_txnpage_idx!=curr_txnpage_idx );
754 0 : if( FD_LIKELY( scratch_txnpage_idx!=ULONG_MAX ) ) {
755 0 : memcpy( page_io( tc, scratch_txnpage_idx, 1 ), tc->scratch_txnpage, sizeof(fd_txncache_txnpage_t) );
756 0 : tc->page_meta[ scratch_txnpage_idx ].dirty = 1U;
757 0 : }
758 0 : scratch_txnpage_idx = curr_txnpage_idx;
759 0 : tc->scratch_txnpage->free = FD_TXNCACHE_TXNS_PER_PAGE;
760 0 : tc->page_meta[ scratch_txnpage_idx ].next = scratch_page_head;
761 0 : scratch_page_head = (uint)scratch_txnpage_idx;
762 0 : scratch_pages_cnt++;
763 0 : }
764 0 : ulong txn_idx = FD_TXNCACHE_TXNS_PER_PAGE-tc->scratch_txnpage->free;
765 0 : memcpy( tc->scratch_txnpage->txns[ txn_idx ], curr_txn, sizeof(*curr_txn) );
766 0 : ulong txn_bucket = fd_txncache_bucket( tc, curr_txn->txnhash );
767 0 : uint head = tc->scratch_heads[ txn_bucket ];
768 0 : tc->scratch_txnpage->txns[ txn_idx ]->blockcache_next = head;
769 0 : ulong txn_gidx = FD_TXNCACHE_TXNS_PER_PAGE*scratch_txnpage_idx+txn_idx;
770 0 : FD_TEST( txn_gidx<UINT_MAX );
771 0 : tc->scratch_heads[ txn_bucket ] = (uint)txn_gidx;
772 0 : tc->scratch_txnpage->free--;
773 0 : } else {
774 : /* Stale transaction. Drop. */
775 0 : continue;
776 0 : }
777 0 : }
778 0 : if( FD_UNLIKELY( curr_txnpage_idx!=scratch_txnpage_idx ) ) {
779 : /* The txnpage is not being used for compaction, free it up. */
780 0 : release_page( tc, curr_txnpage_idx );
781 0 : }
782 0 : }
783 0 : if( FD_LIKELY( scratch_txnpage_idx!=ULONG_MAX ) ) {
784 0 : memcpy( page_io( tc, scratch_txnpage_idx, 1 ), tc->scratch_txnpage, sizeof(fd_txncache_txnpage_t) );
785 0 : tc->page_meta[ scratch_txnpage_idx ].dirty = 1U;
786 0 : }
787 0 : blockcache->shmem->page_head = scratch_page_head;
788 0 : blockcache->shmem->pages_cnt = scratch_pages_cnt;
789 0 : memcpy( blockcache->heads, tc->scratch_heads, tc->shmem->bucket_cnt*sizeof(blockcache->heads[0]) );
790 0 : }
791 :
792 : static void
793 : purge_stale_on_fork( fd_txncache_t * tc,
794 0 : blockcache_t * fork ) {
795 0 : purge_stale_on_blockcache( tc, fork );
796 :
797 0 : fd_txncache_fork_id_t sibling_idx = fork->shmem->child_id;
798 0 : while( sibling_idx.val!=USHORT_MAX ) {
799 0 : blockcache_t * sibling = &tc->blockcache_pool[ sibling_idx.val ];
800 0 : purge_stale_on_fork( tc, sibling );
801 0 : sibling_idx = sibling->shmem->sibling_id;
802 0 : }
803 0 : }
804 :
805 : static void
806 0 : purge_stale( fd_txncache_t * tc ) {
807 0 : fd_txncache_blockcache_shmem_t * root_shmem = root_slist_ele_peek_head( tc->shmem->root_ll, tc->blockcache_shmem_pool );
808 0 : FD_TEST( root_shmem );
809 0 : blockcache_t * root = &tc->blockcache_pool[ blockcache_pool_idx( tc->blockcache_shmem_pool, root_shmem ) ];
810 0 : ulong free_before = tc->shmem->txnpages_free_cnt;
811 : /* One might think that an optimization here is to stop the descent on
812 : the latest rooted blockcache. There could be no pruned minority
813 : forks from that point on. As a result, there could be no stale
814 : transactions in any blockcache descending from that.
815 : Unfortunately, frontier eviction means that any blockcache in the
816 : fork tree can have stale transactions. */
817 0 : purge_stale_on_fork( tc, root );
818 0 : FD_LOG_WARNING(( "purge_stale: txnpages_free %lu -> %lu", free_before, tc->shmem->txnpages_free_cnt ));
819 0 : }
820 :
821 : void
822 : fd_txncache_insert( fd_txncache_t * tc,
823 : fd_txncache_fork_id_t fork_id,
824 : uchar const * blockhash,
825 0 : uchar const * txnhash ) {
826 0 : long t0 = fd_tickcount();
827 0 : lock_read( tc );
828 :
829 0 : blockcache_t const * fork = &tc->blockcache_pool[ fork_id.val ];
830 0 : FD_TEST( fork->shmem->frozen<=1 );
831 0 : FD_TEST( fork->shmem->frozen>=0 );
832 0 : blockcache_t * blockcache = blockhash_on_fork( tc, fork, blockhash );
833 0 : FD_TEST( blockcache );
834 :
835 0 : ulong fetched_idx = ULONG_MAX;
836 0 : int miss = 0;
837 0 : int write = 0;
838 0 : for(;;) {
839 : /* First look for an in-memory page (write==0) */
840 0 : ulong txnpage_idx;
841 0 : fd_txncache_txnpage_t * txnpage = fd_txncache_ensure_txnpage( tc, blockcache, 0, &txnpage_idx );
842 0 : if( FD_UNLIKELY( !txnpage && txnpage_idx!=ULONG_MAX ) ) miss = 1;
843 0 : if( FD_UNLIKELY( !txnpage && txnpage_idx!=ULONG_MAX && txnpage_idx!=fetched_idx ) ) {
844 : /* The page is not resident. Load it without blocking others. */
845 0 : ulong seq = tc->shmem->release_seq;
846 0 : fd_rwlock_unread( tc->shmem->lock );
847 0 : page_fetch( tc, txnpage_idx, seq );
848 0 : fetched_idx = txnpage_idx;
849 0 : lock_read( tc );
850 0 : blockcache = blockhash_on_fork( tc, fork, blockhash );
851 0 : FD_TEST( blockcache );
852 0 : continue;
853 0 : }
854 0 : if( FD_UNLIKELY( !txnpage ) ) {
855 : /* The page was evicted again after fetching it without the lock,
856 : or no page is available. Retry under the write lock, which
857 : loads pages in place.
858 :
859 : Because of sizing invariants when creating the structure, it is
860 : not typically possible to fill it, unless there are stale
861 : transactions from minority forks that were purged floating
862 : around, in which case we can purge them here and try again.
863 : Under the write lock there are no concurrent allocators, so a
864 : page still unavailable after the purge means the cache is
865 : undersized for the caller's usage. */
866 0 : fd_rwlock_unread( tc->shmem->lock );
867 0 : lock_write( tc );
868 0 : blockcache = blockhash_on_fork( tc, fork, blockhash );
869 0 : FD_TEST( blockcache );
870 : /* Load or allocate a frame under exclusive access. */
871 0 : txnpage = fd_txncache_ensure_txnpage( tc, blockcache, 1, &txnpage_idx );
872 0 : if( FD_UNLIKELY( !txnpage ) ) {
873 : /* Slow path: there is no available txnpage in-memory or on
874 : disk. Try to purge any stale entries and try one more time.
875 : If that fails, capacity has been reached. */
876 0 : purge_stale( tc );
877 0 : txnpage = fd_txncache_ensure_txnpage( tc, blockcache, 1, &txnpage_idx );
878 0 : if( FD_UNLIKELY( !txnpage ) )
879 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)",
880 0 : blockcache->shmem->pages_cnt, tc->shmem->txnpages_per_blockhash_max,
881 0 : tc->shmem->txnpages_free_cnt, tc->shmem->txnpages_max,
882 0 : tc->shmem->active_slots_max, tc->shmem->txn_per_slot_max ));
883 0 : }
884 0 : FD_TEST( fd_txncache_insert_txn( tc, blockcache, txnpage, txnpage_idx, fork_id, txnhash ) );
885 0 : write = 1;
886 0 : break;
887 0 : }
888 :
889 : /* Fast path: insert the transaction into an in-memory page. */
890 0 : int success = fd_txncache_insert_txn( tc, blockcache, txnpage, txnpage_idx, fork_id, txnhash );
891 0 : if( FD_LIKELY( success ) ) break;
892 :
893 0 : FD_SPIN_PAUSE();
894 0 : }
895 :
896 0 : if( FD_UNLIKELY( write ) ) fd_rwlock_unwrite( tc->shmem->lock );
897 0 : else fd_rwlock_unread ( tc->shmem->lock );
898 :
899 0 : tc->metrics->insert_miss_cnt += (ulong)miss;
900 0 : tc->metrics->stalled_cnt += (ulong)write;
901 0 : fd_histf_sample( tc->metrics->insert_ticks, (ulong)(fd_tickcount()-t0) );
902 0 : }
903 :
904 : int
905 : fd_txncache_query( fd_txncache_t * tc,
906 : fd_txncache_fork_id_t fork_id,
907 : uchar const * blockhash,
908 0 : uchar const * txnhash ) {
909 0 : long t0 = fd_tickcount();
910 : /* Prevent fork removal and pruning while allowing inserts. */
911 0 : lock_read( tc );
912 :
913 : /* Find the blockhash cache visible from the requested fork. */
914 0 : blockcache_t const * fork = &tc->blockcache_pool[ fork_id.val ];
915 0 : FD_TEST( fork->shmem->frozen>=0 );
916 0 : blockcache_t const * blockcache = blockhash_on_fork( tc, fork, blockhash );
917 0 : FD_TEST( blockcache );
918 0 : FD_TEST( blockcache->shmem->frozen==2 );
919 :
920 0 : int found = 0;
921 0 : int write = 0;
922 0 : int miss = 0;
923 0 : ulong fetch_cnt = 0UL;
924 :
925 : /* Select the bucket using the stored portion of the txn hash. */
926 0 : ulong txnhash_offset = blockcache->shmem->txnhash_offset;
927 0 : ulong head_hash = fd_txncache_bucket( tc, txnhash+txnhash_offset );
928 : /* Walk the bucket's list. UINT_MAX marks the end. */
929 0 : for( uint head=blockcache->heads[ head_hash ]; head!=UINT_MAX; ) {
930 : /* If the page is not in-memory, try to load it optimistically
931 : without the lock and check again. Try fetching upto the number
932 : of pages in the blockcache, or frames if fewer; if there's a lot
933 : of contention, fall back to use the write-lock to do the i/o. On
934 : average, this means that we will try to do 1 optimsitic load
935 : before falling back to the write lock. */
936 0 : fd_txncache_txnpage_t * page = page_io( tc, head/FD_TXNCACHE_TXNS_PER_PAGE, write );
937 0 : if( FD_UNLIKELY( !page ) ) {
938 0 : miss = 1;
939 0 : ulong seq = tc->shmem->release_seq;
940 0 : int fetch = fetch_cnt<fd_ulong_min( blockcache->shmem->pages_cnt, tc->shmem->frames_max );
941 0 : fd_rwlock_unread( tc->shmem->lock );
942 0 : if( FD_LIKELY( fetch ) ) {
943 : /* Optimistically do I/O without the lock and check again. */
944 0 : page_fetch( tc, head/FD_TXNCACHE_TXNS_PER_PAGE, seq );
945 0 : fetch_cnt++;
946 0 : lock_read( tc );
947 0 : } else {
948 : /* Acquire write lock to load the page in place. */
949 0 : lock_write( tc );
950 0 : write = 1;
951 0 : }
952 0 : blockcache = blockhash_on_fork( tc, fork, blockhash );
953 0 : FD_TEST( blockcache );
954 0 : head = blockcache->heads[ head_hash ];
955 0 : continue;
956 0 : }
957 0 : fd_txncache_single_txn_t const * txn = page->txns[ head%FD_TXNCACHE_TXNS_PER_PAGE ];
958 0 : head = txn->blockcache_next;
959 :
960 : /* Accept transactions from this fork or its ancestors. Reject
961 : removed forks and old records whose fork ID has been reused. */
962 0 : blockcache_t const * txn_fork = &tc->blockcache_pool[ txn->fork_id.val ];
963 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;
964 : /* Different hashes can share a bucket, so compare the stored hash. */
965 0 : if( FD_LIKELY( descends && !memcmp( txnhash+txnhash_offset, txn->txnhash, 20UL ) ) ) {
966 0 : found = 1;
967 0 : break;
968 0 : }
969 0 : }
970 :
971 0 : if( FD_UNLIKELY( write ) ) fd_rwlock_unwrite( tc->shmem->lock );
972 0 : else fd_rwlock_unread ( tc->shmem->lock );
973 :
974 0 : tc->metrics->query_miss_cnt += (ulong)miss;
975 0 : tc->metrics->stalled_cnt += (ulong)write;
976 0 : fd_histf_sample( tc->metrics->query_ticks, (ulong)(fd_tickcount()-t0) );
977 0 : return found;
978 0 : }
979 :
980 : ulong
981 : fd_txncache_snapshot_roots( fd_txncache_t * tc,
982 0 : fd_txncache_root_t * out ) {
983 0 : lock_read( tc );
984 0 : ulong cnt = 0UL;
985 0 : for( ulong it = root_slist_iter_init( tc->shmem->root_ll, tc->blockcache_shmem_pool );
986 0 : !root_slist_iter_done( it, tc->shmem->root_ll, tc->blockcache_shmem_pool );
987 0 : it = root_slist_iter_next( it, tc->shmem->root_ll, tc->blockcache_shmem_pool ) ) {
988 0 : ulong blockcache_idx = root_slist_iter_idx( it, tc->shmem->root_ll, tc->blockcache_shmem_pool );
989 0 : fd_txncache_blockcache_shmem_t const * blockcache_shmem = &tc->blockcache_shmem_pool[ blockcache_idx ];
990 0 : FD_TEST( cnt<FD_TXNCACHE_MAX_ROOTS );
991 0 : fd_txncache_root_t * root = &out[ cnt++ ];
992 0 : root->blockcache_idx = blockcache_idx;
993 0 : root->txnhash_offset = blockcache_shmem->txnhash_offset;
994 0 : root->generation = blockcache_shmem->generation;
995 0 : memcpy( root->blockhash, blockcache_shmem->blockhash.uc, 32UL );
996 0 : }
997 0 : fd_rwlock_unread( tc->shmem->lock );
998 0 : return cnt;
999 0 : }
1000 :
1001 : static fd_txncache_single_txn_t const *
1002 : bucket_chain_txn( fd_txncache_t * tc,
1003 : uint txn_gidx,
1004 0 : int write_lock ) {
1005 0 : if( FD_UNLIKELY( txn_gidx==UINT_MAX ) ) return NULL;
1006 0 : fd_txncache_txnpage_t const * page = page_io( tc, txn_gidx/FD_TXNCACHE_TXNS_PER_PAGE, write_lock );
1007 0 : FD_TEST( page );
1008 0 : return page->txns[ txn_gidx%FD_TXNCACHE_TXNS_PER_PAGE ];
1009 0 : }
1010 :
1011 : /* Locks tc and positions iter at the head of its current bucket. */
1012 :
1013 : static void
1014 0 : blockcache_iter_bucket_open( fd_txncache_blockcache_iter_t * iter ) {
1015 0 : fd_txncache_t * tc = iter->tc;
1016 0 : fd_rwlock_t * lock = tc->shmem->lock;
1017 0 : uint const * heads = tc->blockcache_pool[ iter->blockcache_idx ].heads;
1018 0 : ulong bucket = iter->bucket;
1019 :
1020 0 : fd_racesan_hook( "txncache:bucket_started" );
1021 0 : lock_read( tc );
1022 :
1023 0 : uint head;
1024 0 : int write_lock = 0;
1025 0 : for( ulong fetch_cnt=0UL;; fetch_cnt++ ) {
1026 : /* Inserts may publish a new head concurrently under the read lock.
1027 : The chain below the head loaded here is immutable until unlock. */
1028 0 : head = FD_VOLATILE_CONST( heads[ bucket ] );
1029 :
1030 : /* Check that every page of the chain is resident. */
1031 0 : ulong miss = ULONG_MAX;
1032 0 : for( uint idx=head; idx!=UINT_MAX; ) {
1033 0 : fd_txncache_txnpage_t const * page = page_io( tc, idx/FD_TXNCACHE_TXNS_PER_PAGE, 0 );
1034 0 : if( FD_UNLIKELY( !page ) ) {
1035 0 : miss = idx/FD_TXNCACHE_TXNS_PER_PAGE;
1036 0 : break;
1037 0 : }
1038 0 : idx = page->txns[ idx%FD_TXNCACHE_TXNS_PER_PAGE ]->blockcache_next;
1039 0 : }
1040 0 : if( FD_LIKELY( miss==ULONG_MAX ) ) break;
1041 :
1042 : /* Load the missing page without the lock and check again, or after
1043 : min(pages, frames) loads hold the write lock, which loads pages
1044 : in place. The chain may be compacted while unlocked, so the head
1045 : is reloaded. */
1046 0 : ulong seq = tc->shmem->release_seq;
1047 0 : int fetch = fetch_cnt<fd_ulong_min( tc->blockcache_pool[ iter->blockcache_idx ].shmem->pages_cnt, tc->shmem->frames_max );
1048 0 : fd_rwlock_unread( lock );
1049 0 : if( !fetch_cnt ) { fd_racesan_hook( "txncache:bucket_iter_miss" ); }
1050 0 : if( FD_LIKELY( fetch ) ) {
1051 0 : page_fetch( tc, miss, seq );
1052 0 : lock_read( tc );
1053 0 : continue;
1054 0 : }
1055 0 : lock_write( tc );
1056 0 : write_lock = 1;
1057 0 : head = heads[ bucket ];
1058 0 : break;
1059 0 : }
1060 :
1061 0 : iter->write_lock = write_lock;
1062 0 : iter->txn = bucket_chain_txn( tc, head, write_lock );
1063 0 : }
1064 :
1065 : static void
1066 0 : blockcache_iter_bucket_close( fd_txncache_blockcache_iter_t * iter ) {
1067 0 : if( iter->write_lock ) fd_rwlock_unwrite( iter->tc->shmem->lock );
1068 0 : else fd_rwlock_unread ( iter->tc->shmem->lock );
1069 0 : }
1070 :
1071 : /* Moves past empty buckets, holding the lock of one bucket at a time. */
1072 :
1073 : static void
1074 0 : blockcache_iter_skip_empty( fd_txncache_blockcache_iter_t * iter ) {
1075 0 : while( !iter->txn && iter->bucket+1UL<iter->tc->shmem->bucket_cnt ) {
1076 0 : blockcache_iter_bucket_close( iter );
1077 0 : iter->bucket++;
1078 0 : blockcache_iter_bucket_open( iter );
1079 0 : }
1080 0 : }
1081 :
1082 : fd_txncache_blockcache_iter_t *
1083 : fd_txncache_blockcache_iter_init( fd_txncache_blockcache_iter_t * iter,
1084 : fd_txncache_t * tc,
1085 0 : ulong blockcache_idx ) {
1086 0 : iter->tc = tc;
1087 0 : iter->blockcache_idx = blockcache_idx;
1088 0 : iter->bucket = 0UL;
1089 0 : blockcache_iter_bucket_open( iter );
1090 0 : blockcache_iter_skip_empty( iter );
1091 0 : return iter;
1092 0 : }
1093 :
1094 : void
1095 0 : fd_txncache_blockcache_iter_next( fd_txncache_blockcache_iter_t * iter ) {
1096 : /* Under the write lock, loading the next page may evict the current
1097 : transaction's frame, so read the link first. */
1098 0 : uint next = iter->txn->blockcache_next;
1099 0 : iter->txn = bucket_chain_txn( iter->tc, next, iter->write_lock );
1100 0 : blockcache_iter_skip_empty( iter );
1101 0 : }
1102 :
1103 : void
1104 0 : fd_txncache_blockcache_iter_fini( fd_txncache_blockcache_iter_t * iter ) {
1105 0 : blockcache_iter_bucket_close( iter );
1106 0 : }
|
