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 1488 : fd_txncache_align( void ) {
11 1488 : return FD_TXNCACHE_ALIGN;
12 1488 : }
13 :
14 : FD_FN_CONST ulong
15 606 : fd_txncache_footprint( ulong max_live_slots ) {
16 606 : ulong max_active_slots = FD_TXNCACHE_MAX_BLOCKHASH_DISTANCE+max_live_slots;
17 :
18 606 : ulong l;
19 606 : l = FD_LAYOUT_INIT;
20 606 : l = FD_LAYOUT_APPEND( l, FD_TXNCACHE_ALIGN, sizeof(fd_txncache_t) );
21 606 : l = FD_LAYOUT_APPEND( l, alignof(blockcache_t), max_active_slots*sizeof(blockcache_t) );
22 606 : return FD_LAYOUT_FINI( l, FD_TXNCACHE_ALIGN );
23 606 : }
24 :
25 : void *
26 : fd_txncache_new( void * ljoin,
27 : fd_txncache_shmem_t * shmem,
28 324 : int spill_fd ) {
29 324 : if( FD_UNLIKELY( !ljoin ) ) {
30 0 : FD_LOG_WARNING(( "NULL ljoin" ));
31 0 : return NULL;
32 0 : }
33 :
34 324 : 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 324 : ulong max_active_slots = shmem->active_slots_max;
40 324 : ulong blockhash_map_chains = fd_ulong_pow2_up( 2UL*shmem->active_slots_max );
41 324 : 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 324 : ulong _txnpages_max = shmem->txnpages_max;
47 :
48 324 : ulong _descends_footprint = descends_set_footprint( max_active_slots );
49 324 : if( FD_UNLIKELY( !_descends_footprint ) ) {
50 0 : FD_LOG_WARNING(( "invalid max_active_slots" ));
51 0 : return NULL;
52 0 : }
53 :
54 324 : FD_SCRATCH_ALLOC_INIT( l, shmem );
55 324 : fd_txncache_shmem_t * tc = FD_SCRATCH_ALLOC_APPEND( l, FD_TXNCACHE_SHMEM_ALIGN, sizeof(fd_txncache_shmem_t) );
56 324 : void * _blockhash_map = FD_SCRATCH_ALLOC_APPEND( l, blockhash_map_align(), blockhash_map_footprint( blockhash_map_chains ) );
57 324 : void * _blockcache_pool = FD_SCRATCH_ALLOC_APPEND( l, blockcache_pool_align(), blockcache_pool_footprint( max_active_slots ) );
58 324 : void * _blockcache_heads = FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), max_active_slots*bucket_cnt*sizeof(uint) );
59 324 : void * _blockcache_descends = FD_SCRATCH_ALLOC_APPEND( l, descends_set_align(), max_active_slots*_descends_footprint );
60 324 : void * _txnpages_free = FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), _txnpages_max*sizeof(uint) );
61 324 : void * _page_meta = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_txncache_page_meta_t), _txnpages_max*sizeof(fd_txncache_page_meta_t) );
62 324 : void * _frame_page = FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), shmem->frames_max*sizeof(uint) );
63 324 : void * _frames_free = FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), shmem->frames_max*sizeof(uint) );
64 324 : void * _txnpages = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_txncache_txnpage_t), shmem->frames_max*sizeof(fd_txncache_txnpage_t) );
65 324 : void * _scratch_heads = FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), bucket_cnt*sizeof(uint) );
66 324 : void * _scratch_txnpage = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_txncache_txnpage_t), sizeof(fd_txncache_txnpage_t) );
67 324 : void * _scratch_input = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_txncache_txnpage_t), sizeof(fd_txncache_txnpage_t) );
68 :
69 324 : FD_SCRATCH_ALLOC_INIT( l2, ljoin );
70 324 : fd_txncache_t * ltc = FD_SCRATCH_ALLOC_APPEND( l2, FD_TXNCACHE_ALIGN, sizeof(fd_txncache_t) );
71 324 : void * _local_blockcache_pool = FD_SCRATCH_ALLOC_APPEND( l2, alignof(blockcache_t), max_active_slots*sizeof(blockcache_t) );
72 :
73 324 : ltc->shmem = tc;
74 :
75 324 : ltc->blockcache_pool = (blockcache_t*)_local_blockcache_pool;
76 324 : ltc->blockcache_shmem_pool = blockcache_pool_join( _blockcache_pool );
77 :
78 56841 : for( ulong i=0UL; i<shmem->active_slots_max; i++ ) {
79 56517 : ltc->blockcache_pool[ i ].heads = (uint *)_blockcache_heads + i*bucket_cnt;
80 56517 : ltc->blockcache_pool[ i ].descends = descends_set_join( (uchar *)_blockcache_descends + i*_descends_footprint );
81 56517 : ltc->blockcache_pool[ i ].shmem = ltc->blockcache_shmem_pool + i;
82 56517 : FD_TEST( ltc->blockcache_pool[ i ].shmem );
83 56517 : }
84 :
85 324 : FD_TEST( ltc->blockcache_shmem_pool );
86 :
87 324 : ltc->blockhash_map = blockhash_map_join( _blockhash_map );
88 324 : FD_TEST( ltc->blockhash_map );
89 :
90 324 : ltc->txnpages_free = _txnpages_free;
91 324 : ltc->page_meta = _page_meta;
92 324 : ltc->frame_page = _frame_page;
93 324 : ltc->frames_free = _frames_free;
94 324 : ltc->txnpages = (fd_txncache_txnpage_t *)_txnpages;
95 324 : ltc->scratch_heads = _scratch_heads;
96 324 : ltc->scratch_txnpage = _scratch_txnpage;
97 324 : ltc->scratch_input = _scratch_input;
98 324 : ltc->spill_fd = spill_fd;
99 :
100 324 : memset( ltc->metrics, 0, sizeof(ltc->metrics) );
101 324 : fd_histf_join( fd_histf_new( ltc->metrics->query_ticks, 1UL, 1UL<<30 ) );
102 324 : fd_histf_join( fd_histf_new( ltc->metrics->insert_ticks, 1UL, 1UL<<30 ) );
103 :
104 324 : return (void *)ltc;
105 324 : }
106 :
107 : fd_txncache_join_metrics_t *
108 126 : fd_txncache_join_metrics( fd_txncache_t * tc ) {
109 126 : return tc->metrics;
110 126 : }
111 :
112 : fd_txncache_t *
113 333 : fd_txncache_join( void * ljoin ) {
114 333 : if( FD_UNLIKELY( !ljoin ) ) {
115 3 : FD_LOG_WARNING(( "NULL ljoin" ));
116 3 : return NULL;
117 3 : }
118 :
119 330 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)ljoin, fd_txncache_align() ) ) ) {
120 3 : FD_LOG_WARNING(( "misaligned ljoin" ));
121 3 : return NULL;
122 3 : }
123 :
124 327 : fd_txncache_t * tc = (fd_txncache_t *)ljoin;
125 :
126 327 : return tc;
127 330 : }
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 32347849 : lock_read( fd_txncache_t * tc ) FD_ACQUIRE_SHARED( tc->shmem->lock ) FD_NO_THREAD_SAFETY_ANALYSIS {
135 32347849 : if( FD_LIKELY( fd_rwlock_tryread( tc->shmem->lock ) ) ) return;
136 3204 : long t0 = fd_tickcount();
137 3204 : fd_rwlock_read( tc->shmem->lock );
138 3204 : FD_ATOMIC_FETCH_AND_ADD( &tc->shmem->lock_read_wait_ticks, (ulong)(fd_tickcount()-t0) );
139 3204 : }
140 :
141 : static inline void
142 244288 : lock_write( fd_txncache_t * tc ) FD_ACQUIRE( tc->shmem->lock ) FD_NO_THREAD_SAFETY_ANALYSIS {
143 244288 : if( FD_LIKELY( fd_rwlock_trywrite( tc->shmem->lock ) ) ) return;
144 3201 : long t0 = fd_tickcount();
145 3201 : fd_rwlock_write( tc->shmem->lock );
146 3201 : tc->shmem->lock_write_wait_ticks += (ulong)(fd_tickcount()-t0);
147 3201 : }
148 :
149 : static void
150 : spill_rw( fd_txncache_t * tc,
151 : ulong off,
152 : uchar * buf,
153 : ulong sz,
154 42292 : int write ) {
155 42292 : FD_TEST( tc->spill_fd>=0 );
156 84584 : while( sz ) {
157 42292 : long n = write ? pwrite( tc->spill_fd, buf, sz, (off_t)off )
158 42292 : : pread ( tc->spill_fd, buf, sz, (off_t)off );
159 42292 : if( FD_UNLIKELY( n<0L && errno==EINTR ) ) continue;
160 42292 : 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 42292 : if( FD_UNLIKELY( !n ) ) FD_LOG_CRIT(( "txncache spill %s at offset %lu", write ? "write made no progress" : "read reached unexpected EOF", off ));
162 42292 : sz -= (ulong)n;
163 42292 : 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 42292 : buf += (ulong)n;
167 42292 : off += (ulong)n;
168 42292 : }
169 42292 : }
170 :
171 : static void
172 : spill_io( fd_txncache_t * tc,
173 : ulong page,
174 : fd_txncache_txnpage_t * data,
175 42274 : int write ) {
176 42274 : long t0 = fd_tickcount();
177 42274 : spill_rw( tc, page*sizeof(*data), (uchar *)data, sizeof(*data), write );
178 42274 : if( write ) FD_ATOMIC_FETCH_AND_ADD( &tc->shmem->page_spilled_cnt, 1UL );
179 14428 : else FD_ATOMIC_FETCH_AND_ADD( &tc->shmem->page_spill_read_cnt, 1UL );
180 42274 : FD_ATOMIC_FETCH_AND_ADD( &tc->shmem->spill_ticks, (ulong)(fd_tickcount()-t0) );
181 42274 : }
182 :
183 : static uint
184 : frame_take( fd_txncache_t * tc,
185 164052137 : uint * victim ) {
186 164052137 : fd_txncache_shmem_t * shmem = tc->shmem;
187 164052137 : *victim = UINT_MAX;
188 164052137 : if( FD_LIKELY( shmem->frames_free_cnt ) ) return tc->frames_free[ --shmem->frames_free_cnt ];
189 492116837 : for( ulong i=0UL; i<2UL*shmem->frames_max; i++ ) {
190 328103815 : uint frame = (uint)shmem->evict_next;
191 328103815 : shmem->evict_next = (shmem->evict_next+1UL)%shmem->frames_max;
192 328103815 : uint owner = tc->frame_page[ frame ];
193 328103815 : FD_TEST( owner!=UINT_MAX );
194 328103815 : fd_txncache_page_meta_t * old = &tc->page_meta[ owner ];
195 328103815 : if( FD_VOLATILE_CONST( old->io ) ) continue;
196 78179 : if( old->ref ) {
197 39085 : old->ref = 0U;
198 39085 : continue;
199 39085 : }
200 39094 : old->frame = UINT_MAX;
201 39094 : shmem->frame_evicted_cnt++;
202 39094 : *victim = owner;
203 39094 : return frame;
204 78179 : }
205 164013022 : return UINT_MAX;
206 164052116 : }
207 :
208 : static inline fd_txncache_txnpage_t *
209 : page_io( fd_txncache_t * tc,
210 : ulong page,
211 2880786478 : int write_lock ) {
212 : /* Fast path: the page is resident. A fetch may publish the frame
213 : concurrently. */
214 2880786478 : fd_txncache_page_meta_t * meta = &tc->page_meta[ page ];
215 2880786478 : uint resident = FD_VOLATILE_CONST( meta->frame );
216 2880931076 : if( FD_LIKELY( resident!=UINT_MAX ) ) {
217 2880931076 : if( FD_UNLIKELY( !meta->ref ) ) meta->ref = 1U;
218 2880931076 : return &tc->txnpages[ resident ];
219 2880931076 : }
220 :
221 : /* Loading changes the frame cache, which needs the write lock. */
222 5 : if( FD_LIKELY( !write_lock ) ) {
223 16102 : FD_ATOMIC_FETCH_AND_ADD( &tc->shmem->page_miss_cnt, 1UL );
224 16102 : return NULL;
225 16102 : }
226 5 : 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 5 : while( FD_UNLIKELY( FD_VOLATILE_CONST( meta->io ) ) ) FD_SPIN_PAUSE();
231 5 : FD_COMPILER_MFENCE();
232 5 : 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 5 : uint victim;
238 5 : uint frame;
239 5 : while( FD_UNLIKELY( (frame=frame_take( tc, &victim ))==UINT_MAX ) ) FD_SPIN_PAUSE();
240 5 : if( victim!=UINT_MAX ) {
241 8999 : fd_txncache_page_meta_t * old = &tc->page_meta[ victim ];
242 8999 : if( old->dirty ) {
243 3090 : spill_io( tc, victim, &tc->txnpages[ frame ], 1 );
244 3090 : old->dirty = 0U;
245 3090 : old->disk_valid = 1;
246 3090 : }
247 8999 : }
248 :
249 : /* Read the page back, or start it empty if it was never written. */
250 5 : fd_txncache_txnpage_t * data = &tc->txnpages[ frame ];
251 5 : if( meta->disk_valid ) {
252 9014 : spill_io( tc, page, data, 0 );
253 5 : } else {
254 5 : data->free = FD_TXNCACHE_TXNS_PER_PAGE;
255 5 : }
256 5 : tc->frame_page[ frame ] = (uint)page;
257 5 : meta->frame = frame;
258 5 : meta->ref = 1U;
259 5 : return data;
260 5 : }
261 :
262 : static void
263 : page_fetch( fd_txncache_t * tc,
264 : ulong page,
265 33158 : ulong seq ) {
266 33158 : fd_txncache_page_meta_t * meta = &tc->page_meta[ page ];
267 33158 : lock_write( tc );
268 :
269 : /* page may have been freed, or is already resident. */
270 33158 : 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 33158 : if( FD_UNLIKELY( meta->io ) ) {
277 514 : fd_rwlock_unwrite( tc->shmem->lock );
278 38949254 : while( FD_UNLIKELY( FD_VOLATILE_CONST( meta->io ) ) ) FD_SPIN_PAUSE();
279 514 : return;
280 514 : }
281 :
282 : /* Every frame is busy. */
283 32644 : uint victim;
284 32644 : uint frame = frame_take( tc, &victim );
285 32644 : if( FD_UNLIKELY( frame==UINT_MAX ) ) {
286 2549 : fd_rwlock_unwrite( tc->shmem->lock );
287 2549 : FD_SPIN_PAUSE();
288 2549 : return;
289 2549 : }
290 :
291 : /* Mark page and the evicted page busy until i/o finishes. */
292 30095 : fd_txncache_page_meta_t * old = victim!=UINT_MAX ? &tc->page_meta[ victim ] : NULL;
293 30095 : int write = old && old->dirty;
294 30095 : if( write ) {
295 24756 : old->dirty = 0U;
296 24756 : old->io = 1U;
297 24756 : }
298 30095 : int read = meta->disk_valid;
299 30095 : meta->io = 1U;
300 30095 : tc->frame_page[ frame ] = (uint)page;
301 30095 : fd_rwlock_unwrite( tc->shmem->lock );
302 :
303 : /* Write back the evicted page, then read page, without the lock. */
304 30095 : fd_racesan_hook( "txncache:fetch_io" );
305 30095 : fd_txncache_txnpage_t * data = &tc->txnpages[ frame ];
306 30095 : if( write ) spill_io( tc, victim, data, 1 );
307 30095 : if( read ) spill_io( tc, page, data, 0 );
308 24681 : 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 30095 : if( write ) {
313 24756 : old->disk_valid = 1;
314 24756 : FD_HW_MFENCE_ST();
315 24756 : FD_VOLATILE( old->io ) = 0U;
316 24756 : }
317 30095 : meta->ref = 1U;
318 30095 : FD_HW_MFENCE_ST();
319 30095 : FD_VOLATILE( meta->frame ) = frame;
320 30095 : FD_HW_MFENCE_ST();
321 30095 : FD_VOLATILE( meta->io ) = 0U;
322 30095 : }
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 48789 : ulong page ) {
328 48789 : fd_txncache_page_meta_t * meta = &tc->page_meta[ page ];
329 48789 : while( FD_UNLIKELY( FD_VOLATILE_CONST( meta->io ) ) ) FD_SPIN_PAUSE();
330 48789 : FD_COMPILER_MFENCE();
331 48789 : tc->shmem->release_seq++;
332 48789 : if( meta->frame!=UINT_MAX ) {
333 24666 : tc->frame_page[ meta->frame ] = UINT_MAX;
334 24666 : tc->frames_free[ tc->shmem->frames_free_cnt++ ] = meta->frame;
335 24666 : meta->frame = UINT_MAX;
336 24666 : }
337 48789 : meta->dirty = 0U;
338 48789 : meta->disk_valid = 0;
339 48789 : meta->next = UINT_MAX;
340 48789 : tc->txnpages_free[ tc->shmem->txnpages_free_cnt++ ] = (uint)page;
341 48789 : }
342 :
343 : void
344 4890 : fd_txncache_reset( fd_txncache_t * tc ) {
345 4890 : lock_write( tc );
346 :
347 4890 : tc->shmem->root_cnt = 0UL;
348 4890 : root_slist_remove_all( tc->shmem->root_ll, tc->blockcache_shmem_pool );
349 :
350 4890 : tc->shmem->txnpages_free_cnt = tc->shmem->txnpages_max;
351 4890 : tc->shmem->frames_free_cnt = tc->shmem->frames_max;
352 4890 : tc->shmem->evict_next = 0UL;
353 4890 : tc->shmem->release_seq++;
354 1269459 : for( ulong i=0UL; i<tc->shmem->txnpages_max; i++ ) {
355 1264569 : while( FD_UNLIKELY( FD_VOLATILE_CONST( tc->page_meta[ i ].io ) ) ) FD_SPIN_PAUSE();
356 1264569 : }
357 4890 : FD_COMPILER_MFENCE();
358 1269459 : for( ulong i=0UL; i<tc->shmem->txnpages_max; i++ ) {
359 1264569 : tc->txnpages_free[ i ] = (uint)i;
360 1264569 : tc->page_meta[ i ].frame = UINT_MAX;
361 1264569 : tc->page_meta[ i ].dirty = 0U;
362 1264569 : tc->page_meta[ i ].disk_valid = 0;
363 1264569 : tc->page_meta[ i ].next = UINT_MAX;
364 1264569 : }
365 1189038 : for( ulong i=0UL; i<tc->shmem->frames_max; i++ ) {
366 1184148 : tc->frame_page[ i ] = UINT_MAX;
367 1184148 : tc->frames_free[ i ] = (uint)i;
368 1184148 : }
369 :
370 4890 : blockcache_pool_reset( tc->blockcache_shmem_pool );
371 4890 : blockhash_map_reset( tc->blockhash_map );
372 :
373 4890 : fd_rwlock_unwrite( tc->shmem->lock );
374 4890 : }
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 108 : int write ) {
401 108 : FD_TEST_ERR( tc->shmem->txnpages_free_cnt==tc->shmem->txnpages_max );
402 108 : FD_TEST( fd_ulong_is_aligned( off, FD_TXNCACHE_SNAPIN_STAGE_ALIGN ) );
403 108 : FD_TEST( fd_ulong_is_aligned( sz, FD_TXNCACHE_SNAPIN_STAGE_ALIGN ) );
404 108 : FD_TEST( fd_ulong_is_aligned( (ulong)buf, FD_TXNCACHE_SNAPIN_STAGE_ALIGN ) );
405 :
406 108 : ulong mem_sz = tc->shmem->frames_max*sizeof(fd_txncache_txnpage_t);
407 108 : uchar * p = (uchar *)buf;
408 108 : if( off<mem_sz ) {
409 102 : ulong n = fd_ulong_min( sz, mem_sz-off );
410 102 : uchar * mem = (uchar *)tc->txnpages+off;
411 102 : if( write ) fd_memcpy( mem, p, n );
412 48 : else fd_memcpy( p, mem, n );
413 102 : off += n;
414 102 : p += n;
415 102 : sz -= n;
416 102 : }
417 108 : if( sz ) spill_rw( tc, off-mem_sz, p, sz, write );
418 108 : }
419 :
420 : FD_FN_PURE static inline ulong
421 : fd_txncache_bucket( fd_txncache_t const * tc,
422 34345620 : uchar const * txnhash ) {
423 34345620 : return fd_ulong_hash( FD_LOAD( ulong, txnhash )^tc->shmem->seed )%tc->shmem->bucket_cnt;
424 34345620 : }
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 22919586 : ulong * out_idx ) {
436 22919586 : fd_txncache_shmem_t * shmem = tc->shmem;
437 22919586 : for(;;) {
438 : /* Use the newest page if it has room. */
439 22919586 : uint head = FD_VOLATILE_CONST( blockcache->shmem->page_head );
440 22919586 : if( FD_LIKELY( head!=UINT_MAX ) ) {
441 22869534 : fd_txncache_txnpage_t * page = page_io( tc, head, write );
442 22869534 : *out_idx = head;
443 22869534 : if( FD_UNLIKELY( !page ) ) return NULL;
444 22863366 : if( FD_LIKELY( FD_VOLATILE_CONST( page->free ) ) ) return page;
445 22863366 : }
446 :
447 : /* Need a new page. Try to make sure we can safely take the lock
448 : and that there's enough capacity. */
449 53937 : if( !write ) {
450 52374 : fd_rwlock_write( shmem->alloc_lock );
451 52374 : 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 52374 : }
456 :
457 53937 : ulong page_cnt = blockcache->shmem->pages_cnt;
458 53937 : if( FD_UNLIKELY( page_cnt>=shmem->txnpages_per_blockhash_max || !shmem->txnpages_free_cnt ) ) {
459 3120 : if( !write ) fd_rwlock_unwrite( shmem->alloc_lock );
460 3120 : *out_idx = ULONG_MAX;
461 3120 : return NULL;
462 3120 : }
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 50817 : ulong idx = tc->txnpages_free[ --shmem->txnpages_free_cnt ];
468 50817 : fd_txncache_txnpage_t * page = NULL;
469 50817 : if( write ) {
470 3 : page = page_io( tc, idx, 1 );
471 50814 : } else if( FD_LIKELY( shmem->frames_free_cnt ) ) {
472 26133 : uint frame = tc->frames_free[ --shmem->frames_free_cnt ];
473 26133 : tc->frame_page[ frame ] = (uint)idx;
474 26133 : tc->page_meta[ idx ].frame = frame;
475 26133 : page = &tc->txnpages[ frame ];
476 26133 : }
477 50817 : if( FD_LIKELY( page ) ) page->free = FD_TXNCACHE_TXNS_PER_PAGE;
478 50817 : tc->page_meta[ idx ].next = head;
479 50817 : blockcache->shmem->pages_cnt = page_cnt+1UL;
480 50817 : FD_HW_MFENCE_ST();
481 50817 : FD_VOLATILE( blockcache->shmem->page_head ) = (uint)idx;
482 50817 : if( !write ) fd_rwlock_unwrite( shmem->alloc_lock );
483 50817 : *out_idx = idx;
484 50817 : return page;
485 53937 : }
486 22919586 : }
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 22885617 : uchar const * txnhash ) {
495 22885617 : for(;;) {
496 22885617 : ushort txnpage_free = txnpage->free;
497 22885617 : if( FD_UNLIKELY( !txnpage_free ) ) return 0;
498 22885617 : 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 22885617 : ulong txn_idx = FD_TXNCACHE_TXNS_PER_PAGE-txnpage_free;
504 22885617 : ulong txnhash_offset = blockcache->shmem->txnhash_offset;
505 22885617 : memcpy( txnpage->txns[ txn_idx ]->txnhash, txnhash+txnhash_offset, 20UL );
506 22885617 : txnpage->txns[ txn_idx ]->fork_id = fork_id;
507 22885617 : txnpage->txns[ txn_idx ]->generation = tc->blockcache_pool[ fork_id.val ].shmem->generation;
508 22885617 : FD_COMPILER_MFENCE();
509 :
510 22885617 : ulong txn_bucket = fd_txncache_bucket( tc, txnhash+txnhash_offset );
511 22885617 : for(;;) {
512 22885617 : uint head = blockcache->heads[ txn_bucket ];
513 22885617 : txnpage->txns[ txn_idx ]->blockcache_next = head;
514 22885617 : FD_COMPILER_MFENCE();
515 22885617 : 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 22885617 : if( FD_UNLIKELY( !tc->page_meta[ txnpage_idx ].dirty ) ) tc->page_meta[ txnpage_idx ].dirty = 1U;
521 22885617 : return 1;
522 22885617 : }
523 22885617 : }
524 :
525 : fd_txncache_fork_id_t
526 : fd_txncache_attach_child( fd_txncache_t * tc,
527 75186 : fd_txncache_fork_id_t parent_fork_id ) {
528 75186 : lock_write( tc );
529 :
530 75186 : FD_TEST( blockcache_pool_free( tc->blockcache_shmem_pool ) );
531 75186 : ulong idx = blockcache_pool_idx_acquire( tc->blockcache_shmem_pool );
532 :
533 75186 : blockcache_t * fork = &tc->blockcache_pool[ idx ];
534 75186 : fd_txncache_fork_id_t fork_id = { .val = (ushort)idx };
535 :
536 75186 : fork->shmem->generation = tc->shmem->blockcache_generation++;
537 75186 : fork->shmem->child_id = (fd_txncache_fork_id_t){ .val = USHORT_MAX };
538 :
539 75186 : if( FD_LIKELY( parent_fork_id.val==USHORT_MAX ) ) {
540 4956 : FD_TEST( blockcache_pool_free( tc->blockcache_shmem_pool )==blockcache_pool_max( tc->blockcache_shmem_pool )-1UL );
541 4956 : fork->shmem->parent_id = (fd_txncache_fork_id_t){ .val = USHORT_MAX };
542 4956 : fork->shmem->sibling_id = (fd_txncache_fork_id_t){ .val = USHORT_MAX };
543 :
544 4956 : descends_set_null( fork->descends );
545 4956 : root_slist_ele_push_tail( tc->shmem->root_ll, fork->shmem, tc->blockcache_shmem_pool );
546 70230 : } else {
547 70230 : 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 70230 : fork->shmem->sibling_id = parent->shmem->child_id;
553 70230 : fork->shmem->parent_id = parent_fork_id;
554 70230 : parent->shmem->child_id = fork_id;
555 :
556 70230 : descends_set_copy( fork->descends, parent->descends );
557 70230 : descends_set_insert( fork->descends, parent_fork_id.val );
558 70230 : }
559 :
560 75186 : fork->shmem->txnhash_offset = 0UL;
561 75186 : fork->shmem->frozen = 0;
562 75186 : memset( fork->heads, 0xFF, tc->shmem->bucket_cnt*sizeof(uint) );
563 75186 : fork->shmem->pages_cnt = 0UL;
564 75186 : fork->shmem->page_head = UINT_MAX;
565 :
566 75186 : fd_rwlock_unwrite( tc->shmem->lock );
567 75186 : return fork_id;
568 75186 : }
569 :
570 : void
571 : fd_txncache_attach_blockhash( fd_txncache_t * tc,
572 : fd_txncache_fork_id_t fork_id,
573 54 : uchar const * blockhash ) {
574 54 : lock_write( tc );
575 :
576 54 : blockcache_t * fork = &tc->blockcache_pool[ fork_id.val ];
577 54 : FD_TEST( !fork->shmem->frozen );
578 54 : fork->shmem->frozen = 1;
579 :
580 54 : memcpy( fork->shmem->blockhash.uc, blockhash, 32UL );
581 :
582 54 : blockhash_map_ele_insert( tc->blockhash_map, fork->shmem, tc->blockcache_shmem_pool );
583 :
584 54 : fd_rwlock_unwrite( tc->shmem->lock );
585 54 : }
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 60945 : uchar const * blockhash ) {
592 60945 : lock_write( tc );
593 :
594 60945 : blockcache_t * fork = &tc->blockcache_pool[ fork_id.val ];
595 60945 : FD_TEST( fork->shmem->frozen<=1 );
596 60945 : FD_TEST( fork->shmem->frozen>=0 );
597 60945 : fork->shmem->txnhash_offset = txnhash_offset;
598 :
599 60945 : memcpy( fork->shmem->blockhash.uc, blockhash, 32UL );
600 :
601 60945 : if( FD_LIKELY( !fork->shmem->frozen ) ) blockhash_map_ele_insert( tc->blockhash_map, fork->shmem, tc->blockcache_shmem_pool );
602 60945 : fork->shmem->frozen = 2;
603 :
604 60945 : fd_rwlock_unwrite( tc->shmem->lock );
605 60945 : }
606 :
607 : static inline void
608 : remove_blockcache( fd_txncache_t * tc,
609 58875 : blockcache_t * blockcache ) {
610 58875 : FD_TEST( blockcache->shmem->frozen>=0 );
611 107169 : for( uint page=blockcache->shmem->page_head; page!=UINT_MAX; ) {
612 48294 : uint next = tc->page_meta[ page ].next;
613 48294 : release_page( tc, page );
614 48294 : page = next;
615 48294 : }
616 58875 : blockcache->shmem->page_head = UINT_MAX;
617 58875 : blockcache->shmem->pages_cnt = 0UL;
618 :
619 58875 : ulong idx = blockcache_pool_idx( tc->blockcache_shmem_pool, blockcache->shmem );
620 9267837 : for( ulong i=0UL; i<tc->shmem->active_slots_max; i++ ) descends_set_remove( tc->blockcache_pool[ i ].descends, idx );
621 :
622 58875 : if( FD_LIKELY( blockcache->shmem->frozen ) ) blockhash_map_ele_remove_fast( tc->blockhash_map, blockcache->shmem, tc->blockcache_shmem_pool );
623 58875 : blockcache->shmem->frozen = -1;
624 58875 : blockcache_pool_ele_release( tc->blockcache_shmem_pool, blockcache->shmem );
625 58875 : }
626 :
627 : static inline void
628 : remove_children( fd_txncache_t * tc,
629 : blockcache_t const * fork,
630 62868 : blockcache_t const * except ) {
631 62868 : fd_txncache_fork_id_t sibling_idx = fork->shmem->child_id;
632 118017 : while( sibling_idx.val!=USHORT_MAX ) {
633 55149 : blockcache_t * sibling = &tc->blockcache_pool[ sibling_idx.val ];
634 :
635 55149 : sibling_idx = sibling->shmem->sibling_id;
636 55149 : if( FD_UNLIKELY( sibling==except ) ) continue;
637 :
638 1998 : remove_children( tc, sibling, except );
639 1998 : remove_blockcache( tc, sibling );
640 1998 : }
641 62868 : }
642 :
643 : void
644 : fd_txncache_cancel_fork( fd_txncache_t * tc,
645 7719 : fd_txncache_fork_id_t fork_id ) {
646 7719 : lock_write( tc );
647 7719 : blockcache_t * fork = &tc->blockcache_pool[ fork_id.val ];
648 7719 : FD_TEST( fork->shmem->parent_id.val!=USHORT_MAX );
649 :
650 : /* The soon-to-be-pruned subtree must be unrooted. */
651 7719 : fd_txncache_blockcache_shmem_t const * latest_root = root_slist_ele_peek_tail_const( tc->shmem->root_ll, tc->blockcache_shmem_pool );
652 7719 : FD_TEST( latest_root );
653 7719 : FD_TEST( descends_set_test( fork->descends, blockcache_pool_idx( tc->blockcache_shmem_pool, latest_root ) ) );
654 :
655 7719 : remove_children( tc, fork, NULL );
656 7719 : remove_blockcache( tc, fork );
657 7719 : ushort * fork_id_p = &(tc->blockcache_pool[ fork->shmem->parent_id.val ].shmem->child_id.val);
658 7725 : while( *fork_id_p!=fork_id.val ) {
659 6 : fork_id_p = &(tc->blockcache_pool[ *fork_id_p ].shmem->sibling_id.val);
660 6 : }
661 7719 : *fork_id_p = fork->shmem->sibling_id.val;
662 7719 : fd_rwlock_unwrite( tc->shmem->lock );
663 7719 : }
664 :
665 : void
666 : fd_txncache_advance_root( fd_txncache_t * tc,
667 53151 : fd_txncache_fork_id_t fork_id ) {
668 53151 : lock_write( tc );
669 :
670 53151 : blockcache_t * fork = &tc->blockcache_pool[ fork_id.val ];
671 53151 : FD_TEST( fork->shmem->parent_id.val!=USHORT_MAX );
672 :
673 53151 : blockcache_t * parent_fork = &tc->blockcache_pool[ fork->shmem->parent_id.val ];
674 53151 : 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 53151 : FD_BASE58_ENCODE_32_BYTES( parent_fork->shmem->blockhash.uc, parent_blockhash_b58 );
685 53151 : FD_BASE58_ENCODE_32_BYTES( fork->shmem->blockhash.uc, fork_blockhash_b58 );
686 53151 : FD_LOG_DEBUG(( "advancing root from %s to %s",
687 53151 : parent_blockhash_b58,
688 53151 : 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 53151 : remove_children( tc, parent_fork, fork );
694 53151 : parent_fork->shmem->child_id = fork_id;
695 53151 : 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 53151 : tc->shmem->root_cnt++;
701 53151 : root_slist_ele_push_tail( tc->shmem->root_ll, fork->shmem, tc->blockcache_shmem_pool );
702 53151 : if( FD_LIKELY( tc->shmem->root_cnt>FD_TXNCACHE_MAX_BLOCKHASH_DISTANCE ) ) {
703 49158 : fd_txncache_blockcache_shmem_t * old_root_shmem = root_slist_ele_pop_head( tc->shmem->root_ll, tc->blockcache_shmem_pool );
704 49158 : FD_TEST( old_root_shmem );
705 49158 : blockcache_t * old_root = &tc->blockcache_pool[ blockcache_pool_idx( tc->blockcache_shmem_pool, old_root_shmem ) ];
706 :
707 49158 : root_slist_ele_peek_head( tc->shmem->root_ll, tc->blockcache_shmem_pool )->parent_id.val = USHORT_MAX;
708 :
709 49158 : remove_blockcache( tc, old_root );
710 49158 : tc->shmem->root_cnt--;
711 49158 : }
712 :
713 53151 : fd_rwlock_unwrite( tc->shmem->lock );
714 53151 : }
715 :
716 : static inline blockcache_t *
717 : blockhash_on_fork( fd_txncache_t * tc,
718 : blockcache_t const * fork,
719 28602573 : uchar const * blockhash ) {
720 28602573 : 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 28602573 : if( FD_UNLIKELY( !candidate ) ) return NULL;
722 :
723 28602573 : while( candidate ) {
724 28602573 : ulong candidate_idx = blockcache_pool_idx( tc->blockcache_shmem_pool, candidate );
725 28602573 : 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 28602573 : }
730 :
731 : static void
732 : purge_stale_on_blockcache( fd_txncache_t * tc,
733 7071 : blockcache_t * blockcache ) {
734 7071 : FD_TEST( blockcache->shmem->frozen>=0 );
735 7071 : memset( tc->scratch_heads, 0xFF, tc->shmem->bucket_cnt*sizeof(tc->scratch_heads[ 0 ]) );
736 7071 : uint scratch_page_head = UINT_MAX;
737 7071 : ulong scratch_pages_cnt = 0UL;
738 7071 : ulong scratch_txnpage_idx = ULONG_MAX;
739 7071 : tc->scratch_txnpage->free = 0;
740 7071 : uint next_txnpage_idx = blockcache->shmem->page_head;
741 9822 : while( next_txnpage_idx!=UINT_MAX ) {
742 2751 : ulong curr_txnpage_idx = next_txnpage_idx;
743 2751 : next_txnpage_idx = tc->page_meta[ curr_txnpage_idx ].next;
744 2751 : fd_txncache_txnpage_t * input = tc->scratch_input;
745 2751 : memcpy( input, page_io( tc, curr_txnpage_idx, 1 ), sizeof(*input) );
746 2751 : ulong curr_txn_cnt = FD_TXNCACHE_TXNS_PER_PAGE-input->free;
747 22538943 : for( ulong j=0UL; j<curr_txn_cnt; j++ ) {
748 22536192 : fd_txncache_single_txn_t const * curr_txn = input->txns[ curr_txn_cnt-j-1UL ];
749 22536192 : blockcache_t const * txn_fork = &tc->blockcache_pool[ curr_txn->fork_id.val ];
750 22536192 : if( FD_LIKELY( txn_fork->shmem->frozen>=0 && txn_fork->shmem->generation==curr_txn->generation ) ) {
751 : /* Valid transaction. Keep. */
752 5778486 : if( FD_UNLIKELY( !tc->scratch_txnpage->free ) ) {
753 2256 : FD_TEST( scratch_txnpage_idx!=curr_txnpage_idx );
754 2256 : if( FD_LIKELY( scratch_txnpage_idx!=ULONG_MAX ) ) {
755 240 : memcpy( page_io( tc, scratch_txnpage_idx, 1 ), tc->scratch_txnpage, sizeof(fd_txncache_txnpage_t) );
756 240 : tc->page_meta[ scratch_txnpage_idx ].dirty = 1U;
757 240 : }
758 2256 : scratch_txnpage_idx = curr_txnpage_idx;
759 2256 : tc->scratch_txnpage->free = FD_TXNCACHE_TXNS_PER_PAGE;
760 2256 : tc->page_meta[ scratch_txnpage_idx ].next = scratch_page_head;
761 2256 : scratch_page_head = (uint)scratch_txnpage_idx;
762 2256 : scratch_pages_cnt++;
763 2256 : }
764 5778486 : ulong txn_idx = FD_TXNCACHE_TXNS_PER_PAGE-tc->scratch_txnpage->free;
765 5778486 : memcpy( tc->scratch_txnpage->txns[ txn_idx ], curr_txn, sizeof(*curr_txn) );
766 5778486 : ulong txn_bucket = fd_txncache_bucket( tc, curr_txn->txnhash );
767 5778486 : uint head = tc->scratch_heads[ txn_bucket ];
768 5778486 : tc->scratch_txnpage->txns[ txn_idx ]->blockcache_next = head;
769 5778486 : ulong txn_gidx = FD_TXNCACHE_TXNS_PER_PAGE*scratch_txnpage_idx+txn_idx;
770 5778486 : FD_TEST( txn_gidx<UINT_MAX );
771 5778486 : tc->scratch_heads[ txn_bucket ] = (uint)txn_gidx;
772 5778486 : tc->scratch_txnpage->free--;
773 16757706 : } else {
774 : /* Stale transaction. Drop. */
775 16757706 : continue;
776 16757706 : }
777 22536192 : }
778 2751 : if( FD_UNLIKELY( curr_txnpage_idx!=scratch_txnpage_idx ) ) {
779 : /* The txnpage is not being used for compaction, free it up. */
780 495 : release_page( tc, curr_txnpage_idx );
781 495 : }
782 2751 : }
783 7071 : if( FD_LIKELY( scratch_txnpage_idx!=ULONG_MAX ) ) {
784 2016 : memcpy( page_io( tc, scratch_txnpage_idx, 1 ), tc->scratch_txnpage, sizeof(fd_txncache_txnpage_t) );
785 2016 : tc->page_meta[ scratch_txnpage_idx ].dirty = 1U;
786 2016 : }
787 7071 : blockcache->shmem->page_head = scratch_page_head;
788 7071 : blockcache->shmem->pages_cnt = scratch_pages_cnt;
789 7071 : memcpy( blockcache->heads, tc->scratch_heads, tc->shmem->bucket_cnt*sizeof(blockcache->heads[0]) );
790 7071 : }
791 :
792 : static void
793 : purge_stale_on_fork( fd_txncache_t * tc,
794 7071 : blockcache_t * fork ) {
795 7071 : purge_stale_on_blockcache( tc, fork );
796 :
797 7071 : fd_txncache_fork_id_t sibling_idx = fork->shmem->child_id;
798 12582 : while( sibling_idx.val!=USHORT_MAX ) {
799 5511 : blockcache_t * sibling = &tc->blockcache_pool[ sibling_idx.val ];
800 5511 : purge_stale_on_fork( tc, sibling );
801 5511 : sibling_idx = sibling->shmem->sibling_id;
802 5511 : }
803 7071 : }
804 :
805 : static void
806 1560 : purge_stale( fd_txncache_t * tc ) {
807 1560 : fd_txncache_blockcache_shmem_t * root_shmem = root_slist_ele_peek_head( tc->shmem->root_ll, tc->blockcache_shmem_pool );
808 1560 : FD_TEST( root_shmem );
809 1560 : blockcache_t * root = &tc->blockcache_pool[ blockcache_pool_idx( tc->blockcache_shmem_pool, root_shmem ) ];
810 1560 : 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 1560 : purge_stale_on_fork( tc, root );
818 1560 : FD_LOG_WARNING(( "purge_stale: txnpages_free %lu -> %lu", free_before, tc->shmem->txnpages_free_cnt ));
819 1560 : }
820 :
821 : void
822 : fd_txncache_insert( fd_txncache_t * tc,
823 : fd_txncache_fork_id_t fork_id,
824 : uchar const * blockhash,
825 22885617 : uchar const * txnhash ) {
826 22885617 : long t0 = fd_tickcount();
827 22885617 : lock_read( tc );
828 :
829 22885617 : blockcache_t const * fork = &tc->blockcache_pool[ fork_id.val ];
830 22885617 : FD_TEST( fork->shmem->frozen<=1 );
831 22885617 : FD_TEST( fork->shmem->frozen>=0 );
832 22885617 : blockcache_t * blockcache = blockhash_on_fork( tc, fork, blockhash );
833 22885617 : FD_TEST( blockcache );
834 :
835 22885617 : ulong fetched_idx = ULONG_MAX;
836 22885617 : int miss = 0;
837 22885617 : int write = 0;
838 22913400 : for(;;) {
839 : /* First look for an in-memory page (write==0) */
840 22913400 : ulong txnpage_idx;
841 22913400 : fd_txncache_txnpage_t * txnpage = fd_txncache_ensure_txnpage( tc, blockcache, 0, &txnpage_idx );
842 22913400 : if( FD_UNLIKELY( !txnpage && txnpage_idx!=ULONG_MAX ) ) miss = 1;
843 22913400 : if( FD_UNLIKELY( !txnpage && txnpage_idx!=ULONG_MAX && txnpage_idx!=fetched_idx ) ) {
844 : /* The page is not resident. Load it without blocking others. */
845 27783 : ulong seq = tc->shmem->release_seq;
846 27783 : fd_rwlock_unread( tc->shmem->lock );
847 27783 : page_fetch( tc, txnpage_idx, seq );
848 27783 : fetched_idx = txnpage_idx;
849 27783 : lock_read( tc );
850 27783 : blockcache = blockhash_on_fork( tc, fork, blockhash );
851 27783 : FD_TEST( blockcache );
852 27783 : continue;
853 27783 : }
854 22885617 : 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 4626 : fd_rwlock_unread( tc->shmem->lock );
867 4626 : lock_write( tc );
868 4626 : blockcache = blockhash_on_fork( tc, fork, blockhash );
869 4626 : FD_TEST( blockcache );
870 : /* Load or allocate a frame under exclusive access. */
871 4626 : txnpage = fd_txncache_ensure_txnpage( tc, blockcache, 1, &txnpage_idx );
872 4626 : 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 1560 : purge_stale( tc );
877 1560 : txnpage = fd_txncache_ensure_txnpage( tc, blockcache, 1, &txnpage_idx );
878 1560 : 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 1560 : blockcache->shmem->pages_cnt, tc->shmem->txnpages_per_blockhash_max,
881 1560 : tc->shmem->txnpages_free_cnt, tc->shmem->txnpages_max,
882 1560 : tc->shmem->active_slots_max, tc->shmem->txn_per_slot_max ));
883 1560 : }
884 4626 : FD_TEST( fd_txncache_insert_txn( tc, blockcache, txnpage, txnpage_idx, fork_id, txnhash ) );
885 4626 : write = 1;
886 4626 : break;
887 4626 : }
888 :
889 : /* Fast path: insert the transaction into an in-memory page. */
890 22880991 : int success = fd_txncache_insert_txn( tc, blockcache, txnpage, txnpage_idx, fork_id, txnhash );
891 22880991 : if( FD_LIKELY( success ) ) break;
892 :
893 0 : FD_SPIN_PAUSE();
894 0 : }
895 :
896 22885617 : if( FD_UNLIKELY( write ) ) fd_rwlock_unwrite( tc->shmem->lock );
897 22880991 : else fd_rwlock_unread ( tc->shmem->lock );
898 :
899 22885617 : tc->metrics->insert_miss_cnt += (ulong)miss;
900 22885617 : tc->metrics->stalled_cnt += (ulong)write;
901 22885617 : fd_histf_sample( tc->metrics->insert_ticks, (ulong)(fd_tickcount()-t0) );
902 22885617 : }
903 :
904 : int
905 : fd_txncache_query( fd_txncache_t * tc,
906 : fd_txncache_fork_id_t fork_id,
907 : uchar const * blockhash,
908 5681517 : uchar const * txnhash ) {
909 5681517 : long t0 = fd_tickcount();
910 : /* Prevent fork removal and pruning while allowing inserts. */
911 5681517 : lock_read( tc );
912 :
913 : /* Find the blockhash cache visible from the requested fork. */
914 5681517 : blockcache_t const * fork = &tc->blockcache_pool[ fork_id.val ];
915 5681517 : FD_TEST( fork->shmem->frozen>=0 );
916 5681517 : blockcache_t const * blockcache = blockhash_on_fork( tc, fork, blockhash );
917 5681517 : FD_TEST( blockcache );
918 5681517 : FD_TEST( blockcache->shmem->frozen==2 );
919 :
920 5681517 : int found = 0;
921 5681517 : int write = 0;
922 5681517 : int miss = 0;
923 5681517 : ulong fetch_cnt = 0UL;
924 :
925 : /* Select the bucket using the stored portion of the txn hash. */
926 5681517 : ulong txnhash_offset = blockcache->shmem->txnhash_offset;
927 5681517 : ulong head_hash = fd_txncache_bucket( tc, txnhash+txnhash_offset );
928 : /* Walk the bucket's list. UINT_MAX marks the end. */
929 157977294 : 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 157975536 : fd_txncache_txnpage_t * page = page_io( tc, head/FD_TXNCACHE_TXNS_PER_PAGE, write );
937 157975536 : if( FD_UNLIKELY( !page ) ) {
938 3030 : miss = 1;
939 3030 : ulong seq = tc->shmem->release_seq;
940 3030 : int fetch = fetch_cnt<fd_ulong_min( blockcache->shmem->pages_cnt, tc->shmem->frames_max );
941 3030 : fd_rwlock_unread( tc->shmem->lock );
942 3030 : if( FD_LIKELY( fetch ) ) {
943 : /* Optimistically do I/O without the lock and check again. */
944 1545 : page_fetch( tc, head/FD_TXNCACHE_TXNS_PER_PAGE, seq );
945 1545 : fetch_cnt++;
946 1545 : lock_read( tc );
947 1545 : } else {
948 : /* Acquire write lock to load the page in place. */
949 1485 : lock_write( tc );
950 1485 : write = 1;
951 1485 : }
952 3030 : blockcache = blockhash_on_fork( tc, fork, blockhash );
953 3030 : FD_TEST( blockcache );
954 3030 : head = blockcache->heads[ head_hash ];
955 3030 : continue;
956 3030 : }
957 157972506 : fd_txncache_single_txn_t const * txn = page->txns[ head%FD_TXNCACHE_TXNS_PER_PAGE ];
958 157972506 : 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 157972506 : blockcache_t const * txn_fork = &tc->blockcache_pool[ txn->fork_id.val ];
963 157972506 : 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 157972506 : if( FD_LIKELY( descends && !memcmp( txnhash+txnhash_offset, txn->txnhash, 20UL ) ) ) {
966 5679759 : found = 1;
967 5679759 : break;
968 5679759 : }
969 157972506 : }
970 :
971 5681517 : if( FD_UNLIKELY( write ) ) fd_rwlock_unwrite( tc->shmem->lock );
972 5680032 : else fd_rwlock_unread ( tc->shmem->lock );
973 :
974 5681517 : tc->metrics->query_miss_cnt += (ulong)miss;
975 5681517 : tc->metrics->stalled_cnt += (ulong)write;
976 5681517 : fd_histf_sample( tc->metrics->query_ticks, (ulong)(fd_tickcount()-t0) );
977 5681517 : return found;
978 5681517 : }
979 :
980 : ulong
981 : fd_txncache_snapshot_roots( fd_txncache_t * tc,
982 90599 : fd_txncache_root_t * out ) {
983 90599 : lock_read( tc );
984 90599 : ulong cnt = 0UL;
985 90599 : for( ulong it = root_slist_iter_init( tc->shmem->root_ll, tc->blockcache_shmem_pool );
986 280023 : !root_slist_iter_done( it, tc->shmem->root_ll, tc->blockcache_shmem_pool );
987 189424 : it = root_slist_iter_next( it, tc->shmem->root_ll, tc->blockcache_shmem_pool ) ) {
988 189424 : ulong blockcache_idx = root_slist_iter_idx( it, tc->shmem->root_ll, tc->blockcache_shmem_pool );
989 189424 : fd_txncache_blockcache_shmem_t const * blockcache_shmem = &tc->blockcache_shmem_pool[ blockcache_idx ];
990 189424 : FD_TEST( cnt<FD_TXNCACHE_MAX_ROOTS );
991 189424 : fd_txncache_root_t * root = &out[ cnt++ ];
992 189424 : root->blockcache_idx = blockcache_idx;
993 189424 : root->txnhash_offset = blockcache_shmem->txnhash_offset;
994 189424 : root->generation = blockcache_shmem->generation;
995 189424 : memcpy( root->blockhash, blockcache_shmem->blockhash.uc, 32UL );
996 189424 : }
997 90599 : fd_rwlock_unread( tc->shmem->lock );
998 90599 : return cnt;
999 90599 : }
1000 :
1001 : static fd_txncache_single_txn_t const *
1002 : bucket_chain_txn( fd_txncache_t * tc,
1003 : uint txn_gidx,
1004 1353732749 : int write_lock ) {
1005 1353732749 : if( FD_UNLIKELY( txn_gidx==UINT_MAX ) ) return NULL;
1006 1350075791 : fd_txncache_txnpage_t const * page = page_io( tc, txn_gidx/FD_TXNCACHE_TXNS_PER_PAGE, write_lock );
1007 1350075791 : FD_TEST( page );
1008 1350075791 : return page->txns[ txn_gidx%FD_TXNCACHE_TXNS_PER_PAGE ];
1009 1350075791 : }
1010 :
1011 : /* Locks tc and positions iter at the head of its current bucket. */
1012 :
1013 : static void
1014 3656958 : blockcache_iter_bucket_open( fd_txncache_blockcache_iter_t * iter ) {
1015 3656958 : fd_txncache_t * tc = iter->tc;
1016 3656958 : fd_rwlock_t * lock = tc->shmem->lock;
1017 3656958 : uint const * heads = tc->blockcache_pool[ iter->blockcache_idx ].heads;
1018 3656958 : ulong bucket = iter->bucket;
1019 :
1020 3656958 : fd_racesan_hook( "txncache:bucket_started" );
1021 3656958 : lock_read( tc );
1022 :
1023 3656958 : uint head;
1024 3656958 : int write_lock = 0;
1025 3660788 : 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 3660788 : head = FD_VOLATILE_CONST( heads[ bucket ] );
1029 :
1030 : /* Check that every page of the chain is resident. */
1031 3660788 : ulong miss = ULONG_MAX;
1032 1353684211 : for( uint idx=head; idx!=UINT_MAX; ) {
1033 1350030327 : fd_txncache_txnpage_t const * page = page_io( tc, idx/FD_TXNCACHE_TXNS_PER_PAGE, 0 );
1034 1350030327 : if( FD_UNLIKELY( !page ) ) {
1035 6904 : miss = idx/FD_TXNCACHE_TXNS_PER_PAGE;
1036 6904 : break;
1037 6904 : }
1038 1350023423 : idx = page->txns[ idx%FD_TXNCACHE_TXNS_PER_PAGE ]->blockcache_next;
1039 1350023423 : }
1040 3660788 : 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 6904 : ulong seq = tc->shmem->release_seq;
1047 6904 : int fetch = fetch_cnt<fd_ulong_min( tc->blockcache_pool[ iter->blockcache_idx ].shmem->pages_cnt, tc->shmem->frames_max );
1048 6904 : fd_rwlock_unread( lock );
1049 6904 : if( !fetch_cnt ) { fd_racesan_hook( "txncache:bucket_iter_miss" ); }
1050 6904 : if( FD_LIKELY( fetch ) ) {
1051 3830 : page_fetch( tc, miss, seq );
1052 3830 : lock_read( tc );
1053 3830 : continue;
1054 3830 : }
1055 3074 : lock_write( tc );
1056 3074 : write_lock = 1;
1057 3074 : head = heads[ bucket ];
1058 3074 : break;
1059 6904 : }
1060 :
1061 3656958 : iter->write_lock = write_lock;
1062 3656958 : iter->txn = bucket_chain_txn( tc, head, write_lock );
1063 3656958 : }
1064 :
1065 : static void
1066 3656958 : blockcache_iter_bucket_close( fd_txncache_blockcache_iter_t * iter ) {
1067 3656958 : if( iter->write_lock ) fd_rwlock_unwrite( iter->tc->shmem->lock );
1068 3653884 : else fd_rwlock_unread ( iter->tc->shmem->lock );
1069 3656958 : }
1070 :
1071 : /* Moves past empty buckets, holding the lock of one bucket at a time. */
1072 :
1073 : static void
1074 1350363176 : blockcache_iter_skip_empty( fd_txncache_blockcache_iter_t * iter ) {
1075 1353732749 : while( !iter->txn && iter->bucket+1UL<iter->tc->shmem->bucket_cnt ) {
1076 3369573 : blockcache_iter_bucket_close( iter );
1077 3369573 : iter->bucket++;
1078 3369573 : blockcache_iter_bucket_open( iter );
1079 3369573 : }
1080 1350363176 : }
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 287385 : ulong blockcache_idx ) {
1086 287385 : iter->tc = tc;
1087 287385 : iter->blockcache_idx = blockcache_idx;
1088 287385 : iter->bucket = 0UL;
1089 287385 : blockcache_iter_bucket_open( iter );
1090 287385 : blockcache_iter_skip_empty( iter );
1091 287385 : return iter;
1092 287385 : }
1093 :
1094 : void
1095 1350075791 : 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 1350075791 : uint next = iter->txn->blockcache_next;
1099 1350075791 : iter->txn = bucket_chain_txn( iter->tc, next, iter->write_lock );
1100 1350075791 : blockcache_iter_skip_empty( iter );
1101 1350075791 : }
1102 :
1103 : void
1104 287385 : fd_txncache_blockcache_iter_fini( fd_txncache_blockcache_iter_t * iter ) {
1105 287385 : blockcache_iter_bucket_close( iter );
1106 287385 : }
|
