Line data Source code
1 : #define _GNU_SOURCE
2 : #include "fd_accdb.h"
3 : #include "fd_accdb_shmem.h"
4 : #include "fd_accdb_private.h"
5 : #include "../../util/fd_hash32.h"
6 :
7 : #if FD_TMPL_USE_HANDHOLDING
8 : #include "../../ballet/txn/fd_txn.h"
9 : #include "../../ballet/base58/fd_base58.h"
10 : #endif
11 : #include "../../util/racesan/fd_racesan_target.h"
12 :
13 : #include "../../disco/events/generated/fd_event_gen.h"
14 : #include "../../disco/sleep/fd_sleep.h"
15 : #include "../runtime/fd_runtime_const.h" /* FD_RUNTIME_ACC_SZ_MAX */
16 :
17 : FD_STATIC_ASSERT( sizeof(fd_accdb_cache_line_t)==FD_ACCDB_CACHE_META_SZ, cache_meta_sz );
18 :
19 0 : #define FD_ACCDB_BOUNCE_SZ (16UL<<20)
20 : FD_STATIC_ASSERT( FD_ACCDB_BOUNCE_SZ>=FD_RUNTIME_ACC_SZ_MAX+sizeof(fd_accdb_disk_meta_t), bounce_fits_any_record );
21 :
22 : /* Preevict batches up to PREEVICT_BATCH_SZ bytes, but stages a larger
23 : record alone, so its buffer holds one record of any size. */
24 0 : #define PREEVICT_BATCH_SZ (256UL<<10)
25 : #define PREEVICT_STAGE_SZ (FD_RUNTIME_ACC_SZ_MAX+sizeof(fd_accdb_disk_meta_t))
26 : FD_STATIC_ASSERT( PREEVICT_BATCH_SZ<=PREEVICT_STAGE_SZ, batch_fits_stage );
27 :
28 : #if FD_HAS_RACESAN
29 : /* Test-only telemetry: background_compact publishes the pubkey + dest
30 : offset of the record it is about to relocation-CAS at the
31 : accdb_compact:pre_offset_cas hook, so test_accdb_racesan can PROVE the
32 : parked relocation is the account it set up (avoiding a vacuous test).
33 : Zero-cost / absent in production (racesan off). */
34 : uchar fd_accdb_dbg_reloc_pubkey[ 32UL ];
35 : ulong fd_accdb_dbg_reloc_dest;
36 : ulong fd_accdb_dbg_reloc_cnt;
37 : ulong fd_accdb_dbg_compact_scanned; /* bytes compaction stepped over */
38 : #endif
39 :
40 : #include <stddef.h>
41 : #include <unistd.h>
42 : #include <fcntl.h>
43 : #include <errno.h>
44 : #include <sys/uio.h>
45 :
46 : struct fd_accdb_fork {
47 : fd_accdb_fork_shmem_t * shmem;
48 : descends_set_t * descends;
49 : };
50 :
51 : typedef struct fd_accdb_fork fd_accdb_fork_t;
52 :
53 487864 : #define FD_ACCDB_ACQUIRE_STATE_IDLE (0)
54 7290 : #define FD_ACCDB_ACQUIRE_STATE_PHASE_A (1)
55 488032 : #define FD_ACCDB_ACQUIRE_STATE_OPEN (2)
56 :
57 : struct __attribute__((aligned(FD_ACCDB_ALIGN))) fd_accdb_private {
58 : int fd;
59 :
60 : int acquire_state;
61 :
62 : fd_accdb_shmem_t * shmem;
63 :
64 : /* Doorbell for the parked accdb tile, NULL when nobody parks */
65 : fd_sleep_t * sleep;
66 : ulong sleep_tile_id;
67 :
68 : fd_accdb_fork_t * fork_pool;
69 : fork_pool_t fork_shmem_pool[1];
70 :
71 : fd_accdb_accmeta_t * acc_pool;
72 : acc_pool_t acc_pool_join[1];
73 : uint * acc_map;
74 :
75 : uchar * cache [ FD_ACCDB_CACHE_CLASS_CNT ];
76 :
77 : fd_accdb_partition_t * partition_pool;
78 : compaction_dlist_t * compaction_dlist[ FD_ACCDB_COMPACTION_LAYER_CNT ];
79 : deferred_free_dlist_t * deferred_free_dlist;
80 :
81 : txn_pool_t txn_pool[1];
82 :
83 : /* Pointer into shmem->joiner_epochs[ my_slot ].val for writer
84 : joiners, or into a private per-tile fseq for read-only joiners.
85 : Set to the current global epoch on entry to an epoch-protected
86 : operation, and ULONG_MAX on exit. Used to determine when
87 : deferred frees are safe. */
88 : ulong * my_epoch_slot;
89 :
90 : /* Read-only pointers to external epoch slots (e.g. fseqs owned by
91 : RO consumer tiles like the rpc tile). Scanned in addition to
92 : shmem->joiner_epochs[] by compaction's deferred-free
93 : reclamation. Borrowed; the caller of fd_accdb_new owns the
94 : storage. */
95 : ulong const * const * external_epoch_slots;
96 : ulong external_epoch_cnt;
97 :
98 : /* Side buffer of acc pool indices that have been CAS-unlinked from
99 : their hash chains but cannot be released back to acc_pool yet,
100 : because concurrent readers (acquire / compact) may still be
101 : traversing the removed nodes via map.next. The batch is released
102 : once all joiner_epochs exceed shmem->deferred_acc_epoch. Indices
103 : are written here (not into pool.next) until after the epoch drain
104 : because pool.next is union-aliased to cache_idx, which a concurrent
105 : cold_load_acc may still write through a captured pointer. Backed
106 : by shmem->deferred_acc_buf_off; cnt and epoch live in shmem too. */
107 : uint * deferred_acc_buf;
108 :
109 : /* Chain of fork pool slots whose IDs are still potentially
110 : referenced by concurrent readers (via descends_set_test or
111 : root_fork_id snapshot). The chain is released back to fork_pool
112 : once all joiner_epochs exceed deferred_fork_epoch. NULL head
113 : means no deferred forks. */
114 : fd_accdb_fork_shmem_t * deferred_fork_head;
115 : fd_accdb_fork_shmem_t * deferred_fork_tail;
116 : ulong deferred_fork_epoch;
117 :
118 : fd_accdb_metrics_t metrics[1];
119 :
120 : ulong prealloc_failed_at;
121 :
122 : /* Compaction bounce buffer. */
123 : uchar * bounce;
124 : ulong carry_off;
125 : ulong carry_sz;
126 : ulong carry_layer;
127 :
128 : /* Preevict write staging, PREEVICT_STAGE_SZ bytes. */
129 : uchar * stage;
130 :
131 : /* Track account addresses changed since a full snap.
132 : Used to determine which accounts should be packed into a full
133 : snapshot (including tombstones for accounts no longer present in
134 : accdb, but present in the full snapshot). */
135 : struct {
136 : uint * chains;
137 : fd_accdb_delta_t * pool;
138 : struct {
139 : uchar const * pubkey;
140 : uint chain;
141 : } scratch[ FD_ACCDB_MAX_ACQUIRE_CNT ];
142 : } delta;
143 :
144 : /* Batch metric updates to avoid shared atomics per write. */
145 : struct {
146 : ulong bytes; /* bytes reserved on partition_idx */
147 : ulong num_ops; /* reservations behind those bytes */
148 : ulong partition_idx; /* set while num_ops>0 */
149 : ulong disk_used_added;
150 : ulong disk_used_removed;
151 : ulong accounts_total_added;
152 : } write_stats __attribute__((aligned(64)));
153 : };
154 :
155 : static inline fd_accdb_cache_line_t *
156 : cache_line( fd_accdb_t * accdb,
157 : ulong cls,
158 8118747 : ulong idx ) {
159 8118747 : return (fd_accdb_cache_line_t *)( accdb->cache[ cls ] + idx * fd_accdb_cache_slot_sz[ cls ] );
160 8118747 : }
161 :
162 : /* Bump the per-partition read counters for the partition that contains
163 : file_offset. Called at preadv2 sites. Writes are counted at
164 : allocate time (see fd_accdb_partition_write_bump) so that they reflect
165 : bytes committed to a partition rather than syscalls — the snapshot
166 : loader bypasses pwritev2 entirely, but every write still goes through
167 : reserve_next_write. */
168 : static inline void
169 : fd_accdb_partition_read_bump( fd_accdb_t * accdb,
170 : ulong file_offset,
171 0 : ulong bytes ) {
172 0 : if( FD_UNLIKELY( !bytes ) ) return;
173 : /* Readonly joiners have no partition_pool join (see
174 : fd_accdb_join_readonly) and do not contribute to per-partition
175 : read telemetry today; their disk reads still show up in the
176 : joiner-local fd_accdb_metrics_t bytes_read/read_ops. */
177 0 : if( FD_UNLIKELY( !accdb->partition_pool ) ) return;
178 0 : ulong partition_idx = file_offset / accdb->shmem->partition_sz;
179 0 : ulong partition_cnt = partition_pool_max( accdb->partition_pool );
180 0 : FD_CHECK_CRIT( partition_idx<partition_cnt, "read partition index out of range" );
181 0 : fd_accdb_partition_t * p = partition_pool_ele( accdb->partition_pool, partition_idx );
182 0 : FD_ATOMIC_FETCH_AND_ADD( &p->bytes_read, bytes );
183 0 : FD_ATOMIC_FETCH_AND_ADD( &p->read_ops, 1UL );
184 0 : }
185 :
186 : /* Bump the per-partition write counters. bytes is how much landed on
187 : this partition, over num_ops reservations. */
188 : static inline void
189 : fd_accdb_partition_write_bump( fd_accdb_t * accdb,
190 : ulong partition_idx,
191 : ulong bytes,
192 0 : ulong num_ops ) {
193 0 : if( FD_UNLIKELY( !bytes ) ) return;
194 0 : ulong partition_cnt = partition_pool_max( accdb->partition_pool );
195 0 : FD_CHECK_CRIT( partition_idx<partition_cnt, "write partition index out of range" );
196 0 : fd_accdb_partition_t * p = partition_pool_ele( accdb->partition_pool, partition_idx );
197 0 : FD_ATOMIC_FETCH_AND_ADD( &p->bytes_written, bytes );
198 0 : FD_ATOMIC_FETCH_AND_ADD( &p->write_ops, num_ops );
199 0 : }
200 :
201 : void
202 0 : fd_accdb_flush_metrics( fd_accdb_t * accdb ) {
203 0 : ulong bytes = accdb->write_stats.bytes;
204 0 : ulong num_ops = accdb->write_stats.num_ops;
205 0 : ulong part_idx = accdb->write_stats.partition_idx;
206 0 : ulong disk_used_added = accdb->write_stats.disk_used_added;
207 0 : ulong disk_used_removed = accdb->write_stats.disk_used_removed;
208 0 : ulong accounts_total_added = accdb->write_stats.accounts_total_added;
209 :
210 0 : memset( &accdb->write_stats, 0, sizeof(accdb->write_stats) );
211 :
212 0 : if( num_ops ) {
213 0 : FD_ATOMIC_FETCH_AND_ADD( &accdb->shmem->shmetrics->disk_current_bytes, bytes );
214 0 : fd_accdb_partition_write_bump( accdb, part_idx, bytes, num_ops );
215 0 : }
216 :
217 0 : if( disk_used_added ) FD_ATOMIC_FETCH_AND_ADD( &accdb->shmem->shmetrics->disk_used_bytes, disk_used_added );
218 0 : if( disk_used_removed ) FD_ATOMIC_FETCH_AND_SUB( &accdb->shmem->shmetrics->disk_used_bytes, disk_used_removed );
219 0 : if( accounts_total_added ) FD_ATOMIC_FETCH_AND_ADD( &accdb->shmem->shmetrics->accounts_total, accounts_total_added );
220 0 : }
221 :
222 : static inline ulong
223 : cache_line_idx( fd_accdb_t * accdb,
224 : ulong cls,
225 6923059 : fd_accdb_cache_line_t const * line ) {
226 6923059 : return (ulong)( (uchar const *)line - accdb->cache[ cls ] ) / fd_accdb_cache_slot_sz[ cls ];
227 6923059 : }
228 :
229 : #if FD_TMPL_USE_HANDHOLDING
230 : static inline int
231 : fd_accdb_ptr_in_region( fd_accdb_t const * accdb,
232 : ulong cls,
233 : void const * ptr ) {
234 : if( FD_UNLIKELY( cls>=FD_ACCDB_CACHE_CLASS_CNT ) ) return 0;
235 :
236 : uchar const * base = accdb->cache[ cls ];
237 : if( FD_UNLIKELY( !base ) ) return 0;
238 :
239 : ulong slot_sz = fd_accdb_cache_slot_sz[ cls ];
240 : ulong region_sz = accdb->shmem->cache_class_max[ cls ] * slot_sz;
241 : uchar const * p = (uchar const *)ptr;
242 :
243 : if( FD_UNLIKELY( p<base || p>=base+region_sz ) ) return 0;
244 : return ( (ulong)( p - base ) % slot_sz )==FD_ACCDB_CACHE_META_SZ;
245 : }
246 : #endif
247 :
248 : FD_FN_CONST ulong
249 48 : fd_accdb_align( void ) {
250 48 : return FD_ACCDB_ALIGN;
251 48 : }
252 :
253 : FD_FN_CONST ulong
254 : fd_accdb_footprint( ulong max_live_slots,
255 12 : int compaction ) {
256 12 : ulong l;
257 12 : l = FD_LAYOUT_INIT;
258 12 : l = FD_LAYOUT_APPEND( l, FD_ACCDB_ALIGN, sizeof(fd_accdb_t) );
259 12 : l = FD_LAYOUT_APPEND( l, alignof(fd_accdb_fork_t), max_live_slots*sizeof(fd_accdb_fork_t) );
260 12 : if( compaction ) l = FD_LAYOUT_APPEND( l, FD_ACCDB_ALIGN, FD_ACCDB_BOUNCE_SZ+PREEVICT_STAGE_SZ );
261 12 : return FD_LAYOUT_FINI( l, FD_ACCDB_ALIGN );
262 12 : }
263 :
264 : void *
265 : fd_accdb_new( void * ljoin,
266 : fd_accdb_shmem_t * shmem,
267 : int fd,
268 : ulong external_epoch_cnt,
269 : ulong const ** external_epoch_slots,
270 : fd_sleep_t * sleep,
271 : ulong sleep_tile_id,
272 12 : int compaction ) {
273 12 : if( FD_UNLIKELY( !ljoin ) ) {
274 0 : FD_LOG_WARNING(( "NULL ljoin" ));
275 0 : return NULL;
276 0 : }
277 :
278 12 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)ljoin, fd_accdb_align() ) ) ) {
279 0 : FD_LOG_WARNING(( "misaligned ljoin" ));
280 0 : return NULL;
281 0 : }
282 :
283 12 : if( FD_UNLIKELY( fd<0 ) ) {
284 0 : FD_LOG_WARNING(( "fd must be a valid file descriptor" ));
285 0 : return NULL;
286 0 : }
287 :
288 12 : ulong max_live_slots = shmem->max_live_slots;
289 12 : ulong max_accounts = shmem->max_accounts;
290 12 : ulong max_account_writes_per_slot = shmem->max_account_writes_per_slot;
291 12 : ulong partition_cnt = shmem->partition_cnt;
292 :
293 12 : ulong chain_cnt = fd_ulong_pow2_up( (max_accounts>>1) + (max_accounts&1UL) );
294 12 : ulong txn_max = max_live_slots * max_account_writes_per_slot;
295 :
296 12 : FD_SCRATCH_ALLOC_INIT( l, shmem );
297 12 : FD_SCRATCH_ALLOC_APPEND( l, FD_ACCDB_SHMEM_ALIGN, sizeof(fd_accdb_shmem_t) );
298 12 : void * _fork_pool_ele = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_accdb_fork_shmem_t), max_live_slots*sizeof(fd_accdb_fork_shmem_t) );
299 12 : void * _descends_sets = FD_SCRATCH_ALLOC_APPEND( l, descends_set_align(), max_live_slots*descends_set_footprint( max_live_slots ) );
300 12 : void * _acc_map = FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), chain_cnt*sizeof(uint) );
301 12 : void * _acc_pool_ele = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_accdb_accmeta_t), max_accounts*sizeof(fd_accdb_accmeta_t) );
302 12 : void * _txn_pool_ele = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_accdb_txn_t), txn_max*sizeof(fd_accdb_txn_t) );
303 12 : void * _partition_pool = FD_SCRATCH_ALLOC_APPEND( l, partition_pool_align(), partition_pool_footprint( partition_cnt ) );
304 0 : void * _compaction_dlists[ FD_ACCDB_COMPACTION_LAYER_CNT ];
305 48 : for( ulong k=0UL; k<FD_ACCDB_COMPACTION_LAYER_CNT; k++ ) {
306 36 : _compaction_dlists[ k ] = FD_SCRATCH_ALLOC_APPEND( l, compaction_dlist_align(), compaction_dlist_footprint() );
307 36 : }
308 12 : void * _deferred_free_dlist = FD_SCRATCH_ALLOC_APPEND( l, deferred_free_dlist_align(), deferred_free_dlist_footprint() );
309 :
310 12 : FD_SCRATCH_ALLOC_INIT( l2, ljoin );
311 12 : fd_accdb_t * accdb = FD_SCRATCH_ALLOC_APPEND( l2, fd_accdb_align(), sizeof(fd_accdb_t) );
312 12 : void * _local_fork_pool = FD_SCRATCH_ALLOC_APPEND( l2, alignof(fd_accdb_fork_t), max_live_slots*sizeof(fd_accdb_fork_t) );
313 12 : uchar * _bounce = compaction ? FD_SCRATCH_ALLOC_APPEND( l2, FD_ACCDB_ALIGN, FD_ACCDB_BOUNCE_SZ+PREEVICT_STAGE_SZ ) : NULL;
314 :
315 0 : accdb->fd = fd;
316 12 : accdb->acquire_state = FD_ACCDB_ACQUIRE_STATE_IDLE;
317 12 : accdb->sleep = sleep;
318 12 : accdb->sleep_tile_id = sleep_tile_id;
319 :
320 12 : accdb->shmem = (fd_accdb_shmem_t *)shmem;
321 12 : FD_TEST( acc_pool_join( accdb->acc_pool_join, shmem->acc_pool, _acc_pool_ele, max_accounts ) );
322 12 : accdb->acc_pool = accdb->acc_pool_join->ele;
323 12 : accdb->acc_map = _acc_map;
324 12 : FD_TEST( txn_pool_join( accdb->txn_pool, shmem->txn_pool, _txn_pool_ele, txn_max ) );
325 108 : for( ulong c=0UL; c<FD_ACCDB_CACHE_CLASS_CNT; c++ ) accdb->cache[ c ] = (uchar *)shmem + shmem->cache_region_off[ c ];
326 12 : accdb->partition_pool = partition_pool_join( _partition_pool );
327 12 : FD_TEST( accdb->partition_pool );
328 48 : for( ulong k=0UL; k<FD_ACCDB_COMPACTION_LAYER_CNT; k++ ) {
329 36 : accdb->compaction_dlist[ k ] = compaction_dlist_join( _compaction_dlists[ k ] );
330 36 : FD_TEST( accdb->compaction_dlist[ k ] );
331 36 : }
332 12 : accdb->deferred_free_dlist = deferred_free_dlist_join( _deferred_free_dlist );
333 12 : FD_TEST( accdb->deferred_free_dlist );
334 :
335 12 : FD_TEST( fork_pool_join( accdb->fork_shmem_pool, shmem->fork_pool, _fork_pool_ele, max_live_slots ) );
336 12 : accdb->fork_pool = _local_fork_pool;
337 204 : for( ulong i=0UL; i<max_live_slots; i++ ) {
338 192 : fd_accdb_fork_t * fork = &accdb->fork_pool[ i ];
339 192 : fork->shmem = fork_pool_ele( accdb->fork_shmem_pool, i );
340 192 : fork->descends = descends_set_join( (uchar *)_descends_sets + i*descends_set_footprint( max_live_slots ) );
341 192 : FD_TEST( fork->shmem );
342 192 : FD_TEST( fork->descends );
343 192 : }
344 :
345 12 : ulong epoch_idx = FD_ATOMIC_FETCH_AND_ADD( &shmem->joiner_cnt, 1UL );
346 12 : FD_TEST( epoch_idx<shmem->joiner_cnt_max );
347 12 : accdb->my_epoch_slot = &shmem->joiner_epochs[ epoch_idx ].val;
348 :
349 12 : accdb->external_epoch_slots = external_epoch_slots;
350 12 : accdb->external_epoch_cnt = external_epoch_cnt;
351 :
352 12 : accdb->bounce = _bounce;
353 12 : accdb->carry_off = 0UL;
354 12 : accdb->carry_sz = 0UL;
355 12 : accdb->carry_layer = 0UL;
356 12 : accdb->stage = _bounce ? _bounce+FD_ACCDB_BOUNCE_SZ : NULL;
357 :
358 12 : accdb->deferred_acc_buf = (uint *)( (uchar *)shmem + shmem->deferred_acc_buf_off );
359 :
360 12 : accdb->delta.chains = (uint *) ( (uchar *)shmem + shmem->delta.chain_off );
361 12 : accdb->delta.pool = (fd_accdb_delta_t *) ( (uchar *)shmem + shmem->delta.ele_off );
362 :
363 12 : accdb->deferred_fork_head = NULL;
364 12 : accdb->deferred_fork_tail = NULL;
365 12 : accdb->deferred_fork_epoch = 0UL;
366 12 : accdb->prealloc_failed_at = ULONG_MAX;
367 :
368 12 : memset( accdb->metrics, 0, sizeof(fd_accdb_metrics_t) );
369 12 : memset( &accdb->write_stats, 0, sizeof(accdb->write_stats) );
370 :
371 12 : return accdb;
372 12 : }
373 :
374 : static inline void wait_cmd( fd_accdb_t * accdb );
375 : static inline void submit_cmd( fd_accdb_t * accdb, uint op, ushort fork_id );
376 :
377 : void
378 0 : fd_accdb_reset( fd_accdb_t * accdb ) {
379 0 : fd_accdb_shmem_t * shmem = accdb->shmem;
380 :
381 : /* Wait for any pending background command (advance_root / purge) on
382 : T2 to finish before clobbering shared state. */
383 0 : wait_cmd( accdb );
384 :
385 : /* Reset pools through the joiner's existing pointers. acc_pool and
386 : txn_pool use POOL_LAZY=1 so reset is O(1). fork_pool and
387 : partition_pool rebuild their free lists in O(max_live_slots) and
388 : O(partition_cnt), both small. */
389 0 : spin_lock_acquire( &shmem->grow_lock );
390 0 : acc_pool_reset( accdb->acc_pool_join );
391 0 : txn_pool_reset( accdb->txn_pool );
392 0 : fork_pool_reset( accdb->fork_shmem_pool );
393 0 : partition_pool_reset( accdb->partition_pool );
394 :
395 : /* Clear hash chains */
396 0 : fd_memset( accdb->acc_map, 0xFF, shmem->chain_cnt*sizeof(uint) );
397 :
398 : /* Empty dlists */
399 0 : for( ulong k=0UL; k<FD_ACCDB_COMPACTION_LAYER_CNT; k++ ) {
400 0 : compaction_dlist_remove_all( accdb->compaction_dlist[ k ], accdb->partition_pool );
401 0 : }
402 0 : deferred_free_dlist_remove_all( accdb->deferred_free_dlist, accdb->partition_pool );
403 :
404 : /* Null descends_sets. */
405 0 : for( ulong i=0UL; i<shmem->max_live_slots; i++ ) {
406 0 : descends_set_null( accdb->fork_pool[ i ].descends );
407 0 : }
408 :
409 : /* Reset shmem scalar fields. */
410 0 : shmem->root_fork_id = (fd_accdb_fork_id_t){ .val = USHORT_MAX };
411 0 : shmem->generation = 0U;
412 0 : shmem->partition_lock = 0;
413 0 : shmem->partition_max = 0UL;
414 0 : shmem->partition_fallocated = 0UL;
415 0 : spin_lock_release( &shmem->grow_lock );
416 :
417 : /* Write heads: sentinel values that force partition-switch on first
418 : write. */
419 0 : for( ulong k=0UL; k<FD_ACCDB_COMPACTION_LAYER_CNT; k++ ) {
420 0 : shmem->whead[ k ] = accdb_offset( shmem->partition_cnt, shmem->partition_sz );
421 0 : shmem->has_partition[ k ] = 0;
422 0 : }
423 :
424 : /* Cache state */
425 0 : for( ulong c=0UL; c<FD_ACCDB_CACHE_CLASS_CNT; c++ ) {
426 0 : shmem->clock_hand[ c ].val = 0UL;
427 0 : shmem->cache_free[ c ].ver_top = (ulong)UINT_MAX;
428 0 : shmem->cache_free_cnt[ c ].val = 0UL;
429 0 : shmem->cache_class_init[ c ].val = 0UL;
430 0 : if( shmem->cache_class_max[ c ]>=shmem->cache_min_reserved*shmem->joiner_cnt_max )
431 0 : shmem->cache_class_used[ c ].val = ULONG_MAX;
432 0 : else
433 0 : shmem->cache_class_used[ c ].val = 0UL;
434 0 : }
435 :
436 : /* Reset every cache slot's metadata to empty sentinels. */
437 0 : for( ulong c=0UL; c<FD_ACCDB_CACHE_CLASS_CNT; c++ ) {
438 0 : ulong slot_sz = fd_accdb_cache_slot_sz[ c ];
439 0 : for( ulong i=0UL; i<shmem->cache_class_max[ c ]; i++ ) {
440 0 : fd_accdb_cache_line_t * line = (fd_accdb_cache_line_t *)( accdb->cache[ c ] + i*slot_sz );
441 0 : line->key.generation = UINT_MAX;
442 0 : line->acc_idx = UINT_MAX;
443 0 : line->refcnt = 0U;
444 0 : line->referenced = 0;
445 0 : line->persisted = 1;
446 0 : line->seq = 0U;
447 0 : }
448 0 : }
449 :
450 : /* Epoch system: reset epoch and all slot values to idle, but
451 : preserve joiner_cnt and each tile's my_epoch_slot pointer so that
452 : tiles which joined during init keep their original slot indices. */
453 0 : shmem->epoch = 1UL;
454 0 : for( ulong i=0UL; i<FD_ACCDB_MAX_JOINERS; i++ ) shmem->joiner_epochs[ i ].val = ULONG_MAX;
455 :
456 : /* Deferred acc buffer. */
457 0 : shmem->deferred_acc_buf_cnt = 0UL;
458 0 : shmem->deferred_acc_epoch = 0UL;
459 :
460 : /* Shared metrics: zero gauges that reflect current state (now empty)
461 : but preserve counters and accounts_capacity. */
462 0 : shmem->shmetrics->accounts_total = 0UL;
463 0 : shmem->shmetrics->disk_allocated_bytes = 0UL;
464 0 : shmem->shmetrics->disk_current_bytes = 0UL;
465 0 : shmem->shmetrics->disk_used_bytes = 0UL;
466 0 : shmem->shmetrics->in_compaction = 0;
467 :
468 : /* Command slot */
469 0 : shmem->cmd_op = FD_ACCDB_CMD_IDLE;
470 0 : shmem->cmd_fork_id = USHORT_MAX;
471 :
472 0 : shmem->snapshot_loading = 0;
473 :
474 0 : FD_COMPILER_MFENCE();
475 :
476 : /* Tell the accdb tile to clear its stale deferred fork chain.
477 : Its deferred_fork_head/tail now reference recycled pool elements;
478 : it must discard them before processing any future advance_root or
479 : purge command. The command is asynchronous; the next advance_root
480 : or purge call will wait for it to complete via wait_cmd. */
481 0 : submit_cmd( accdb, FD_ACCDB_CMD_CLEAR_DEFERRED, 0 );
482 :
483 : /* Reset local state */
484 0 : accdb->deferred_fork_head = NULL;
485 0 : accdb->deferred_fork_tail = NULL;
486 0 : accdb->deferred_fork_epoch = 0UL;
487 0 : accdb->acquire_state = FD_ACCDB_ACQUIRE_STATE_IDLE;
488 0 : memset( &accdb->write_stats, 0, sizeof(accdb->write_stats) );
489 0 : }
490 :
491 : void
492 0 : fd_accdb_snapshot_load_begin( fd_accdb_t * accdb ) {
493 0 : FD_CHECK_CRIT( fd_accdb_snapshot_sync_state( &accdb->shmem->snapshot_sync )==FD_ACCDB_SNAPSHOT_SYNC_IDLE,
494 0 : "snapshot load started while snapshot production active" );
495 0 : FD_VOLATILE( accdb->shmem->snapshot_loading ) = 1;
496 0 : }
497 :
498 : static inline void
499 : change_partition( fd_accdb_t * accdb,
500 : accdb_offset_t const * offset_before,
501 : accdb_offset_t * out_offset,
502 : int * has_partition,
503 : uchar layer );
504 :
505 : void
506 0 : fd_accdb_snapshot_load_end( fd_accdb_t * accdb ) {
507 0 : fd_accdb_flush_metrics( accdb );
508 :
509 : /* Once the snapshot is done loading, the cold partition we were
510 : writing into is left open as the layer 2 head, and we switch the
511 : layer 0 head to a new hot partition. */
512 0 : ulong const cold = FD_ACCDB_COMPACTION_LAYER_CNT-1UL;
513 0 : if( FD_LIKELY( accdb->shmem->has_partition[ 0 ] ) ) {
514 0 : if( FD_LIKELY( !accdb->shmem->has_partition[ cold ] ) ) {
515 0 : accdb_offset_t snap_head = accdb->shmem->whead[ 0 ];
516 0 : int no_head = 0;
517 0 : change_partition( accdb, &snap_head, &accdb->shmem->whead[ 0 ], &no_head, 0 );
518 0 : spin_lock_acquire( &accdb->shmem->partition_lock );
519 0 : accdb->shmem->whead[ cold ] = snap_head;
520 0 : accdb->shmem->has_partition[ cold ] = 1;
521 0 : spin_lock_release( &accdb->shmem->partition_lock );
522 0 : } else {
523 0 : change_partition( accdb, &accdb->shmem->whead[ 0 ], &accdb->shmem->whead[ 0 ], &accdb->shmem->has_partition[ 0 ], 0 );
524 0 : }
525 0 : }
526 :
527 0 : spin_lock_acquire( &accdb->shmem->partition_lock );
528 0 : if( FD_LIKELY( accdb->shmem->has_partition[ 0 ] ) ) {
529 0 : ulong new_idx = packed_partition_idx( &accdb->shmem->whead[ 0 ] );
530 0 : fd_accdb_partition_t * newp = partition_pool_ele( accdb->partition_pool, new_idx );
531 0 : FD_VOLATILE( newp->layer ) = 0;
532 0 : }
533 :
534 0 : FD_VOLATILE( accdb->shmem->snapshot_loading ) = 0;
535 :
536 : /* Sweep all partitions written during the load — any that crossed
537 : the fragmentation threshold while enqueue was suppressed are
538 : re-checked now and pushed onto the compaction queue. */
539 0 : ulong partition_max = accdb->shmem->partition_max;
540 0 : for( ulong p=0UL; p<partition_max; p++ ) {
541 0 : fd_accdb_shmem_try_enqueue_compaction( accdb->shmem, p );
542 0 : }
543 :
544 0 : spin_lock_release( &accdb->shmem->partition_lock );
545 0 : }
546 :
547 : static inline uint
548 : delta_chain( fd_accdb_shmem_t const * accdb,
549 0 : uchar const pubkey[ 32 ] ) {
550 0 : uint hash = (uint)fd_hash32( pubkey, accdb->delta.seed );
551 0 : return hash & accdb->delta.chain_mask;
552 0 : }
553 :
554 : static int
555 : delta_insert( fd_accdb_t * accdb,
556 0 : uchar const pubkey[ 32 ] ) {
557 : /* FIXME consider batch inserting */
558 0 : fd_accdb_shmem_t * shmem = accdb->shmem;
559 0 : if( FD_UNLIKELY( shmem->delta.head >= shmem->delta.ele_max ) ) return 0;
560 :
561 0 : uint * chains = accdb->delta.chains;
562 0 : uint * chain = &chains[ delta_chain( shmem, pubkey ) ];
563 0 : fd_accdb_delta_t * pool = accdb->delta.pool;
564 :
565 0 : uint head = *chain;
566 0 : for( uint cur=head; cur!=UINT_MAX; cur=pool[ cur ].next ) {
567 0 : if( FD_UNLIKELY( !memcmp( pool[ cur ].pubkey, pubkey, 32UL ) ) ) return 1;
568 0 : }
569 :
570 0 : ulong idx = shmem->delta.head++;
571 0 : fd_accdb_delta_t * delta = &pool[ idx ];
572 0 : delta->next = head;
573 0 : memcpy( delta->pubkey, pubkey, 32UL );
574 0 : *chain = (uint)idx;
575 0 : return 1;
576 0 : }
577 :
578 : int
579 : fd_accdb_snapshot_recover_delta( fd_accdb_t * accdb,
580 0 : fd_accdb_fork_id_t fork_id ) {
581 0 : if( FD_UNLIKELY( fork_id.val>=fork_pool_ele_max( accdb->fork_shmem_pool ) ) ) {
582 0 : FD_LOG_CRIT(( "fd_accdb_snapshot_recover_delta: invalid fork id %u (capacity %lu)",
583 0 : (uint)fork_id.val, fork_pool_ele_max( accdb->fork_shmem_pool ) ));
584 0 : }
585 :
586 0 : uint txn_idx = accdb->fork_pool[ fork_id.val ].shmem->txn_head;
587 0 : while( txn_idx!=UINT_MAX ) {
588 0 : fd_accdb_txn_t const * txn = txn_pool_ele( accdb->txn_pool, (ulong)txn_idx );
589 0 : fd_accdb_accmeta_t const * acc = &accdb->acc_pool[ txn->acc_pool_idx ];
590 0 : if( FD_UNLIKELY( !delta_insert( accdb, acc->key.pubkey ) ) ) return -1;
591 0 : txn_idx = txn->fork.next;
592 0 : }
593 0 : return 0;
594 0 : }
595 :
596 : void
597 : fd_accdb_snapshot_save_whead( fd_accdb_t * accdb,
598 0 : fd_accdb_snapshot_recovery_t * out ) {
599 : /* Flush metrics to update disk_current_bytes. */
600 0 : fd_accdb_flush_metrics( accdb );
601 :
602 0 : out->whead_val = FD_VOLATILE_CONST( accdb->shmem->whead[ 0 ].val );
603 0 : out->has_partition = FD_VOLATILE_CONST( accdb->shmem->has_partition[ 0 ] );
604 0 : out->partition_max = FD_VOLATILE_CONST( accdb->shmem->partition_max );
605 0 : out->disk_current_bytes = FD_VOLATILE_CONST( accdb->shmem->shmetrics->disk_current_bytes );
606 :
607 0 : if( out->has_partition ) {
608 0 : accdb_offset_t whead = { .val = out->whead_val };
609 0 : ulong idx = packed_partition_idx( &whead );
610 0 : fd_accdb_partition_t * part = partition_pool_ele( accdb->partition_pool, idx );
611 0 : out->savepoint_bytes_freed = FD_VOLATILE_CONST( part->bytes_freed );
612 0 : } else {
613 0 : out->savepoint_bytes_freed = 0UL;
614 0 : }
615 0 : }
616 :
617 : void
618 : fd_accdb_snapshot_revert_whead( fd_accdb_t * accdb,
619 0 : fd_accdb_snapshot_recovery_t const * recover ) {
620 0 : fd_accdb_shmem_t * shmem = accdb->shmem;
621 :
622 : /* Partitions are about to be released, so flush metrics first. */
623 0 : fd_accdb_flush_metrics( accdb );
624 :
625 : /* Wait for any pending background command (purge) on T2 to finish
626 : before releasing partitions. */
627 0 : wait_cmd( accdb );
628 :
629 0 : ulong cur_partition_max = shmem->partition_max;
630 :
631 : /* Materialize the active partition's write_offset from the whead
632 : before releasing. Closed partitions have write_offset set by
633 : change_partition, but the last active partition still has
634 : write_offset == 0 from its initialization. The real byte offset
635 : is encoded in whead[0]. */
636 0 : if( shmem->has_partition[ 0 ] && cur_partition_max>recover->partition_max ) {
637 0 : ulong active_idx = packed_partition_idx( &shmem->whead[ 0 ] );
638 0 : if( active_idx>=recover->partition_max && active_idx<cur_partition_max ) {
639 0 : fd_accdb_partition_t * active = partition_pool_ele( accdb->partition_pool, active_idx );
640 0 : active->write_offset = packed_partition_offset( &shmem->whead[ 0 ] );
641 0 : }
642 0 : }
643 :
644 : /* Release partitions that have been previously allocated. Must hold
645 : partition_lock because partition_pool_ele_release mutates the
646 : pool free list. Before releasing, unlink any partition that sits
647 : on a compaction dlist (queued flag).
648 :
649 : Release in descending index order so that the LIFO free list
650 : re-acquires them in ascending order (P, P+1, P+2, ...). This
651 : preserves the ascending partition index order expected when the
652 : released suffix is allocated again. */
653 0 : spin_lock_acquire( &shmem->partition_lock );
654 0 : for( ulong p=cur_partition_max; p>recover->partition_max; p-- ) {
655 0 : fd_accdb_partition_t * part = partition_pool_ele( accdb->partition_pool, p-1UL );
656 0 : if( FD_UNLIKELY( part->queued ) ) {
657 0 : compaction_dlist_ele_remove( accdb->compaction_dlist[ part->layer ], part, accdb->partition_pool );
658 0 : }
659 0 : FD_VOLATILE( part->write_offset ) = 0UL;
660 0 : partition_pool_ele_release( accdb->partition_pool, part );
661 0 : }
662 :
663 0 : shmem->whead[ 0 ].val = recover->whead_val;
664 0 : shmem->has_partition[ 0 ] = recover->has_partition;
665 0 : shmem->partition_max = recover->partition_max;
666 :
667 : /* disk_used_bytes is NOT saved/restored here. It is implicitly
668 : reverted by purge_inner -> acc_unlink, which decrements
669 : disk_used_bytes for each unlinked entry. The caller must
670 : complete the purge before calling revert_whead. */
671 :
672 0 : shmem->shmetrics->disk_current_bytes = recover->disk_current_bytes;
673 :
674 0 : if( recover->has_partition ) {
675 0 : accdb_offset_t sp_off = (accdb_offset_t){ .val = recover->whead_val };
676 0 : ulong sp_idx = packed_partition_idx( &sp_off );
677 0 : fd_accdb_partition_t * sp = partition_pool_ele( accdb->partition_pool, sp_idx );
678 0 : sp->bytes_freed = recover->savepoint_bytes_freed;
679 0 : sp->write_offset = 0UL;
680 0 : }
681 :
682 0 : spin_lock_release( &shmem->partition_lock );
683 0 : }
684 :
685 : fd_accdb_t *
686 12 : fd_accdb_join( void * shaccdb ) {
687 12 : if( FD_UNLIKELY( !shaccdb ) ) {
688 0 : FD_LOG_WARNING(( "NULL shaccdb" ));
689 0 : return NULL;
690 0 : }
691 :
692 12 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shaccdb, fd_accdb_align() ) ) ) {
693 0 : FD_LOG_WARNING(( "misaligned shaccdb" ));
694 0 : return NULL;
695 0 : }
696 :
697 12 : return (fd_accdb_t*)shaccdb;
698 12 : }
699 :
700 : fd_accdb_t *
701 : fd_accdb_join_readonly( void * ljoin,
702 : fd_accdb_shmem_t * shmem,
703 : ulong * my_epoch_slot_rw,
704 0 : int fd_ro ) {
705 0 : if( FD_UNLIKELY( !ljoin ) ) {
706 0 : FD_LOG_WARNING(( "NULL ljoin" ));
707 0 : return NULL;
708 0 : }
709 :
710 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)ljoin, fd_accdb_align() ) ) ) {
711 0 : FD_LOG_WARNING(( "misaligned ljoin" ));
712 0 : return NULL;
713 0 : }
714 :
715 0 : if( FD_UNLIKELY( !my_epoch_slot_rw ) ) {
716 0 : FD_LOG_WARNING(( "NULL my_epoch_slot_rw" ));
717 0 : return NULL;
718 0 : }
719 :
720 0 : ulong max_live_slots = shmem->max_live_slots;
721 0 : ulong max_accounts = shmem->max_accounts;
722 0 : ulong max_account_writes_per_slot = shmem->max_account_writes_per_slot;
723 0 : ulong partition_cnt = shmem->partition_cnt;
724 :
725 0 : ulong chain_cnt = fd_ulong_pow2_up( (max_accounts>>1) + (max_accounts&1UL) );
726 0 : ulong txn_max = max_live_slots * max_account_writes_per_slot;
727 :
728 : /* Recompute the same shmem scratch layout that fd_accdb_shmem_new
729 : used. All FD_SCRATCH_ALLOC_APPEND calls here only compute pointer
730 : offsets — they do not write to shmem. */
731 0 : FD_SCRATCH_ALLOC_INIT( l, shmem );
732 0 : FD_SCRATCH_ALLOC_APPEND( l, FD_ACCDB_SHMEM_ALIGN, sizeof(fd_accdb_shmem_t) );
733 0 : void * _fork_pool_ele = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_accdb_fork_shmem_t), max_live_slots*sizeof(fd_accdb_fork_shmem_t) );
734 0 : void * _descends_sets = FD_SCRATCH_ALLOC_APPEND( l, descends_set_align(), max_live_slots*descends_set_footprint( max_live_slots ) );
735 0 : void * _acc_map = FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), chain_cnt*sizeof(uint) );
736 0 : void * _acc_pool_ele = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_accdb_accmeta_t), max_accounts*sizeof(fd_accdb_accmeta_t) );
737 0 : FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_accdb_txn_t), txn_max*sizeof(fd_accdb_txn_t) );
738 0 : FD_SCRATCH_ALLOC_APPEND( l, partition_pool_align(), partition_pool_footprint( partition_cnt ) );
739 0 : for( ulong k=0UL; k<FD_ACCDB_COMPACTION_LAYER_CNT; k++ ) {
740 0 : FD_SCRATCH_ALLOC_APPEND( l, compaction_dlist_align(), compaction_dlist_footprint() );
741 0 : }
742 0 : FD_SCRATCH_ALLOC_APPEND( l, deferred_free_dlist_align(), deferred_free_dlist_footprint() );
743 :
744 0 : FD_SCRATCH_ALLOC_INIT( l2, ljoin );
745 0 : fd_accdb_t * accdb = FD_SCRATCH_ALLOC_APPEND( l2, fd_accdb_align(), sizeof(fd_accdb_t) );
746 0 : void * _local_fork_pool = FD_SCRATCH_ALLOC_APPEND( l2, alignof(fd_accdb_fork_t), max_live_slots*sizeof(fd_accdb_fork_t) );
747 :
748 0 : accdb->fd = fd_ro;
749 0 : accdb->acquire_state = FD_ACCDB_ACQUIRE_STATE_IDLE;
750 0 : accdb->shmem = shmem;
751 0 : FD_TEST( acc_pool_join( accdb->acc_pool_join, shmem->acc_pool, _acc_pool_ele, max_accounts ) );
752 0 : accdb->acc_pool = accdb->acc_pool_join->ele;
753 0 : accdb->acc_map = _acc_map;
754 0 : for( ulong c=0UL; c<FD_ACCDB_CACHE_CLASS_CNT; c++ ) accdb->cache[ c ] = (uchar *)shmem + shmem->cache_region_off[ c ];
755 :
756 : /* Writer-only structures: leave NULL so any accidental writer-path
757 : call from a readonly joiner crashes loudly rather than corrupting
758 : state. */
759 0 : accdb->partition_pool = NULL;
760 0 : for( ulong k=0UL; k<FD_ACCDB_COMPACTION_LAYER_CNT; k++ ) accdb->compaction_dlist[ k ] = NULL;
761 0 : accdb->deferred_free_dlist = NULL;
762 0 : accdb->delta.chains = NULL;
763 0 : accdb->delta.pool = NULL;
764 :
765 0 : FD_TEST( fork_pool_join( accdb->fork_shmem_pool, shmem->fork_pool, _fork_pool_ele, max_live_slots ) );
766 0 : accdb->fork_pool = _local_fork_pool;
767 0 : for( ulong i=0UL; i<max_live_slots; i++ ) {
768 0 : fd_accdb_fork_t * fork = &accdb->fork_pool[ i ];
769 0 : fork->shmem = fork_pool_ele( accdb->fork_shmem_pool, i );
770 0 : fork->descends = descends_set_join( (uchar *)_descends_sets + i*descends_set_footprint( max_live_slots ) );
771 0 : FD_TEST( fork->shmem );
772 0 : FD_TEST( fork->descends );
773 0 : }
774 :
775 : /* my_epoch_slot_rw points at memory owned by this joiner (e.g. a
776 : private per-tile fseq) that the joiner can write to. The
777 : accdb tile sees it via its external_epoch_slots[] array (mapped
778 : read-only) and includes it in its compaction epoch scan.
779 : Storing through this pointer is the only side effect a readonly
780 : joiner has on shared state. */
781 0 : accdb->my_epoch_slot = my_epoch_slot_rw;
782 :
783 : /* Readonly joiners do not own external slots themselves; only the
784 : compaction tile / writer joiners do. */
785 0 : accdb->external_epoch_slots = NULL;
786 0 : accdb->external_epoch_cnt = 0UL;
787 0 : accdb->bounce = NULL;
788 0 : accdb->stage = NULL;
789 :
790 0 : accdb->deferred_acc_buf = NULL;
791 0 : accdb->deferred_fork_head = NULL;
792 0 : accdb->deferred_fork_tail = NULL;
793 0 : accdb->deferred_fork_epoch = 0UL;
794 0 : accdb->prealloc_failed_at = ULONG_MAX;
795 :
796 0 : memset( accdb->metrics, 0, sizeof(fd_accdb_metrics_t) );
797 0 : memset( &accdb->write_stats, 0, sizeof(accdb->write_stats) );
798 :
799 0 : return accdb;
800 0 : }
801 :
802 : /* T1 -> T2 cmd channel. Two states on cmd_op:
803 :
804 : IDLE - no cmd in flight
805 : non-IDLE - cmd pending; T2 will process it then flip back to IDLE
806 :
807 : T1 submits by writing fork_id then cmd_op (non-IDLE). T2 processes
808 : by reading fork_id then writing cmd_op = IDLE. T1 waits for IDLE
809 : before submitting again, so T2 never sees a half-written cmd and
810 : never re-processes the same cmd. */
811 :
812 : static inline void
813 128101 : wait_cmd( fd_accdb_t * accdb ) {
814 128101 : fd_accdb_shmem_t * shmem = accdb->shmem;
815 128101 : while( FD_VOLATILE_CONST( shmem->cmd_op )!=FD_ACCDB_CMD_IDLE ) FD_SPIN_PAUSE();
816 128101 : FD_COMPILER_MFENCE();
817 128101 : }
818 :
819 : static inline void
820 : submit_cmd( fd_accdb_t * accdb,
821 : uint op,
822 64051 : ushort fork_id ) {
823 64051 : fd_accdb_shmem_t * shmem = accdb->shmem;
824 64051 : FD_VOLATILE( shmem->cmd_fork_id ) = fork_id;
825 64051 : FD_COMPILER_MFENCE();
826 64051 : FD_VOLATILE( shmem->cmd_op ) = op;
827 64051 : if( FD_UNLIKELY( accdb->sleep ) ) fd_sleep_ring( accdb->sleep, accdb->sleep_tile_id );
828 64051 : }
829 :
830 : fd_accdb_fork_id_t
831 : fd_accdb_attach_child( fd_accdb_t * accdb,
832 64065 : fd_accdb_fork_id_t parent_fork_id ) {
833 : /* fork_pool_acquire is not NULL-checked: replay gates attaches on
834 : fd_banks_can_start_bank, and wait_cmd ensures the prior
835 : advance_root has fully run on T2, so
836 : live + deferred forks <= max_live_slots. */
837 64065 : wait_cmd( accdb );
838 :
839 64065 : if( FD_UNLIKELY( accdb->shmem->generation==UINT_MAX ) ) FD_LOG_ERR(( "accdb ran out of generation sequence numbers, restart required" ));
840 :
841 64065 : fd_accdb_fork_shmem_t * acquired = fork_pool_acquire( accdb->fork_shmem_pool );
842 64065 : if( FD_UNLIKELY( !acquired ) ) {
843 0 : submit_cmd( accdb, FD_ACCDB_CMD_DRAIN_DEFERRED, USHORT_MAX );
844 0 : wait_cmd( accdb );
845 0 : acquired = fork_pool_acquire( accdb->fork_shmem_pool );
846 0 : FD_CHECK_CRIT( acquired, "accdb fork pool exhausted after deferred drain" );
847 0 : }
848 64065 : ulong idx = fork_pool_idx( accdb->fork_shmem_pool, acquired );
849 :
850 64065 : fd_accdb_fork_t * fork = &accdb->fork_pool[ idx ];
851 64065 : fd_accdb_fork_id_t fork_id = { .val = (ushort)idx };
852 :
853 64065 : fork->shmem->child_id = (fd_accdb_fork_id_t){ .val = USHORT_MAX };
854 :
855 64065 : if( FD_UNLIKELY( parent_fork_id.val==USHORT_MAX ) ) {
856 12 : fork->shmem->parent_id = (fd_accdb_fork_id_t){ .val = USHORT_MAX };
857 12 : fork->shmem->sibling_id = (fd_accdb_fork_id_t){ .val = USHORT_MAX };
858 :
859 12 : descends_set_null( fork->descends );
860 12 : accdb->shmem->root_fork_id = fork_id;
861 64053 : } else {
862 64053 : fd_accdb_fork_t * parent = &accdb->fork_pool[ parent_fork_id.val ];
863 64053 : fork->shmem->parent_id = parent_fork_id;
864 :
865 64053 : descends_set_copy( fork->descends, parent->descends );
866 64053 : descends_set_insert( fork->descends, parent_fork_id.val );
867 :
868 : /* Atomically prepend to parent's child list. T2 (background_purge)
869 : may concurrently unlink a different child from the same list, so
870 : we must CAS here. */
871 64053 : FD_COMPILER_MFENCE();
872 64053 : for(;;) {
873 63928 : ushort old_head = FD_VOLATILE_CONST( parent->shmem->child_id.val );
874 63928 : fork->shmem->sibling_id = (fd_accdb_fork_id_t){ .val = old_head };
875 63928 : FD_COMPILER_MFENCE();
876 63983 : if( FD_LIKELY( FD_ATOMIC_CAS( &parent->shmem->child_id.val, old_head, fork_id.val )==old_head ) ) break;
877 >1844*10^16 : FD_SPIN_PAUSE();
878 >1844*10^16 : }
879 64053 : }
880 :
881 64065 : fork->shmem->generation = accdb->shmem->generation++;
882 64065 : fork->shmem->txn_head = UINT_MAX;
883 :
884 64065 : FD_TEST( !descends_set_test( fork->descends, fork_id.val ) );
885 :
886 64065 : return fork_id;
887 64065 : }
888 :
889 : /* evict_clear_acc_cache_ref atomically tears down acc->cache_idx and
890 : acc->executable_size.CACHE_VALID for an acc that is being evicted
891 : from cache line (size_class, line_idx). The caller must already
892 : hold an exclusive claim on the line (line->refcnt ==
893 : FD_ACCDB_EVICT_SENTINEL) so that no concurrent thread can pin the
894 : line.
895 :
896 : The naive sequence (clear cache_idx, clear VALID) lets a reader in
897 : cold_load_acc see VALID=1 and read a stale INVAL cache_idx, which
898 : decodes to an OOB cache_line pointer. The reverse sequence (clear
899 : VALID, clear cache_idx) lets a concurrent cold_load_acc observe
900 : VALID=0/CLAIM=0 and start publishing a *new* cache_idx + VALID=1
901 : between our two stores; our later cache_idx=INVAL would then
902 : stomp on the cold-loader's published idx.
903 :
904 : We close both races by acquiring CACHE_CLAIM_BIT before mutating
905 : acc->cache_idx. cold_load_acc spins while CLAIM is held, so it
906 : cannot enter the publish path concurrently. If CLAIM is already
907 : held, a cold-loader is already mid-publish; in that case
908 : acc->cache_idx is being repointed away from our line, and we must
909 : not touch it. After mutation we release CLAIM.
910 :
911 : Verifies acc->cache_idx still encodes (size_class, line_idx) before
912 : clobbering, in case the acc was concurrently re-published into a
913 : different line (e.g. by a previous cold_load_acc completing before
914 : we arrived). */
915 :
916 : static inline void
917 : evict_clear_acc_cache_ref( fd_accdb_accmeta_t * accmeta,
918 : ulong size_class,
919 90 : ulong line_idx ) {
920 90 : uint expected_cidx = FD_ACCDB_ACC_CIDX_PACK( (uint)size_class, (uint)line_idx );
921 :
922 : /* CAS-acquire CLAIM. A holder with VALID clear is a cold-loader
923 : publishing a different line, or a clearer that already severed the
924 : binding: bail. A holder with VALID set is another evictor,
925 : acc_unlink or the deferred-free drain, none of which block while
926 : holding CLAIM: wait, so whoever evicts the current line clears the
927 : binding. */
928 90 : for(;;) {
929 90 : uint cur = FD_VOLATILE_CONST( accmeta->executable_size );
930 90 : if( FD_UNLIKELY( cur & FD_ACCDB_SIZE_CACHE_CLAIM_BIT ) ) {
931 0 : if( !FD_ACCDB_SIZE_CACHE_VALID( cur ) ) return;
932 0 : fd_racesan_hook( "accdb_evict_clear:claim_held" );
933 0 : FD_SPIN_PAUSE();
934 0 : continue;
935 0 : }
936 90 : uint nxt = cur | FD_ACCDB_SIZE_CACHE_CLAIM_BIT;
937 90 : if( FD_LIKELY( FD_ATOMIC_CAS( &accmeta->executable_size, cur, nxt )==cur ) ) break;
938 0 : fd_racesan_hook( "accdb_evict_clear:claim_wait" );
939 0 : FD_SPIN_PAUSE();
940 0 : }
941 :
942 90 : fd_racesan_hook( "accdb_evict_clear:post_claim" );
943 :
944 : /* CLAIM held. If accmeta->cache_idx still points at our line, clear
945 : VALID and INVAL the cache_idx. Otherwise the accmeta was already
946 : re-published into a different line; leave it alone. */
947 90 : uint es = FD_VOLATILE_CONST( accmeta->executable_size );
948 90 : if( FD_LIKELY( FD_ACCDB_SIZE_CACHE_VALID( es ) &&
949 90 : FD_VOLATILE_CONST( accmeta->cache_idx )==expected_cidx ) ) {
950 90 : FD_ATOMIC_FETCH_AND_AND( &accmeta->executable_size, ~FD_ACCDB_SIZE_CACHE_VALID_BIT );
951 90 : fd_racesan_hook( "accdb_evict_clear:pre_inval" );
952 90 : FD_VOLATILE( accmeta->cache_idx ) = FD_ACCDB_ACC_CIDX_INVAL;
953 90 : }
954 :
955 : /* Release CLAIM. */
956 90 : FD_ATOMIC_FETCH_AND_AND( &accmeta->executable_size, ~FD_ACCDB_SIZE_CACHE_CLAIM_BIT );
957 90 : }
958 :
959 : /* cache_free_push pushes a fully-freed cache line onto the per-class
960 : CAS free list (Treiber stack). The caller must have already
961 : invalidated the line (key.generation==UINT_MAX, acc_idx==UINT_MAX)
962 : and set persisted=1 before pushing. Those stores must also be
963 : ordered before the store that releases refcnt to 0. */
964 :
965 : static inline void
966 : cache_free_push( fd_accdb_t * accdb,
967 : ulong size_class,
968 3254575 : fd_accdb_cache_line_t * line ) {
969 3254575 : ulong line_idx = cache_line_idx( accdb, size_class, line );
970 3254575 : for(;;) {
971 3236807 : ulong old_vt = FD_VOLATILE_CONST( accdb->shmem->cache_free[ size_class ].ver_top );
972 3236807 : uint old_top = (uint)( old_vt & (ulong)UINT_MAX );
973 3236807 : uint old_ver = (uint)( old_vt >> 32 );
974 3236807 : line->next = old_top;
975 3236807 : FD_COMPILER_MFENCE();
976 3236807 : ulong new_vt = ((ulong)(uint)( old_ver+1U ) << 32) | (ulong)(uint)line_idx;
977 3236807 : if( FD_LIKELY( FD_ATOMIC_CAS( &accdb->shmem->cache_free[ size_class ].ver_top, old_vt, new_vt )==old_vt ) ) {
978 3224864 : FD_ATOMIC_FETCH_AND_ADD( &accdb->shmem->cache_free_cnt[ size_class ].val, 1UL );
979 3224864 : return;
980 3224864 : }
981 11943 : FD_SPIN_PAUSE();
982 11943 : }
983 3254575 : }
984 :
985 : /* cache_free_pop pops a line from the per-class CAS free list. Returns
986 : NULL if the list is empty. */
987 :
988 : static inline fd_accdb_cache_line_t *
989 : cache_free_pop( fd_accdb_t * accdb,
990 3191767 : ulong size_class ) {
991 3191767 : for(;;) {
992 3191767 : ulong old_vt = FD_VOLATILE_CONST( accdb->shmem->cache_free[ size_class ].ver_top );
993 3191767 : uint old_top = (uint)( old_vt & (ulong)UINT_MAX );
994 3191767 : if( FD_UNLIKELY( old_top==UINT_MAX ) ) return NULL;
995 3185945 : uint old_ver = (uint)( old_vt >> 32 );
996 3185945 : fd_accdb_cache_line_t * top = cache_line( accdb, size_class, (ulong)old_top );
997 3185945 : uint next = FD_VOLATILE_CONST( top->next );
998 3185945 : ulong new_vt = ((ulong)(uint)( old_ver+1U ) << 32) | (ulong)next;
999 3225182 : if( FD_LIKELY( FD_ATOMIC_CAS( &accdb->shmem->cache_free[ size_class ].ver_top, old_vt, new_vt )==old_vt ) ) {
1000 3225182 : ulong freec = FD_ATOMIC_FETCH_AND_SUB( &accdb->shmem->cache_free_cnt[ size_class ].val, 1UL )-1UL;
1001 3225182 : if( FD_UNLIKELY( accdb->sleep ) ) {
1002 0 : ulong max_c = accdb->shmem->cache_class_max[ size_class ];
1003 0 : ulong init = fd_ulong_min( FD_VOLATILE_CONST( accdb->shmem->cache_class_init[ size_class ].val ), max_c );
1004 0 : ulong avail = max_c-(init>freec ? init-freec : 0UL);
1005 0 : if( FD_UNLIKELY( avail+1UL==accdb->shmem->cache_free_low_water[ size_class ] ) ) fd_sleep_ring( accdb->sleep, accdb->sleep_tile_id );
1006 0 : }
1007 3225182 : return top;
1008 3225182 : }
1009 >1844*10^16 : FD_SPIN_PAUSE();
1010 >1844*10^16 : }
1011 3191767 : }
1012 :
1013 : /* cache_try_pin attempts a lock-free pin of a cache-hit line. Returns
1014 : the line if successfully pinned, or NULL if the line is being evicted
1015 : or was recycled (ABA). */
1016 :
1017 : static inline fd_accdb_cache_line_t *
1018 : cache_try_pin( fd_accdb_cache_line_t * line,
1019 : uchar const pubkey[ 32 ],
1020 178918 : uint generation ) {
1021 178918 : for(;;) {
1022 178918 : uint old_rc = FD_VOLATILE_CONST( line->refcnt );
1023 178918 : if( FD_UNLIKELY( old_rc==FD_ACCDB_EVICT_SENTINEL ) ) return NULL;
1024 : /* No saturation guard needed: refcnt is a uint and at most
1025 : FD_ACCDB_MAX_JOINERS (256) threads can pin concurrently,
1026 : so old_rc+1 can never reach FD_ACCDB_EVICT_SENTINEL
1027 : (UINT_MAX) or wrap. */
1028 178971 : if( FD_LIKELY( FD_ATOMIC_CAS( &line->refcnt, old_rc, old_rc+1U )==old_rc ) ) {
1029 : /* Pinned. ABA check: verify the key hasn't changed under us. */
1030 178971 : fd_racesan_hook( "accdb_try_pin:post_cas" );
1031 178971 : FD_COMPILER_MFENCE();
1032 178971 : if( FD_UNLIKELY( line->key.generation!=generation ||
1033 178971 : memcmp( line->key.pubkey, pubkey, 32UL ) ) ) {
1034 0 : FD_ATOMIC_FETCH_AND_SUB( &line->refcnt, 1U );
1035 0 : return NULL;
1036 0 : }
1037 178971 : line->referenced = 1;
1038 178971 : fd_racesan_hook( "cache_try_pin:pinned" );
1039 178971 : return line;
1040 178971 : }
1041 >1844*10^16 : FD_SPIN_PAUSE();
1042 >1844*10^16 : }
1043 178918 : }
1044 :
1045 : /* wait_for_epoch_drain spins until every joiner's published epoch
1046 : exceeds tag, meaning all readers that were active at epoch=tag have
1047 : since exited their critical sections. */
1048 :
1049 : static void
1050 : wait_for_epoch_drain( fd_accdb_t * accdb,
1051 72382 : ulong tag ) {
1052 72382 : for(;;) {
1053 72378 : ulong min_epoch = ULONG_MAX;
1054 72378 : ulong joiner_cnt = FD_VOLATILE_CONST( accdb->shmem->joiner_cnt );
1055 144755 : for( ulong t=0UL; t<joiner_cnt; t++ ) {
1056 72377 : ulong e = FD_VOLATILE_CONST( accdb->shmem->joiner_epochs[ t ].val );
1057 72377 : if( FD_LIKELY( e<min_epoch ) ) min_epoch = e;
1058 72377 : }
1059 72378 : for( ulong t=0UL; t<accdb->external_epoch_cnt; t++ ) {
1060 0 : ulong e = FD_VOLATILE_CONST( *accdb->external_epoch_slots[ t ] );
1061 0 : if( FD_LIKELY( e<min_epoch ) ) min_epoch = e;
1062 0 : }
1063 72382 : if( FD_LIKELY( tag<min_epoch ) ) break;
1064 >1844*10^16 : fd_racesan_hook( "accdb_epoch_drain:wait" );
1065 >1844*10^16 : FD_SPIN_PAUSE();
1066 >1844*10^16 : }
1067 72382 : }
1068 :
1069 : /* drain_deferred_frees releases back to their respective pools any acc
1070 : batch and/or fork slots that were unlinked in a prior advance_root /
1071 : purge call. The resources cannot be released immediately because
1072 : concurrent readers may still reference them. We wait until every
1073 : joiner's published epoch exceeds the tag stamped when each resource
1074 : was unlinked.
1075 :
1076 : Must be called before creating new deferred batches (there is at most
1077 : one of each outstanding at a time). */
1078 :
1079 : static void
1080 64055 : drain_deferred_frees( fd_accdb_t * accdb ) {
1081 64055 : if( FD_UNLIKELY( accdb->deferred_fork_head ) ) {
1082 64046 : wait_for_epoch_drain( accdb, accdb->deferred_fork_epoch );
1083 64046 : fork_pool_release_chain( accdb->fork_shmem_pool, accdb->deferred_fork_head, accdb->deferred_fork_tail );
1084 64046 : accdb->deferred_fork_head = NULL;
1085 64046 : accdb->deferred_fork_tail = NULL;
1086 64046 : }
1087 :
1088 64055 : ulong n = accdb->shmem->deferred_acc_buf_cnt;
1089 64055 : if( FD_LIKELY( !n ) ) return;
1090 8261 : wait_for_epoch_drain( accdb, accdb->shmem->deferred_acc_epoch );
1091 :
1092 : /* All readers that could have been holding a captured pointer to any
1093 : of these accs at unlink time have now exited their epoch sections.
1094 : It is safe to materialize pool.next links and hand the chain to
1095 : acc_pool_release_chain. */
1096 8261 : uint * buf = accdb->deferred_acc_buf;
1097 8261 : fd_accdb_accmeta_t * acc_pool = accdb->acc_pool;
1098 :
1099 : /* Late-publish sweep: a concurrent acquire evictor may have published
1100 : a new offset into one of these accmetas after acc_unlink's
1101 : xchg-to-INVAL but before exiting its epoch. Now that the epoch has
1102 : drained, any such publish is complete and visible. Free the
1103 : orphaned disk bytes here, before the accmeta is released to the
1104 : pool and its fields recycled. */
1105 8261 : ulong acc_pool_cap = acc_pool_ele_max( accdb->acc_pool_join );
1106 377085 : for( ulong i=0UL; i<n; i++ ) {
1107 368824 : FD_TEST( (ulong)buf[ i ]<acc_pool_cap );
1108 368824 : fd_accdb_accmeta_t * accmeta = &acc_pool[ buf[ i ] ];
1109 368824 : ulong off = fd_accdb_acc_offset( accmeta );
1110 368824 : if( FD_UNLIKELY( off!=FD_ACCDB_OFF_INVAL ) ) {
1111 0 : ulong entry_sz = (ulong)FD_ACCDB_SIZE_DATA(accmeta->executable_size)+sizeof(fd_accdb_disk_meta_t);
1112 0 : fd_accdb_shmem_bytes_freed( accdb->shmem, off, entry_sz );
1113 0 : FD_ATOMIC_FETCH_AND_SUB( &accdb->shmem->shmetrics->disk_used_bytes, entry_sz );
1114 0 : }
1115 : /* A reader that captured this accmeta before the unlink may have
1116 : cold-loaded it since (tombstones skip the offset wait), publishing
1117 : a line and VALID, and an evictor of that line may be clearing the
1118 : binding right now under CLAIM. Take CLAIM so that evictor has
1119 : finished its cache_idx store before pool.next overwrites it, then
1120 : leave the slot with VALID clear. */
1121 368824 : for(;;) {
1122 368626 : uint cur = FD_VOLATILE_CONST( accmeta->executable_size );
1123 368626 : if( FD_UNLIKELY( cur & FD_ACCDB_SIZE_CACHE_CLAIM_BIT ) ) {
1124 0 : fd_racesan_hook( "accdb_drain:claim_wait" );
1125 0 : FD_SPIN_PAUSE();
1126 0 : continue;
1127 0 : }
1128 368739 : if( FD_LIKELY( FD_ATOMIC_CAS( &accmeta->executable_size, cur, cur | FD_ACCDB_SIZE_CACHE_CLAIM_BIT )==cur ) ) break;
1129 >1844*10^16 : FD_SPIN_PAUSE();
1130 >1844*10^16 : }
1131 368824 : FD_ATOMIC_FETCH_AND_AND( &accmeta->executable_size, ~(FD_ACCDB_SIZE_CACHE_VALID_BIT|FD_ACCDB_SIZE_CACHE_CLAIM_BIT) );
1132 368824 : }
1133 :
1134 368181 : for( ulong i=0UL; i+1UL<n; i++ ) {
1135 359920 : acc_pool[ buf[ i ] ].pool.next = acc_pool_private_cidx( (ulong)buf[ i+1UL ] );
1136 359920 : }
1137 8261 : fd_accdb_accmeta_t * head = &acc_pool[ buf[ 0UL ] ];
1138 8261 : fd_accdb_accmeta_t * tail = &acc_pool[ buf[ n-1UL ] ];
1139 8261 : acc_pool_release_chain( accdb->acc_pool_join, head, tail );
1140 8261 : accdb->shmem->deferred_acc_buf_cnt = 0UL;
1141 8261 : }
1142 :
1143 : /* deferred_acc_append records an unlinked acc index in the side buffer
1144 : for later release after wait_for_epoch_drain. T2 is the sole writer.
1145 : The chain link from acc->pool.next is NOT laid down here: pool.next
1146 : is union-aliased to cache_idx, and a concurrent cold_load_acc may
1147 : still publish through a captured pointer until the epoch drains.
1148 : Materialization of the chain happens in drain_deferred_frees. */
1149 :
1150 : static inline void
1151 : deferred_acc_append( fd_accdb_t * accdb,
1152 368625 : uint acc_idx ) {
1153 368625 : fd_accdb_shmem_t * shmem = accdb->shmem;
1154 368625 : FD_TEST( shmem->deferred_acc_buf_cnt<shmem->deferred_acc_buf_max );
1155 368625 : accdb->deferred_acc_buf[ shmem->deferred_acc_buf_cnt++ ] = acc_idx;
1156 368625 : }
1157 :
1158 : /* acc_unlink unlinks an account from its hash map chain, frees any
1159 : associated disk bytes, and invalidates a stale cache reference. Does
1160 : NOT release the acc pool slot — the caller is responsible for that
1161 : (or for batching releases).
1162 :
1163 : prev is the previous element in the map chain (UINT_MAX if acc_idx is
1164 : the head).
1165 :
1166 : CONCURRENCY: The chain link being removed is swapped out with a CAS
1167 : so that a concurrent fd_accdb_release prepending to the same chain
1168 : cannot lose its update. If a head-removal CAS fails (a new node was
1169 : prepended since we loaded the head), we re-walk from the new head to
1170 : find the target as an interior node. Interior CAS cannot fail from
1171 : inserts (inserts only touch the head) and only one remover exists at
1172 : a time (advance_root / purge are serialized). */
1173 :
1174 : static inline void
1175 : acc_unlink( fd_accdb_t * accdb,
1176 : uint map_idx,
1177 : uint prev,
1178 368360 : uint acc_idx ) {
1179 368360 : fd_accdb_accmeta_t * accmeta = &accdb->acc_pool[ acc_idx ];
1180 :
1181 : /* Atomically capture and clear the offset. Two races to defuse:
1182 :
1183 : (1) A concurrent fd_accdb_acquire_inner that is CLOCK-evicting the
1184 : cache line currently holding this acc's data may have already
1185 : xchg'd the offset to INVAL in step 5-6 and freed the old disk
1186 : bytes. Without atomicity we would re-read the old offset and
1187 : free those same bytes a second time. The xchg here serializes:
1188 : whoever wins sees the real offset and frees; the loser sees
1189 : INVAL and skips.
1190 :
1191 : (2) That same evictor may also be mid-flight to publish a NEW
1192 : offset in step 9 (after step 5-6's free but before step 9's
1193 : store). That late publish lands on an accmeta that is about
1194 : to be chain-unlinked and deferred-released. drain_deferred_
1195 : frees sweeps the deferred buffer after epoch drain to catch
1196 : the late publish and free the orphaned bytes. */
1197 368360 : uint es = FD_VOLATILE_CONST( accmeta->executable_size );
1198 368360 : ulong entry_sz = (ulong)FD_ACCDB_SIZE_DATA(es)+sizeof(fd_accdb_disk_meta_t);
1199 :
1200 : /* Prefetch the cache line the reclaim below will CAS */
1201 368360 : if( FD_UNLIKELY( FD_ACCDB_SIZE_CACHE_VALID( es ) ) ) {
1202 368293 : uint cidx = FD_VOLATILE_CONST( accmeta->cache_idx );
1203 368293 : ulong cls = FD_ACCDB_ACC_CIDX_CLASS( cidx ) & (FD_ACCDB_CACHE_CLASS_CNT-1UL);
1204 368293 : ulong idx = FD_ACCDB_ACC_CIDX_IDX( cidx );
1205 368328 : if( FD_LIKELY( cidx!=FD_ACCDB_ACC_CIDX_INVAL && idx<accdb->shmem->cache_class_max[ cls ] ) ) __builtin_prefetch( cache_line( accdb, cls, idx ), 1, 3 );
1206 368293 : }
1207 :
1208 368360 : ulong old_offset = fd_accdb_acc_xchg_offset( accmeta, FD_ACCDB_OFF_INVAL );
1209 368360 : if( FD_LIKELY( old_offset!=FD_ACCDB_OFF_INVAL ) ) {
1210 0 : fd_accdb_shmem_bytes_freed( accdb->shmem, old_offset, entry_sz );
1211 0 : FD_ATOMIC_FETCH_AND_SUB( &accdb->shmem->shmetrics->disk_used_bytes, entry_sz );
1212 0 : }
1213 368360 : FD_ATOMIC_FETCH_AND_SUB( &accdb->shmem->shmetrics->accounts_total, 1UL );
1214 368360 : accdb->metrics->accounts_deleted++;
1215 :
1216 368360 : if( FD_LIKELY( prev==UINT_MAX ) ) {
1217 : /* Head removal — CAS may fail if a concurrent insert prepended a
1218 : new node. On failure the target is now interior. */
1219 367861 : for(;;) {
1220 367227 : uint old_head = FD_VOLATILE_CONST( accdb->acc_map[ map_idx ] );
1221 367358 : if( FD_LIKELY( old_head==acc_idx ) ) {
1222 367631 : if( FD_LIKELY( FD_ATOMIC_CAS( &accdb->acc_map[ map_idx ], acc_idx, accmeta->map.next )==acc_idx ) ) break;
1223 >1844*10^16 : FD_SPIN_PAUSE();
1224 >1844*10^16 : continue;
1225 367358 : }
1226 : /* Head changed — walk from new head to find prev for interior
1227 : removal. The target must still be in the chain because only
1228 : this thread removes elements. */
1229 >1844*10^16 : prev = old_head;
1230 >1844*10^16 : while( FD_VOLATILE_CONST( accdb->acc_pool[ prev ].map.next )!=acc_idx ) prev = FD_VOLATILE_CONST( accdb->acc_pool[ prev ].map.next );
1231 >1844*10^16 : FD_ATOMIC_CAS( &accdb->acc_pool[ prev ].map.next, acc_idx, accmeta->map.next );
1232 >1844*10^16 : break;
1233 367227 : }
1234 367861 : } else {
1235 499 : FD_ATOMIC_CAS( &accdb->acc_pool[ prev ].map.next, acc_idx, accmeta->map.next );
1236 499 : }
1237 :
1238 368360 : fd_racesan_hook( "accdb_acc_unlink:post_splice" );
1239 :
1240 : /* If the freed acc still has a cached location, invalidate it and
1241 : try to reclaim the cache line so the eviction path does not try
1242 : to write back stale data from a recycled pool slot. Lock-free:
1243 : CAS the refcnt 0 -> EVICT_SENTINEL to claim it exclusively, then
1244 : push to the CAS free list. If the line is pinned (refcnt>0),
1245 : skip, the pinner's release will handle it.
1246 :
1247 : Acquire CACHE_CLAIM_BIT before touching acc->cache_idx /
1248 : CACHE_VALID — see evict_clear_acc_cache_ref for the protocol.
1249 : Without CLAIM, a concurrent cold_load_acc can publish a fresh
1250 : (cache_idx, VALID=1) pair into this acc between our two stores,
1251 : and our subsequent cache_idx=INVAL stomps onto the freelist
1252 : pool.next field (the union sibling of cache_idx), corrupting the
1253 : pool. Unlike evict_clear_acc_cache_ref, we cannot bail when CLAIM
1254 : is held: this acc is being permanently unlinked, so we must
1255 : spin-wait for the cold-loader to release CLAIM and then invalidate
1256 : whatever cache_idx is current. */
1257 368360 : uint cur_es;
1258 368360 : for(;;) {
1259 367742 : cur_es = FD_VOLATILE_CONST( accmeta->executable_size );
1260 367742 : if( FD_UNLIKELY( cur_es & FD_ACCDB_SIZE_CACHE_CLAIM_BIT ) ) { FD_SPIN_PAUSE(); continue; }
1261 367742 : uint nxt_es = cur_es | FD_ACCDB_SIZE_CACHE_CLAIM_BIT;
1262 368360 : if( FD_LIKELY( FD_ATOMIC_CAS( &accmeta->executable_size, cur_es, nxt_es )==cur_es ) ) break;
1263 >1844*10^16 : FD_SPIN_PAUSE();
1264 >1844*10^16 : }
1265 :
1266 368360 : uint cidx = FD_ACCDB_ACC_CIDX_INVAL;
1267 368360 : int had_valid = FD_ACCDB_SIZE_CACHE_VALID( cur_es );
1268 368456 : if( FD_UNLIKELY( had_valid ) ) {
1269 368456 : cidx = FD_VOLATILE_CONST( accmeta->cache_idx );
1270 : /* Clear VALID before INVAL'ing cache_idx — matches the order in
1271 : evict_clear_acc_cache_ref so cold_load_acc's "VALID=1 +
1272 : cidx=INVAL" spin path resolves on the next iteration when it
1273 : observes VALID=0. */
1274 368456 : FD_ATOMIC_FETCH_AND_AND( &accmeta->executable_size, ~FD_ACCDB_SIZE_CACHE_VALID_BIT );
1275 368456 : FD_VOLATILE( accmeta->cache_idx ) = FD_ACCDB_ACC_CIDX_INVAL;
1276 368456 : }
1277 :
1278 : /* Release CLAIM. */
1279 368360 : FD_ATOMIC_FETCH_AND_AND( &accmeta->executable_size, ~FD_ACCDB_SIZE_CACHE_CLAIM_BIT );
1280 :
1281 368665 : if( FD_UNLIKELY( had_valid ) ) {
1282 368665 : fd_accdb_cache_line_t * stale = cache_line( accdb, FD_ACCDB_ACC_CIDX_CLASS( cidx ), FD_ACCDB_ACC_CIDX_IDX( cidx ) );
1283 368665 : fd_racesan_hook( "acc_unlink:pre_reclaim_cas" );
1284 368665 : uint old_rc = FD_ATOMIC_CAS( &stale->refcnt, 0U, FD_ACCDB_EVICT_SENTINEL );
1285 368665 : fd_racesan_hook( "acc_unlink:post_reclaim_cas" );
1286 368665 : if( FD_LIKELY( !old_rc ) ) {
1287 : /* Claimed. Validate key (ABA, slot could have been recycled
1288 : between our read of cache_idx and the CAS). */
1289 368556 : if( FD_LIKELY( stale->key.generation==accmeta->key.generation &&
1290 368556 : !memcmp( stale->key.pubkey, accmeta->key.pubkey, 32UL ) ) ) {
1291 368296 : ulong sc = FD_ACCDB_ACC_CIDX_CLASS( cidx );
1292 368296 : stale->key.generation = UINT_MAX;
1293 368296 : stale->persisted = 1;
1294 368296 : stale->acc_idx = UINT_MAX;
1295 368296 : FD_COMPILER_MFENCE();
1296 368296 : FD_VOLATILE( stale->refcnt ) = 0;
1297 368296 : cache_free_push( accdb, sc, stale );
1298 368296 : } else {
1299 : /* Wrong line (ABA). Release claim. */
1300 260 : FD_VOLATILE( stale->refcnt ) = 0;
1301 260 : }
1302 368556 : }
1303 109 : else if( FD_LIKELY( old_rc!=FD_ACCDB_EVICT_SENTINEL ) ) {
1304 : /* The CAS lost to a non-sentinel refcnt, but that does not prove
1305 : `stale` is still our line. Between capturing cidx and here we
1306 : released the claim, so we could have evicted `stale` and
1307 : recycled it to an unrelated account. */
1308 0 : fd_accdb_cache_line_t * mine = cache_try_pin( stale, accmeta->key.pubkey, accmeta->key.generation );
1309 0 : if( FD_LIKELY( mine ) ) {
1310 : /* Genuinely our line, still pinned by a reader. The accmeta
1311 : slot is about to be deferred-released and recycled; if a
1312 : later writeback of this dirty line fires, it would pair the
1313 : recycled accmeta's pubkey with the old owner/data. Set
1314 : persisted so the writeback gate never fires. */
1315 0 : FD_VOLATILE( mine->persisted ) = 1;
1316 :
1317 : /* Only the tombstone self-unlink may be pinned here old-version
1318 : and purge unlinks are never pinned, because a reader on a
1319 : live fork resolves to the newest version, not the one these
1320 : unlink. */
1321 0 : FD_TEST( accmeta->lamports==0UL );
1322 :
1323 0 : FD_ATOMIC_FETCH_AND_SUB( &mine->refcnt, 1U );
1324 0 : }
1325 : /* Else was recycled to a foreign account. Nothing to neutralize,
1326 : leave the line alone. */
1327 109 : } else {
1328 : /* A foreground evictor already claimed this line. It holds its
1329 : epoch acquire and writeback, so drain_deferred_frees cannot
1330 : recycle the slot before it finishes. Its writeback names the
1331 : old account correctly, no poison. */
1332 109 : }
1333 368665 : }
1334 368360 : }
1335 :
1336 : /* fork_slot_defer removes fork_id from every descends_set and chains
1337 : the fork pool slot onto the deferred fork chain for later release.
1338 : The slot must not be released immediately because concurrent readers
1339 : may still reference the fork ID via descends_set or stale chain
1340 : walks.
1341 :
1342 : The eager descends_set_remove here is safe despite being a
1343 : non-atomic RMW that races with concurrent descends_set_test in
1344 : fd_accdb_acquire, for two reasons:
1345 :
1346 : (a) Rooted parent forks: after advance_root publishes the new
1347 : root_fork_id, any acquire loads root_generation >=
1348 : parent->generation. Every account from the old parent has
1349 : generation <= parent->generation, so the
1350 : "generation > root_generation" gate in the chain walk is
1351 : never satisfied and the parent's bit is never tested.
1352 :
1353 : (b) Purged / pruned sibling forks: a purged fork is by
1354 : definition not an ancestor of any live fork, so its bit
1355 : was never set in any live fork's descends_set. Clearing
1356 : it is a literal no-op.
1357 :
1358 : Fork-id ABA after slot reuse is also safe: the fork pool slot
1359 : is not released until drain_deferred_frees, which waits until
1360 : all epoch-protected readers have exited. On x86 (TSO), the
1361 : synchronization chain (T2: bit clear -> epoch FAA; reader:
1362 : epoch load -> epoch_slot store -> mfence -> bit read) guarantees
1363 : that any reader entering a new epoch section after the drain
1364 : will observe the cleared bit before the slot is recycled by
1365 : attach_child. */
1366 :
1367 : static inline void
1368 : fork_slot_defer( fd_accdb_t * accdb,
1369 : fd_accdb_fork_id_t fork_id,
1370 : fd_accdb_fork_shmem_t ** fork_head,
1371 64103 : fd_accdb_fork_shmem_t ** fork_tail ) {
1372 1087414 : for( ulong i=0UL; i<accdb->shmem->max_live_slots; i++ ) descends_set_remove( accdb->fork_pool[ i ].descends, fork_id.val );
1373 64103 : fd_accdb_fork_shmem_t * shmem = fork_pool_ele( accdb->fork_shmem_pool, (ulong)fork_id.val );
1374 64103 : if( *fork_tail ) (*fork_tail)->pool.next = fork_pool_private_cidx( (ulong)fork_id.val );
1375 64103 : else *fork_head = shmem;
1376 64103 : *fork_tail = shmem;
1377 64103 : }
1378 :
1379 : /* chain_prewalk prefetches the hash chain walks of up to CHAIN_PREWALK_W
1380 : txn records from txn in lock-step, over the span the caller will walk.
1381 : Read only. Fills txns[] and map_idxs[] and returns the count. */
1382 :
1383 791458 : #define CHAIN_PREWALK_W (16UL)
1384 :
1385 : static ulong
1386 : chain_prewalk( fd_accdb_t * accdb,
1387 : uint txn,
1388 : int purge,
1389 : fd_accdb_txn_t ** txns,
1390 27686 : uint * map_idxs ) {
1391 27686 : uint acc_idx [ CHAIN_PREWALK_W ];
1392 27686 : uint cur [ CHAIN_PREWALK_W ];
1393 27686 : int from_head[ CHAIN_PREWALK_W ];
1394 :
1395 27686 : ulong cnt = 0UL;
1396 395729 : for( ; cnt<CHAIN_PREWALK_W && txn!=UINT_MAX; cnt++ ) {
1397 368043 : fd_accdb_txn_t * txne = txn_pool_ele( accdb->txn_pool, (ulong)txn );
1398 368043 : txns [ cnt ] = txne;
1399 368043 : acc_idx[ cnt ] = txne->acc_pool_idx;
1400 368043 : __builtin_prefetch( &accdb->acc_pool[ txne->acc_pool_idx ], 0, 3 );
1401 368043 : txn = txne->fork.next;
1402 368043 : }
1403 395561 : for( ulong i=0UL; i<cnt; i++ ) {
1404 367875 : fd_accdb_accmeta_t const * acc = &accdb->acc_pool[ acc_idx[ i ] ];
1405 367875 : map_idxs [ i ] = (uint)(fd_hash32( acc->key.pubkey, accdb->shmem->seed ) & (accdb->shmem->chain_cnt-1UL));
1406 367875 : from_head[ i ] = purge || acc->lamports==0UL;
1407 367875 : if( from_head[ i ] ) __builtin_prefetch( &accdb->acc_map[ map_idxs[ i ] ], 1, 3 );
1408 367875 : }
1409 396272 : for( ulong i=0UL; i<cnt; i++ ) {
1410 368586 : if( from_head[ i ] ) cur[ i ] = FD_VOLATILE_CONST( accdb->acc_map[ map_idxs[ i ] ] );
1411 0 : else cur[ i ] = FD_VOLATILE_CONST( accdb->acc_pool[ acc_idx[ i ] ].map.next );
1412 368586 : }
1413 :
1414 30316 : for(;;) {
1415 30316 : ulong live = 0UL;
1416 439546 : for( ulong i=0UL; i<cnt; i++ ) {
1417 409230 : if( cur[ i ]==UINT_MAX || ( purge && cur[ i ]==acc_idx[ i ] ) ) continue;
1418 2725 : cur[ i ] = FD_VOLATILE_CONST( accdb->acc_pool[ cur[ i ] ].map.next );
1419 2725 : live++;
1420 2725 : }
1421 30316 : if( !live ) break;
1422 30316 : }
1423 27686 : return cnt;
1424 27686 : }
1425 :
1426 : static void
1427 : purge_inner( fd_accdb_t * accdb,
1428 : fd_accdb_fork_id_t fork_id,
1429 : fd_accdb_fork_shmem_t ** fork_head,
1430 64119 : fd_accdb_fork_shmem_t ** fork_tail ) {
1431 64119 : fd_accdb_fork_t * fork = &accdb->fork_pool[ fork_id.val ];
1432 :
1433 64119 : fd_accdb_fork_id_t child = fork->shmem->child_id;
1434 64119 : while( child.val!=USHORT_MAX ) {
1435 0 : fd_accdb_fork_id_t next = accdb->fork_pool[ child.val ].shmem->sibling_id;
1436 0 : purge_inner( accdb, child, fork_head, fork_tail );
1437 0 : child = next;
1438 0 : }
1439 :
1440 64119 : uint txn = fork->shmem->txn_head;
1441 64119 : if( txn!=UINT_MAX ) {
1442 8340 : fd_accdb_txn_t * txn_head = txn_pool_ele( accdb->txn_pool, (ulong)txn );
1443 8340 : fd_accdb_txn_t * txn_tail = NULL;
1444 36037 : while( txn!=UINT_MAX ) {
1445 27697 : fd_accdb_txn_t * txns [ CHAIN_PREWALK_W ];
1446 27697 : uint map_idxs[ CHAIN_PREWALK_W ];
1447 27697 : ulong cnt = chain_prewalk( accdb, txn, 1, txns, map_idxs );
1448 396216 : for( ulong w=0UL; w<cnt; w++ ) {
1449 368519 : fd_accdb_txn_t * txne = txns[ w ];
1450 :
1451 368519 : uint acc_idx = txne->acc_pool_idx;
1452 368519 : uint acc_map_idx = map_idxs[ w ];
1453 :
1454 368519 : uint prev = UINT_MAX;
1455 368519 : uint cur = FD_VOLATILE_CONST( accdb->acc_map[ acc_map_idx ] );
1456 368519 : while( cur!=acc_idx ) {
1457 0 : prev = cur;
1458 0 : cur = FD_VOLATILE_CONST( accdb->acc_pool[ cur ].map.next );
1459 0 : }
1460 :
1461 368519 : fd_racesan_hook( "accdb_purge:pre_unlink" );
1462 368519 : acc_unlink( accdb, acc_map_idx, prev, acc_idx );
1463 368519 : deferred_acc_append( accdb, acc_idx );
1464 :
1465 368519 : txn_tail = txne;
1466 368519 : txn = txne->fork.next;
1467 368519 : }
1468 27697 : }
1469 8340 : txn_pool_release_chain( accdb->txn_pool, txn_head, txn_tail );
1470 8340 : }
1471 :
1472 64119 : fork_slot_defer( accdb, fork_id, fork_head, fork_tail );
1473 64119 : }
1474 :
1475 : static inline void
1476 : remove_children( fd_accdb_t * accdb,
1477 : fd_accdb_fork_t * fork,
1478 : fd_accdb_fork_t * except,
1479 : fd_accdb_fork_shmem_t ** fork_head,
1480 0 : fd_accdb_fork_shmem_t ** fork_tail ) {
1481 0 : fd_accdb_fork_id_t sibling_idx = fork->shmem->child_id;
1482 0 : while( sibling_idx.val!=USHORT_MAX ) {
1483 0 : fd_accdb_fork_t * sibling = &accdb->fork_pool[ sibling_idx.val ];
1484 0 : fd_accdb_fork_id_t cur_idx = sibling_idx;
1485 :
1486 0 : sibling_idx = sibling->shmem->sibling_id;
1487 0 : if( FD_UNLIKELY( sibling==except ) ) continue;
1488 :
1489 0 : purge_inner( accdb, cur_idx, fork_head, fork_tail );
1490 0 : }
1491 0 : }
1492 :
1493 : static void
1494 : background_advance_root( fd_accdb_t * accdb,
1495 0 : fd_accdb_fork_id_t fork_id ) {
1496 0 : drain_deferred_frees( accdb );
1497 :
1498 : /* The caller guarantees that rooting is sequential: each call
1499 : advances the root by exactly one slot (the immediate child of the
1500 : current root). Skipping levels is not supported. */
1501 0 : fd_accdb_fork_t * fork = &accdb->fork_pool[ fork_id.val ];
1502 0 : FD_TEST( fork->shmem->parent_id.val==accdb->shmem->root_fork_id.val );
1503 0 : FD_TEST( fork->shmem->parent_id.val!=USHORT_MAX );
1504 :
1505 0 : fd_accdb_fork_t * parent_fork = &accdb->fork_pool[ fork->shmem->parent_id.val ];
1506 :
1507 : /* Accumulate freed fork pool slots across remove_children and the
1508 : old-version cleanup below into a chain that will be deferred-
1509 : released after the epoch bump. Freed acc pool slots are recorded
1510 : in the shmem side buffer via deferred_acc_append (they cannot be
1511 : chained via pool.next yet — see comment on the side buffer). */
1512 0 : fd_accdb_fork_shmem_t * fork_head = NULL;
1513 0 : fd_accdb_fork_shmem_t * fork_tail = NULL;
1514 :
1515 : /* When a fork is rooted, any competing forks can be immediately
1516 : removed as they will not be needed again. This includes child
1517 : forks of the pruned siblings as well. */
1518 0 : remove_children( accdb, parent_fork, fork, &fork_head, &fork_tail );
1519 :
1520 : /* And for any accounts which were updated in the newly rooted slot,
1521 : we will now never need to access any older version, so we can
1522 : discard any slots earlier than the one we are rooting. */
1523 0 : uint txn = fork->shmem->txn_head;
1524 0 : if( txn!=UINT_MAX ) {
1525 0 : fd_accdb_txn_t * txn_head = txn_pool_ele( accdb->txn_pool, (ulong)txn );
1526 0 : fd_accdb_txn_t * txn_tail = NULL;
1527 0 : while( txn!=UINT_MAX ) {
1528 0 : fd_accdb_txn_t * txns [ CHAIN_PREWALK_W ];
1529 0 : uint map_idxs[ CHAIN_PREWALK_W ];
1530 0 : ulong cnt = chain_prewalk( accdb, txn, 0, txns, map_idxs );
1531 0 : for( ulong w=0UL; w<cnt; w++ ) {
1532 0 : fd_accdb_txn_t * txne = txns[ w ];
1533 :
1534 0 : fd_accdb_accmeta_t const * new_acc = &accdb->acc_pool[ txne->acc_pool_idx ];
1535 0 : uint acc_map_idx = map_idxs[ w ];
1536 :
1537 0 : delta_insert( accdb, new_acc->key.pubkey );
1538 :
1539 : /* Walk from new_acc, not the head: everything ahead of it was
1540 : prepended by a non-ancestor fork. A tombstone needs its
1541 : predecessor so still walks from the head. */
1542 0 : uint prev = UINT_MAX;
1543 0 : uint new_acc_prev = UINT_MAX; /* prev of new_acc on the chain when we encounter it (UINT_MAX if head or never seen) */
1544 0 : int new_acc_seen = 0;
1545 0 : uint acc;
1546 0 : FD_TEST( new_acc->key.generation==fork->shmem->generation && fd_accdb_acc_fork_id( new_acc )==fork_id.val );
1547 0 : if( FD_LIKELY( new_acc->lamports ) ) {
1548 : #if FD_TMPL_USE_HANDHOLDING
1549 : uint chk = FD_VOLATILE_CONST( accdb->acc_map[ acc_map_idx ] );
1550 : while( chk!=UINT_MAX && chk!=txne->acc_pool_idx ) {
1551 : fd_accdb_accmeta_t const * ahead = &accdb->acc_pool[ chk ];
1552 : FD_TEST( !( (ahead->key.generation<=parent_fork->shmem->generation || descends_set_test( fork->descends, fd_accdb_acc_fork_id(ahead) ) ) && !memcmp( new_acc->key.pubkey, ahead->key.pubkey, 32UL ) ) );
1553 : chk = FD_VOLATILE_CONST( ahead->map.next );
1554 : }
1555 : FD_TEST( chk==txne->acc_pool_idx );
1556 : #endif
1557 0 : new_acc_seen = 1;
1558 0 : prev = txne->acc_pool_idx;
1559 0 : acc = FD_VOLATILE_CONST( new_acc->map.next );
1560 0 : } else {
1561 0 : acc = FD_VOLATILE_CONST( accdb->acc_map[ acc_map_idx ] );
1562 0 : FD_TEST( acc!=UINT_MAX );
1563 0 : }
1564 0 : while( acc!=UINT_MAX ) {
1565 0 : if( FD_UNLIKELY( acc==txne->acc_pool_idx ) ) {
1566 0 : new_acc_prev = prev;
1567 0 : new_acc_seen = 1;
1568 0 : prev = acc;
1569 0 : acc = FD_VOLATILE_CONST( new_acc->map.next );
1570 0 : continue;
1571 0 : }
1572 :
1573 0 : fd_accdb_accmeta_t const * cur_acc = &accdb->acc_pool[ acc ];
1574 0 : uint cur_next = FD_VOLATILE_CONST( cur_acc->map.next );
1575 :
1576 0 : if( FD_LIKELY( (cur_acc->key.generation<=parent_fork->shmem->generation || descends_set_test( fork->descends, fd_accdb_acc_fork_id(cur_acc) ) ) && !memcmp( new_acc->key.pubkey, cur_acc->key.pubkey, 32UL ) ) ) {
1577 0 : uint next = cur_next;
1578 0 : fd_racesan_hook( "accdb_advance:pre_unlink" );
1579 0 : acc_unlink( accdb, acc_map_idx, prev, acc );
1580 0 : deferred_acc_append( accdb, acc );
1581 0 : acc = next;
1582 0 : } else {
1583 0 : prev = acc;
1584 0 : acc = cur_next;
1585 0 : }
1586 0 : }
1587 :
1588 : /* If the newly rooted version is a tombstone (lamports==0, e.g.
1589 : account was closed), drop it from the index too: no fork can
1590 : reach it anymore, and keeping it around just wastes a hash
1591 : slot and the disk bytes it occupies. */
1592 0 : if( FD_UNLIKELY( new_acc_seen && new_acc->lamports==0UL ) ) {
1593 0 : uint new_acc_idx = (uint)txne->acc_pool_idx;
1594 0 : acc_unlink( accdb, acc_map_idx, new_acc_prev, new_acc_idx );
1595 0 : deferred_acc_append( accdb, new_acc_idx );
1596 0 : }
1597 :
1598 0 : txn_tail = txne;
1599 0 : txn = txne->fork.next;
1600 0 : }
1601 0 : }
1602 0 : txn_pool_release_chain( accdb->txn_pool, txn_head, txn_tail );
1603 0 : }
1604 :
1605 0 : uint parent_txn = parent_fork->shmem->txn_head;
1606 0 : if( parent_txn!=UINT_MAX ) {
1607 0 : fd_accdb_txn_t * parent_head = txn_pool_ele( accdb->txn_pool, (ulong)parent_txn );
1608 0 : fd_accdb_txn_t * parent_tail = NULL;
1609 0 : while( parent_txn!=UINT_MAX ) {
1610 0 : fd_accdb_txn_t * t = txn_pool_ele( accdb->txn_pool, (ulong)parent_txn );
1611 0 : parent_tail = t;
1612 0 : parent_txn = t->fork.next;
1613 0 : }
1614 0 : txn_pool_release_chain( accdb->txn_pool, parent_head, parent_tail );
1615 0 : }
1616 :
1617 : /* Remove the parent from all descends_sets and chain it for deferred
1618 : release, so that when the slot is eventually recycled to a new
1619 : fork, no concurrent reader can mistake the new fork for the old
1620 : ancestor. Entries from the freed parent are still visible via the
1621 : generation <= root_generation fast path in reads. */
1622 0 : fd_accdb_fork_id_t old_parent_id = fork->shmem->parent_id;
1623 0 : fork_slot_defer( accdb, old_parent_id, &fork_head, &fork_tail );
1624 :
1625 0 : fork->shmem->parent_id = (fd_accdb_fork_id_t){ .val = USHORT_MAX };
1626 0 : fork->shmem->sibling_id = (fd_accdb_fork_id_t){ .val = USHORT_MAX };
1627 0 : fork->shmem->txn_head = UINT_MAX;
1628 0 : descends_set_null( fork->descends );
1629 :
1630 : /* Publish the new root_fork_id BEFORE bumping the epoch and deferring
1631 : the parent slot. On x86-64 (TSO) a concurrent reader that still
1632 : loads the old root_fork_id is guaranteed to see the parent shmem in
1633 : its original (not-yet-recycled) state because the slot has not been
1634 : released yet. A reader that loads the new root_fork_id uses the
1635 : new fork. */
1636 0 : fd_racesan_hook( "accdb_advance:pre_publish_root" );
1637 0 : accdb->shmem->root_fork_id = fork_id;
1638 0 : FD_COMPILER_MFENCE();
1639 0 : fd_racesan_hook( "accdb_advance:post_publish_root" );
1640 :
1641 : /* Bump epoch and defer both the acc batch and parent fork slot. They
1642 : will be released at the next drain_deferred_frees call once all
1643 : concurrent readers have exited. The acc batch lives in the shmem
1644 : side buffer; only its epoch tag needs setting here. */
1645 0 : ulong tag = FD_ATOMIC_FETCH_AND_ADD( &accdb->shmem->epoch, 1UL );
1646 0 : if( FD_LIKELY( accdb->shmem->deferred_acc_buf_cnt ) ) {
1647 0 : accdb->shmem->deferred_acc_epoch = tag;
1648 0 : }
1649 0 : if( FD_LIKELY( fork_head ) ) {
1650 0 : accdb->deferred_fork_head = fork_head;
1651 0 : accdb->deferred_fork_tail = fork_tail;
1652 0 : accdb->deferred_fork_epoch = tag;
1653 0 : }
1654 0 : }
1655 :
1656 : int
1657 0 : fd_accdb_cmd_pending( fd_accdb_t const * accdb ) {
1658 0 : return FD_VOLATILE_CONST( accdb->shmem->cmd_op )!=FD_ACCDB_CMD_IDLE;
1659 0 : }
1660 :
1661 : void
1662 : fd_accdb_advance_root( fd_accdb_t * accdb,
1663 0 : fd_accdb_fork_id_t fork_id ) {
1664 0 : FD_CHECK_CRIT( fd_accdb_snapshot_sync_state( &accdb->shmem->snapshot_sync )!=FD_ACCDB_SNAPSHOT_SYNC_RUNNING,
1665 0 : "fd_accdb_advance_root called during snapshot production" );
1666 0 : wait_cmd( accdb );
1667 0 : submit_cmd( accdb, FD_ACCDB_CMD_ADVANCE_ROOT, fork_id.val );
1668 0 : }
1669 :
1670 : /* background_purge does the heavy lifting of purge on T2: unlink the
1671 : fork from the parent's child list, drain deferred frees, recursively
1672 : purge the fork subtree, and defer-release the freed acc pool
1673 : elements. The sibling-list unlink is done here (not on T1) because
1674 : advance_root / remove_children also mutate sibling lists on T2, and
1675 : T2 is single-threaded so plain stores are safe. */
1676 :
1677 : static void
1678 : background_purge( fd_accdb_t * accdb,
1679 64059 : fd_accdb_fork_id_t fork_id ) {
1680 : /* Unlink fork_id from its parent's child list. This runs on T2
1681 : which is the sole mutator of sibling lists (advance_root and
1682 : remove_children also run on T2), so plain stores are safe. */
1683 64059 : fd_accdb_fork_t * fork = &accdb->fork_pool[ fork_id.val ];
1684 64059 : fd_accdb_fork_id_t parent_id = fork->shmem->parent_id;
1685 64059 : if( FD_LIKELY( parent_id.val!=USHORT_MAX ) ) {
1686 64056 : fd_accdb_fork_t * parent = &accdb->fork_pool[ parent_id.val ];
1687 64058 : if( FD_UNLIKELY( parent->shmem->child_id.val==fork_id.val ) ) {
1688 64058 : parent->shmem->child_id = fork->shmem->sibling_id;
1689 >1844*10^16 : } else {
1690 >1844*10^16 : fd_accdb_fork_id_t prev_id = parent->shmem->child_id;
1691 >1844*10^16 : while( prev_id.val!=USHORT_MAX ) {
1692 0 : fd_accdb_fork_t * prev = &accdb->fork_pool[ prev_id.val ];
1693 0 : if( prev->shmem->sibling_id.val==fork_id.val ) {
1694 0 : prev->shmem->sibling_id = fork->shmem->sibling_id;
1695 0 : break;
1696 0 : }
1697 0 : prev_id = prev->shmem->sibling_id;
1698 0 : }
1699 >1844*10^16 : }
1700 64056 : }
1701 :
1702 64059 : drain_deferred_frees( accdb );
1703 :
1704 64059 : fd_accdb_fork_shmem_t * fork_head = NULL;
1705 64059 : fd_accdb_fork_shmem_t * fork_tail = NULL;
1706 64059 : purge_inner( accdb, fork_id, &fork_head, &fork_tail );
1707 :
1708 64059 : ulong tag = FD_ATOMIC_FETCH_AND_ADD( &accdb->shmem->epoch, 1UL );
1709 64059 : if( FD_LIKELY( accdb->shmem->deferred_acc_buf_cnt ) ) {
1710 8340 : accdb->shmem->deferred_acc_epoch = tag;
1711 8340 : }
1712 64134 : if( FD_LIKELY( fork_head ) ) {
1713 64134 : accdb->deferred_fork_head = fork_head;
1714 64134 : accdb->deferred_fork_tail = fork_tail;
1715 64134 : accdb->deferred_fork_epoch = tag;
1716 64134 : }
1717 64059 : }
1718 :
1719 : void
1720 : fd_accdb_purge( fd_accdb_t * accdb,
1721 64040 : fd_accdb_fork_id_t fork_id ) {
1722 64040 : FD_TEST( fork_id.val!=accdb->shmem->root_fork_id.val );
1723 :
1724 64040 : wait_cmd( accdb );
1725 64040 : submit_cmd( accdb, FD_ACCDB_CMD_PURGE, fork_id.val );
1726 64040 : }
1727 :
1728 : static inline fd_accdb_cache_line_t *
1729 : acquire_cache_line( fd_accdb_t * accdb,
1730 : ulong size_class,
1731 3230345 : uint * out_evicted_acc_idx ) {
1732 : /* Priority 1: CAS free list — already invalidated,
1733 : persisted==1, generation==UINT_MAX. Cheapest path. */
1734 3230345 : fd_accdb_cache_line_t * result = cache_free_pop( accdb, size_class );
1735 3249893 : if( FD_LIKELY( result ) ) {
1736 3249893 : while( FD_UNLIKELY( FD_ATOMIC_CAS( &result->refcnt, 0U, 1U )!=0U ) ) {
1737 0 : fd_racesan_hook( "accdb_freepop:refcnt_wait" );
1738 0 : FD_SPIN_PAUSE();
1739 0 : }
1740 3249893 : result->referenced = 0;
1741 3249893 : FD_VOLATILE( result->seq ) = result->seq+1U;
1742 3249893 : FD_COMPILER_MFENCE();
1743 3249893 : *out_evicted_acc_idx = UINT_MAX;
1744 3249893 : return result;
1745 3249893 : }
1746 :
1747 : /* Priority 2: Lazy initial allocation — atomic FAA with undo on
1748 : overflow. Safe for concurrent callers. */
1749 >1844*10^16 : ulong old_init = FD_ATOMIC_FETCH_AND_ADD( &accdb->shmem->cache_class_init[ size_class ].val, 1UL );
1750 >1844*10^16 : if( FD_LIKELY( old_init<accdb->shmem->cache_class_max[ size_class ] ) ) {
1751 5970 : result = cache_line( accdb, size_class, old_init );
1752 5970 : result->refcnt = 1;
1753 5970 : result->persisted = 1;
1754 5970 : result->referenced = 0;
1755 5970 : result->acc_idx = UINT_MAX;
1756 5970 : result->key.generation = UINT_MAX;
1757 5970 : *out_evicted_acc_idx = UINT_MAX;
1758 5970 : return result;
1759 5970 : }
1760 >1844*10^16 : FD_ATOMIC_FETCH_AND_SUB( &accdb->shmem->cache_class_init[ size_class ].val, 1UL );
1761 :
1762 : /* Priority 3: CLOCK sweep ... scan forward giving second chances. */
1763 >1844*10^16 : for(;;) {
1764 0 : ulong hand = FD_ATOMIC_FETCH_AND_ADD( &accdb->shmem->clock_hand[ size_class ].val, 1UL ) % accdb->shmem->cache_class_max[ size_class ];
1765 0 : fd_accdb_cache_line_t * line = cache_line( accdb, size_class, hand );
1766 :
1767 0 : if( FD_UNLIKELY( line->key.generation==UINT_MAX && line->acc_idx==UINT_MAX ) ) continue;
1768 :
1769 0 : fd_racesan_hook( "accdb_clock:pre_refcnt" );
1770 0 : uint rc = FD_VOLATILE_CONST( line->refcnt );
1771 0 : if( FD_UNLIKELY( rc!=0U ) ) continue; /* Pinned or being evicted */
1772 :
1773 0 : if( FD_UNLIKELY( line->referenced ) ) {
1774 0 : line->referenced = 0;
1775 0 : continue; /* Second chance */
1776 0 : }
1777 :
1778 0 : if( FD_UNLIKELY( FD_ATOMIC_CAS( &line->refcnt, 0U, FD_ACCDB_EVICT_SENTINEL )!=0U ) ) continue;
1779 :
1780 0 : if( FD_UNLIKELY( line->acc_idx==UINT_MAX && line->key.generation==UINT_MAX ) ) {
1781 0 : FD_VOLATILE( line->refcnt ) = 0;
1782 0 : continue;
1783 0 : }
1784 :
1785 : /* The line is now claimed for eviction (refcnt==EVICT_SENTINEL). A
1786 : concurrent acc_unlink that targets this same line's accmeta will
1787 : observe the sentinel here and take its do-nothing branch — see the
1788 : test_accdb_racesan SENTINEL case. */
1789 0 : fd_racesan_hook( "clock_evict:post_sentinel" );
1790 :
1791 0 : if( FD_LIKELY( line->acc_idx!=UINT_MAX ) ) {
1792 0 : evict_clear_acc_cache_ref( &accdb->acc_pool[ line->acc_idx ], size_class, hand );
1793 0 : }
1794 0 : *out_evicted_acc_idx = line->persisted ? UINT_MAX : line->acc_idx;
1795 0 : line->key.generation = UINT_MAX;
1796 0 : FD_VOLATILE( line->seq ) = line->seq+1U;
1797 0 : FD_COMPILER_MFENCE();
1798 0 : line->refcnt = 1;
1799 0 : line->referenced = 0;
1800 0 : return line;
1801 0 : }
1802 :
1803 >1844*10^16 : FD_TEST( 0 );
1804 >1844*10^16 : return NULL;
1805 >1844*10^16 : }
1806 :
1807 : static int
1808 : grow_locked( fd_accdb_t * accdb,
1809 : ulong partition_idx,
1810 : uchar layer,
1811 : ulong prior_partition_idx,
1812 0 : int preallocated ) {
1813 0 : for( ulong p=accdb->shmem->partition_fallocated; p<=partition_idx; p++ ) {
1814 0 : FD_LOG_INFO(( "growing accounts database from %lu GiB to %lu GiB", p*accdb->shmem->partition_sz/(1UL<<30UL), (p+1UL)*accdb->shmem->partition_sz/(1UL<<30UL) ));
1815 :
1816 0 : long fallocate_nanos = -fd_log_wallclock();
1817 0 : int result = fallocate( accdb->fd, 0, (long)(p*accdb->shmem->partition_sz), (long)accdb->shmem->partition_sz );
1818 0 : fallocate_nanos += fd_log_wallclock();
1819 0 : if( FD_UNLIKELY( -1==result ) ) {
1820 0 : if( FD_LIKELY( preallocated ) ) return errno;
1821 0 : if( FD_LIKELY( errno==ENOSPC ) ) FD_LOG_ERR(( "fallocate() failed (%d-%s). The accounts database filled "
1822 0 : "the disk it is on, trying to grow from %lu GiB to %lu GiB. Please "
1823 0 : "free up disk space and restart the validator.",
1824 0 : errno, fd_io_strerror( errno ), p*accdb->shmem->partition_sz/(1UL<<30UL), (p+1UL)*accdb->shmem->partition_sz/(1UL<<30UL) ));
1825 0 : else FD_LOG_ERR(( "fallocate() failed (%d-%s)", errno, fd_io_strerror( errno ) ));
1826 0 : }
1827 :
1828 0 : FD_VOLATILE( accdb->shmem->partition_fallocated ) = p+1UL;
1829 0 : accdb->shmem->shmetrics->disk_allocated_bytes = (p+1UL)*accdb->shmem->partition_sz;
1830 :
1831 0 : fd_event_accdb_partition_added_t ev = {
1832 0 : .partition_idx = p,
1833 0 : .prior_partition_idx = p==partition_idx ? prior_partition_idx : ULONG_MAX,
1834 0 : .layer = layer,
1835 0 : .old_partition_max = p,
1836 0 : .new_partition_max = p+1UL,
1837 0 : .partition_sz = accdb->shmem->partition_sz,
1838 0 : .disk_allocated_bytes = (p+1UL)*accdb->shmem->partition_sz,
1839 0 : .duration_fallocate_nanos = (ulong)fd_long_max( fallocate_nanos, 0L ),
1840 0 : .preallocated = preallocated,
1841 0 : };
1842 0 : fd_event_report_accdb_partition_added( &ev );
1843 0 : }
1844 0 : return 0;
1845 0 : }
1846 :
1847 : static inline void
1848 : change_partition( fd_accdb_t * accdb,
1849 : accdb_offset_t const * offset_before,
1850 : accdb_offset_t * out_offset,
1851 : int * has_partition,
1852 0 : uchar layer ) {
1853 : /* New data will not fit in the current partition, so we need to
1854 : move to the next one. */
1855 0 : spin_lock_acquire( &accdb->shmem->partition_lock );
1856 0 : ulong partition_idx_before = packed_partition_idx( offset_before );
1857 0 : ulong partition_offset_before = packed_partition_offset( offset_before );
1858 0 : if( FD_LIKELY( *has_partition ) ) {
1859 0 : fd_accdb_partition_t * before = partition_pool_ele( accdb->partition_pool, partition_idx_before );
1860 0 : before->write_offset = partition_offset_before;
1861 0 : }
1862 :
1863 : /* Single rdtsc per partition lifecycle event: stamp the closing
1864 : partition's filled time and the new partition's created time off
1865 : the same sample. */
1866 0 : long now_ticks = (long)fd_tickcount();
1867 :
1868 0 : ulong free_size = accdb->shmem->partition_sz - partition_offset_before;
1869 0 : if( FD_LIKELY( *has_partition ) ) {
1870 0 : fd_accdb_partition_t * old = partition_pool_ele( accdb->partition_pool, partition_idx_before );
1871 0 : FD_ATOMIC_FETCH_AND_ADD( &old->bytes_freed, free_size );
1872 0 : FD_VOLATILE( old->filled_ticks ) = now_ticks;
1873 : /* The tail slack is now committed dead — count it as current
1874 : (written-through) so fragmentation reflects it. */
1875 0 : FD_ATOMIC_FETCH_AND_ADD( &accdb->shmem->shmetrics->disk_current_bytes, free_size );
1876 0 : }
1877 :
1878 0 : if( FD_UNLIKELY( !partition_pool_free( accdb->partition_pool ) ) ) FD_LOG_ERR(( "accounts database file is at capacity" ));
1879 0 : fd_accdb_partition_t * partition = partition_pool_ele_acquire( accdb->partition_pool );
1880 0 : partition->bytes_freed = 0UL;
1881 0 : partition->marked_compaction = 0;
1882 0 : partition->layer = layer;
1883 0 : partition->read_ops = 0UL;
1884 0 : partition->bytes_read = 0UL;
1885 0 : partition->write_ops = 0UL;
1886 0 : partition->bytes_written = 0UL;
1887 0 : partition->write_offset = 0UL;
1888 0 : partition->compaction_offset = 0UL;
1889 0 : partition->created_ticks = now_ticks;
1890 0 : partition->filled_ticks = 0L;
1891 0 : partition->queued = 0;
1892 0 : partition->compacting_now = 0;
1893 :
1894 0 : ulong new_partition_idx = partition_pool_idx( accdb->partition_pool, partition );
1895 0 : int had_partition = *has_partition;
1896 :
1897 : /* Snapshot-load tiering: accounts loaded from a snapshot never get
1898 : a second write, so compaction-driven promotion never fires and
1899 : they would otherwise live in Hot forever. When snapshot_loading
1900 : is set, tag the new partition as Cold up front. We do not set
1901 : has_partition[Cold] / whead[Cold] — those are owned by the
1902 : compaction tile and represent the live Cold write head, which is
1903 : independent of snapshot-loaded partitions that happen to be
1904 : labeled Cold. */
1905 0 : if( FD_UNLIKELY( FD_VOLATILE_CONST( accdb->shmem->snapshot_loading ) && layer==0 ) ) {
1906 0 : FD_VOLATILE( partition->layer ) = FD_ACCDB_COMPACTION_LAYER_CNT-1UL;
1907 0 : }
1908 :
1909 0 : if( FD_UNLIKELY( new_partition_idx>=accdb->shmem->partition_max ) ) FD_VOLATILE( accdb->shmem->partition_max ) = new_partition_idx+1UL;
1910 0 : spin_lock_release( &accdb->shmem->partition_lock );
1911 :
1912 : /* Back the partition before publishing the head, so no writer
1913 : reserves into it while fallocate is running. Layer-0 writers spin
1914 : on the head until then. */
1915 0 : if( FD_UNLIKELY( new_partition_idx>=FD_VOLATILE_CONST( accdb->shmem->partition_fallocated ) ) ) {
1916 0 : spin_lock_acquire( &accdb->shmem->grow_lock );
1917 0 : grow_locked( accdb, new_partition_idx, layer, had_partition ? partition_idx_before : ULONG_MAX, 0 );
1918 0 : spin_lock_release( &accdb->shmem->grow_lock );
1919 0 : }
1920 :
1921 0 : spin_lock_acquire( &accdb->shmem->partition_lock );
1922 0 : *out_offset = accdb_offset( new_partition_idx, 0UL );
1923 0 : FD_COMPILER_MFENCE();
1924 0 : *has_partition = 1;
1925 0 : if( FD_LIKELY( had_partition && partition_idx_before!=new_partition_idx ) ) {
1926 0 : fd_accdb_shmem_try_enqueue_compaction( accdb->shmem, partition_idx_before );
1927 0 : }
1928 0 : spin_lock_release( &accdb->shmem->partition_lock );
1929 0 : }
1930 :
1931 : /* Reserve sz bytes in the layer-0 write head and set
1932 : out_partition_idx to where they landed. Bumps no counters. */
1933 :
1934 : static inline ulong
1935 : reserve_next_write( fd_accdb_t * accdb,
1936 : ulong sz,
1937 0 : ulong * out_partition_idx ) {
1938 0 : for(;;) {
1939 0 : accdb_offset_t offset = { .val = FD_ATOMIC_FETCH_AND_ADD( &accdb->shmem->whead[ 0 ].val, sz ) };
1940 0 : if( FD_LIKELY( packed_partition_offset( &offset )+sz<=accdb->shmem->partition_sz ) ) {
1941 0 : *out_partition_idx = packed_partition_idx( &offset );
1942 0 : return packed_partition_file_offset( &offset, accdb->shmem->partition_sz );
1943 0 : }
1944 :
1945 0 : if( FD_UNLIKELY( packed_partition_offset( &offset )>accdb->shmem->partition_sz ) ) {
1946 : /* This can happen if another thread also raced to allocate the
1947 : next write and won. Wait for the partition switch to finish
1948 : before retrying, so we do not keep doing fetch-and-adds that
1949 : advance the offset further past the boundary.
1950 :
1951 : A switch is detected by the head moving to a different
1952 : partition index OR its offset dropping back to a valid position
1953 : (a switch resets the offset to 0). We must not key the wait
1954 : solely on the index changing: the initial write head is a
1955 : sentinel whose packed index can coincide with a real pool
1956 : index. */
1957 0 : ulong stale_partition = packed_partition_idx( &offset );
1958 0 : for(;;) {
1959 0 : accdb_offset_t cur = { .val = FD_VOLATILE_CONST( accdb->shmem->whead[ 0 ].val ) };
1960 0 : if( packed_partition_idx( &cur )!=stale_partition ) break;
1961 0 : if( packed_partition_offset( &cur )<=accdb->shmem->partition_sz ) break;
1962 0 : FD_SPIN_PAUSE();
1963 0 : }
1964 0 : continue;
1965 0 : }
1966 :
1967 0 : change_partition( accdb, &offset, &accdb->shmem->whead[ 0 ], &accdb->shmem->has_partition[ 0 ], 0 );
1968 0 : }
1969 0 : }
1970 :
1971 : /* Reserve sz bytes. Layer-0 write metrics are deferred until
1972 : explicitly flushed. */
1973 :
1974 : static inline ulong
1975 : allocate_next_write( fd_accdb_t * accdb,
1976 0 : ulong sz ) {
1977 0 : ulong partition_idx;
1978 0 : ulong file_offset = reserve_next_write( accdb, sz, &partition_idx );
1979 :
1980 : /* Very rarely the reservation crosses into a new partition. Since
1981 : stats are aggregated per-partition, any accumulated stats are
1982 : flushed and reset to accommodate the new partition. */
1983 0 : if( FD_LIKELY( accdb->write_stats.num_ops ) &&
1984 0 : FD_UNLIKELY( accdb->write_stats.partition_idx!=partition_idx ) ) {
1985 0 : fd_accdb_flush_metrics( accdb );
1986 0 : }
1987 :
1988 0 : accdb->write_stats.partition_idx = partition_idx;
1989 0 : accdb->write_stats.bytes += sz;
1990 0 : accdb->write_stats.num_ops++;
1991 0 : return file_offset;
1992 0 : }
1993 :
1994 : ulong
1995 : fd_accdb_snapshot_reserve_write( fd_accdb_t * accdb,
1996 : ulong sz,
1997 0 : ulong dead_tail_sz ) {
1998 0 : FD_TEST( sz && sz<=accdb->shmem->partition_sz && dead_tail_sz<=sz );
1999 0 : ulong off = allocate_next_write( accdb, sz );
2000 0 : if( dead_tail_sz ) fd_accdb_shmem_bytes_freed( accdb->shmem, off+sz-dead_tail_sz, dead_tail_sz );
2001 0 : return off;
2002 0 : }
2003 :
2004 : /* Compaction write allocation. Single-threaded: only the compaction
2005 : tile calls these, so the compaction write heads do not need atomic
2006 : fetch-and-add. dest_layer is the target layer (1..N-1). */
2007 :
2008 : static inline ulong
2009 : allocate_next_compaction_write( fd_accdb_t * accdb,
2010 : ulong sz,
2011 0 : ulong dest_layer ) {
2012 0 : accdb_offset_t offset = accdb->shmem->whead[ dest_layer ];
2013 0 : if( FD_UNLIKELY( !accdb->shmem->has_partition[ dest_layer ] ||
2014 0 : packed_partition_offset( &offset )+sz>accdb->shmem->partition_sz ) ) {
2015 0 : change_partition( accdb, &offset, &accdb->shmem->whead[ dest_layer ], &accdb->shmem->has_partition[ dest_layer ], (uchar)dest_layer );
2016 0 : offset = accdb->shmem->whead[ dest_layer ];
2017 0 : }
2018 0 : accdb->shmem->whead[ dest_layer ].val += sz;
2019 0 : FD_ATOMIC_FETCH_AND_ADD( &accdb->shmem->shmetrics->disk_current_bytes, sz );
2020 0 : ulong file_offset = packed_partition_file_offset( &offset, accdb->shmem->partition_sz );
2021 0 : fd_accdb_partition_write_bump( accdb, packed_partition_idx( &offset ), sz, 1UL );
2022 0 : return file_offset;
2023 0 : }
2024 :
2025 : /* fd_accdb_compact relocates one record from the oldest partition
2026 : queued for compaction at src_layer into the write head for the
2027 : next colder tier, or the same tier for the deepest layer. It is
2028 : designed to be called repeatedly from a dedicated compaction tile.
2029 : If there is work to do, *charge_busy is set to 1; otherwise 0 is
2030 : left unchanged and the call returns immediately.
2031 :
2032 : src_layer must be in 0..FD_ACCDB_COMPACTION_LAYER_CNT-1. */
2033 :
2034 : static void
2035 : background_compact( fd_accdb_t * accdb,
2036 : ulong src_layer,
2037 0 : int * charge_busy ) {
2038 0 : FD_TEST( accdb->bounce );
2039 :
2040 0 : FD_COMPILER_MFENCE();
2041 0 : FD_VOLATILE( *accdb->my_epoch_slot ) = FD_VOLATILE_CONST( accdb->shmem->epoch );
2042 0 : FD_HW_MFENCE(); /* StoreLoad: epoch store must be globally visible
2043 : before any subsequent loads so the deferred
2044 : reclamation scan does not miss us. */
2045 :
2046 : /* Reclaim any deferred-free partitions whose epoch has been observed
2047 : by all joiners (i.e. no epoch-publishing joiner could still be
2048 : referencing data in them). Scan writer slots [0, joiner_cnt)
2049 : plus each external (read-only) joiner's private epoch fseq. */
2050 0 : ulong min_epoch = ULONG_MAX;
2051 0 : ulong joiner_cnt = FD_VOLATILE_CONST( accdb->shmem->joiner_cnt );
2052 0 : for( ulong t=0UL; t<joiner_cnt; t++ ) {
2053 0 : ulong e = FD_VOLATILE_CONST( accdb->shmem->joiner_epochs[ t ].val );
2054 0 : if( FD_LIKELY( e<min_epoch ) ) min_epoch = e;
2055 0 : }
2056 0 : for( ulong t=0UL; t<accdb->external_epoch_cnt; t++ ) {
2057 0 : ulong e = FD_VOLATILE_CONST( *accdb->external_epoch_slots[ t ] );
2058 0 : if( FD_LIKELY( e<min_epoch ) ) min_epoch = e;
2059 0 : }
2060 0 : for(;;) {
2061 0 : if( FD_LIKELY( deferred_free_dlist_is_empty( accdb->deferred_free_dlist, accdb->partition_pool ) ) ) break;
2062 0 : fd_accdb_partition_t * p = deferred_free_dlist_ele_peek_head( accdb->deferred_free_dlist, accdb->partition_pool );
2063 0 : if( FD_LIKELY( p->epoch_tag>=min_epoch ) ) break;
2064 :
2065 0 : fd_racesan_hook( "accdb_reclaim:pre_free_partition" );
2066 :
2067 0 : spin_lock_acquire( &accdb->shmem->partition_lock );
2068 0 : deferred_free_dlist_ele_pop_head( accdb->deferred_free_dlist, accdb->partition_pool );
2069 0 : FD_ATOMIC_FETCH_AND_SUB( &accdb->shmem->shmetrics->disk_current_bytes, accdb->shmem->partition_sz );
2070 0 : FD_VOLATILE( p->write_offset ) = 0UL;
2071 0 : partition_pool_ele_release( accdb->partition_pool, p );
2072 0 : spin_lock_release( &accdb->shmem->partition_lock );
2073 0 : }
2074 :
2075 0 : if( FD_LIKELY( compaction_dlist_is_empty( accdb->compaction_dlist[ src_layer ], accdb->partition_pool ) ) ) {
2076 0 : FD_COMPILER_MFENCE();
2077 0 : FD_VOLATILE( *accdb->my_epoch_slot ) = ULONG_MAX;
2078 0 : return;
2079 0 : }
2080 0 : fd_accdb_partition_t * compact = compaction_dlist_ele_peek_head( accdb->compaction_dlist[ src_layer ], accdb->partition_pool );
2081 0 : if( FD_UNLIKELY( !compact ) ) {
2082 0 : FD_COMPILER_MFENCE();
2083 0 : FD_VOLATILE( *accdb->my_epoch_slot ) = ULONG_MAX;
2084 0 : return;
2085 0 : }
2086 :
2087 : /* Wait until all epoch-publishing joiners that were active when this
2088 : partition was enqueued for compaction have exited, ensuring any
2089 : in-flight pwritev2 to this partition has completed before we start
2090 : reading from it. */
2091 0 : if( FD_UNLIKELY( compact->compaction_ready_epoch>=min_epoch ) ) {
2092 0 : FD_COMPILER_MFENCE();
2093 0 : FD_VOLATILE( *accdb->my_epoch_slot ) = ULONG_MAX;
2094 0 : return;
2095 0 : }
2096 :
2097 0 : *charge_busy = 1;
2098 :
2099 0 : if( FD_UNLIKELY( !compact->compacting_now ) ) {
2100 0 : compact->compaction_start_wallclock = fd_log_wallclock();
2101 0 : compact->compaction_accounts_relocated = 0UL;
2102 0 : compact->compaction_bytes_relocated = 0UL;
2103 0 : compact->compaction_dead_records = 0UL;
2104 0 : }
2105 0 : FD_VOLATILE( compact->queued ) = 0;
2106 0 : FD_VOLATILE( compact->compacting_now ) = 1;
2107 :
2108 0 : ulong compact_base = partition_pool_idx( accdb->partition_pool, compact )*accdb->shmem->partition_sz;
2109 0 : ulong dest_layer = fd_ulong_min( src_layer+1UL, FD_ACCDB_COMPACTION_LAYER_CNT-1UL );
2110 :
2111 0 : ulong want = fd_ulong_min( FD_ACCDB_BOUNCE_SZ, compact->write_offset-compact->compaction_offset );
2112 0 : ulong carry = fd_ulong_if( accdb->carry_off==compact_base+compact->compaction_offset, fd_ulong_min( accdb->carry_sz, want ), 0UL );
2113 0 : accdb->carry_sz = 0UL;
2114 0 : ulong bytes_read = carry;
2115 0 : while( bytes_read<want ) {
2116 0 : long result = pread( accdb->fd, accdb->bounce+bytes_read, want-bytes_read, (long)(compact_base+compact->compaction_offset+bytes_read) );
2117 0 : if( FD_UNLIKELY( -1==result && (errno==EINTR || errno==EAGAIN || errno==EWOULDBLOCK ) ) ) continue;
2118 0 : else if( FD_UNLIKELY( -1==result ) ) FD_LOG_ERR(( "pread() failed (%d-%s)", errno, fd_io_strerror( errno ) ));
2119 0 : else if( FD_UNLIKELY( !result ) ) FD_LOG_ERR(( "accounts database is corrupt, data expected at offset %lu with size %lu exceeded file extents",
2120 0 : compact_base+compact->compaction_offset+bytes_read, want ));
2121 0 : fd_accdb_partition_read_bump( accdb, compact_base+compact->compaction_offset+bytes_read, (ulong)result );
2122 0 : bytes_read += (ulong)result;
2123 0 : }
2124 :
2125 0 : ulong span = 0UL;
2126 0 : for(;;) {
2127 0 : if( FD_UNLIKELY( span+sizeof(fd_accdb_disk_meta_t)>bytes_read ) ) break;
2128 0 : ulong record_sz = sizeof(fd_accdb_disk_meta_t) + (ulong)((fd_accdb_disk_meta_t const *)(accdb->bounce+span))->size;
2129 0 : if( FD_UNLIKELY( span+record_sz>bytes_read ) ) break;
2130 0 : span += record_sz;
2131 0 : }
2132 0 : if( FD_UNLIKELY( !span ) ) FD_LOG_ERR(( "accounts database is corrupt, record at offset %lu larger than %lu bytes",
2133 0 : compact_base+compact->compaction_offset, FD_ACCDB_BOUNCE_SZ ));
2134 :
2135 0 : fd_accdb_accmeta_t * live_accmeta[ IOV_MAX ];
2136 0 : ulong live_packed [ IOV_MAX ];
2137 0 : ulong live_src [ IOV_MAX ];
2138 0 : ulong live_sz [ IOV_MAX ];
2139 0 : ulong live_dst [ IOV_MAX ];
2140 0 : struct iovec iov [ IOV_MAX ];
2141 :
2142 0 : ulong live_cnt = 0UL;
2143 0 : ulong live_bytes = 0UL;
2144 0 : for( ulong off=0UL; off<span; ) {
2145 0 : fd_accdb_disk_meta_t const * meta = (fd_accdb_disk_meta_t const *)(accdb->bounce+off);
2146 :
2147 0 : fd_accdb_accmeta_t * accmeta = NULL;
2148 0 : ulong source_packed = 0UL;
2149 0 : uint acc_idx = FD_VOLATILE_CONST( accdb->acc_map[ fd_hash32( meta->pubkey, accdb->shmem->seed )&(accdb->shmem->chain_cnt-1UL) ] );
2150 0 : while( acc_idx!=UINT_MAX ) {
2151 0 : fd_accdb_accmeta_t * candidate = &accdb->acc_pool[ acc_idx ];
2152 0 : uint next_idx = FD_VOLATILE_CONST( candidate->map.next );
2153 0 : ulong candidate_packed = FD_VOLATILE_CONST( candidate->offset_fork );
2154 0 : if( FD_LIKELY( (candidate_packed & FD_ACCDB_OFF_MASK)==compact_base+compact->compaction_offset+off ) ) {
2155 0 : accmeta = candidate;
2156 0 : source_packed = candidate_packed;
2157 0 : break;
2158 0 : }
2159 0 : acc_idx = next_idx;
2160 0 : }
2161 :
2162 0 : ulong record_sz = sizeof(fd_accdb_disk_meta_t) + (ulong)meta->size;
2163 0 : if( FD_UNLIKELY( !accmeta ) ) compact->compaction_dead_records++; /* index entry gone, extent is garbage */
2164 0 : else {
2165 0 : live_accmeta[ live_cnt ] = accmeta;
2166 0 : live_packed [ live_cnt ] = source_packed;
2167 0 : live_src [ live_cnt ] = off;
2168 0 : live_sz [ live_cnt ] = record_sz;
2169 0 : live_cnt++;
2170 0 : live_bytes += record_sz;
2171 0 : }
2172 0 : off += record_sz;
2173 :
2174 0 : if( FD_UNLIKELY( live_cnt==IOV_MAX ) ) {
2175 0 : span = off;
2176 0 : break;
2177 0 : }
2178 0 : }
2179 :
2180 0 : if( FD_LIKELY( live_cnt ) ) {
2181 0 : ulong dest_base = allocate_next_compaction_write( accdb, live_bytes, dest_layer );
2182 0 : ulong dst = dest_base;
2183 0 : for( ulong i=0UL; i<live_cnt; i++ ) { live_dst[ i ] = dst; dst += live_sz[ i ]; }
2184 :
2185 0 : for( ulong i=0UL; i<live_cnt; i++ ) iov[ i ] = (struct iovec){ .iov_base = accdb->bounce+live_src[ i ], .iov_len = live_sz[ i ] };
2186 0 : ulong written = 0UL; ulong iov_idx = 0UL;
2187 0 : while( written<live_bytes ) {
2188 0 : long result = pwritev2( accdb->fd, iov+iov_idx, (int)(live_cnt-iov_idx), (long)(dest_base+written), 0 );
2189 0 : if( FD_UNLIKELY( -1==result && (errno==EINTR || errno==EAGAIN || errno==EWOULDBLOCK ) ) ) continue;
2190 0 : else if( FD_UNLIKELY( -1==result ) ) FD_LOG_ERR(( "pwritev2() failed (%d-%s)", errno, fd_io_strerror( errno ) ));
2191 0 : else if( FD_UNLIKELY( !result ) ) FD_LOG_ERR(( "accounts database is corrupt, pwritev2() returned 0 at offset %lu", dest_base+written ));
2192 0 : written += (ulong)result;
2193 0 : accdb->metrics->copy_ops++;
2194 0 : ulong adv = (ulong)result;
2195 0 : while( adv && iov_idx<live_cnt ) {
2196 0 : if( adv>=iov[ iov_idx ].iov_len ) {
2197 0 : adv -= iov[ iov_idx ].iov_len;
2198 0 : iov_idx++;
2199 0 : }
2200 0 : else {
2201 0 : iov[ iov_idx ].iov_base = (uchar *)iov[ iov_idx ].iov_base+adv;
2202 0 : iov[ iov_idx ].iov_len -= adv;
2203 0 : adv = 0UL;
2204 0 : }
2205 0 : }
2206 0 : }
2207 :
2208 0 : accdb->shmem->shmetrics->accounts_relocated += live_cnt;
2209 0 : accdb->shmem->shmetrics->accounts_relocated_bytes += live_bytes;
2210 0 : compact->compaction_accounts_relocated += live_cnt;
2211 0 : compact->compaction_bytes_relocated += live_bytes;
2212 :
2213 : /* Data is written before the offsets are published, so a
2214 : concurrent acquire never preadv2s an unwritten location. */
2215 0 : FD_COMPILER_MFENCE();
2216 :
2217 0 : for( ulong i=0UL; i<live_cnt; i++ ) {
2218 : /* CAS the offset from the exact source record we copied to the
2219 : new destination. If a concurrent release overwrote the offset
2220 : to FD_ACCDB_OFF_INVAL (dirty sentinel for a new commit), or
2221 : later published a newer on-disk location, the CAS fails and
2222 : the relocated copy is dead on arrival: account it as freed so
2223 : compaction reclaims it. The full packed offset_fork is CASed
2224 : so the fork_id is preserved. */
2225 0 : ulong new_packed = ( live_packed[ i ] & ~FD_ACCDB_OFF_MASK ) | ( live_dst[ i ] & FD_ACCDB_OFF_MASK );
2226 : #if FD_HAS_RACESAN
2227 : fd_memcpy( fd_accdb_dbg_reloc_pubkey, live_accmeta[ i ]->key.pubkey, 32UL );
2228 : fd_accdb_dbg_reloc_dest = live_dst[ i ];
2229 : fd_accdb_dbg_reloc_cnt++;
2230 : #endif
2231 0 : fd_racesan_hook( "accdb_compact:pre_offset_cas" );
2232 0 : if( FD_UNLIKELY( FD_ATOMIC_CAS( &live_accmeta[ i ]->offset_fork, live_packed[ i ], new_packed )!=live_packed[ i ] ) ) {
2233 0 : fd_accdb_shmem_bytes_freed( accdb->shmem, live_dst[ i ], live_sz[ i ] );
2234 0 : }
2235 0 : }
2236 0 : }
2237 :
2238 0 : fd_racesan_hook( "accdb_compact:post_relocate" );
2239 :
2240 0 : compact->compaction_offset += span;
2241 : #if FD_HAS_RACESAN
2242 : fd_accdb_dbg_compact_scanned += span;
2243 : #endif
2244 :
2245 0 : if( FD_LIKELY( span<bytes_read ) ) {
2246 0 : memmove( accdb->bounce, accdb->bounce+span, bytes_read-span );
2247 0 : accdb->carry_off = compact_base+compact->compaction_offset;
2248 0 : accdb->carry_sz = bytes_read-span;
2249 0 : accdb->carry_layer = src_layer;
2250 0 : }
2251 :
2252 0 : if( FD_UNLIKELY( compact->compaction_offset>=compact->write_offset ) ) {
2253 0 : FD_LOG_INFO(( "compaction of partition %lu completed", partition_pool_idx( accdb->partition_pool, compact ) ));
2254 :
2255 0 : fd_event_accdb_compaction_completed_t ev = {
2256 0 : .partition_idx = partition_pool_idx( accdb->partition_pool, compact ),
2257 0 : .src_layer = (uchar)src_layer,
2258 0 : .dest_layer = (uchar)fd_ulong_min( src_layer+1UL, FD_ACCDB_COMPACTION_LAYER_CNT-1UL ),
2259 0 : .bytes_scanned = compact->write_offset,
2260 0 : .bytes_freed = compact->bytes_freed,
2261 0 : .accounts_relocated = compact->compaction_accounts_relocated,
2262 0 : .bytes_relocated = compact->compaction_bytes_relocated,
2263 0 : .dead_records = compact->compaction_dead_records,
2264 0 : .start_time = (ulong)compact->compaction_start_wallclock,
2265 0 : .end_time = (ulong)fd_log_wallclock(),
2266 0 : };
2267 0 : fd_event_report_accdb_compaction_completed( &ev );
2268 :
2269 : /* Ensure the new acc->offset_fork stores above are visible to other
2270 : cores before the source partition is moved to the deferred-free
2271 : list. On x86 (TSO) hardware store ordering already guarantees
2272 : this, but the compiler fence prevents the compiler from sinking
2273 : the offset store past the inlined pool/dlist mutations below. */
2274 0 : FD_COMPILER_MFENCE();
2275 :
2276 : /* Bump the global epoch and tag this partition so the reclamation
2277 : scan knows when all epoch-publishing joiners that could reference
2278 : data in this partition have exited. */
2279 0 : ulong tag = FD_ATOMIC_FETCH_AND_ADD( &accdb->shmem->epoch, 1UL );
2280 0 : compact->epoch_tag = tag;
2281 :
2282 : /* partition_lock serializes these dlist/pool mutations with
2283 : concurrent push_tail in fd_accdb_shmem_bytes_freed and
2284 : partition_pool_ele_acquire in change_partition. Neither fd_dlist
2285 : nor fd_pool are thread-safe, so all mutations must be under the
2286 : same lock. */
2287 0 : spin_lock_acquire( &accdb->shmem->partition_lock );
2288 :
2289 0 : accdb->shmem->shmetrics->partitions_freed++;
2290 0 : compaction_dlist_ele_pop_head( accdb->compaction_dlist[ src_layer ], accdb->partition_pool );
2291 0 : FD_VOLATILE( compact->compacting_now ) = 0;
2292 0 : FD_VOLATILE( compact->queued ) = 0;
2293 0 : deferred_free_dlist_ele_push_tail( accdb->deferred_free_dlist, compact, accdb->partition_pool );
2294 :
2295 0 : accdb->shmem->shmetrics->compactions_completed++;
2296 0 : if( FD_LIKELY( compaction_dlist_is_empty( accdb->compaction_dlist[ src_layer ], accdb->partition_pool ) ) ) {
2297 0 : accdb->shmem->shmetrics->in_compaction = 0;
2298 0 : } else {
2299 0 : fd_accdb_partition_t * next = compaction_dlist_ele_peek_head( accdb->compaction_dlist[ src_layer ], accdb->partition_pool );
2300 0 : FD_LOG_INFO(( "compaction of layer %lu partition %lu started", src_layer, partition_pool_idx( accdb->partition_pool, next ) ));
2301 0 : }
2302 :
2303 0 : spin_lock_release( &accdb->shmem->partition_lock );
2304 0 : }
2305 :
2306 0 : accdb->metrics->bytes_read += bytes_read-carry;
2307 0 : accdb->metrics->bytes_written += live_bytes;
2308 :
2309 0 : FD_COMPILER_MFENCE();
2310 0 : FD_VOLATILE( *accdb->my_epoch_slot ) = ULONG_MAX;
2311 0 : }
2312 :
2313 : /* cold_load_acc resolves the cache slot for `acc` when STEP 1's
2314 : cache_try_pin failed. It uses bit 29 of executable_size as a
2315 : single-claimer lock so that two concurrent acquirers cannot each
2316 : install their own cache slot for the same acc (which would orphan
2317 : one slot with a dangling line->acc_idx and eventually corrupt
2318 : acc->cache_valid via CLOCK).
2319 :
2320 : Protocol per acc:
2321 : - If cache_valid is set, retry cache_try_pin (another thread
2322 : finished the cold-load while we were here). On success, mark
2323 : exists_in_cache so STEP 4 will not write back the slot.
2324 : - If claim is set, spin (another thread is mid-cold-load).
2325 : - Otherwise CAS-set the claim bit. Winner allocates a cache
2326 : line, populates the placeholder (acc_idx=UINT_MAX), publishes
2327 : cache_idx, then atomically (CAS-loop) sets cache_valid and
2328 : clears claim.
2329 :
2330 : The eviction sites that clear cache_valid must use FETCH_AND with
2331 : ~CACHE_VALID_BIT (preserving the claim bit) to interact correctly
2332 : with this protocol. */
2333 :
2334 : static fd_accdb_cache_line_t *
2335 : cold_load_acc( fd_accdb_t * accdb,
2336 : fd_accdb_accmeta_t * accmeta,
2337 : uchar const * pubkey,
2338 : int * out_exists_in_cache,
2339 0 : uint * out_evicted_acc_idx ) {
2340 0 : for(;;) {
2341 0 : uint old_es = FD_VOLATILE_CONST( accmeta->executable_size );
2342 0 : int valid = FD_ACCDB_SIZE_CACHE_VALID( old_es );
2343 0 : int claimed = FD_ACCDB_SIZE_CACHE_CLAIM( old_es );
2344 :
2345 0 : if( FD_UNLIKELY( valid ) ) {
2346 : /* old_es snapshot saw VALID=1 but a concurrent
2347 : evict_clear_acc_cache_ref may have cleared VALID and stored
2348 : cache_idx=INVAL between our snapshot and this load. Decoding
2349 : INVAL would yield a wild cache_line pointer; retry the loop
2350 : instead (next iteration will see VALID=0). */
2351 0 : uint cidx = FD_VOLATILE_CONST( accmeta->cache_idx );
2352 0 : if( FD_UNLIKELY( cidx==FD_ACCDB_ACC_CIDX_INVAL ) ) { FD_SPIN_PAUSE(); continue; }
2353 0 : fd_accdb_cache_line_t * hit = cache_line( accdb, FD_ACCDB_ACC_CIDX_CLASS( cidx ), FD_ACCDB_ACC_CIDX_IDX( cidx ) );
2354 0 : fd_racesan_hook( "accdb_cold_load:pre_try_pin" );
2355 0 : fd_accdb_cache_line_t * pinned = cache_try_pin( hit, pubkey, accmeta->key.generation );
2356 0 : if( FD_LIKELY( pinned ) ) {
2357 0 : *out_exists_in_cache = 1;
2358 0 : *out_evicted_acc_idx = UINT_MAX;
2359 0 : return pinned;
2360 0 : }
2361 0 : FD_SPIN_PAUSE();
2362 0 : continue;
2363 0 : }
2364 :
2365 0 : if( FD_UNLIKELY( claimed ) ) {
2366 0 : fd_racesan_hook( "accdb_cold_load:claim_wait" );
2367 0 : FD_SPIN_PAUSE();
2368 0 : continue;
2369 0 : }
2370 :
2371 0 : if( FD_UNLIKELY( FD_ATOMIC_CAS( &accmeta->executable_size, old_es, old_es | FD_ACCDB_SIZE_CACHE_CLAIM_BIT )!=old_es ) ) {
2372 0 : FD_SPIN_PAUSE();
2373 0 : continue;
2374 0 : }
2375 :
2376 : /* We hold the claim. Allocate a cache line and publish. */
2377 0 : ulong size_class = fd_accdb_cache_class( FD_ACCDB_SIZE_DATA( old_es ) );
2378 0 : fd_accdb_cache_line_t * line = acquire_cache_line( accdb, size_class, out_evicted_acc_idx );
2379 0 : fd_memcpy( line->key.pubkey, accmeta->key.pubkey, 32UL );
2380 0 : line->key.generation = accmeta->key.generation;
2381 : /* Leave acc_idx at UINT_MAX (the "loading" sentinel) until step 12
2382 : publishes it after the preadv2 fence. Concurrent threads that
2383 : pin via cache_idx will spin on this in step 13. */
2384 0 : line->acc_idx = UINT_MAX;
2385 0 : FD_COMPILER_MFENCE();
2386 0 : FD_VOLATILE( accmeta->cache_idx ) = FD_ACCDB_ACC_CIDX_PACK( (uint)size_class, (uint)cache_line_idx( accdb, size_class, line ) );
2387 0 : FD_COMPILER_MFENCE();
2388 :
2389 0 : fd_racesan_hook( "accdb_cold_load:pre_valid" );
2390 :
2391 : /* Atomically set CACHE_VALID_BIT and clear CACHE_CLAIM_BIT.
2392 : Eviction may have flipped CACHE_VALID_BIT on us between our
2393 : claim and now (it preserves CLAIM but can clear VALID); the
2394 : CAS loop tolerates that. The data length and exec bits stay
2395 : unchanged. */
2396 0 : for(;;) {
2397 0 : uint cur = FD_VOLATILE_CONST( accmeta->executable_size );
2398 0 : uint nxt = (cur & ~FD_ACCDB_SIZE_CACHE_CLAIM_BIT) | FD_ACCDB_SIZE_CACHE_VALID_BIT;
2399 0 : if( FD_LIKELY( FD_ATOMIC_CAS( &accmeta->executable_size, cur, nxt )==cur ) ) break;
2400 0 : FD_SPIN_PAUSE();
2401 0 : }
2402 :
2403 0 : *out_exists_in_cache = 0;
2404 0 : return line;
2405 0 : }
2406 0 : }
2407 :
2408 480737 : #define RESERVATION_TYPE_SIMPLE (0)
2409 7290 : #define RESERVATION_TYPE_MAYBE_PROGRAMDATA (1)
2410 7295 : #define RESERVATION_TYPE_ALREADY_RESERVED (2)
2411 :
2412 : static void
2413 : fd_accdb_acquire_inner( fd_accdb_t * accdb,
2414 : fd_accdb_fork_id_t fork_id,
2415 : int reservation_type,
2416 : ulong reserved_cnt,
2417 : ulong pubkeys_cnt,
2418 : uchar const * const * pubkeys,
2419 : int const * writable,
2420 494849 : fd_acc_t * out_accs ) {
2421 494849 : accdb->metrics->acquire_calls++;
2422 :
2423 494849 : ulong max_acquire_cnt = accdb->shmem->bundle_enabled ? FD_ACCDB_MAX_ACQUIRE_CNT : FD_ACCDB_MAX_TX_ACCOUNT_LOCKS;
2424 494849 : FD_TEST( pubkeys_cnt<=max_acquire_cnt );
2425 :
2426 494849 : FD_TEST( FD_VOLATILE_CONST( *accdb->my_epoch_slot )==ULONG_MAX );
2427 :
2428 494849 : FD_COMPILER_MFENCE();
2429 494849 : FD_VOLATILE( *accdb->my_epoch_slot ) = FD_VOLATILE_CONST( accdb->shmem->epoch );
2430 494849 : FD_HW_MFENCE(); /* StoreLoad: epoch store must be globally visible
2431 : before any subsequent loads so the deferred
2432 : reclamation scan does not miss us */
2433 :
2434 : // STEP 1.
2435 : // Locate each account in the fork and index structure, to determine
2436 : // if it already exists, its size and other metadata, and which
2437 : // specific slot (generation) it was last written in.
2438 :
2439 494849 : fd_accdb_fork_t * fork = &accdb->fork_pool[ fork_id.val ];
2440 494849 : uint root_generation = accdb->fork_pool[ accdb->shmem->root_fork_id.val ].shmem->generation;
2441 :
2442 494849 : fd_racesan_hook( "accdb_acquire:post_root_gen" );
2443 :
2444 494849 : fd_accdb_accmeta_t * accmetas[ FD_ACCDB_MAX_ACQUIRE_CNT ];
2445 494849 : ulong acc_map_idxs[ FD_ACCDB_MAX_ACQUIRE_CNT ];
2446 :
2447 : /* Walk the hash chain for each pubkey and take the first visible
2448 : match. Correctness relies on newer entries always being prepended
2449 : to the chain head, which is guaranteed because replay processes
2450 : writes in slot order and release always inserts at the head.
2451 :
2452 : CONCURRENCY: This chain walk runs epoch-protected. A concurrent
2453 : fd_accdb_release may prepend a new node to the same chain while
2454 : we walk it. This is safe on x86-64 (TSO): the releasing thread
2455 : stores all acc fields (pubkey, generation, map.next, ...) before
2456 : publishing the new head via a CAS on acc_map[idx], and TSO
2457 : guarantees a reading core that observes the new head also observes
2458 : all prior stores to the node. A reader that does not yet see the
2459 : new head simply sees an older (still valid) version of the chain.
2460 : On weakly-ordered architectures an explicit acquire fence would be
2461 : needed before the chain walk and a release fence in
2462 : fd_accdb_release before the head-pointer store. Multiple
2463 : concurrent releases serialize on the CAS of the chain head. */
2464 1058671 : for( ulong i=0UL; i<pubkeys_cnt; i++ ) {
2465 563822 : acc_map_idxs[ i ] = fd_hash32( pubkeys[ i ], accdb->shmem->seed )&(accdb->shmem->chain_cnt-1UL);
2466 563822 : uint acc = FD_VOLATILE_CONST( accdb->acc_map[ acc_map_idxs[ i ] ] );
2467 612463 : while( acc!=UINT_MAX ) {
2468 227462 : fd_accdb_accmeta_t const * candidate_acc = &accdb->acc_pool[ acc ];
2469 227462 : uint next_acc = FD_VOLATILE_CONST( candidate_acc->map.next );
2470 :
2471 227462 : fd_racesan_hook( "accdb_acquire:post_next" );
2472 :
2473 227462 : if( FD_UNLIKELY( (candidate_acc->key.generation>root_generation &&
2474 227462 : fd_accdb_acc_fork_id(candidate_acc)!=fork_id.val &&
2475 227462 : !descends_set_test( fork->descends, fd_accdb_acc_fork_id(candidate_acc) )) ) ||
2476 227462 : memcmp( pubkeys[ i ], candidate_acc->key.pubkey, 32UL ) ) {
2477 48641 : acc = next_acc;
2478 48641 : continue;
2479 48641 : }
2480 :
2481 178821 : break;
2482 227462 : }
2483 563822 : if( FD_UNLIKELY( acc==UINT_MAX ) ) accmetas[ i ] = NULL;
2484 181650 : else accmetas[ i ] = &accdb->acc_pool[ acc ];
2485 :
2486 : #if FD_TMPL_USE_HANDHOLDING
2487 : if( FD_UNLIKELY( accmetas[ i ] ) ) {
2488 : fd_accdb_accmeta_t const * sel = accmetas[ i ];
2489 : FD_TEST( !memcmp( sel->key.pubkey, pubkeys[ i ], 32UL ) );
2490 : FD_TEST( sel->key.generation<=root_generation ||
2491 : fd_accdb_acc_fork_id( sel )==fork_id.val ||
2492 : descends_set_test( fork->descends, fd_accdb_acc_fork_id( sel ) ) );
2493 : FD_TEST( sel->key.generation<=FD_VOLATILE_CONST( accdb->shmem->generation ) );
2494 : }
2495 : #endif
2496 :
2497 563822 : if( FD_UNLIKELY( accmetas[ i ] && !writable[ i ] && !accmetas[ i ]->lamports ) ) accmetas[ i ] = NULL;
2498 :
2499 : /* Attribute this acquired account to a size class for per-class
2500 : rate metrics. Use the account's current size class when known;
2501 : otherwise (new account) bucket as class 0. */
2502 563822 : ulong acq_class = 0UL;
2503 563822 : if( FD_LIKELY( accmetas[ i ] ) ) acq_class = fd_accdb_cache_class( FD_ACCDB_SIZE_DATA( accmetas[ i ]->executable_size ) );
2504 563822 : if( FD_LIKELY( writable[ i ] ) ) accdb->metrics->writable_accounts_acquired_per_class[ acq_class ]++;
2505 159385 : else accdb->metrics->accounts_acquired_per_class[ acq_class ]++;
2506 563822 : }
2507 :
2508 : // STEP 2.
2509 : // The two-phase programdata acquire (acquire_a then acquire_b)
2510 : // works as follows: acquire_a (RESERVATION_TYPE_MAYBE_PROGRAMDATA)
2511 : // over-reserves one slot in every live size class per candidate
2512 : // account (reserved_cnt total per class), because it does not yet
2513 : // know which accounts have programdata or what size class it lands
2514 : // in. acquire_b then resolves the actual programdata pubkeys and
2515 : // re-enters here with RESERVATION_TYPE_ALREADY_RESERVED to refund
2516 : // the surplus. Keep one reservation per found programdata account
2517 : // in its own size class (consumed later by release) and give the
2518 : // rest back.
2519 494849 : if( FD_UNLIKELY( reservation_type==RESERVATION_TYPE_ALREADY_RESERVED ) ) {
2520 7291 : ulong refund[ FD_ACCDB_CACHE_CLASS_CNT ] = {0};
2521 65613 : for( ulong j=0UL; j<FD_ACCDB_CACHE_CLASS_CNT; j++ ) {
2522 58322 : if( FD_LIKELY( accdb->shmem->cache_class_used[ j ].val!=ULONG_MAX ) ) refund[ j ] = reserved_cnt;
2523 58322 : }
2524 7311 : for( ulong i=0UL; i<pubkeys_cnt; i++ ) {
2525 20 : if( FD_LIKELY( accmetas[ i ] ) ) {
2526 20 : ulong cls = fd_accdb_cache_class( FD_ACCDB_SIZE_DATA( accmetas[ i ]->executable_size ) );
2527 20 : if( FD_LIKELY( accdb->shmem->cache_class_used[ cls ].val!=ULONG_MAX ) ) {
2528 0 : FD_TEST( refund[ cls ]>0UL );
2529 0 : refund[ cls ]--;
2530 0 : }
2531 20 : }
2532 20 : }
2533 65611 : for( ulong k=0UL; k<FD_ACCDB_CACHE_CLASS_CNT; k++ ) {
2534 58320 : if( FD_UNLIKELY( refund[ k ] ) ) FD_ATOMIC_FETCH_AND_SUB( &accdb->shmem->cache_class_used[ k ].val, refund[ k ] );
2535 58320 : }
2536 7291 : }
2537 :
2538 : // STEP 3.
2539 : // We are potentially going to need to read the account data off of
2540 : // disk into the cache, if the account(s) are not in the cache so
2541 : // reserve the necessary cache space. This is done with an "atomic
2542 : // subtract" spin loop on the cache class counters, which is
2543 : // actually faster than doing a real CAS on a packed ulong.
2544 : //
2545 : // For reads, we only need space to copy the account data into a
2546 : // single right-sized cache line, but for writes ... we need to
2547 : // reserve one of every size class. The reason is we are going to
2548 : // need a 10MiB staging buffer for the executor to write to (it may
2549 : // grow the account, so needs the max size class). Even if the
2550 : // account is already in the 10MiB cache class, we need another one
2551 : // because a transaction can fail half way, so we need scratch space
2552 : // to be able to unwind.
2553 : //
2554 : // So we acquire one of each size class. Then when the transaction
2555 : // finishes, if it succeeded, we will copy the data back to the
2556 : // whichever size-class is now right-sized post execution.
2557 494849 : if( FD_LIKELY( reservation_type==RESERVATION_TYPE_SIMPLE || reservation_type==RESERVATION_TYPE_MAYBE_PROGRAMDATA ) ) {
2558 489227 : ulong requested_buckets[ FD_ACCDB_CACHE_CLASS_CNT ] = {0};
2559 1048977 : for( ulong i=0UL; i<pubkeys_cnt; i++ ) {
2560 559750 : if( FD_LIKELY( accmetas[ i ] || writable[ i ] ) ) {
2561 544666 : if( FD_LIKELY( accmetas[ i ] ) ) {
2562 178917 : if( FD_UNLIKELY( accdb->shmem->cache_class_used[ fd_accdb_cache_class( FD_ACCDB_SIZE_DATA( accmetas[ i ]->executable_size ) ) ].val!=ULONG_MAX ) ) {
2563 0 : requested_buckets[ fd_accdb_cache_class( FD_ACCDB_SIZE_DATA( accmetas[ i ]->executable_size ) ) ]++;
2564 0 : }
2565 178917 : }
2566 544666 : if( FD_UNLIKELY( writable[ i ] ) ) {
2567 3634046 : for( ulong j=0UL; j<FD_ACCDB_CACHE_CLASS_CNT; j++ ) {
2568 3230079 : if( FD_UNLIKELY( accdb->shmem->cache_class_used[ j ].val!=ULONG_MAX ) ) {
2569 0 : requested_buckets[ j ]++;
2570 0 : }
2571 3230079 : }
2572 403967 : }
2573 544666 : }
2574 :
2575 559750 : if( FD_LIKELY( reservation_type==RESERVATION_TYPE_MAYBE_PROGRAMDATA ) ) {
2576 : /* Any account could also have an implied reference to a
2577 : programdata account, which we don't know yet ... so we need to
2578 : reserve worst case space if they all went to the same size
2579 : class. This reservation runs unconditionally per pubkey (not
2580 : gated on accmetas/writable) so that acquire_b can refund based on
2581 : pubkeys_cnt without needing to re-derive the live-account set. */
2582 704349 : for( ulong j=0UL; j<FD_ACCDB_CACHE_CLASS_CNT; j++ ) {
2583 626080 : if( FD_UNLIKELY( accdb->shmem->cache_class_used[ j ].val!=ULONG_MAX ) ) {
2584 0 : requested_buckets[ j ]++;
2585 0 : }
2586 626080 : }
2587 78269 : }
2588 559750 : }
2589 :
2590 : /* TODO: This over-reserves cache slots for writable accounts that
2591 : already exist. For each such account we reserve one line in the
2592 : account's size class (for the read into cache) AND one line in
2593 : every size class (for the write destination buffers). But if the
2594 : account is already resident in cache (which is the common case
2595 : for hot accounts), the read-into-cache line is unnecessary — we
2596 : will get a cache hit in step 4 and never use it. The fix is to
2597 : probe acc->cache_idx here and skip the per-account size class
2598 : reservation per-account size class reservation when a hit is
2599 : found. This would reduce peak reservation by up to one line per
2600 : writable account per acquire batch, lowering contention on the
2601 : cache class counters and allowing smaller cache provisioning. */
2602 :
2603 : /* Reserve cache slots by atomically incrementing the shared used
2604 : counters. If any class exceeds its max, the reservation
2605 : overflowed — subtract back partial grabs and retry. */
2606 489227 : for(;;) {
2607 486739 : int acquire_failed = 0;
2608 486739 : ulong grabbed[ FD_ACCDB_CACHE_CLASS_CNT ] = {0};
2609 4364950 : for( ulong i=0UL; i<FD_ACCDB_CACHE_CLASS_CNT; i++ ) {
2610 3880909 : if( FD_LIKELY( !requested_buckets[ i ] ) ) continue;
2611 >1844*10^16 : ulong new_used = FD_ATOMIC_ADD_AND_FETCH( &accdb->shmem->cache_class_used[ i ].val, requested_buckets[ i ] );
2612 >1844*10^16 : if( FD_UNLIKELY( new_used>accdb->shmem->cache_class_max[ i ] ) ) {
2613 0 : FD_ATOMIC_FETCH_AND_SUB( &accdb->shmem->cache_class_used[ i ].val, requested_buckets[ i ] );
2614 0 : acquire_failed = 1;
2615 >1844*10^16 : } else {
2616 >1844*10^16 : grabbed[ i ] = requested_buckets[ i ];
2617 >1844*10^16 : }
2618 >1844*10^16 : if( FD_UNLIKELY( acquire_failed ) ) {
2619 0 : accdb->metrics->acquire_failed++;
2620 0 : for( ulong j=0UL; j<i; j++ ) {
2621 0 : if( grabbed[ j ] ) FD_ATOMIC_FETCH_AND_SUB( &accdb->shmem->cache_class_used[ j ].val, grabbed[ j ] );
2622 0 : }
2623 0 : FD_SPIN_PAUSE();
2624 0 : break;
2625 0 : }
2626 >1844*10^16 : }
2627 487879 : if( FD_LIKELY( !acquire_failed ) ) break;
2628 486739 : }
2629 489227 : }
2630 :
2631 : // STEP 4.
2632 : // For any accounts that are not in cache, we now need to actually
2633 : // retrieve the cache pointers from our structures. Space has been
2634 : // reserved already, so this step is guaranteed to succeed, and is
2635 : // just pulling the cache lines out of the free lists and marking
2636 : // them as in-use.
2637 : //
2638 : // This step is fully lock-free. Cache hits are pinned with an
2639 : // atomic CAS on refcnt (cache_try_pin). Eviction uses the CLOCK
2640 : // algorithm. The CAS free list provides immediate recycling of
2641 : // fully-freed lines.
2642 :
2643 494849 : int exists_in_cache[ FD_ACCDB_MAX_ACQUIRE_CNT ];
2644 494849 : fd_accdb_cache_line_t * original_cache_line[ FD_ACCDB_MAX_ACQUIRE_CNT ];
2645 494849 : fd_accdb_cache_line_t * destination_cache_lines[ FD_ACCDB_MAX_ACQUIRE_CNT ][ FD_ACCDB_CACHE_CLASS_CNT ];
2646 :
2647 : /* Saved acc_pool indices of evicted dirty cache lines. These are
2648 : captured before clearing acc_idx to UINT_MAX on the line struct, so
2649 : that the sentinel protocol (step 14) works correctly while the
2650 : evicted account metadata is still available for writeback in steps
2651 : 4 and 6. */
2652 494849 : uint evicted_dest_acc[ FD_ACCDB_MAX_ACQUIRE_CNT ][ FD_ACCDB_CACHE_CLASS_CNT ];
2653 494849 : uint evicted_orig_acc[ FD_ACCDB_MAX_ACQUIRE_CNT ];
2654 :
2655 1053352 : for( ulong i=0UL; i<pubkeys_cnt; i++ ) {
2656 558503 : if( FD_UNLIKELY( !accmetas[ i ] && !writable[ i ] ) ) continue;
2657 :
2658 543563 : original_cache_line[ i ] = NULL;
2659 543563 : if( FD_LIKELY( accmetas[ i ] ) ) {
2660 178989 : if( FD_LIKELY( FD_ACCDB_SIZE_CACHE_VALID( FD_VOLATILE_CONST( accmetas[ i ]->executable_size ) ) ) ) {
2661 : /* Concurrent evict_clear_acc_cache_ref clears VALID then stores
2662 : cache_idx=INVAL. We may have observed VALID=1 just before the
2663 : writer cleared it, so cidx can read as INVAL here; decoding it
2664 : would yield a wild cache_line pointer. Skip on INVAL. Any
2665 : other stale cidx is harmless: cache_try_pin's ABA generation
2666 : check rejects a recycled line. */
2667 178989 : uint cidx = FD_VOLATILE_CONST( accmetas[ i ]->cache_idx );
2668 178997 : if( FD_LIKELY( cidx!=FD_ACCDB_ACC_CIDX_INVAL ) ) {
2669 178997 : fd_accdb_cache_line_t * hit = cache_line( accdb, FD_ACCDB_ACC_CIDX_CLASS( cidx ), FD_ACCDB_ACC_CIDX_IDX( cidx ) );
2670 178997 : fd_racesan_hook( "accdb_acquire:pre_try_pin" );
2671 178997 : original_cache_line[ i ] = cache_try_pin( hit, pubkeys[ i ], accmetas[ i ]->key.generation );
2672 : #if FD_TMPL_USE_HANDHOLDING
2673 : if( FD_LIKELY( original_cache_line[ i ] ) ) {
2674 : FD_TEST( original_cache_line[ i ]->key.generation==accmetas[ i ]->key.generation &&
2675 : !memcmp( original_cache_line[ i ]->key.pubkey, pubkeys[ i ], 32UL ) );
2676 : uint rc = FD_VOLATILE_CONST( original_cache_line[ i ]->refcnt );
2677 : FD_TEST( rc>0U && rc!=FD_ACCDB_EVICT_SENTINEL );
2678 : }
2679 : #endif
2680 178997 : }
2681 178989 : }
2682 178976 : }
2683 543563 : exists_in_cache[ i ] = original_cache_line[ i ]!=NULL;
2684 :
2685 543563 : if( FD_UNLIKELY( writable[ i ] ) ) {
2686 3648396 : for( ulong j=0UL; j<FD_ACCDB_CACHE_CLASS_CNT; j++ ) destination_cache_lines[ i ][ j ] = acquire_cache_line( accdb, j, &evicted_dest_acc[ i ][ j ] );
2687 403336 : if( FD_UNLIKELY( accmetas[ i ] && !original_cache_line[ i ] ) ) {
2688 0 : original_cache_line[ i ] = cold_load_acc( accdb, accmetas[ i ], pubkeys[ i ], &exists_in_cache[ i ], &evicted_orig_acc[ i ] );
2689 0 : }
2690 403336 : } else {
2691 140227 : if( FD_UNLIKELY( !original_cache_line[ i ] ) ) {
2692 0 : original_cache_line[ i ] = cold_load_acc( accdb, accmetas[ i ], pubkeys[ i ], &exists_in_cache[ i ], &evicted_orig_acc[ i ] );
2693 0 : }
2694 140227 : }
2695 543563 : }
2696 :
2697 : // STEP 5.
2698 : // For any cache lines we have retrieved, which we might potentially
2699 : // be about to trash (by writing stuff in there), we need to write
2700 : // them back to disk first if they are dirty. This is the process of
2701 : // "persisting" (a/k/a evicting) whatever was previously in the
2702 : // cache line we are about to use.
2703 : //
2704 : // This step does not actually persist the data to disk, it just
2705 : // constructs a series of iovecs (write instructions) which will be
2706 : // used later to do the actual write. The reason is that we want to
2707 : // batch all the writes together into a single writev call, to
2708 : // minimize overhead, and also keep the actual writes at the end of
2709 : // the function and independent of the specific control flow, so
2710 : // that they could be offloaded to another thread of made
2711 : // asynchronous (e.g. with io_uring) in the future without needing
2712 : // to change the rest of the logic.
2713 :
2714 494849 : int write_ops_cnt = 0;
2715 494849 : int write_meta_cnt = 0;
2716 494849 : ulong total_write_sz = 0UL;
2717 494849 : fd_accdb_disk_meta_t write_metas[ (FD_ACCDB_CACHE_CLASS_CNT+1UL)*FD_ACCDB_MAX_ACQUIRE_CNT ];
2718 494849 : struct iovec write_ops[ 2UL*(FD_ACCDB_CACHE_CLASS_CNT+1UL)*FD_ACCDB_MAX_ACQUIRE_CNT ];
2719 :
2720 1058270 : for( ulong i=0UL; i<pubkeys_cnt; i++ ) {
2721 563421 : if( FD_UNLIKELY( !accmetas[ i ] && !writable[ i ] ) ) continue;
2722 :
2723 548482 : if( FD_UNLIKELY( writable[ i ] ) ) {
2724 3625986 : for( ulong j=0UL; j<FD_ACCDB_CACHE_CLASS_CNT; j++ ) {
2725 3220612 : if( FD_LIKELY( evicted_dest_acc[ i ][ j ]==UINT_MAX ) ) continue;
2726 2233 : accdb->metrics->accounts_evicted++;
2727 2233 : accdb->metrics->accounts_evicted_per_class[ j ]++;
2728 :
2729 2233 : fd_accdb_accmeta_t const * evicted = &accdb->acc_pool[ evicted_dest_acc[ i ][ j ] ];
2730 2233 : fd_racesan_hook( "writeback:pre_synth" );
2731 2233 : total_write_sz += sizeof(fd_accdb_disk_meta_t) + FD_ACCDB_SIZE_DATA( evicted->executable_size );
2732 2233 : FD_TEST( write_meta_cnt<(int)(sizeof(write_metas)/sizeof(write_metas[0])) );
2733 2233 : fd_memcpy( write_metas[ write_meta_cnt ].pubkey, evicted->key.pubkey, 32UL );
2734 2233 : write_metas[ write_meta_cnt ].size = FD_ACCDB_SIZE_DATA( evicted->executable_size );
2735 2233 : write_metas[ write_meta_cnt ].generation = evicted->key.generation;
2736 2233 : fd_memcpy( write_metas[ write_meta_cnt ].owner, destination_cache_lines[ i ][ j ]->owner, 32UL );
2737 2233 : write_ops[ write_ops_cnt++ ] = (struct iovec){ .iov_base = &write_metas[ write_meta_cnt ], .iov_len = sizeof(fd_accdb_disk_meta_t) };
2738 2233 : write_meta_cnt++;
2739 2233 : write_ops[ write_ops_cnt++ ] = (struct iovec){ .iov_base = destination_cache_lines[ i ][ j ]+1UL, .iov_len = FD_ACCDB_SIZE_DATA( evicted->executable_size ) };
2740 2233 : }
2741 405374 : if( FD_UNLIKELY( accmetas[ i ] && !exists_in_cache[ i ] && evicted_orig_acc[ i ]!=UINT_MAX ) ) {
2742 0 : fd_accdb_accmeta_t const * evicted = &accdb->acc_pool[ evicted_orig_acc[ i ] ];
2743 0 : accdb->metrics->accounts_evicted++;
2744 0 : accdb->metrics->accounts_evicted_per_class[ fd_accdb_cache_class( FD_ACCDB_SIZE_DATA( evicted->executable_size ) ) ]++;
2745 :
2746 0 : total_write_sz += sizeof(fd_accdb_disk_meta_t) + FD_ACCDB_SIZE_DATA( evicted->executable_size );
2747 0 : FD_TEST( write_meta_cnt<(int)(sizeof(write_metas)/sizeof(write_metas[0])) );
2748 0 : fd_memcpy( write_metas[ write_meta_cnt ].pubkey, evicted->key.pubkey, 32UL );
2749 0 : write_metas[ write_meta_cnt ].size = FD_ACCDB_SIZE_DATA( evicted->executable_size );
2750 0 : write_metas[ write_meta_cnt ].generation = evicted->key.generation;
2751 0 : fd_memcpy( write_metas[ write_meta_cnt ].owner, original_cache_line[ i ]->owner, 32UL );
2752 0 : write_ops[ write_ops_cnt++ ] = (struct iovec){ .iov_base = &write_metas[ write_meta_cnt ], .iov_len = sizeof(fd_accdb_disk_meta_t) };
2753 0 : write_meta_cnt++;
2754 0 : write_ops[ write_ops_cnt++ ] = (struct iovec){ .iov_base = original_cache_line[ i ]+1UL, .iov_len = FD_ACCDB_SIZE_DATA( evicted->executable_size ) };
2755 0 : }
2756 405374 : } else {
2757 143108 : if( FD_LIKELY( exists_in_cache[ i ] || evicted_orig_acc[ i ]==UINT_MAX ) ) continue;
2758 1671 : fd_accdb_accmeta_t const * evicted = &accdb->acc_pool[ evicted_orig_acc[ i ] ];
2759 1671 : accdb->metrics->accounts_evicted++;
2760 1671 : accdb->metrics->accounts_evicted_per_class[ fd_accdb_cache_class( FD_ACCDB_SIZE_DATA( evicted->executable_size ) ) ]++;
2761 1671 : total_write_sz += sizeof(fd_accdb_disk_meta_t) + FD_ACCDB_SIZE_DATA( evicted->executable_size );
2762 1671 : FD_TEST( write_meta_cnt<(int)(sizeof(write_metas)/sizeof(write_metas[0])) );
2763 1671 : fd_memcpy( write_metas[ write_meta_cnt ].pubkey, evicted->key.pubkey, 32UL );
2764 1671 : write_metas[ write_meta_cnt ].size = FD_ACCDB_SIZE_DATA( evicted->executable_size );
2765 1671 : write_metas[ write_meta_cnt ].generation = evicted->key.generation;
2766 1671 : fd_memcpy( write_metas[ write_meta_cnt ].owner, original_cache_line[ i ]->owner, 32UL );
2767 1671 : write_ops[ write_ops_cnt++ ] = (struct iovec){ .iov_base = &write_metas[ write_meta_cnt ], .iov_len = sizeof(fd_accdb_disk_meta_t) };
2768 1671 : write_meta_cnt++;
2769 1671 : write_ops[ write_ops_cnt++ ] = (struct iovec){ .iov_base = original_cache_line[ i ]+1UL, .iov_len = FD_ACCDB_SIZE_DATA( evicted->executable_size ) };
2770 1671 : }
2771 548482 : }
2772 :
2773 : // STEP 6-7.
2774 : // Compute the file offset for the writes we are about to do and
2775 : // build the pending offset table. The common case is a single
2776 : // atomic fetch-add on the write head, reserving a contiguous
2777 : // region. If the total eviction batch is too large to fit in one
2778 : // partition (extremely unlikely — requires many dirty 10MiB
2779 : // evictions), fall back to per-entry allocation so that each
2780 : // individual write fits in a single partition.
2781 : //
2782 : // The actual stores to evicted->offset_fork and line->persisted
2783 : // are deferred until after pwritev2 completes (Step 9-10), so
2784 : // a concurrent acquire spinning on offset==FD_ACCDB_OFF_INVAL
2785 : // does not proceed to preadv2 from a location that hasn't been
2786 : // written.
2787 494849 : int pending_cnt = 0;
2788 494849 : fd_accdb_accmeta_t * pending_accs [ (FD_ACCDB_CACHE_CLASS_CNT+1UL)*FD_ACCDB_MAX_ACQUIRE_CNT ];
2789 494849 : ulong pending_offs [ (FD_ACCDB_CACHE_CLASS_CNT+1UL)*FD_ACCDB_MAX_ACQUIRE_CNT ];
2790 494849 : fd_accdb_cache_line_t * pending_lines[ (FD_ACCDB_CACHE_CLASS_CNT+1UL)*FD_ACCDB_MAX_ACQUIRE_CNT ];
2791 :
2792 494849 : ulong file_offset;
2793 494849 : int batch_contiguous;
2794 494849 : if( FD_LIKELY( total_write_sz && total_write_sz<=accdb->shmem->partition_sz ) ) {
2795 0 : file_offset = allocate_next_write( accdb, total_write_sz );
2796 0 : batch_contiguous = 1;
2797 494849 : } else {
2798 494849 : file_offset = 0UL;
2799 494849 : batch_contiguous = 0;
2800 494849 : }
2801 :
2802 494849 : ulong cumulative_offset = 0UL;
2803 1055878 : for( ulong i=0UL; i<pubkeys_cnt; i++ ) {
2804 561029 : if( FD_UNLIKELY( !accmetas[ i ] && !writable[ i ] ) ) continue;
2805 :
2806 546091 : if( FD_UNLIKELY( writable[ i ] ) ) {
2807 3597171 : for( ulong j=0UL; j<FD_ACCDB_CACHE_CLASS_CNT; j++ ) {
2808 3196173 : if( FD_LIKELY( evicted_dest_acc[ i ][ j ]==UINT_MAX ) ) continue;
2809 :
2810 >1844*10^16 : fd_accdb_accmeta_t * evicted = &accdb->acc_pool[ evicted_dest_acc[ i ][ j ] ];
2811 >1844*10^16 : ulong entry_sz = sizeof(fd_accdb_disk_meta_t) + (ulong)FD_ACCDB_SIZE_DATA( evicted->executable_size );
2812 : /* xchg-to-INVAL atomically captures the old offset and prevents
2813 : a concurrent acc_unlink from also reading and freeing it (the
2814 : xchg there will see INVAL and skip). Step 10 republishes the
2815 : new offset; the spinner at line ~2082 tolerates the transient
2816 : INVAL. Same pattern as the overwrite path at line ~2388. */
2817 >1844*10^16 : ulong old_off = fd_accdb_acc_xchg_offset( evicted, FD_ACCDB_OFF_INVAL );
2818 >1844*10^16 : if( FD_LIKELY( old_off!=FD_ACCDB_OFF_INVAL ) ) {
2819 0 : fd_accdb_shmem_bytes_freed( accdb->shmem, old_off, entry_sz );
2820 0 : FD_ATOMIC_FETCH_AND_SUB( &accdb->shmem->shmetrics->disk_used_bytes, entry_sz );
2821 0 : }
2822 >1844*10^16 : FD_TEST( pending_cnt<(int)(sizeof(pending_accs)/sizeof(pending_accs[0])) );
2823 >1844*10^16 : pending_accs [ pending_cnt ] = evicted;
2824 >1844*10^16 : if( FD_LIKELY( batch_contiguous ) ) pending_offs[ pending_cnt ] = file_offset + cumulative_offset;
2825 >1844*10^16 : else pending_offs[ pending_cnt ] = allocate_next_write( accdb, entry_sz );
2826 >1844*10^16 : pending_lines[ pending_cnt ] = destination_cache_lines[ i ][ j ];
2827 >1844*10^16 : pending_cnt++;
2828 >1844*10^16 : cumulative_offset += entry_sz;
2829 >1844*10^16 : FD_ATOMIC_FETCH_AND_ADD( &accdb->shmem->shmetrics->disk_used_bytes, entry_sz );
2830 >1844*10^16 : }
2831 404373 : if( FD_UNLIKELY( accmetas[ i ] && !exists_in_cache[ i ] && evicted_orig_acc[ i ]!=UINT_MAX ) ) {
2832 0 : fd_accdb_accmeta_t * evicted = &accdb->acc_pool[ evicted_orig_acc[ i ] ];
2833 0 : ulong entry_sz = sizeof(fd_accdb_disk_meta_t) + (ulong)FD_ACCDB_SIZE_DATA( evicted->executable_size );
2834 0 : ulong old_off = fd_accdb_acc_xchg_offset( evicted, FD_ACCDB_OFF_INVAL );
2835 0 : if( FD_LIKELY( old_off!=FD_ACCDB_OFF_INVAL ) ) {
2836 0 : fd_accdb_shmem_bytes_freed( accdb->shmem, old_off, entry_sz );
2837 0 : FD_ATOMIC_FETCH_AND_SUB( &accdb->shmem->shmetrics->disk_used_bytes, entry_sz );
2838 0 : }
2839 0 : FD_TEST( pending_cnt<(int)(sizeof(pending_accs)/sizeof(pending_accs[0])) );
2840 0 : pending_accs [ pending_cnt ] = evicted;
2841 0 : if( FD_LIKELY( batch_contiguous ) ) pending_offs[ pending_cnt ] = file_offset + cumulative_offset;
2842 0 : else pending_offs[ pending_cnt ] = allocate_next_write( accdb, entry_sz );
2843 0 : pending_lines[ pending_cnt ] = original_cache_line[ i ];
2844 0 : pending_cnt++;
2845 0 : cumulative_offset += entry_sz;
2846 0 : FD_ATOMIC_FETCH_AND_ADD( &accdb->shmem->shmetrics->disk_used_bytes, entry_sz );
2847 0 : }
2848 404373 : } else {
2849 141718 : if( FD_LIKELY( exists_in_cache[ i ] || evicted_orig_acc[ i ]==UINT_MAX ) ) continue;
2850 :
2851 286 : fd_accdb_accmeta_t * evicted = &accdb->acc_pool[ evicted_orig_acc[ i ] ];
2852 286 : ulong entry_sz = sizeof(fd_accdb_disk_meta_t) + (ulong)FD_ACCDB_SIZE_DATA( evicted->executable_size );
2853 286 : ulong old_off = fd_accdb_acc_xchg_offset( evicted, FD_ACCDB_OFF_INVAL );
2854 286 : if( FD_LIKELY( old_off!=FD_ACCDB_OFF_INVAL ) ) {
2855 0 : fd_accdb_shmem_bytes_freed( accdb->shmem, old_off, entry_sz );
2856 0 : FD_ATOMIC_FETCH_AND_SUB( &accdb->shmem->shmetrics->disk_used_bytes, entry_sz );
2857 0 : }
2858 286 : FD_TEST( pending_cnt<(int)(sizeof(pending_accs)/sizeof(pending_accs[0])) );
2859 286 : pending_accs [ pending_cnt ] = evicted;
2860 286 : if( FD_LIKELY( batch_contiguous ) ) pending_offs[ pending_cnt ] = file_offset + cumulative_offset;
2861 286 : else pending_offs[ pending_cnt ] = allocate_next_write( accdb, entry_sz );
2862 286 : pending_lines[ pending_cnt ] = original_cache_line[ i ];
2863 286 : pending_cnt++;
2864 286 : cumulative_offset += entry_sz;
2865 286 : FD_ATOMIC_FETCH_AND_ADD( &accdb->shmem->shmetrics->disk_used_bytes, entry_sz );
2866 286 : }
2867 546091 : }
2868 :
2869 : // STEP 8.
2870 : // Fill the output entries with cache pointers and metadata based on
2871 : // the accounts we have located and the cache lines we have
2872 : // reserved.
2873 :
2874 1052878 : for( ulong i=0UL; i<pubkeys_cnt; i++ ) {
2875 558029 : if( FD_UNLIKELY( !accmetas[ i ] && !writable[ i ] ) ) {
2876 14939 : out_accs[ i ].data = NULL;
2877 14939 : out_accs[ i ].data_len = 0UL;
2878 14939 : out_accs[ i ].lamports = 0UL;
2879 14939 : out_accs[ i ].executable = 0;
2880 14939 : memset( out_accs[ i ].owner, 0, 32UL );
2881 14939 : fd_memcpy( out_accs[ i ].pubkey, pubkeys[ i ], 32UL );
2882 14939 : out_accs[ i ].prior_lamports = 0UL;
2883 14939 : out_accs[ i ].prior_data_len = 0UL;
2884 14939 : out_accs[ i ].prior_executable = 0;
2885 14939 : memset( out_accs[ i ].prior_owner, 0, 32UL );
2886 14939 : out_accs[ i ].prior_data = NULL;
2887 14939 : out_accs[ i ].commit = 0;
2888 14939 : out_accs[ i ].pd_write = 0;
2889 14939 : out_accs[ i ]._writable = 0;
2890 14939 : out_accs[ i ]._original_size_class = ULONG_MAX;
2891 14939 : out_accs[ i ]._original_cache_idx = ULONG_MAX;
2892 14939 : continue;
2893 14939 : }
2894 :
2895 543090 : if( FD_LIKELY( !writable[ i ] ) ) out_accs[ i ].data = (uchar *)(original_cache_line[ i ]+1UL);
2896 401712 : else out_accs[ i ].data = (uchar *)(destination_cache_lines[ i ][ 7UL ]+1UL);
2897 : /* Tombstone reset: agave's account loader returns AccountSharedData::default()
2898 : (System owner, empty data, exec=0) for any account with lamports==0.
2899 : https://github.com/anza-xyz/agave/blob/v2.3.1/svm/src/account_loader.rs#L199-L228 */
2900 543090 : fd_racesan_hook( "accdb_acquire:pre_step7_meta" );
2901 543090 : int tombstone = accmetas[ i ] && accmetas[ i ]->lamports==0UL;
2902 543090 : out_accs[ i ].data_len = ( accmetas[ i ] && !tombstone ) ? FD_ACCDB_SIZE_DATA( accmetas[ i ]->executable_size ) : 0UL;
2903 543090 : out_accs[ i ].executable = ( accmetas[ i ] && !tombstone ) ? FD_ACCDB_SIZE_EXEC( accmetas[ i ]->executable_size ) : 0;
2904 543090 : fd_racesan_hook( "accdb_acquire:mid_step7_meta" );
2905 543090 : out_accs[ i ].lamports = accmetas[ i ] ? accmetas[ i ]->lamports : 0UL;
2906 543090 : if( FD_UNLIKELY( !accmetas[ i ] ) ) {
2907 364538 : memset( out_accs[ i ].owner, 0, 32UL );
2908 364538 : memset( out_accs[ i ].prior_owner, 0, 32UL );
2909 364538 : }
2910 : /* For accmetas[i] != NULL, both owners are copied from the cache
2911 : line below in step 15, after step 12 has populated it from disk
2912 : for cold loads. */
2913 :
2914 543090 : out_accs[ i ].prior_lamports = out_accs[ i ].lamports;
2915 543090 : out_accs[ i ].prior_data_len = out_accs[ i ].data_len;
2916 543090 : out_accs[ i ].prior_executable = out_accs[ i ].executable;
2917 543090 : out_accs[ i ].prior_data = (uchar *)(original_cache_line[ i ] ? (original_cache_line[ i ]+1UL) : NULL);
2918 :
2919 543090 : out_accs[ i ].commit = 0;
2920 543090 : out_accs[ i ].pd_write = 0;
2921 543090 : out_accs[ i ]._writable = writable[ i ];
2922 543090 : if( FD_UNLIKELY( writable[ i ] && accmetas[ i ] ) ) out_accs[ i ]._overwrite = accdb->fork_pool[ fork_id.val ].shmem->generation==accmetas[ i ]->key.generation;
2923 505496 : else out_accs[ i ]._overwrite = 0;
2924 :
2925 543090 : FD_TEST( out_accs[ i ].data_len<=(10UL<<20) );
2926 543090 : FD_TEST( !out_accs[ i ]._overwrite || accdb->fork_pool[ fork_id.val ].shmem->generation==accmetas[ i ]->key.generation );
2927 :
2928 : #if FD_TMPL_USE_HANDHOLDING
2929 : if( FD_UNLIKELY( !writable[ i ] && accmetas[ i ] && !tombstone ) ) {
2930 : ulong cls = fd_accdb_cache_class( FD_ACCDB_SIZE_DATA( accmetas[ i ]->executable_size ) );
2931 : FD_TEST( fd_accdb_ptr_in_region( accdb, cls, out_accs[ i ].data ) );
2932 : }
2933 : #endif
2934 :
2935 543090 : if( FD_UNLIKELY( writable[ i ] ) ) {
2936 401282 : out_accs[ i ]._fork_id = fork_id.val;
2937 401282 : out_accs[ i ]._generation = fork->shmem->generation;
2938 401282 : out_accs[ i ]._acc_map_idx = acc_map_idxs[ i ];
2939 401282 : }
2940 543090 : fd_memcpy( out_accs[ i ].pubkey, pubkeys[ i ], 32UL );
2941 :
2942 543090 : if( FD_UNLIKELY( !accmetas[ i ] ) ) {
2943 363907 : out_accs[ i ]._original_size_class = ULONG_MAX;
2944 363907 : out_accs[ i ]._original_cache_idx = ULONG_MAX;
2945 363907 : } else {
2946 179183 : out_accs[ i ]._original_size_class = fd_accdb_cache_class( FD_ACCDB_SIZE_DATA( accmetas[ i ]->executable_size ) );
2947 179183 : out_accs[ i ]._original_cache_idx = cache_line_idx( accdb, out_accs[ i ]._original_size_class, original_cache_line[ i ] );
2948 179183 : }
2949 :
2950 543090 : if( FD_UNLIKELY( writable[ i ] ) ) {
2951 3573790 : for( ulong j=0UL; j<FD_ACCDB_CACHE_CLASS_CNT; j++ ) {
2952 3172641 : out_accs[ i ]._write.destination_cache_idx[ j ] = cache_line_idx( accdb, j, destination_cache_lines[ i ][ j ] );
2953 3172641 : }
2954 401149 : }
2955 543090 : }
2956 :
2957 : // STEP 9.
2958 : // Write the dirty eviction data to disk and publish the new offsets
2959 : // BEFORE constructing read iovecs. This is critical: step 4 may
2960 : // have evicted a dirty cache line belonging to another account in
2961 : // the same batch whose acc->offset is still FD_ACCDB_OFF_INVAL.
2962 : // The read-iovec loop below spin-waits on
2963 : // offset!=FD_ACCDB_OFF_INVAL, so publishing evicted offsets first
2964 : // prevents an intra-batch deadlock where the thread waits on an
2965 : // offset that only it can resolve.
2966 494849 : if( FD_LIKELY( batch_contiguous ) ) {
2967 : /* Fast path: all evictions fit in one contiguous region. Use the
2968 : pre-built iovec array for a single batched pwritev2 call. */
2969 0 : ulong bytes_written = 0UL;
2970 0 : struct iovec * write_ptr = write_ops;
2971 0 : while( FD_LIKELY( bytes_written<total_write_sz ) ) {
2972 0 : long result = pwritev2( accdb->fd, write_ptr, fd_int_min( write_ops_cnt, IOV_MAX ), (long)(file_offset+bytes_written), 0 );
2973 0 : if( FD_UNLIKELY( -1==result && (errno==EINTR || errno==EAGAIN || errno==EWOULDBLOCK ) ) ) continue;
2974 0 : else if( FD_UNLIKELY( -1==result ) ) FD_LOG_ERR(( "pwritev2() failed (%d-%s)", errno, fd_io_strerror( errno ) ));
2975 0 : else if( FD_UNLIKELY( !result ) ) FD_LOG_ERR(( "accounts database is corrupt, pwritev2() returned 0 at offset %lu with %lu bytes remaining",
2976 0 : file_offset+bytes_written, total_write_sz-bytes_written ));
2977 0 : bytes_written += (ulong)result;
2978 0 : accdb->metrics->bytes_written += (ulong)result;
2979 0 : accdb->metrics->write_ops++;
2980 :
2981 0 : while( write_ops_cnt && (ulong)result>=(ulong)write_ptr[ 0 ].iov_len ) {
2982 0 : result -= (long)write_ptr[ 0 ].iov_len;
2983 0 : write_ptr++;
2984 0 : write_ops_cnt--;
2985 0 : }
2986 0 : if( FD_LIKELY( write_ops_cnt ) ) {
2987 0 : write_ptr[ 0 ].iov_base = (uchar *)write_ptr[ 0 ].iov_base + result;
2988 0 : write_ptr[ 0 ].iov_len -= (ulong)result;
2989 0 : }
2990 0 : }
2991 494849 : } else {
2992 : /* Slow path: total eviction batch exceeds a single partition.
2993 : Write each entry individually using its own allocated offset.
2994 : This path is only taken in extreme edge cases (many concurrent
2995 : dirty 10 MiB evictions). */
2996 494849 : struct iovec * wp = write_ops;
2997 494849 : for( int k=0; k<pending_cnt; k++ ) {
2998 0 : ulong entry_sz = sizeof(fd_accdb_disk_meta_t) + (ulong)FD_ACCDB_SIZE_DATA( pending_accs[ k ]->executable_size );
2999 0 : ulong entry_off = pending_offs[ k ];
3000 0 : struct iovec entry_iovs[2] = { wp[0], wp[1] };
3001 0 : wp += 2;
3002 :
3003 0 : ulong written = 0UL;
3004 0 : while( FD_LIKELY( written<entry_sz ) ) {
3005 0 : long result = pwritev2( accdb->fd, entry_iovs, 2, (long)(entry_off+written), 0 );
3006 0 : if( FD_UNLIKELY( -1==result && (errno==EINTR || errno==EAGAIN || errno==EWOULDBLOCK ) ) ) continue;
3007 0 : else if( FD_UNLIKELY( -1==result ) ) FD_LOG_ERR(( "pwritev2() failed (%d-%s)", errno, fd_io_strerror( errno ) ));
3008 0 : else if( FD_UNLIKELY( !result ) ) FD_LOG_ERR(( "accounts database is corrupt, pwritev2() returned 0 at offset %lu with %lu bytes remaining", entry_off+written, entry_sz-written ));
3009 0 : written += (ulong)result;
3010 0 : accdb->metrics->bytes_written += (ulong)result;
3011 0 : accdb->metrics->write_ops++;
3012 :
3013 0 : for( int v=0; v<2; v++ ) {
3014 0 : if( (ulong)result>=(ulong)entry_iovs[ v ].iov_len ) {
3015 0 : result -= (long)entry_iovs[ v ].iov_len;
3016 0 : entry_iovs[ v ].iov_len = 0UL;
3017 0 : } else {
3018 0 : entry_iovs[ v ].iov_base = (uchar *)entry_iovs[ v ].iov_base + result;
3019 0 : entry_iovs[ v ].iov_len -= (ulong)result;
3020 0 : break;
3021 0 : }
3022 0 : }
3023 0 : }
3024 0 : }
3025 494849 : }
3026 :
3027 : // STEP 10.
3028 : // Now that the data is on disk, publish the evicted account offsets
3029 : // so concurrent acquire threads spinning on
3030 : // offset==FD_ACCDB_OFF_INVAL can proceed. The fence ensures
3031 : // pwritev2 data is globally visible before the offset stores.
3032 494849 : FD_COMPILER_MFENCE();
3033 494849 : for( int k=0; k<pending_cnt; k++ ) {
3034 0 : pending_accs[ k ]->offset_fork = fd_accdb_acc_pack_offset_fork( pending_offs[ k ], fd_accdb_acc_fork_id(pending_accs[ k ]) );
3035 0 : pending_lines[ k ]->persisted = 1;
3036 0 : }
3037 :
3038 : // STEP 11.
3039 : // Now construct iovecs for any reads we need to do of accounts into
3040 : // the cache. For reading accounts, we read them directly into the
3041 : // sole cache line we took (and maybe just evicted). For writing
3042 : // accounts, we read them into the right sized cache line, and later
3043 : // it will be copied to the staging buffer. This is to prevent
3044 : // repeatedly reading the same account off disk into cache, if it is
3045 : // being written cold multiple times and every write fails.
3046 :
3047 494849 : ulong read_ops_cnt = 0UL;
3048 494849 : ulong read_offsets[ FD_ACCDB_CACHE_CLASS_CNT*FD_ACCDB_MAX_ACQUIRE_CNT ];
3049 494849 : uchar * read_bases[ FD_ACCDB_CACHE_CLASS_CNT*FD_ACCDB_MAX_ACQUIRE_CNT ];
3050 494849 : ulong read_sizes[ FD_ACCDB_CACHE_CLASS_CNT*FD_ACCDB_MAX_ACQUIRE_CNT ];
3051 494849 : struct iovec read_ops[ FD_ACCDB_CACHE_CLASS_CNT*FD_ACCDB_MAX_ACQUIRE_CNT ];
3052 :
3053 1054653 : for( ulong i=0UL; i<pubkeys_cnt; i++ ) {
3054 559804 : if( FD_UNLIKELY( !accmetas[ i ] || exists_in_cache[ i ] ) ) continue;
3055 :
3056 342 : accdb->metrics->accounts_not_found_per_class[ fd_accdb_cache_class( FD_ACCDB_SIZE_DATA( accmetas[ i ]->executable_size ) ) ]++;
3057 :
3058 : /* Tombstones (lamports==0) have no on-disk payload to read, and
3059 : background_advance_root may unlink the acc and never assign it a
3060 : disk offset, so the offset_fork spin below would hang forever.
3061 : Step 15's tombstone reset zeros the owner for these accounts.
3062 :
3063 : Exception: when re-creating a closed account on the same fork,
3064 : wait anyway, or a concurrent evictor will later store the
3065 : closed account's disk offset into the new account. */
3066 342 : int tombstone = !accmetas[ i ]->lamports;
3067 342 : if( FD_UNLIKELY( tombstone && !out_accs[ i ]._overwrite ) ) continue;
3068 :
3069 : /* We are guaranteed that if an account is in the cache, the bytes
3070 : are available (all cache operations are atomic via refcnt CAS),
3071 : but we are not guaranteed that if something is _not_ in the cache
3072 : that it has been written back to disk yet. In particular, if we
3073 : are trying to read an account that another thread is in the
3074 : process of evicting, we know they removed it from the cache, but
3075 : we don't know exactly when they will have written it back fully
3076 : to disk, so we may need to wait for that here.
3077 :
3078 : Compaction may concurrently relocate this record, but
3079 : epoch-based safe reclamation guarantees the source partition
3080 : is not freed until all epoch-protected operations that could
3081 : have snapshotted the old offset have exited. So the data at the
3082 : snapshotted offset remains stable for the duration of our
3083 : read and no post-read validation is needed. */
3084 342 : ulong off_packed = FD_VOLATILE_CONST( accmetas[ i ]->offset_fork );
3085 342 : while( FD_UNLIKELY( (off_packed & FD_ACCDB_OFF_MASK)==FD_ACCDB_OFF_INVAL ) ) {
3086 0 : fd_racesan_hook( "accdb_acquire:offset_wait" );
3087 0 : FD_SPIN_PAUSE();
3088 0 : off_packed = FD_VOLATILE_CONST( accmetas[ i ]->offset_fork );
3089 0 : }
3090 342 : if( FD_UNLIKELY( tombstone ) ) continue;
3091 342 : fd_racesan_hook( "accdb_coldload:pre_iovec" );
3092 :
3093 342 : read_offsets[ read_ops_cnt ] = fd_accdb_acc_offset(accmetas[ i ]) + offsetof(fd_accdb_disk_meta_t, owner);
3094 342 : read_bases[ read_ops_cnt ] = original_cache_line[ i ]->owner;
3095 342 : read_sizes[ read_ops_cnt ] = 32UL + FD_ACCDB_SIZE_DATA( accmetas[ i ]->executable_size );
3096 342 : read_ops[ read_ops_cnt++ ] = (struct iovec){ .iov_base = original_cache_line[ i ]->owner, .iov_len = 32UL + FD_ACCDB_SIZE_DATA( accmetas[ i ]->executable_size ) };
3097 342 : }
3098 :
3099 : // STEP 12.
3100 : // Almost done... now do the actual reads of accounts into cache,
3101 : // using the iovecs we constructed. This is basically the same loop
3102 : // as the writes, but with preadv2 instead of pwritev2, and that the
3103 : // reads are not necessarily all contiguous, but occur at random
3104 : // offsets.
3105 : //
3106 : // CONCURRENCY: The compaction tile may concurrently relocate a
3107 : // record we are about to read (both are epoch-protected). Epoch-
3108 : // based safe reclamation guarantees the source partition is not
3109 : // freed until all epoch-protected operations that could have
3110 : // snapshotted the old offset have exited, so the data at the
3111 : // remains stable for the duration of this read — no post-read
3112 : // validation or retry is needed.
3113 494849 : for( ulong i=0UL; i<read_ops_cnt; i++ ) {
3114 0 : ulong bytes_read = 0UL;
3115 0 : while( FD_LIKELY( bytes_read<read_sizes[ i ] ) ) {
3116 0 : long result = preadv2( accdb->fd, &read_ops[ i ], 1, (long)(read_offsets[ i ]+bytes_read), 0 );
3117 0 : if( FD_UNLIKELY( -1==result && (errno==EINTR || errno==EAGAIN || errno==EWOULDBLOCK ) ) ) continue;
3118 0 : else if( FD_UNLIKELY( -1==result ) ) FD_LOG_ERR(( "preadv2() failed (%d-%s)", errno, fd_io_strerror( errno ) ));
3119 0 : else if( FD_UNLIKELY( !result ) ) FD_LOG_ERR(( "accounts database is corrupt, data expected at offset %lu with size %lu exceeded file extents",
3120 0 : read_offsets[ i ]+bytes_read, read_sizes[ i ] ));
3121 0 : fd_accdb_partition_read_bump( accdb, read_offsets[ i ]+bytes_read, (ulong)result );
3122 0 : bytes_read += (ulong)result;
3123 0 : accdb->metrics->bytes_read += (ulong)result;
3124 0 : accdb->metrics->read_ops++;
3125 :
3126 0 : read_ops[ i ].iov_base = read_bases[ i ] + bytes_read;
3127 0 : read_ops[ i ].iov_len = read_sizes[ i ] - bytes_read;
3128 0 : }
3129 0 : }
3130 :
3131 : // STEP 13.
3132 : // Publish the real acc index for any cache lines we just loaded
3133 : // from disk, so concurrent threads spinning on acc_idx==UINT_MAX
3134 : // can proceed. The fence ensures all preadv2 data is visible
3135 : // before the sentinel is cleared.
3136 494849 : FD_COMPILER_MFENCE();
3137 1053695 : for( ulong i=0UL; i<pubkeys_cnt; i++ ) {
3138 558939 : if( FD_UNLIKELY( !accmetas[ i ] || exists_in_cache[ i ] ) ) continue;
3139 >1844*10^16 : FD_VOLATILE( original_cache_line[ i ]->acc_idx ) = (uint)( accmetas[ i ] - accdb->acc_pool );
3140 >1844*10^16 : FD_TEST( FD_VOLATILE_CONST( original_cache_line[ i ]->acc_idx )==(uint)( accmetas[ i ] - accdb->acc_pool ) );
3141 >1844*10^16 : }
3142 :
3143 : // STEP 14.
3144 : // Spin-wait for any cache lines found via acc->cache_idx that are
3145 : // still being loaded by another thread's preadv2. The loading
3146 : // thread sets acc_idx to UINT_MAX before publishing cache_idx
3147 : // and publishes the real acc index after its read completes.
3148 : // This step is placed as late as possible to give the loading
3149 : // thread maximum time to finish before we need to spin.
3150 1054263 : for( ulong i=0UL; i<pubkeys_cnt; i++ ) {
3151 558974 : if( FD_UNLIKELY( !accmetas[ i ] && !writable[ i ] ) ) continue;
3152 :
3153 544039 : if( FD_UNLIKELY( !original_cache_line[ i ] ) ) continue;
3154 178855 : if( FD_LIKELY( FD_VOLATILE_CONST( original_cache_line[ i ]->acc_idx )!=UINT_MAX ) ) goto step13_check;
3155 >1844*10^16 : accdb->metrics->accounts_waited++;
3156 >1844*10^16 : while( FD_UNLIKELY( FD_VOLATILE_CONST( original_cache_line[ i ]->acc_idx )==UINT_MAX ) ) {
3157 0 : fd_racesan_hook( "accdb_acquire:step14_load_wait" );
3158 0 : FD_SPIN_PAUSE();
3159 0 : }
3160 >1844*10^16 : step13_check:;
3161 : #if FD_TMPL_USE_HANDHOLDING
3162 : FD_TEST( original_cache_line[ i ]->key.generation==accmetas[ i ]->key.generation &&
3163 : !memcmp( original_cache_line[ i ]->key.pubkey, pubkeys[ i ], 32UL ) );
3164 : #endif
3165 178842 : }
3166 :
3167 : // STEP 15.
3168 : // Now that all reads from disk into original_cache_line have
3169 : // completed (and any concurrent loaders have published their
3170 : // acc_idx in step 14), copy the owner into the output entries.
3171 : // This must happen here rather than in step 8 because the cache
3172 : // line owner is only valid post-read for cold loads.
3173 1054105 : for( ulong i=0UL; i<pubkeys_cnt; i++ ) {
3174 558816 : if( FD_UNLIKELY( !accmetas[ i ] ) ) continue;
3175 178344 : fd_racesan_hook( "accdb_acquire:pre_step14_owner" );
3176 : /* Tombstone reset: see STEP 7 comment. */
3177 178344 : if( FD_UNLIKELY( accmetas[ i ]->lamports==0UL ) ) {
3178 0 : memset( out_accs[ i ].owner, 0, 32UL );
3179 0 : memset( out_accs[ i ].prior_owner, 0, 32UL );
3180 178344 : } else {
3181 178344 : fd_memcpy( out_accs[ i ].owner, original_cache_line[ i ]->owner, 32UL );
3182 178344 : fd_memcpy( out_accs[ i ].prior_owner, original_cache_line[ i ]->owner, 32UL );
3183 178344 : }
3184 178344 : }
3185 :
3186 : // STEP 16.
3187 : // Finally, copy any accounts we are writing into the staging
3188 : // buffers, so they occupy a 10MiB cache line for the execution
3189 : // system.
3190 1053720 : for( ulong i=0UL; i<pubkeys_cnt; i++ ) {
3191 558431 : if( FD_UNLIKELY( !accmetas[ i ] || !writable[ i ] ) ) continue;
3192 :
3193 37370 : ulong copy_sz = (ulong)FD_ACCDB_SIZE_DATA( accmetas[ i ]->executable_size );
3194 37370 : fd_memcpy( destination_cache_lines[ i ][ 7UL ]+1UL, original_cache_line[ i ]+1UL, copy_sz );
3195 37370 : accdb->metrics->bytes_copied += copy_sz;
3196 37370 : }
3197 :
3198 495289 : FD_COMPILER_MFENCE();
3199 495289 : FD_VOLATILE( *accdb->my_epoch_slot ) = ULONG_MAX;
3200 495289 : }
3201 :
3202 : void
3203 : fd_accdb_acquire( fd_accdb_t * accdb,
3204 : fd_accdb_fork_id_t fork_id,
3205 : ulong pubkeys_cnt,
3206 : uchar const * const * pubkeys,
3207 : int * writable,
3208 480737 : fd_acc_t * out_accs ) {
3209 480737 : FD_TEST( accdb->acquire_state==FD_ACCDB_ACQUIRE_STATE_IDLE );
3210 480737 : accdb->acquire_state = FD_ACCDB_ACQUIRE_STATE_OPEN;
3211 480737 : fd_accdb_acquire_inner( accdb, fork_id, RESERVATION_TYPE_SIMPLE, 0UL, pubkeys_cnt, pubkeys, writable, out_accs );
3212 480737 : }
3213 :
3214 : void
3215 : fd_accdb_acquire_a( fd_accdb_t * accdb,
3216 : fd_accdb_fork_id_t fork_id,
3217 : ulong pubkeys_cnt,
3218 : uchar const * const * pubkeys,
3219 : int * writable,
3220 7290 : fd_acc_t * out_accs ) {
3221 7290 : FD_TEST( accdb->acquire_state==FD_ACCDB_ACQUIRE_STATE_IDLE );
3222 7290 : accdb->acquire_state = FD_ACCDB_ACQUIRE_STATE_PHASE_A;
3223 7290 : fd_accdb_acquire_inner( accdb, fork_id, RESERVATION_TYPE_MAYBE_PROGRAMDATA, 0UL, pubkeys_cnt, pubkeys, writable, out_accs );
3224 7290 : }
3225 :
3226 : void
3227 : fd_accdb_acquire_b( fd_accdb_t * accdb,
3228 : fd_accdb_fork_id_t fork_id,
3229 : ulong reserved_cnt,
3230 : ulong pubkeys_cnt,
3231 : uchar const * const * pubkeys,
3232 7295 : fd_acc_t * out_accs ) {
3233 7295 : static int const readonly[ FD_ACCDB_MAX_ACQUIRE_CNT ] = {0};
3234 7295 : FD_TEST( accdb->acquire_state==FD_ACCDB_ACQUIRE_STATE_PHASE_A );
3235 7295 : accdb->acquire_state = FD_ACCDB_ACQUIRE_STATE_OPEN;
3236 7295 : fd_accdb_acquire_inner( accdb, fork_id, RESERVATION_TYPE_ALREADY_RESERVED, reserved_cnt, pubkeys_cnt, pubkeys, readonly, out_accs );
3237 7295 : }
3238 :
3239 : /* release_inner drains one group of acquired accs but does NOT change the
3240 : handle's acquire_state. The public fd_accdb_release / fd_accdb_release_ab
3241 : wrappers below own the state transition (a single-phase release closes
3242 : the bracket; release_ab drains both phase groups then closes). */
3243 : static void
3244 : release_inner( fd_accdb_t * accdb,
3245 : ulong accs_cnt,
3246 487502 : fd_acc_t * accs ) {
3247 487502 : FD_TEST( accdb->acquire_state==FD_ACCDB_ACQUIRE_STATE_OPEN );
3248 :
3249 487502 : {
3250 487502 : ulong prev = FD_VOLATILE_CONST( *accdb->my_epoch_slot );
3251 487502 : FD_TEST( prev==ULONG_MAX || prev<=FD_VOLATILE_CONST( accdb->shmem->epoch ) );
3252 487502 : }
3253 :
3254 487502 : FD_COMPILER_MFENCE();
3255 487502 : FD_VOLATILE( *accdb->my_epoch_slot ) = FD_VOLATILE_CONST( accdb->shmem->epoch );
3256 487502 : FD_HW_MFENCE(); /* StoreLoad: epoch store must be globally visible
3257 : before any subsequent loads so the deferred
3258 : reclamation scan does not miss us. */
3259 :
3260 : // STEP 1.
3261 : // For each cache line which was written to in the 10MiB staging
3262 : // buffer, we may need to copy to the data out to a right sized
3263 : // cache line. Figuring out the target cache line is non-obvious,
3264 : // but follows the more complete logic below this, we just pull the
3265 : // memcpy out so they are not done inside the cache lock.
3266 :
3267 1051150 : for( ulong i=0UL; i<accs_cnt; i++ ) {
3268 563648 : if( FD_UNLIKELY( accs[ i ]._original_size_class==ULONG_MAX && !accs[ i ]._writable ) ) continue;
3269 :
3270 : #if FD_TMPL_USE_HANDHOLDING
3271 : if( FD_LIKELY( accs[ i ]._original_size_class!=ULONG_MAX ) ) {
3272 : FD_TEST( accs[ i ]._original_cache_idx<accdb->shmem->cache_class_max[ accs[ i ]._original_size_class ] );
3273 : }
3274 : if( FD_UNLIKELY( accs[ i ].commit ) ) FD_TEST( accs[ i ]._writable );
3275 : #endif
3276 :
3277 548711 : if( FD_LIKELY( !accs[ i ]._writable || !accs[ i ].commit ) ) continue;
3278 : #if FD_TMPL_USE_HANDHOLDING
3279 : if( FD_UNLIKELY( accs[ i ]._overwrite ) ) {
3280 : FD_TEST( accs[ i ]._writable );
3281 : FD_TEST( accs[ i ]._original_cache_idx!=ULONG_MAX );
3282 : FD_TEST( accs[ i ]._original_size_class!=ULONG_MAX );
3283 : }
3284 : #endif
3285 :
3286 372247 : ulong original_size_class = accs[ i ]._original_size_class;
3287 372247 : ulong new_size_class = fd_accdb_cache_class( accs[ i ].data_len );
3288 372247 : if( FD_UNLIKELY( new_size_class==7UL ) ) continue;
3289 :
3290 372220 : fd_accdb_cache_line_t * target_cache_line;
3291 372220 : if( FD_LIKELY( original_size_class==new_size_class && accs[ i ]._overwrite ) ) target_cache_line = cache_line( accdb, original_size_class, accs[ i ]._original_cache_idx );
3292 368003 : else target_cache_line = cache_line( accdb, new_size_class, accs[ i ]._write.destination_cache_idx[ new_size_class ] );
3293 :
3294 372220 : fd_accdb_cache_line_t * staging_line = cache_line( accdb, 7UL, accs[ i ]._write.destination_cache_idx[ 7UL ] );
3295 :
3296 372220 : fd_racesan_hook( "accdb_commit:pre_owner_write" );
3297 :
3298 : #if FD_TMPL_USE_HANDHOLDING
3299 : if( FD_UNLIKELY( original_size_class==new_size_class && accs[ i ]._overwrite ) ) {
3300 : uint rc = FD_VOLATILE_CONST( target_cache_line->refcnt );
3301 : FD_TEST( target_cache_line->key.generation==accs[ i ]._generation &&
3302 : !memcmp( target_cache_line->key.pubkey, accs[ i ].pubkey, 32UL ) &&
3303 : rc>0U &&
3304 : rc!=FD_ACCDB_EVICT_SENTINEL );
3305 : }
3306 : #endif
3307 :
3308 372220 : fd_memcpy( target_cache_line->owner, accs[ i ].owner, 32UL );
3309 372220 : fd_memcpy( target_cache_line+1UL, staging_line+1UL, accs[ i ].data_len );
3310 372220 : accdb->metrics->bytes_copied += accs[ i ].data_len;
3311 372220 : }
3312 :
3313 : // STEP 2.
3314 : // Now update the metadata structures and free lists to reflect the
3315 : // fact that we are done with these cache lines. This is fully
3316 : // atomic with CLOCK.
3317 :
3318 1046534 : for( ulong i=0UL; i<accs_cnt; i++ ) {
3319 559032 : if( FD_UNLIKELY( accs[ i ]._original_size_class==ULONG_MAX && !accs[ i ]._writable ) ) continue;
3320 :
3321 544092 : ulong original_size_class = accs[ i ]._original_size_class;
3322 544092 : fd_accdb_cache_line_t * original_cache_line = accs[ i ]._original_cache_idx==ULONG_MAX ? NULL : cache_line( accdb, original_size_class, accs[ i ]._original_cache_idx );
3323 : /* For overwrite commits, defer the refcnt decrement on
3324 : original_cache_line until after invalidation completes. If
3325 : we dropped refcnt to 0 here, a concurrent CLOCK sweep could
3326 : CAS(refcnt, 0, EVICT_SENTINEL) and steal the line before we
3327 : get to invalidate it, causing data corruption.
3328 : Non-overwrite and non-commit paths unpin
3329 : immediately because they never invalidate the original line. */
3330 544092 : if( FD_LIKELY( original_cache_line ) ) {
3331 : #if FD_TMPL_USE_HANDHOLDING
3332 : FD_TEST( original_cache_line->refcnt>0U );
3333 : #endif
3334 179142 : if( FD_LIKELY( !accs[ i ]._writable || !accs[ i ].commit || !accs[ i ]._overwrite ) ) {
3335 174801 : FD_ATOMIC_FETCH_AND_SUB( &original_cache_line->refcnt, 1U );
3336 174801 : }
3337 179142 : }
3338 :
3339 544092 : if( FD_LIKELY( !accs[ i ]._writable ) ) {
3340 : /* For readonly accounts, mark as recently used so the CLOCK
3341 : algorithm gives it a second chance before eviction. */
3342 : #if FD_TMPL_USE_HANDHOLDING
3343 : FD_TEST( original_cache_line );
3344 : #endif
3345 141605 : original_cache_line->referenced = 1;
3346 141605 : continue;
3347 141605 : }
3348 :
3349 402487 : fd_accdb_cache_line_t * destination_cache_lines[ FD_ACCDB_CACHE_CLASS_CNT ];
3350 3575586 : for( ulong j=0UL; j<FD_ACCDB_CACHE_CLASS_CNT; j++ ) destination_cache_lines[ j ] = cache_line( accdb, j, accs[ i ]._write.destination_cache_idx[ j ] );
3351 402487 : int destination_committed[ FD_ACCDB_CACHE_CLASS_CNT ] = {0};
3352 :
3353 402487 : if( FD_LIKELY( !accs[ i ].commit ) ) {
3354 : /* If it's writable but it didn't commit, all of the destination
3355 : cache lines (including the staging buffer which is trashed) are
3356 : unused and can be pushed to the CAS free list for immediate
3357 : reuse. Whatever buffer it was accessing also gets marked as
3358 : recently used. */
3359 34894 : if( FD_LIKELY( original_cache_line ) ) original_cache_line->referenced = 1;
3360 314048 : for( ulong j=0UL; j<FD_ACCDB_CACHE_CLASS_CNT; j++ ) {
3361 : /* acquire_cache_line via CLOCK leaves line->acc_idx pointing
3362 : at the prior owner. cache_free_push consumers (CLOCK,
3363 : background_preevict) skip lines only when acc_idx==UINT_MAX
3364 : AND gen==UINT_MAX; if we leave the stale acc_idx, a future
3365 : CLOCK pick would call line 849/853 against the wrong acc
3366 : and corrupt its cache_idx/valid. */
3367 279154 : destination_cache_lines[ j ]->acc_idx = UINT_MAX;
3368 279154 : destination_cache_lines[ j ]->key.generation = UINT_MAX;
3369 279154 : destination_cache_lines[ j ]->persisted = 1;
3370 279154 : while( FD_UNLIKELY( FD_ATOMIC_CAS( &destination_cache_lines[ j ]->refcnt, 1U, 0U )!=1U ) ) {
3371 0 : fd_racesan_hook( "accdb_release:dest_refcnt_wait" );
3372 0 : FD_SPIN_PAUSE();
3373 0 : }
3374 279154 : cache_free_push( accdb, j, destination_cache_lines[ j ] );
3375 279154 : }
3376 34894 : continue;
3377 34894 : }
3378 :
3379 367593 : ulong new_size_class = fd_accdb_cache_class( accs[ i ].data_len );
3380 367593 : uint original_acc_idx = original_cache_line ? original_cache_line->acc_idx : UINT_MAX;
3381 367593 : fd_accdb_cache_line_t * committed_line;
3382 :
3383 : /* For overwrites, invalidate the on-disk offset BEFORE removing
3384 : the cache acc. This ensures a concurrent acquire that misses
3385 : the cache will see offset==FD_ACCDB_OFF_INVAL and spin-wait,
3386 : rather than reading stale on-disk bytes from the old location.
3387 : The CAS-loop exchange also serializes with a concurrent
3388 : compaction CAS (old_offset -> dest_offset). */
3389 367593 : ulong old_offset = FD_ACCDB_OFF_INVAL;
3390 367593 : if( FD_LIKELY( accs[ i ]._overwrite ) ) {
3391 4298 : fd_accdb_accmeta_t * ow_accmeta = &accdb->acc_pool[ original_acc_idx ];
3392 4298 : fd_racesan_hook( "accdb_overwrite:pre_xchg_offset" );
3393 4298 : old_offset = fd_accdb_acc_xchg_offset( ow_accmeta, FD_ACCDB_OFF_INVAL );
3394 4298 : if( FD_LIKELY( old_offset!=FD_ACCDB_OFF_INVAL ) ) {
3395 0 : fd_accdb_shmem_bytes_freed( accdb->shmem, old_offset, (ulong)FD_ACCDB_SIZE_DATA(ow_accmeta->executable_size)+sizeof(fd_accdb_disk_meta_t) );
3396 0 : FD_ATOMIC_FETCH_AND_SUB( &accdb->shmem->shmetrics->disk_used_bytes, (ulong)FD_ACCDB_SIZE_DATA(ow_accmeta->executable_size)+sizeof(fd_accdb_disk_meta_t) );
3397 0 : }
3398 4298 : }
3399 :
3400 367593 : if( FD_UNLIKELY( new_size_class==7UL ) ) {
3401 : /* The account belongs in the largest size class, and we already
3402 : have it resident in a 10MiB buffer anyway, so no need to copy
3403 : back. If we are "overwriting" (same generation as the account
3404 : came from), then the original can be discarded (pushed to
3405 : the CAS free list) and removed from the cache. */
3406 27 : destination_cache_lines[ 7UL ]->persisted = 0;
3407 27 : destination_committed[ 7UL ] = 1;
3408 27 : if( FD_LIKELY( accs[ i ]._overwrite ) ) {
3409 : /* Atomically clear acc.VALID and acc.cache_idx BEFORE freeing
3410 : the line, so a reader cannot observe acc.VALID=1 with
3411 : acc.cache_idx pointing at a line that has been recycled to
3412 : another acc. evict_clear_acc_cache_ref uses the CLAIM
3413 : protocol to serialize with cold_load_acc. */
3414 0 : evict_clear_acc_cache_ref( &accdb->acc_pool[ original_acc_idx ], original_size_class, accs[ i ]._original_cache_idx );
3415 :
3416 : /* Convert our own pin directly into the eviction claim
3417 : (CAS refcnt 1 -> EVICT_SENTINEL) so refcnt never passes
3418 : through 0 while acc_idx/key.generation are still valid:
3419 : in such a window background_preevict could claim and free
3420 : the line, and a claim of our own would then succeed on
3421 : the free-listed line and push it a second time. The CAS
3422 : fails if a concurrent reader that pinned the line via
3423 : cache_try_pin BEFORE evict_clear_acc_cache_ref completed
3424 : still holds a reference (its ABA check on
3425 : line->key.generation is not synchronized with our writes
3426 : to that field); then just drop our pin and leave
3427 : acc_idx/key.generation intact so CLOCK reclaims the line
3428 : once the reader unpins. That reclaim's
3429 : evict_clear_acc_cache_ref is a no-op, but its dirty-
3430 : writeback gate keys on persisted/acc_idx and would
3431 : republish the pre-overwrite bytes over the committed
3432 : version, so set persisted while still pinned. */
3433 0 : original_cache_line->persisted = 1;
3434 0 : fd_racesan_hook( "accdb_release:pre_discard_claim" );
3435 0 : if( FD_LIKELY( FD_ATOMIC_CAS( &original_cache_line->refcnt, 1U, FD_ACCDB_EVICT_SENTINEL )==1U ) ) {
3436 0 : original_cache_line->acc_idx = UINT_MAX;
3437 0 : original_cache_line->key.generation = UINT_MAX;
3438 0 : FD_COMPILER_MFENCE();
3439 0 : FD_VOLATILE( original_cache_line->refcnt ) = 0;
3440 0 : cache_free_push( accdb, original_size_class, original_cache_line );
3441 0 : } else {
3442 0 : FD_ATOMIC_FETCH_AND_SUB( &original_cache_line->refcnt, 1U );
3443 0 : }
3444 0 : }
3445 27 : committed_line = destination_cache_lines[ 7UL ];
3446 367566 : } else {
3447 : /* The account started in some arbitrary size class, transited
3448 : through a 10MiB staging buffer, and is now being written back
3449 : to some arbitrary (non-10MiB) size class, so we need to copy it
3450 : there. The staging buffer is discarded. If we are going to
3451 : a different size class, and we are "overwriting" (same
3452 : generation), then the original can also be discarded, but if
3453 : we are staying in the same size class, we can reuse the cache
3454 : line in place. */
3455 367566 : fd_accdb_cache_line_t * target_cache_line;
3456 367566 : if( FD_LIKELY( original_size_class==new_size_class ) ) {
3457 4215 : if( FD_LIKELY( accs[ i ]._overwrite ) ) {
3458 : /* a reader holding a stale acc->cache_idx from this line's
3459 : previous life may hold a transient cache_try_pin pin, so
3460 : our own pin only bounds refcnt from below. */
3461 4215 : uint ow_rc = FD_VOLATILE_CONST( original_cache_line->refcnt );
3462 4215 : FD_TEST( ow_rc>0U && ow_rc!=FD_ACCDB_EVICT_SENTINEL );
3463 4215 : original_cache_line->key.generation = UINT_MAX;
3464 : /* Keep refcnt>=1 through the reuse window so CLOCK cannot
3465 : steal the line between invalidation and re-publish. The
3466 : pin is released in the destination cleanup loop after
3467 : acc->cache_idx has been republished. */
3468 4215 : original_cache_line->acc_idx = UINT_MAX;
3469 4215 : target_cache_line = original_cache_line;
3470 4215 : } else {
3471 0 : target_cache_line = destination_cache_lines[ new_size_class ];
3472 0 : destination_committed[ new_size_class ] = 1;
3473 0 : }
3474 363351 : } else {
3475 363351 : if( FD_LIKELY( accs[ i ]._overwrite ) ) {
3476 : /* Atomically clear acc.VALID and acc.cache_idx BEFORE freeing
3477 : the line, so a reader cannot observe acc.VALID=1 with
3478 : acc.cache_idx pointing at a line that has been recycled to
3479 : another acc. evict_clear_acc_cache_ref uses the CLAIM
3480 : protocol to serialize with cold_load_acc. See the
3481 : size_class==7 path above for the refcnt claim and
3482 : persisted rationale. */
3483 90 : evict_clear_acc_cache_ref( &accdb->acc_pool[ original_acc_idx ], original_size_class, accs[ i ]._original_cache_idx );
3484 90 : original_cache_line->persisted = 1;
3485 90 : fd_racesan_hook( "accdb_release:pre_discard_claim" );
3486 90 : if( FD_LIKELY( FD_ATOMIC_CAS( &original_cache_line->refcnt, 1U, FD_ACCDB_EVICT_SENTINEL )==1U ) ) {
3487 90 : original_cache_line->acc_idx = UINT_MAX;
3488 90 : original_cache_line->key.generation = UINT_MAX;
3489 90 : FD_COMPILER_MFENCE();
3490 90 : FD_VOLATILE( original_cache_line->refcnt ) = 0;
3491 90 : cache_free_push( accdb, original_size_class, original_cache_line );
3492 90 : } else {
3493 0 : FD_ATOMIC_FETCH_AND_SUB( &original_cache_line->refcnt, 1U );
3494 0 : }
3495 90 : }
3496 :
3497 363351 : destination_committed[ new_size_class ] = 1;
3498 363351 : target_cache_line = destination_cache_lines[ new_size_class ];
3499 363351 : }
3500 :
3501 367566 : target_cache_line->persisted = 0;
3502 : /* If target is the original cache line (overwrite, same size
3503 : class), mark as referenced directly since the cleanup loop
3504 : only handles destination lines. */
3505 367566 : if( FD_LIKELY( !destination_committed[ new_size_class ] ) ) target_cache_line->referenced = 1;
3506 367566 : committed_line = target_cache_line;
3507 367566 : }
3508 :
3509 : /* For non-overwrite commits, the original cache line (if any) still
3510 : holds valid ancestor data but is no longer pinned. Mark it as
3511 : recently used so the CLOCK algorithm retains it. */
3512 367593 : if( FD_UNLIKELY( !accs[ i ]._overwrite && original_cache_line ) ) {
3513 0 : original_cache_line->referenced = 1;
3514 0 : }
3515 :
3516 : /* Handle every destination cache line: committed ones keep
3517 : refcnt>=1 until acc->cache_idx is published (the deferred
3518 : unpin happens after the publish below), uncommitted ones are
3519 : fully freed to the CAS free list. */
3520 3312070 : for( ulong j=0UL; j<FD_ACCDB_CACHE_CLASS_CNT; j++ ) {
3521 2944477 : if( destination_committed[ j ] ) {
3522 357847 : destination_cache_lines[ j ]->referenced = 1;
3523 2586630 : } else {
3524 : /* See note above (no-commit path): clear stale acc_idx/gen
3525 : before pushing, otherwise CLOCK can pick this line and
3526 : stomp the prior owner's cache_idx/valid. CAS on refcnt for
3527 : the same stray-pin reason as the no-commit path. */
3528 2586630 : destination_cache_lines[ j ]->acc_idx = UINT_MAX;
3529 2586630 : destination_cache_lines[ j ]->key.generation = UINT_MAX;
3530 2586630 : destination_cache_lines[ j ]->persisted = 1;
3531 2586630 : while( FD_UNLIKELY( FD_ATOMIC_CAS( &destination_cache_lines[ j ]->refcnt, 1U, 0U )!=1U ) ) {
3532 0 : fd_racesan_hook( "accdb_release:dest_refcnt_wait" );
3533 0 : FD_SPIN_PAUSE();
3534 0 : }
3535 2586630 : cache_free_push( accdb, j, destination_cache_lines[ j ] );
3536 2586630 : }
3537 2944477 : }
3538 :
3539 : /* Update the accounts index for this committed write. For an
3540 : overwrite (same fork+generation), update the existing acc
3541 : in place. Otherwise allocate a new acc, prepend it
3542 : to the hash chain, and record the write in a txn linked to
3543 : the fork so advance_root can clean up old versions. */
3544 367593 : if( FD_LIKELY( accs[ i ]._overwrite ) ) {
3545 4309 : accdb->metrics->accounts_committed_overwrite_per_class[ new_size_class ]++;
3546 4309 : committed_line->acc_idx = original_acc_idx;
3547 :
3548 4309 : fd_accdb_accmeta_t * accmeta = &accdb->acc_pool[ original_acc_idx ];
3549 : /* The offset was already atomically swapped to FD_ACCDB_OFF_INVAL
3550 : and bytes freed above, so just update the metadata and
3551 : re-publish the cache location. CAS-loop preserves CLAIM bit
3552 : (a concurrent evict_clear_acc_cache_ref or acc_unlink may
3553 : hold it) and clears VALID; a plain store would clobber CLAIM
3554 : and break those protocols. */
3555 4309 : uint pd = accs[ i ].pd_write ? FD_ACCDB_SIZE_PD_WRITE_BIT : 0U;
3556 4309 : for(;;) {
3557 4309 : uint cur = FD_VOLATILE_CONST( accmeta->executable_size );
3558 4309 : uint nxt = (cur & (FD_ACCDB_SIZE_CACHE_CLAIM_BIT|FD_ACCDB_SIZE_PD_WRITE_BIT))
3559 4309 : | FD_ACCDB_SIZE_PACK( (uint)accs[ i ].data_len, accs[ i ].executable )
3560 4309 : | pd;
3561 4309 : if( FD_LIKELY( FD_ATOMIC_CAS( &accmeta->executable_size, cur, nxt )==cur ) ) break;
3562 0 : FD_SPIN_PAUSE();
3563 0 : }
3564 4309 : accmeta->lamports = accs[ i ].lamports;
3565 4309 : fd_racesan_hook( "accdb_overwrite:mid_inplace" );
3566 :
3567 4309 : fd_memcpy( committed_line->owner, accs[ i ].owner, 32UL );
3568 4309 : fd_memcpy( committed_line->key.pubkey, accmeta->key.pubkey, 32UL );
3569 4309 : committed_line->key.generation = accmeta->key.generation;
3570 4309 : committed_line->acc_idx = original_acc_idx;
3571 4309 : FD_VOLATILE( accmeta->cache_idx ) = FD_ACCDB_ACC_CIDX_PACK( (uint)new_size_class, (uint)cache_line_idx( accdb, new_size_class, committed_line ) );
3572 : /* Atomic OR so a concurrent evict_clear_acc_cache_ref's CLAIM
3573 : clear (FETCH_AND_AND with ~CLAIM) cannot be lost by an RMW
3574 : race with a plain |= store. */
3575 4309 : FD_ATOMIC_FETCH_AND_OR( &accmeta->executable_size, FD_ACCDB_SIZE_CACHE_VALID_BIT );
3576 :
3577 : /* Now that acc->cache_idx is published, unpin so CLOCK can
3578 : eventually evict it. For same-size overwrites, committed_line
3579 : IS the reused original_cache_line. For cross-size overwrites,
3580 : committed_line is a destination line whose refcnt decrement was
3581 : deferred from the cleanup loop. */
3582 4309 : FD_ATOMIC_FETCH_AND_SUB( &committed_line->refcnt, 1U );
3583 4309 : committed_line->referenced = 1;
3584 363284 : } else {
3585 363284 : accdb->metrics->accounts_committed_new_per_class[ new_size_class ]++;
3586 363284 : fd_accdb_accmeta_t * accmeta = acc_pool_acquire( accdb->acc_pool_join );
3587 363284 : if( FD_UNLIKELY( !accmeta ) ) FD_LOG_ERR(( "accounts database ran out of space, increase [accounts.max_accounts], current value is %lu", acc_pool_ele_max( accdb->acc_pool_join ) ));
3588 363284 : ulong acc_idx = acc_pool_idx( accdb->acc_pool_join, accmeta );
3589 363284 : fd_memcpy( accmeta->key.pubkey, accs[ i ].pubkey, 32UL );
3590 363284 : accmeta->lamports = accs[ i ].lamports;
3591 363284 : accmeta->executable_size = FD_ACCDB_SIZE_PACK( (uint)accs[ i ].data_len, accs[ i ].executable )
3592 363284 : | (accs[ i ].pd_write ? FD_ACCDB_SIZE_PD_WRITE_BIT : 0U);
3593 363284 : accmeta->key.generation = accs[ i ]._generation;
3594 363284 : accmeta->offset_fork = fd_accdb_acc_pack_offset_fork( FD_ACCDB_OFF_INVAL, accs[ i ]._fork_id );
3595 :
3596 : /* Publish in the cache BEFORE the acc_map head so that a
3597 : concurrent acquire that finds this acc in the hash chain will
3598 : also find a cache hit, rather than inserting a conflicting
3599 : placeholder cache acc. */
3600 363284 : committed_line->acc_idx = (uint)acc_idx;
3601 363284 : fd_memcpy( committed_line->owner, accs[ i ].owner, 32UL );
3602 363284 : fd_memcpy( committed_line->key.pubkey, accmeta->key.pubkey, 32UL );
3603 363284 : committed_line->key.generation = accmeta->key.generation;
3604 363284 : FD_VOLATILE( accmeta->cache_idx ) = FD_ACCDB_ACC_CIDX_PACK( (uint)new_size_class, (uint)cache_line_idx( accdb, new_size_class, committed_line ) );
3605 : /* Atomic OR so a concurrent evict_clear_acc_cache_ref's CLAIM
3606 : clear (FETCH_AND_AND with ~CLAIM) cannot be lost by an RMW
3607 : race with a plain |= store. */
3608 363284 : FD_ATOMIC_FETCH_AND_OR( &accmeta->executable_size, FD_ACCDB_SIZE_CACHE_VALID_BIT );
3609 :
3610 : /* Now that acc->cache_idx is published, unpin it so
3611 : CLOCK can eventually evict it. */
3612 363284 : FD_ATOMIC_FETCH_AND_SUB( &committed_line->refcnt, 1U );
3613 363284 : committed_line->referenced = 1;
3614 :
3615 : /* CAS loop to prepend to the hash chain. Succeeds on the first
3616 : try in most cases, but a concurrent acc_unlink CAS removing
3617 : the old head can change acc_map[idx] between our load and
3618 : CAS. Multiple concurrent releases may also race on the head
3619 : pointer — the CAS retry handles this. */
3620 368511 : for(;;) {
3621 368511 : uint old_head = FD_VOLATILE_CONST( accdb->acc_map[ accs[ i ]._acc_map_idx ] );
3622 368511 : accmeta->map.next = old_head;
3623 368511 : FD_COMPILER_MFENCE();
3624 368511 : fd_racesan_hook( "accdb_release:pre_chain_cas" );
3625 368511 : if( FD_LIKELY( FD_ATOMIC_CAS( &accdb->acc_map[ accs[ i ]._acc_map_idx ], old_head, (uint)acc_idx )==old_head ) ) break;
3626 239 : FD_SPIN_PAUSE();
3627 239 : }
3628 :
3629 : /* CONCURRENCY: The cache acc is published before the acc_map
3630 : head so that a concurrent fd_accdb_acquire reader that
3631 : observes the new head also finds a cache hit, preventing
3632 : duplicate cache insertion.
3633 :
3634 : (1) The CAS on acc_map[idx] serializes head-pointer mutations
3635 : from concurrent releases onto the same chain without any
3636 : external lock.
3637 :
3638 : (2) The FD_COMPILER_MFENCE above ensures stores to the acc node
3639 : fields (pubkey, lamports, size, generation, fork_id,
3640 : offset, map.next) are ordered before the CAS that publishes
3641 : the new head. On x86-64 (TSO), hardware also guarantees
3642 : this, but the compiler fence is needed to prevent the
3643 : compiler from reordering the stores. A reader that
3644 : observes the new head is guaranteed to see a fully
3645 : initialized node. A reader that has not yet seen the new
3646 : head simply traverses the previous (still valid) chain.
3647 :
3648 : (3) A concurrent acc_unlink (advance_root / purge) may CAS the
3649 : head away between our load and CAS here. The CAS retry
3650 : loop handles this. */
3651 :
3652 363284 : fd_accdb_txn_t * txn = txn_pool_acquire( accdb->txn_pool );
3653 363284 : FD_TEST( txn ); /* Sized so it always succeeds */
3654 363284 : txn->acc_pool_idx = (uint)acc_idx;
3655 363284 : uint txn_idx = (uint)txn_pool_idx( accdb->txn_pool, txn );
3656 366233 : for(;;) {
3657 366233 : uint old_head = FD_VOLATILE_CONST( accdb->fork_pool[ accs[ i ]._fork_id ].shmem->txn_head );
3658 366233 : txn->fork.next = old_head;
3659 366967 : if( FD_LIKELY( FD_ATOMIC_CAS( &accdb->fork_pool[ accs[ i ]._fork_id ].shmem->txn_head, old_head, txn_idx )==old_head ) ) break;
3660 >1844*10^16 : FD_SPIN_PAUSE();
3661 >1844*10^16 : }
3662 :
3663 363284 : FD_ATOMIC_FETCH_AND_ADD( &accdb->shmem->shmetrics->accounts_total, 1UL );
3664 363284 : }
3665 367593 : }
3666 :
3667 : // STEP 3.
3668 : // Finally, we release the cache class reservations we took at the
3669 : // beginning when we acquired these cache lines. Credits return
3670 : // directly to the shared pool so other threads can use them
3671 : // immediately.
3672 :
3673 487502 : ulong refund[ FD_ACCDB_CACHE_CLASS_CNT ] = {0};
3674 1046574 : for( ulong i=0UL; i<accs_cnt; i++ ) {
3675 559072 : if( FD_LIKELY( accs[ i ]._original_size_class!=ULONG_MAX ) ) {
3676 179055 : if( FD_UNLIKELY( accdb->shmem->cache_class_used[ accs[ i ]._original_size_class ].val!=ULONG_MAX ) ) {
3677 0 : refund[ accs[ i ]._original_size_class ]++;
3678 0 : }
3679 179055 : }
3680 559072 : if( FD_UNLIKELY( accs[ i ]._writable ) ) {
3681 3628815 : for( ulong j=0UL; j<FD_ACCDB_CACHE_CLASS_CNT; j++ ) {
3682 3225523 : if( FD_UNLIKELY( accdb->shmem->cache_class_used[ j ].val!=ULONG_MAX ) ) {
3683 0 : refund[ j ]++;
3684 0 : }
3685 3225523 : }
3686 403292 : }
3687 559072 : }
3688 4396109 : for( ulong k=0UL; k<FD_ACCDB_CACHE_CLASS_CNT; k++ ) {
3689 3908607 : if( FD_UNLIKELY( refund[ k ] ) ) FD_ATOMIC_FETCH_AND_SUB( &accdb->shmem->cache_class_used[ k ].val, refund[ k ] );
3690 3908607 : }
3691 :
3692 487502 : FD_COMPILER_MFENCE();
3693 487502 : FD_VOLATILE( *accdb->my_epoch_slot ) = ULONG_MAX;
3694 487502 : }
3695 :
3696 : void
3697 : fd_accdb_release( fd_accdb_t * accdb,
3698 : ulong accs_cnt,
3699 480553 : fd_acc_t * accs ) {
3700 480553 : FD_TEST( accdb->acquire_state==FD_ACCDB_ACQUIRE_STATE_OPEN );
3701 480553 : release_inner( accdb, accs_cnt, accs );
3702 480553 : accdb->acquire_state = FD_ACCDB_ACQUIRE_STATE_IDLE;
3703 480553 : }
3704 :
3705 : void
3706 : fd_accdb_release_ab( fd_accdb_t * accdb,
3707 : ulong accs_cnt,
3708 : fd_acc_t * accs,
3709 : ulong execs_cnt,
3710 7299 : fd_acc_t * execs ) {
3711 7299 : FD_TEST( accdb->acquire_state==FD_ACCDB_ACQUIRE_STATE_OPEN );
3712 7299 : release_inner( accdb, accs_cnt, accs );
3713 7299 : if( FD_LIKELY( execs_cnt ) ) release_inner( accdb, execs_cnt, execs );
3714 7299 : accdb->acquire_state = FD_ACCDB_ACQUIRE_STATE_IDLE;
3715 7299 : }
3716 :
3717 : fd_acc_t
3718 : fd_accdb_read_one( fd_accdb_t * accdb,
3719 : fd_accdb_fork_id_t fork_id,
3720 113104 : uchar const * pubkey ) {
3721 113104 : fd_acc_t acc;
3722 113104 : fd_accdb_acquire( accdb, fork_id, 1UL, &pubkey, (int[]){0}, &acc );
3723 113104 : return acc;
3724 113104 : }
3725 :
3726 : void
3727 : fd_accdb_unread_one( fd_accdb_t * accdb,
3728 112971 : fd_acc_t * acc ) {
3729 112971 : fd_accdb_release( accdb, 1UL, acc );
3730 112971 : }
3731 :
3732 : fd_acc_t
3733 : fd_accdb_write_one( fd_accdb_t * accdb,
3734 : fd_accdb_fork_id_t fork_id,
3735 368406 : uchar const * pubkey ) {
3736 368406 : fd_acc_t acc;
3737 368406 : fd_accdb_acquire( accdb, fork_id, 1UL, &pubkey, (int[]){1}, &acc );
3738 368406 : return acc;
3739 368406 : }
3740 :
3741 : void
3742 : fd_accdb_unwrite_one( fd_accdb_t * accdb,
3743 368075 : fd_acc_t * acc ) {
3744 368075 : fd_accdb_release( accdb, 1UL, acc );
3745 368075 : }
3746 :
3747 : int
3748 : fd_accdb_read_one_nocache( fd_accdb_t * accdb,
3749 : fd_accdb_fork_id_t fork_id,
3750 : uchar const * pubkey,
3751 : ulong * out_lamports,
3752 : int * out_executable,
3753 : uchar * out_owner,
3754 : uchar * out_data,
3755 0 : ulong * out_data_len ) {
3756 : /* Publish epoch — protects against compaction freeing the partition
3757 : under us during the preadv2 path. This is the only write the
3758 : readonly joiner makes into accdb shmem (and the pointer it stores
3759 : through is mapped through a separately-mmap'd writable page that
3760 : aliases shmem->joiner_epochs[idx]). */
3761 0 : FD_COMPILER_MFENCE();
3762 0 : FD_VOLATILE( *accdb->my_epoch_slot ) = FD_VOLATILE_CONST( accdb->shmem->epoch );
3763 0 : FD_HW_MFENCE();
3764 :
3765 : /// STEP 1.
3766 : /// Walk the hash chain at acc_map[hash(pubkey)] using the same
3767 : // visibility test as fd_accdb_acquire_inner. See that function
3768 : // for the detailed safety argument under concurrent prepend.
3769 0 : uint root_generation = accdb->fork_pool[ accdb->shmem->root_fork_id.val ].shmem->generation;
3770 0 : fd_accdb_fork_t * fork = &accdb->fork_pool[ fork_id.val ];
3771 0 : ulong hash = fd_hash32( pubkey, accdb->shmem->seed )&(accdb->shmem->chain_cnt-1UL);
3772 0 : uint acc_idx = FD_VOLATILE_CONST( accdb->acc_map[ hash ] );
3773 0 : fd_accdb_accmeta_t const * accmeta = NULL;
3774 0 : while( acc_idx!=UINT_MAX ) {
3775 0 : fd_accdb_accmeta_t const * candidate = &accdb->acc_pool[ acc_idx ];
3776 0 : uint next_idx = FD_VOLATILE_CONST( candidate->map.next );
3777 0 : if( FD_UNLIKELY( (candidate->key.generation>root_generation &&
3778 0 : fd_accdb_acc_fork_id(candidate)!=fork_id.val &&
3779 0 : !descends_set_test( fork->descends, fd_accdb_acc_fork_id(candidate) )) ) ||
3780 0 : memcmp( pubkey, candidate->key.pubkey, 32UL ) ) {
3781 0 : acc_idx = next_idx;
3782 0 : continue;
3783 0 : }
3784 0 : accmeta = candidate;
3785 0 : break;
3786 0 : }
3787 :
3788 0 : if( FD_UNLIKELY( !accmeta ) ) {
3789 0 : accdb->metrics->accounts_acquired_per_class[ 0 ]++;
3790 0 : *out_lamports = 0UL;
3791 0 : FD_COMPILER_MFENCE();
3792 0 : FD_VOLATILE( *accdb->my_epoch_slot ) = ULONG_MAX;
3793 0 : return FD_ACCDB_READ_ONE_NOCACHE_MISS;
3794 0 : }
3795 :
3796 : /// STEP 2.
3797 : /// Snapshot acc fields. The acc element's metadata is effectively
3798 : /// immutable from the perspective of cross-fork readers (see the
3799 : /// comment block in fd_accdb.h about cross-fork reads). */
3800 0 : uint snap_es = FD_VOLATILE_CONST( accmeta->executable_size );
3801 0 : uint snap_gen = accmeta->key.generation;
3802 0 : ulong snap_lamports = accmeta->lamports;
3803 0 : uint snap_cidx = FD_VOLATILE_CONST( accmeta->cache_idx );
3804 0 : ulong data_len = (ulong)FD_ACCDB_SIZE_DATA( snap_es );
3805 0 : int executable = FD_ACCDB_SIZE_EXEC( snap_es );
3806 :
3807 0 : accdb->metrics->accounts_acquired_per_class[ fd_accdb_cache_class( data_len ) ]++;
3808 :
3809 0 : if( FD_UNLIKELY( !snap_lamports ) ) {
3810 0 : *out_lamports = 0UL;
3811 0 : FD_COMPILER_MFENCE();
3812 0 : FD_VOLATILE( *accdb->my_epoch_slot ) = ULONG_MAX;
3813 0 : return FD_ACCDB_READ_ONE_NOCACHE_MISS;
3814 0 : }
3815 :
3816 : /// STEP 3.
3817 : /// Cache hit fast path with try-read-test (ABA) loop. Same
3818 : /// primitives as cache_try_pin: re-check key.generation + pubkey
3819 : /// before and after the bulk copy, and bail to the disk path if the
3820 : /// line was claimed for eviction (refcnt ==
3821 : /// FD_ACCDB_EVICT_SENTINEL). No CAS on refcnt, we never pin the
3822 : /// line.
3823 0 : if( FD_LIKELY( FD_ACCDB_SIZE_CACHE_VALID( snap_es ) && snap_cidx!=FD_ACCDB_ACC_CIDX_INVAL ) ) {
3824 0 : ulong cls = FD_ACCDB_ACC_CIDX_CLASS( snap_cidx );
3825 0 : ulong idx = FD_ACCDB_ACC_CIDX_IDX ( snap_cidx );
3826 0 : fd_accdb_cache_line_t * line = cache_line( accdb, cls, idx );
3827 :
3828 0 : for(;;) {
3829 0 : uint seq0 = FD_VOLATILE_CONST( line->seq );
3830 0 : FD_COMPILER_MFENCE();
3831 0 : uint gen0 = FD_VOLATILE_CONST( line->key.generation );
3832 0 : uint rc0 = FD_VOLATILE_CONST( line->refcnt );
3833 0 : uint ai0 = FD_VOLATILE_CONST( line->acc_idx );
3834 0 : if( FD_UNLIKELY( rc0==FD_ACCDB_EVICT_SENTINEL ) ) goto miss;
3835 0 : if( FD_UNLIKELY( gen0!=snap_gen ) ) goto miss;
3836 0 : if( FD_UNLIKELY( memcmp( line->key.pubkey, pubkey, 32UL ) ) ) goto miss;
3837 : /* acc_idx==UINT_MAX is the "loading" sentinel set by cold_load_acc
3838 : before the preadv2 fills the line. CACHE_VALID can be observed
3839 : set while the bytes are still stale, so fall to the disk path
3840 : (which spins on offset_fork and reads from the file) rather
3841 : than copying garbage. */
3842 0 : if( FD_UNLIKELY( ai0==UINT_MAX ) ) goto miss;
3843 :
3844 0 : FD_COMPILER_MFENCE();
3845 0 : fd_racesan_hook( "accdb_nocache:pre_copy" );
3846 0 : memcpy( out_owner, line->owner, 32UL );
3847 0 : memcpy( out_data, (uchar const *)(line+1UL), data_len );
3848 0 : fd_racesan_hook( "accdb_nocache:post_copy" );
3849 0 : FD_COMPILER_MFENCE();
3850 :
3851 0 : uint gen1 = FD_VOLATILE_CONST( line->key.generation );
3852 0 : uint rc1 = FD_VOLATILE_CONST( line->refcnt );
3853 0 : uint ai1 = FD_VOLATILE_CONST( line->acc_idx );
3854 0 : uint seq1 = FD_VOLATILE_CONST( line->seq );
3855 0 : if( FD_UNLIKELY( seq1!=seq0 ) ) goto miss;
3856 0 : if( FD_UNLIKELY( rc1==FD_ACCDB_EVICT_SENTINEL ) ) goto miss;
3857 0 : if( FD_UNLIKELY( gen1!=snap_gen ) ) goto miss;
3858 0 : if( FD_UNLIKELY( memcmp( line->key.pubkey, pubkey, 32UL ) ) ) goto miss;
3859 0 : if( FD_UNLIKELY( ai1==UINT_MAX ) ) goto miss;
3860 :
3861 0 : *out_lamports = snap_lamports;
3862 0 : *out_executable = executable;
3863 0 : *out_data_len = data_len;
3864 0 : accdb->metrics->bytes_copied += data_len;
3865 0 : FD_COMPILER_MFENCE();
3866 0 : FD_VOLATILE( *accdb->my_epoch_slot ) = ULONG_MAX;
3867 0 : return FD_ACCDB_READ_ONE_NOCACHE_CACHE;
3868 0 : }
3869 0 : }
3870 :
3871 0 : miss:;
3872 0 : accdb->metrics->accounts_not_found_per_class[ fd_accdb_cache_class( FD_ACCDB_SIZE_DATA( snap_es ) ) ]++;
3873 :
3874 : /// STEP 4.
3875 : /// Disk path. Spin until the writer publishes a real offset
3876 : /// (matches STEP 10 of fd_accdb_acquire_inner). Compaction may
3877 : /// concurrently relocate the record, but our published epoch
3878 : /// prevents the source partition from being freed until we exit
3879 : /// our critical section, so the bytes at the snapshotted offset
3880 : /// remain stable for the duration of the read.
3881 0 : fd_racesan_hook( "accdb_nocache:pre_offset" );
3882 0 : ulong off_packed = FD_VOLATILE_CONST( accmeta->offset_fork );
3883 0 : if( FD_UNLIKELY( (off_packed & FD_ACCDB_OFF_MASK)==FD_ACCDB_OFF_INVAL ) ) {
3884 0 : accdb->metrics->accounts_waited++;
3885 0 : while( FD_UNLIKELY( ((off_packed=FD_VOLATILE_CONST( accmeta->offset_fork )) & FD_ACCDB_OFF_MASK)==FD_ACCDB_OFF_INVAL ) ) {
3886 0 : fd_racesan_hook( "accdb_nocache:offset_wait" );
3887 0 : FD_SPIN_PAUSE();
3888 0 : }
3889 0 : }
3890 0 : ulong off = off_packed & FD_ACCDB_OFF_MASK;
3891 0 : fd_racesan_hook( "accdb_nocache:pre_preadv2" );
3892 :
3893 0 : struct iovec iovs[ 2 ] = {
3894 0 : { .iov_base = out_owner, .iov_len = 32UL },
3895 0 : { .iov_base = out_data, .iov_len = data_len },
3896 0 : };
3897 0 : ulong total = 32UL+data_len;
3898 0 : ulong start = off+offsetof( fd_accdb_disk_meta_t, owner );
3899 0 : ulong got = 0UL;
3900 0 : int nio = data_len ? 2 : 1;
3901 0 : while( FD_LIKELY( got<total ) ) {
3902 0 : long result = preadv2( accdb->fd, iovs, nio, (long)(start+got), 0 );
3903 0 : if( FD_UNLIKELY( -1==result && (errno==EINTR || errno==EAGAIN || errno==EWOULDBLOCK) ) ) continue;
3904 0 : else if( FD_UNLIKELY( -1==result ) ) FD_LOG_ERR(( "preadv2() failed (%d-%s)", errno, fd_io_strerror( errno ) ));
3905 0 : else if( FD_UNLIKELY( !result ) ) FD_LOG_ERR(( "accounts database is corrupt, data expected at offset %lu with size %lu exceeded file extents", start+got, total ));
3906 0 : fd_accdb_partition_read_bump( accdb, start+got, (ulong)result );
3907 0 : got += (ulong)result;
3908 0 : accdb->metrics->bytes_read += (ulong)result;
3909 0 : accdb->metrics->read_ops++;
3910 :
3911 0 : long r = result;
3912 0 : for( int v=0; v<nio; v++ ) {
3913 0 : if( (ulong)r>=iovs[ v ].iov_len ) {
3914 0 : r -= (long)iovs[ v ].iov_len;
3915 0 : iovs[ v ].iov_len = 0UL;
3916 0 : } else {
3917 0 : iovs[ v ].iov_base = (uchar *)iovs[ v ].iov_base + r;
3918 0 : iovs[ v ].iov_len -= (ulong)r;
3919 0 : break;
3920 0 : }
3921 0 : }
3922 0 : }
3923 :
3924 0 : *out_lamports = snap_lamports;
3925 0 : *out_executable = executable;
3926 0 : *out_data_len = data_len;
3927 :
3928 0 : FD_COMPILER_MFENCE();
3929 0 : FD_VOLATILE( *accdb->my_epoch_slot ) = ULONG_MAX;
3930 0 : return FD_ACCDB_READ_ONE_NOCACHE_DISK;
3931 0 : }
3932 :
3933 : int
3934 : fd_accdb_exists( fd_accdb_t * accdb,
3935 : fd_accdb_fork_id_t fork_id,
3936 225546 : uchar const * pubkey ) {
3937 225546 : FD_COMPILER_MFENCE();
3938 225546 : FD_VOLATILE( *accdb->my_epoch_slot ) = FD_VOLATILE_CONST( accdb->shmem->epoch );
3939 225546 : FD_HW_MFENCE();
3940 :
3941 225546 : uint root_generation = accdb->fork_pool[ accdb->shmem->root_fork_id.val ].shmem->generation;
3942 225546 : fd_accdb_fork_t * fork = &accdb->fork_pool[ fork_id.val ];
3943 225546 : ulong hash = fd_hash32( pubkey, accdb->shmem->seed )&(accdb->shmem->chain_cnt-1UL);
3944 225546 : uint acc = FD_VOLATILE_CONST( accdb->acc_map[ hash ] );
3945 236518 : while( acc!=UINT_MAX ) {
3946 10993 : fd_accdb_accmeta_t const * candidate_acc = &accdb->acc_pool[ acc ];
3947 10993 : uint next_acc = FD_VOLATILE_CONST( candidate_acc->map.next );
3948 :
3949 10997 : if( FD_UNLIKELY( (candidate_acc->key.generation>root_generation && fd_accdb_acc_fork_id(candidate_acc)!=fork_id.val && !descends_set_test( fork->descends, fd_accdb_acc_fork_id(candidate_acc) )) ) || memcmp( pubkey, candidate_acc->key.pubkey, 32UL ) ) {
3950 10972 : acc = next_acc;
3951 10972 : continue;
3952 10972 : }
3953 :
3954 21 : break;
3955 10993 : }
3956 :
3957 225546 : int result;
3958 225546 : if( FD_UNLIKELY( acc==UINT_MAX ) ) result = 0;
3959 297 : else result = !!FD_VOLATILE_CONST( accdb->acc_pool[ acc ].lamports );
3960 :
3961 225546 : FD_COMPILER_MFENCE();
3962 225546 : FD_VOLATILE( *accdb->my_epoch_slot ) = ULONG_MAX;
3963 225546 : return result;
3964 225546 : }
3965 :
3966 : int
3967 : fd_accdb_probe_pd_this_fork( fd_accdb_t * accdb,
3968 : fd_accdb_fork_id_t fork_id,
3969 : uchar const * pubkey,
3970 : int * out_pd_write,
3971 : ulong * out_data_len,
3972 5702 : ulong * out_lamports ) {
3973 5702 : FD_COMPILER_MFENCE();
3974 5702 : FD_VOLATILE( *accdb->my_epoch_slot ) = FD_VOLATILE_CONST( accdb->shmem->epoch );
3975 5702 : FD_HW_MFENCE();
3976 :
3977 5702 : uint root_generation = accdb->fork_pool[ accdb->shmem->root_fork_id.val ].shmem->generation;
3978 5702 : fd_accdb_fork_t * fork = &accdb->fork_pool[ fork_id.val ];
3979 5702 : ulong hash = fd_hash32( pubkey, accdb->shmem->seed )&(accdb->shmem->chain_cnt-1UL);
3980 5702 : uint acc = FD_VOLATILE_CONST( accdb->acc_map[ hash ] );
3981 6127 : while( acc!=UINT_MAX ) {
3982 445 : fd_accdb_accmeta_t const * candidate_acc = &accdb->acc_pool[ acc ];
3983 445 : uint next_acc = FD_VOLATILE_CONST( candidate_acc->map.next );
3984 :
3985 445 : if( FD_UNLIKELY( (candidate_acc->key.generation>root_generation && fd_accdb_acc_fork_id(candidate_acc)!=fork_id.val && !descends_set_test( fork->descends, fd_accdb_acc_fork_id(candidate_acc) )) ) || memcmp( pubkey, candidate_acc->key.pubkey, 32UL ) ) {
3986 425 : acc = next_acc;
3987 425 : continue;
3988 425 : }
3989 :
3990 20 : break;
3991 445 : }
3992 :
3993 5702 : int pd = 0;
3994 5702 : int gen_match = 0;
3995 5702 : ulong len = 0UL;
3996 5702 : ulong lamports = 0UL;
3997 5702 : if( FD_LIKELY( acc!=UINT_MAX ) ) {
3998 20 : fd_accdb_accmeta_t const * m = &accdb->acc_pool[ acc ];
3999 20 : uint es = FD_VOLATILE_CONST( m->executable_size );
4000 20 : gen_match = ( m->key.generation==fork->shmem->generation );
4001 20 : pd = gen_match && FD_ACCDB_SIZE_PD_WRITE( es );
4002 20 : len = FD_ACCDB_SIZE_DATA( es );
4003 20 : lamports = FD_VOLATILE_CONST( m->lamports );
4004 20 : }
4005 :
4006 5702 : FD_COMPILER_MFENCE();
4007 5702 : FD_VOLATILE( *accdb->my_epoch_slot ) = ULONG_MAX;
4008 :
4009 5702 : *out_pd_write = pd;
4010 5702 : if( gen_match ) {
4011 20 : *out_data_len = len;
4012 20 : *out_lamports = lamports;
4013 20 : }
4014 5702 : return gen_match;
4015 5702 : }
4016 :
4017 : ulong
4018 : fd_accdb_lamports( fd_accdb_t * accdb,
4019 : fd_accdb_fork_id_t fork_id,
4020 0 : uchar const * pubkey ) {
4021 0 : FD_COMPILER_MFENCE();
4022 0 : FD_VOLATILE( *accdb->my_epoch_slot ) = FD_VOLATILE_CONST( accdb->shmem->epoch );
4023 0 : FD_HW_MFENCE();
4024 :
4025 0 : uint root_generation = accdb->fork_pool[ accdb->shmem->root_fork_id.val ].shmem->generation;
4026 0 : fd_accdb_fork_t * fork = &accdb->fork_pool[ fork_id.val ];
4027 0 : ulong hash = fd_hash32( pubkey, accdb->shmem->seed )&(accdb->shmem->chain_cnt-1UL);
4028 0 : uint acc = FD_VOLATILE_CONST( accdb->acc_map[ hash ] );
4029 0 : while( acc!=UINT_MAX ) {
4030 0 : fd_accdb_accmeta_t const * candidate_acc = &accdb->acc_pool[ acc ];
4031 0 : uint next_acc = FD_VOLATILE_CONST( candidate_acc->map.next );
4032 :
4033 0 : if( FD_UNLIKELY( (candidate_acc->key.generation>root_generation && fd_accdb_acc_fork_id(candidate_acc)!=fork_id.val && !descends_set_test( fork->descends, fd_accdb_acc_fork_id(candidate_acc) )) ) || memcmp( pubkey, candidate_acc->key.pubkey, 32UL ) ) {
4034 0 : acc = next_acc;
4035 0 : continue;
4036 0 : }
4037 :
4038 0 : break;
4039 0 : }
4040 :
4041 0 : ulong result;
4042 0 : if( FD_UNLIKELY( acc==UINT_MAX ) ) result = 0UL;
4043 0 : else result = FD_VOLATILE_CONST( accdb->acc_pool[ acc ].lamports );
4044 :
4045 0 : FD_COMPILER_MFENCE();
4046 0 : FD_VOLATILE( *accdb->my_epoch_slot ) = ULONG_MAX;
4047 0 : return result;
4048 0 : }
4049 :
4050 : /* Dirty lines are staged in accdb->stage and written with one
4051 : pwritev2 per batch, so the kernel's inode lock is taken once per
4052 : batch instead of once per account. Evictions are performed
4053 : per-class in batches of up to PREEVICT_TICK_BUDGET records or
4054 : PREEVICT_BATCH_SZ bytes. While a line sits in a batch its account
4055 : can be read neither from the cache nor from disk, so a reader waits
4056 : for the batch write to finish. */
4057 :
4058 0 : #define PREEVICT_TICK_BUDGET (256UL)
4059 :
4060 : /* cache_line_free invalidates a claimed line and returns it to its
4061 : class free list. */
4062 :
4063 : static inline void
4064 : cache_line_free( fd_accdb_t * accdb,
4065 : ulong size_class,
4066 0 : fd_accdb_cache_line_t * line ) {
4067 0 : line->persisted = 1;
4068 0 : line->acc_idx = UINT_MAX;
4069 0 : line->key.generation = UINT_MAX;
4070 0 : FD_COMPILER_MFENCE();
4071 0 : FD_VOLATILE( line->refcnt ) = 0;
4072 0 : cache_free_push( accdb, size_class, line );
4073 0 : }
4074 :
4075 : static void
4076 : preevict_flush( fd_accdb_t * accdb,
4077 : ulong size_class,
4078 : fd_accdb_cache_line_t ** pend_line,
4079 : ulong const * pend_sz,
4080 : ulong pend_cnt,
4081 0 : ulong stage_sz ) {
4082 0 : fd_accdb_shmem_t * shmem = accdb->shmem;
4083 :
4084 : /* advertise to external observers that writes are in progress */
4085 0 : FD_COMPILER_MFENCE();
4086 0 : FD_VOLATILE( *accdb->my_epoch_slot ) = FD_VOLATILE_CONST( shmem->epoch );
4087 0 : FD_HW_MFENCE();
4088 :
4089 0 : fd_racesan_hook( "preevict:pre_synth" );
4090 :
4091 : /* Invalidate the records so compaction does not relocate them while
4092 : we are evicting. */
4093 0 : ulong freed = 0UL;
4094 0 : for( ulong i=0UL; i<pend_cnt; i++ ) {
4095 0 : fd_accdb_accmeta_t * accmeta = &accdb->acc_pool[ pend_line[ i ]->acc_idx ];
4096 0 : ulong old_offset = fd_accdb_acc_xchg_offset( accmeta, FD_ACCDB_OFF_INVAL );
4097 0 : if( FD_UNLIKELY( old_offset!=FD_ACCDB_OFF_INVAL ) ) {
4098 0 : fd_accdb_shmem_bytes_freed( shmem, old_offset, pend_sz[ i ] );
4099 0 : freed += pend_sz[ i ];
4100 0 : }
4101 0 : }
4102 0 : FD_ATOMIC_FETCH_AND_SUB( &shmem->shmetrics->disk_used_bytes, freed );
4103 :
4104 0 : ulong file_off = allocate_next_write( accdb, stage_sz );
4105 0 : ulong written = 0UL;
4106 0 : while( written<stage_sz ) {
4107 0 : struct iovec iov = { .iov_base = accdb->stage+written, .iov_len = stage_sz-written };
4108 0 : long result = pwritev2( accdb->fd, &iov, 1, (long)(file_off+written), 0 );
4109 0 : if( FD_UNLIKELY( result==-1 && errno==EINTR ) ) continue;
4110 0 : else if( FD_UNLIKELY( result<=0 ) ) FD_LOG_ERR(( "pwritev2() failed (%d-%s)", errno, fd_io_strerror( errno ) ));
4111 0 : written += (ulong)result;
4112 0 : accdb->metrics->bytes_written += (ulong)result;
4113 0 : accdb->metrics->write_ops++;
4114 0 : }
4115 0 : FD_ATOMIC_FETCH_AND_ADD( &shmem->shmetrics->disk_used_bytes, stage_sz );
4116 :
4117 0 : fd_racesan_hook( "preevict:post_write" );
4118 :
4119 0 : FD_COMPILER_MFENCE();
4120 0 : ulong off = file_off;
4121 0 : for( ulong i=0UL; i<pend_cnt; i++ ) {
4122 0 : fd_accdb_cache_line_t * line = pend_line[ i ];
4123 0 : fd_accdb_accmeta_t * accmeta = &accdb->acc_pool[ line->acc_idx ];
4124 0 : accmeta->offset_fork = fd_accdb_acc_pack_offset_fork( off, fd_accdb_acc_fork_id( accmeta ) );
4125 0 : off += pend_sz[ i ];
4126 0 : FD_COMPILER_MFENCE();
4127 0 : cache_line_free( accdb, size_class, line );
4128 0 : }
4129 0 : accdb->metrics->accounts_preevicted += pend_cnt;
4130 0 : accdb->metrics->accounts_preevicted_per_class[ size_class ] += pend_cnt;
4131 :
4132 0 : FD_COMPILER_MFENCE();
4133 0 : FD_VOLATILE( *accdb->my_epoch_slot ) = ULONG_MAX;
4134 0 : }
4135 :
4136 : /* cache_bg_evict pre-evicts cache lines in the background to keep the
4137 : per-class CAS free lists populated ahead of demand. For each class
4138 : whose immediately available capacity has dropped below low_water,
4139 : a bounded CLOCK sweep claims lines, writes dirty ones to disk, and
4140 : pushes them onto the free list until available capacity reaches
4141 : target. Immediately available capacity includes both the CAS free
4142 : list and the never-initialized tail of the class, since foreground
4143 : allocators can consume either path without evicting resident data.
4144 :
4145 : Budget: at most 256 CLOCK ticks per class per invocation to keep the
4146 : background loop responsive. The function is called every tick of
4147 : fd_accdb_background, so large refills happen across several ticks
4148 : rather than blocking. The low_water / target thresholds are static
4149 : per-class watermarks computed at initialization; pre-eviction only
4150 : converts resident lines into free-list entries and does not consume
4151 : cache-slot reservations.
4152 :
4153 : force: when non-zero, ignore the watermark and sweep every line in
4154 : every class. Always 0 in normal operation; used only by
4155 : test_accdb_racesan to deterministically exercise the writeback path
4156 : without manufacturing real cache pressure. */
4157 :
4158 : static void
4159 : background_preevict( fd_accdb_t * accdb,
4160 : int * charge_busy,
4161 0 : int force ) {
4162 0 : fd_accdb_shmem_t * shmem = accdb->shmem;
4163 :
4164 : /* A batch has to fit one disk partition. */
4165 0 : FD_TEST( accdb->stage );
4166 0 : ulong batch_max = fd_ulong_min( PREEVICT_BATCH_SZ, shmem->partition_sz );
4167 :
4168 0 : fd_accdb_cache_line_t * pend_line[ PREEVICT_TICK_BUDGET ];
4169 0 : ulong pend_sz [ PREEVICT_TICK_BUDGET ];
4170 0 : ulong pend_cnt = 0UL;
4171 0 : ulong stage_sz = 0UL;
4172 :
4173 0 : for( ulong size_class=0UL; size_class<FD_ACCDB_CACHE_CLASS_CNT; size_class++ ) {
4174 0 : ulong target = shmem->cache_free_target[ size_class ];
4175 0 : ulong max_c = shmem->cache_class_max[ size_class ];
4176 0 : ulong init = fd_ulong_min( FD_VOLATILE_CONST( shmem->cache_class_init[ size_class ].val ), max_c );
4177 0 : ulong freec = FD_VOLATILE_CONST( shmem->cache_free_cnt[ size_class ].val );
4178 0 : ulong live = init>freec ? init-freec : 0UL;
4179 0 : ulong avail = max_c-live;
4180 0 : if( FD_LIKELY( !force && avail>=shmem->cache_free_low_water[ size_class ] ) ) continue;
4181 :
4182 0 : *charge_busy = 1;
4183 :
4184 0 : ulong budget = force ? init : PREEVICT_TICK_BUDGET;
4185 0 : ulong evicted = 0UL;
4186 0 : if( FD_UNLIKELY( force ) ) target = max_c; /* sweep everything */
4187 :
4188 0 : for( ulong tick=0UL; tick<budget && avail+evicted<target; tick++ ) {
4189 : /* Only sweep the lazily initialized prefix. cache_class_init
4190 : may transiently exceed max_c during the acquire_cache_line
4191 : overflow/undo path, so clamp it before using it as the wrap
4192 : bound. */
4193 0 : init = fd_ulong_min( FD_VOLATILE_CONST( shmem->cache_class_init[ size_class ].val ), max_c );
4194 0 : if( FD_UNLIKELY( !init ) ) break;
4195 :
4196 0 : ulong hand = FD_ATOMIC_FETCH_AND_ADD( &shmem->clock_hand[ size_class ].val, 1UL ) % init;
4197 :
4198 0 : fd_accdb_cache_line_t * line = cache_line( accdb, size_class, hand );
4199 :
4200 0 : if( FD_UNLIKELY( line->key.generation==UINT_MAX && line->acc_idx==UINT_MAX ) ) continue;
4201 :
4202 0 : uint rc = FD_VOLATILE_CONST( line->refcnt );
4203 0 : if( FD_UNLIKELY( rc ) ) continue;
4204 :
4205 0 : if( FD_UNLIKELY( line->referenced ) ) {
4206 0 : line->referenced = 0;
4207 0 : continue;
4208 0 : }
4209 :
4210 0 : if( FD_UNLIKELY( FD_ATOMIC_CAS( &line->refcnt, 0U, FD_ACCDB_EVICT_SENTINEL )!=0U ) ) continue;
4211 :
4212 0 : if( FD_UNLIKELY( line->acc_idx==UINT_MAX && line->key.generation==UINT_MAX ) ) {
4213 0 : FD_VOLATILE( line->refcnt ) = 0;
4214 0 : continue;
4215 0 : }
4216 :
4217 0 : uint acc_idx = line->acc_idx;
4218 : #if FD_TMPL_USE_HANDHOLDING
4219 : uint line_gen FD_FN_UNUSED = line->key.generation;
4220 : #endif
4221 0 : if( FD_LIKELY( acc_idx!=UINT_MAX ) ) {
4222 0 : evict_clear_acc_cache_ref( &accdb->acc_pool[ acc_idx ], size_class, hand );
4223 0 : }
4224 0 : line->key.generation = UINT_MAX;
4225 0 : if( FD_UNLIKELY( line->persisted || acc_idx==UINT_MAX ) ) {
4226 0 : cache_line_free( accdb, size_class, line );
4227 0 : evicted++;
4228 0 : continue;
4229 0 : }
4230 :
4231 0 : fd_accdb_accmeta_t * accmeta = &accdb->acc_pool[ acc_idx ];
4232 : #if FD_TMPL_USE_HANDHOLDING
4233 : FD_TEST( line_gen==accmeta->key.generation &&
4234 : !memcmp( line->key.pubkey, accmeta->key.pubkey, 32UL ) );
4235 : #endif
4236 0 : uint data_sz = FD_ACCDB_SIZE_DATA( accmeta->executable_size );
4237 0 : ulong entry_sz = sizeof(fd_accdb_disk_meta_t)+data_sz;
4238 :
4239 0 : if( FD_UNLIKELY( pend_cnt && ( stage_sz+entry_sz>batch_max || pend_cnt==PREEVICT_TICK_BUDGET ) ) ) {
4240 0 : preevict_flush( accdb, size_class, pend_line, pend_sz, pend_cnt, stage_sz );
4241 0 : pend_cnt = 0UL;
4242 0 : stage_sz = 0UL;
4243 0 : }
4244 :
4245 0 : fd_accdb_disk_meta_t * meta = (fd_accdb_disk_meta_t *)( accdb->stage+stage_sz );
4246 0 : fd_memcpy( meta->pubkey, accmeta->key.pubkey, 32UL );
4247 0 : meta->size = data_sz;
4248 0 : meta->generation = accmeta->key.generation;
4249 0 : fd_memcpy( meta->owner, line->owner, 32UL );
4250 0 : fd_memcpy( meta+1, line+1UL, data_sz );
4251 :
4252 0 : pend_line[ pend_cnt ] = line;
4253 0 : pend_sz [ pend_cnt ] = entry_sz;
4254 0 : pend_cnt++;
4255 0 : stage_sz += entry_sz;
4256 0 : evicted++;
4257 0 : }
4258 :
4259 0 : if( pend_cnt ) {
4260 0 : preevict_flush( accdb, size_class, pend_line, pend_sz, pend_cnt, stage_sz );
4261 0 : pend_cnt = 0UL;
4262 0 : stage_sz = 0UL;
4263 0 : }
4264 0 : }
4265 0 : }
4266 :
4267 : int
4268 : fd_accdb_snapshot_write_batch( fd_accdb_t * accdb,
4269 : fd_accdb_fork_id_t fork_id,
4270 : ulong cnt,
4271 : uchar const * const pubkeys[],
4272 : ulong const slots[],
4273 : ulong const lamports[],
4274 : ulong const data_lens[],
4275 : int const executables[],
4276 : ulong const file_offsets[],
4277 : ulong * accounts_ignored,
4278 : ulong * accounts_replaced,
4279 : ulong * accounts_loaded,
4280 : ulong * out_replaced_lamports,
4281 : ulong * out_ignored_lamports,
4282 0 : uchar * results ) {
4283 0 : #define CHAIN_LOCKED (UINT_MAX-1U)
4284 :
4285 0 : FD_TEST( cnt>0UL && cnt<=8UL );
4286 :
4287 0 : int incremental = fork_id.val!=USHORT_MAX;
4288 :
4289 0 : fd_accdb_fork_t * fork = NULL;
4290 0 : uint fork_gen = 0U;
4291 0 : if( FD_UNLIKELY( incremental ) ) {
4292 0 : fork = &accdb->fork_pool[ fork_id.val ];
4293 0 : fork_gen = fork->shmem->generation;
4294 0 : }
4295 :
4296 0 : ulong seed = accdb->shmem->seed;
4297 0 : ulong chain_msk = accdb->shmem->chain_cnt - 1UL;
4298 0 : if( FD_UNLIKELY( !incremental && accdb->shmem->root_fork_id.val==USHORT_MAX ) ) {
4299 0 : FD_LOG_ERR(( "snapshot_write_batch called without a root fork attached" ));
4300 0 : }
4301 0 : uint gen = incremental ? 0U : accdb->fork_pool[ accdb->shmem->root_fork_id.val ].shmem->generation;
4302 :
4303 0 : ulong ignored = 0UL;
4304 0 : ulong replaced = 0UL;
4305 0 : ulong loaded = 0UL;
4306 0 : ulong cross_replaced = 0UL; /* cross-fork overrides (subset of replaced) */
4307 0 : ulong replaced_lamports = 0UL;
4308 0 : ulong ignored_lamports = 0UL;
4309 0 : int result = 0;
4310 :
4311 : /* Snapshot slots are stored in the 32-bit cache_idx scratch field
4312 : during loading. Reject anything that would truncate. */
4313 0 : for( ulong i=0UL; i<cnt; i++ ) {
4314 0 : if( FD_UNLIKELY( slots[ i ]>UINT_MAX ) ) FD_LOG_ERR(( "snapshot slot %lu exceeds 2^32-1, accdb format must be widened", slots[ i ] ));
4315 0 : }
4316 :
4317 : /* Phase 1: compute hashes and prefetch chain heads. */
4318 :
4319 0 : ulong hashes[ 8 ];
4320 0 : for( ulong i=0UL; i<cnt; i++ ) {
4321 0 : hashes[ i ] = fd_hash32( pubkeys[ i ], seed ) & chain_msk;
4322 0 : __builtin_prefetch( &accdb->acc_map[ hashes[ i ] ], 1, 1 );
4323 0 : }
4324 :
4325 : /* Phase 1b: reject same-slot duplicate pubkeys. */
4326 :
4327 0 : for( ulong i=1UL; i<cnt; i++ ) {
4328 0 : for( ulong j=0UL; j<i; j++ ) {
4329 0 : if( FD_UNLIKELY( hashes[ j ]==hashes[ i ] && slots[ j ]==slots[ i ] &&
4330 0 : !memcmp( pubkeys[ j ], pubkeys[ i ], 32UL ) ) ) {
4331 0 : FD_LOG_WARNING(( "corrupt snapshot: duplicate pubkey within a single batch (entries %lu and %lu, slots %lu and %lu)", j, i, slots[ j ], slots[ i ] ));
4332 0 : return -1;
4333 0 : }
4334 0 : }
4335 0 : }
4336 :
4337 0 : fd_accdb_accmeta_t * acquired[ 8 ];
4338 0 : if( FD_UNLIKELY( !acc_pool_acquire_batch( accdb->acc_pool_join, cnt, acquired ) ) ) {
4339 0 : FD_LOG_ERR(( "accounts database ran out of space during snapshot loading, increase [accounts.max_accounts], current value is %lu", acc_pool_ele_max( accdb->acc_pool_join ) ));
4340 0 : }
4341 :
4342 : /* Phase 2: walk and commit under the account chain lock. */
4343 :
4344 0 : ulong used_bytes_added = 0UL;
4345 0 : ulong used_bytes_removed = 0UL;
4346 0 : ulong acquired_used = 0UL;
4347 :
4348 0 : for( ulong i=0UL; i<cnt; i++ ) {
4349 0 : ulong entry_sz = sizeof(fd_accdb_disk_meta_t)+data_lens[ i ];
4350 0 : int skip = 0;
4351 :
4352 0 : uint chain_head;
4353 0 : for(;;) {
4354 0 : chain_head = FD_VOLATILE_CONST( accdb->acc_map[ hashes[ i ] ] );
4355 0 : if( FD_UNLIKELY( chain_head==CHAIN_LOCKED ) ) {
4356 0 : FD_SPIN_PAUSE();
4357 0 : continue;
4358 0 : }
4359 0 : if( FD_LIKELY( FD_ATOMIC_CAS( &accdb->acc_map[ hashes[ i ] ], chain_head, CHAIN_LOCKED )==chain_head ) ) {
4360 0 : break;
4361 0 : }
4362 0 : FD_SPIN_PAUSE();
4363 0 : }
4364 :
4365 0 : fd_accdb_accmeta_t * existing = NULL;
4366 0 : fd_accdb_accmeta_t * cross_existing = NULL; /* cross-fork dup (incremental only) */
4367 0 : uint next_acc = chain_head;
4368 0 : while( next_acc!=UINT_MAX ) {
4369 0 : fd_accdb_accmeta_t * candidate = &accdb->acc_pool[ next_acc ];
4370 :
4371 0 : if( FD_LIKELY( candidate->map.next!=UINT_MAX ) ) {
4372 0 : __builtin_prefetch( &accdb->acc_pool[ candidate->map.next ], 0, 1 );
4373 0 : }
4374 :
4375 0 : if( FD_UNLIKELY( !memcmp( pubkeys[ i ], candidate->key.pubkey, 32UL ) ) ) {
4376 0 : if( FD_LIKELY( (ulong)candidate->cache_idx>slots[ i ] ) ) {
4377 0 : skip = 1;
4378 0 : } else if( FD_UNLIKELY( (ulong)candidate->cache_idx==slots[ i ] ) ) {
4379 0 : FD_LOG_WARNING(( "corrupt snapshot: duplicate account at slot %lu", slots[ i ] ));
4380 0 : FD_COMPILER_MFENCE();
4381 0 : FD_VOLATILE( accdb->acc_map[ hashes[ i ] ] ) = chain_head;
4382 0 : result = -1;
4383 0 : goto fini;
4384 0 : } else if( FD_UNLIKELY( incremental ) && candidate->key.generation!=fork_gen ) {
4385 0 : cross_existing = candidate;
4386 0 : } else {
4387 0 : existing = candidate;
4388 0 : }
4389 0 : break;
4390 0 : }
4391 0 : next_acc = candidate->map.next;
4392 0 : }
4393 :
4394 0 : if( FD_UNLIKELY( skip ) ) {
4395 0 : results[ i ] = FD_ACCDB_SNAPSHOT_WRITE_IGNORED;
4396 0 : FD_COMPILER_MFENCE();
4397 0 : FD_VOLATILE( accdb->acc_map[ hashes[ i ] ] ) = chain_head;
4398 0 : fd_accdb_shmem_bytes_freed( accdb->shmem, file_offsets[ i ], entry_sz );
4399 0 : ignored_lamports += lamports[ i ];
4400 0 : ignored++;
4401 0 : continue;
4402 0 : }
4403 :
4404 0 : fd_accdb_accmeta_t * accmeta;
4405 0 : uint new_head = chain_head;
4406 :
4407 0 : fd_accdb_accmeta_t const * prev = existing ? existing : cross_existing;
4408 0 : if( !prev || !prev->lamports ) results[ i ] = FD_ACCDB_SNAPSHOT_WRITE_LOADED;
4409 0 : else if( cross_existing ) results[ i ] = FD_ACCDB_SNAPSHOT_WRITE_REPLACED_CROSS;
4410 0 : else results[ i ] = FD_ACCDB_SNAPSHOT_WRITE_REPLACED;
4411 :
4412 0 : if( FD_UNLIKELY( existing ) ) {
4413 0 : accmeta = existing;
4414 0 : ulong old_sz = sizeof(fd_accdb_disk_meta_t) + FD_ACCDB_SIZE_DATA( accmeta->executable_size );
4415 0 : fd_accdb_shmem_bytes_freed( accdb->shmem, fd_accdb_acc_offset( accmeta ), old_sz );
4416 0 : used_bytes_removed += old_sz;
4417 0 : replaced_lamports += accmeta->lamports;
4418 0 : replaced++;
4419 0 : } else {
4420 0 : accmeta = acquired[ acquired_used++ ];
4421 :
4422 0 : uint acc_idx = (uint)acc_pool_idx( accdb->acc_pool_join, accmeta );
4423 :
4424 0 : fd_memcpy( accmeta->key.pubkey, pubkeys[ i ], 32UL );
4425 0 : accmeta->key.generation = incremental ? fork_gen : gen;
4426 0 : accmeta->map.next = chain_head;
4427 0 : new_head = acc_idx;
4428 :
4429 0 : if( FD_UNLIKELY( incremental ) ) {
4430 0 : fd_accdb_txn_t * txn = txn_pool_acquire( accdb->txn_pool );
4431 0 : if( FD_UNLIKELY( !txn ) ) FD_LOG_ERR(( "txn pool exhausted during incremental snapshot loading" ));
4432 0 : txn->acc_pool_idx = acc_idx;
4433 0 : uint txn_idx = (uint)txn_pool_idx( accdb->txn_pool, txn );
4434 0 : for(;;) {
4435 0 : uint old_head = FD_VOLATILE_CONST( fork->shmem->txn_head );
4436 0 : txn->fork.next = old_head;
4437 0 : if( FD_LIKELY( FD_ATOMIC_CAS( &fork->shmem->txn_head, old_head, txn_idx )==old_head ) ) break;
4438 0 : FD_SPIN_PAUSE();
4439 0 : }
4440 0 : }
4441 :
4442 0 : if( FD_UNLIKELY( cross_existing ) ) {
4443 0 : replaced_lamports += cross_existing->lamports;
4444 0 : replaced++;
4445 0 : cross_replaced++;
4446 0 : } else {
4447 0 : loaded++;
4448 0 : }
4449 0 : }
4450 :
4451 0 : accmeta->cache_idx = (uint)slots[ i ];
4452 0 : accmeta->lamports = lamports[ i ];
4453 0 : accmeta->executable_size = FD_ACCDB_SIZE_PACK( (uint)data_lens[ i ], executables[ i ] );
4454 0 : ulong file_off = file_offsets[ i ];
4455 0 : accmeta->offset_fork = incremental ? fd_accdb_acc_pack_offset_fork( file_off, fork_id.val ) : file_off;
4456 :
4457 0 : FD_COMPILER_MFENCE();
4458 0 : FD_VOLATILE( accdb->acc_map[ hashes[ i ] ] ) = new_head;
4459 :
4460 0 : used_bytes_added += entry_sz;
4461 0 : }
4462 :
4463 0 : fini:
4464 0 : for( ulong i=acquired_used; i<cnt; i++ ) {
4465 0 : acc_pool_release( accdb->acc_pool_join, acquired[ i ] );
4466 0 : }
4467 :
4468 0 : accdb->write_stats.disk_used_added += used_bytes_added;
4469 0 : accdb->write_stats.disk_used_removed += used_bytes_removed;
4470 0 : accdb->write_stats.accounts_total_added += loaded + cross_replaced;
4471 :
4472 0 : *accounts_ignored = ignored;
4473 0 : *accounts_replaced = replaced;
4474 0 : *accounts_loaded = loaded;
4475 0 : *out_replaced_lamports = replaced_lamports;
4476 0 : *out_ignored_lamports = ignored_lamports;
4477 :
4478 0 : return result;
4479 :
4480 0 : #undef CHAIN_LOCKED
4481 0 : }
4482 :
4483 :
4484 : static void
4485 0 : delta_reset( fd_accdb_t * accdb ) {
4486 0 : if( !accdb->shmem->delta.head ) return; /* clean */
4487 0 : uint * chains = accdb->delta.chains;
4488 0 : ulong chain_cnt = accdb->shmem->delta.chain_cnt;
4489 0 : for( ulong i=0UL; i<chain_cnt; i++ ) chains[ i ] = UINT_MAX;
4490 0 : accdb->shmem->delta.head = 0UL;
4491 0 : }
4492 :
4493 : static int
4494 0 : delta_is_valid( fd_accdb_shmem_t const * accdb ) {
4495 0 : return accdb->delta.head < accdb->delta.ele_max;
4496 0 : }
4497 :
4498 : void
4499 : fd_accdb_background( fd_accdb_t * accdb,
4500 64053 : int * charge_busy ) {
4501 64053 : fd_accdb_shmem_t * shmem = accdb->shmem;
4502 :
4503 64053 : ulong * snap_sync_p = &shmem->snapshot_sync;
4504 64053 : ulong snap_sync = fd_accdb_snapshot_sync_state( snap_sync_p );
4505 :
4506 64053 : uint op = FD_VOLATILE_CONST( shmem->cmd_op );
4507 64058 : if( FD_UNLIKELY( op!=FD_ACCDB_CMD_IDLE ) ) {
4508 64058 : fd_accdb_fork_id_t fork_id = { .val = FD_VOLATILE_CONST( shmem->cmd_fork_id ) };
4509 :
4510 64058 : switch( op ) {
4511 0 : case FD_ACCDB_CMD_ADVANCE_ROOT:
4512 0 : background_advance_root( accdb, fork_id );
4513 0 : break;
4514 64057 : case FD_ACCDB_CMD_PURGE:
4515 64057 : background_purge( accdb, fork_id );
4516 64057 : break;
4517 0 : case FD_ACCDB_CMD_DRAIN_DEFERRED:
4518 0 : drain_deferred_frees( accdb );
4519 0 : break;
4520 0 : case FD_ACCDB_CMD_CLEAR_DEFERRED: {
4521 : /* Posted by fd_accdb_reset after it clobbers shared pools.
4522 : T2's deferred fork chain now points at recycled elements;
4523 : discard the stale pointers. Epoch slots are preserved
4524 : across reset so no re-join is needed. */
4525 0 : accdb->deferred_fork_head = NULL;
4526 0 : accdb->deferred_fork_tail = NULL;
4527 0 : accdb->deferred_fork_epoch = 0UL;
4528 0 : accdb->carry_sz = 0UL;
4529 0 : break;
4530 0 : }
4531 0 : default:
4532 0 : FD_LOG_ERR(( "unexpected accdb cmd_op %u", op ));
4533 64058 : }
4534 :
4535 64130 : FD_COMPILER_MFENCE();
4536 64130 : FD_VOLATILE( shmem->cmd_op ) = FD_ACCDB_CMD_IDLE;
4537 64130 : *charge_busy = 1;
4538 64130 : return;
4539 64058 : }
4540 :
4541 >1844*10^16 : if( FD_UNLIKELY( snap_sync!=FD_ACCDB_SNAPSHOT_SYNC_IDLE ) ) {
4542 0 : switch( snap_sync ) {
4543 0 : case FD_ACCDB_SNAPSHOT_SYNC_RUNNING:
4544 : /* while producing a snapshot, don't do compaction work */
4545 0 : background_preevict( accdb, charge_busy, 0 );
4546 0 : return;
4547 0 : case FD_ACCDB_SNAPSHOT_SYNC_DONE:
4548 0 : fd_accdb_snapshot_sync_advance( snap_sync_p, FD_ACCDB_SNAPSHOT_SYNC_IDLE );
4549 0 : break;
4550 0 : case FD_ACCDB_SNAPSHOT_SYNC_START_FULL:
4551 0 : FD_CHECK_CRIT( !FD_VOLATILE_CONST( shmem->snapshot_loading ),
4552 0 : "snapshot production requested during snapshot load" );
4553 0 : delta_reset( accdb );
4554 0 : fd_accdb_snapshot_sync_advance( snap_sync_p, FD_ACCDB_SNAPSHOT_SYNC_RUNNING );
4555 0 : *charge_busy = 1;
4556 0 : return;
4557 0 : case FD_ACCDB_SNAPSHOT_SYNC_START_INCR:
4558 0 : FD_CHECK_CRIT( !FD_VOLATILE_CONST( shmem->snapshot_loading ),
4559 0 : "snapshot production requested during snapshot load" );
4560 0 : if( delta_is_valid( accdb->shmem ) ) {
4561 0 : fd_accdb_snapshot_sync_advance( snap_sync_p, FD_ACCDB_SNAPSHOT_SYNC_RUNNING );
4562 0 : } else {
4563 : /* cannot produce incrementals because delta ran out of space,
4564 : therefore don't know which accounts changed */
4565 0 : fd_accdb_snapshot_sync_advance( snap_sync_p, FD_ACCDB_SNAPSHOT_SYNC_FAIL );
4566 0 : }
4567 0 : *charge_busy = 1;
4568 0 : return;
4569 0 : case FD_ACCDB_SNAPSHOT_SYNC_FAIL:
4570 : /* wait for client to acknowledge */
4571 0 : break;
4572 0 : default:
4573 0 : FD_LOG_CRIT(( "corrupt snapshot_sync state %lu", snap_sync ));
4574 0 : }
4575 0 : }
4576 :
4577 >1844*10^16 : background_preevict( accdb, charge_busy, 0 );
4578 :
4579 : /* Keep one partition allocated ahead so a write head rotating into a
4580 : new partition never waits on fallocate. */
4581 >1844*10^16 : if( FD_UNLIKELY( FD_VOLATILE_CONST( shmem->partition_max )>=FD_VOLATILE_CONST( shmem->partition_fallocated ) &&
4582 >1844*10^16 : FD_VOLATILE_CONST( shmem->partition_fallocated )!=accdb->prealloc_failed_at ) ) {
4583 0 : spin_lock_acquire( &shmem->grow_lock );
4584 0 : ulong fallocated = shmem->partition_fallocated;
4585 0 : int grow = fallocated<shmem->partition_cnt && FD_VOLATILE_CONST( shmem->partition_max )>=fallocated;
4586 0 : int err = grow ? grow_locked( accdb, fallocated, 0, ULONG_MAX, 1 ) : 0;
4587 0 : spin_lock_release( &shmem->grow_lock );
4588 0 : if( FD_UNLIKELY( err ) ) {
4589 0 : FD_LOG_WARNING(( "fallocate() failed (%d-%s) preallocating the accounts database beyond %lu GiB. "
4590 0 : "The validator will exit soon when the disk fills.",
4591 0 : err, fd_io_strerror( err ), fallocated*shmem->partition_sz/(1UL<<30UL) ));
4592 0 : accdb->prealloc_failed_at = fallocated;
4593 0 : } else if( FD_LIKELY( grow ) ) {
4594 0 : *charge_busy = 1;
4595 0 : return;
4596 0 : }
4597 0 : }
4598 :
4599 >1844*10^16 : for( ulong k=0UL; k<FD_ACCDB_COMPACTION_LAYER_CNT; k++ ) {
4600 0 : if( FD_UNLIKELY( accdb->carry_sz && k!=accdb->carry_layer ) ) continue; /* would overwrite the carried tail */
4601 0 : background_compact( accdb, k, charge_busy );
4602 0 : }
4603 >1844*10^16 : }
4604 :
4605 : fd_accdb_shmem_metrics_t const *
4606 0 : fd_accdb_shmetrics( fd_accdb_t * accdb ) {
4607 0 : return accdb->shmem->shmetrics;
4608 0 : }
4609 :
4610 : fd_accdb_metrics_t const *
4611 0 : fd_accdb_metrics( fd_accdb_t * accdb ) {
4612 0 : return accdb->metrics;
4613 0 : }
4614 :
4615 : void
4616 : fd_accdb_cache_class_occupancy( fd_accdb_t * accdb,
4617 : ulong * used,
4618 : ulong * max,
4619 0 : ulong * reserved ) {
4620 0 : for( ulong c=0UL; c<FD_ACCDB_CACHE_CLASS_CNT; c++ ) {
4621 0 : ulong cap = accdb->shmem->cache_class_max[ c ];
4622 0 : ulong init = FD_VOLATILE_CONST( accdb->shmem->cache_class_init[ c ].val );
4623 0 : ulong freec = FD_VOLATILE_CONST( accdb->shmem->cache_free_cnt [ c ].val );
4624 0 : ulong live = init>freec ? init-freec : 0UL;
4625 0 : if( live>cap ) live = cap;
4626 0 : max [ c ] = cap;
4627 0 : used [ c ] = live;
4628 0 : reserved[ c ] = FD_VOLATILE_CONST( accdb->shmem->cache_class_used[ c ].val );
4629 0 : }
4630 0 : }
4631 :
4632 : void
4633 : fd_accdb_cache_class_thresholds( fd_accdb_t * accdb,
4634 : ulong * free_target,
4635 0 : ulong * free_low_water ) {
4636 0 : for( ulong c=0UL; c<FD_ACCDB_CACHE_CLASS_CNT; c++ ) {
4637 0 : free_target [ c ] = accdb->shmem->cache_free_target [ c ];
4638 0 : free_low_water[ c ] = accdb->shmem->cache_free_low_water[ c ];
4639 0 : }
4640 0 : }
4641 :
4642 : #if FD_HAS_RACESAN
4643 :
4644 : /* Force pre-eviction (ignore the watermark) so a deterministic
4645 : single-threaded test can exercise the writeback path without
4646 : manufacturing real cache pressure. Sweeps several times: CLOCK needs
4647 : two visits to evict a recently-touched line (clear the "referenced"
4648 : bit, then evict), and the clock hand position carries across calls, so
4649 : one or two sweeps is not enough to guarantee every eligible line is
4650 : flushed back. */
4651 : void
4652 : fd_accdb_debug_force_preevict( fd_accdb_t * accdb ) {
4653 : for( ulong iter=0UL; iter<8UL; iter++ ) {
4654 : int charge_busy = 0;
4655 : background_preevict( accdb, &charge_busy, 1 );
4656 : }
4657 : }
4658 :
4659 : /* Locate the resident cache line currently holding `pubkey` (most recent
4660 : generation if multiple). Returns 1 and fills out_class/out_idx on a
4661 : hit, 0 if no resident line matches. Test-only helper so the test can
4662 : target a specific line without seeing the opaque fd_accdb struct. */
4663 :
4664 : int
4665 : fd_accdb_debug_find_line( fd_accdb_t * accdb,
4666 : uchar const * pubkey,
4667 : ulong * out_class,
4668 : ulong * out_idx ) {
4669 : int found = 0;
4670 : uint best_gen = 0U;
4671 : for( ulong c=0UL; c<FD_ACCDB_CACHE_CLASS_CNT; c++ ) {
4672 : ulong init = FD_VOLATILE_CONST( accdb->shmem->cache_class_init[ c ].val );
4673 : ulong max_c = accdb->shmem->cache_class_max[ c ];
4674 : if( init>max_c ) init = max_c;
4675 : for( ulong idx=0UL; idx<init; idx++ ) {
4676 : fd_accdb_cache_line_t * line = cache_line( accdb, c, idx );
4677 : if( line->key.generation==UINT_MAX ) continue;
4678 : if( memcmp( line->key.pubkey, pubkey, 32UL ) ) continue;
4679 : if( !found || line->key.generation>=best_gen ) {
4680 : best_gen = line->key.generation;
4681 : *out_class = c;
4682 : *out_idx = idx;
4683 : found = 1;
4684 : }
4685 : }
4686 : }
4687 : return found;
4688 : }
4689 :
4690 : /* Address of one cache line. struct fd_accdb_private is local to this
4691 : translation unit, so a test cannot reach accdb->cache[] itself. */
4692 :
4693 : void *
4694 : fd_accdb_debug_line_addr( fd_accdb_t * accdb,
4695 : ulong size_class,
4696 : ulong line_idx ) {
4697 : return cache_line( accdb, size_class, line_idx );
4698 : }
4699 :
4700 : uint *
4701 : fd_accdb_debug_acc_pool_next( fd_accdb_t * accdb,
4702 : uint acc_idx ) {
4703 : FD_TEST( (ulong)acc_idx<acc_pool_ele_max( accdb->acc_pool_join ) );
4704 : return &accdb->acc_pool[ acc_idx ].pool.next;
4705 : }
4706 :
4707 : /* Deterministically evict a single specified cache line via the
4708 : foreground evictor's claim sequence (CAS refcnt 0->EVICT_SENTINEL),
4709 : then write the dirty line back exactly as fd_accdb_acquire_inner's
4710 : STEP-4 / background_ preevict do (pubkey from accmeta, owner+data
4711 : from the line). Mirrors acquire_cache_line's CLOCK-claim path
4712 : (fd_accdb.c) so a racesan test can reproduce, without a 640+-slot
4713 : cache-pressure rig, the interleaving where acc_unlink observes
4714 : EVICT_SENTINEL on the line it is unlinking.
4715 :
4716 : The fd_racesan_hook("clock_evict:post_sentinel") fires right after
4717 : the sentinel is installed (matching the production foreground path),
4718 : so the test can suspend this fiber holding the sentinel while another
4719 : fiber drives acc_unlink to its reclaim CAS. Returns the captured
4720 : evicted acc_idx (UINT_MAX if the line was clean / unbound). */
4721 :
4722 : uint
4723 : fd_accdb_debug_clock_evict_line( fd_accdb_t * accdb,
4724 : ulong size_class,
4725 : ulong line_idx ) {
4726 : fd_accdb_shmem_t * shmem = accdb->shmem;
4727 : fd_accdb_cache_line_t * line = cache_line( accdb, size_class, line_idx );
4728 :
4729 : /* Claim for eviction, same as acquire_cache_line's CLOCK path. */
4730 : if( FD_UNLIKELY( FD_ATOMIC_CAS( &line->refcnt, 0U, FD_ACCDB_EVICT_SENTINEL )!=0U ) ) return UINT_MAX;
4731 :
4732 : if( FD_UNLIKELY( line->acc_idx==UINT_MAX && line->key.generation==UINT_MAX ) ) {
4733 : FD_VOLATILE( line->refcnt ) = 0;
4734 : return UINT_MAX;
4735 : }
4736 :
4737 : fd_racesan_hook( "clock_evict:post_sentinel" );
4738 :
4739 : uint acc_idx = line->acc_idx;
4740 : if( FD_LIKELY( acc_idx!=UINT_MAX ) ) {
4741 : evict_clear_acc_cache_ref( &accdb->acc_pool[ acc_idx ], size_class, line_idx );
4742 : }
4743 : uint evicted_acc_idx = line->persisted ? UINT_MAX : acc_idx;
4744 : line->key.generation = UINT_MAX;
4745 : fd_racesan_hook( "clock_evict:pre_synth" );
4746 :
4747 : /* Write back the dirty line, exactly like the production writeback
4748 : sites: this is the synthesis that would emit a pubkey=NEW/owner=OLD
4749 : poison record if the accmeta slot had been recycled out from under
4750 : us. In the SENTINEL-vs-acc_unlink race this proves no poison: the
4751 : epoch the evictor holds blocks drain_deferred_frees, so the slot is
4752 : never recycled while we are here. */
4753 : if( FD_UNLIKELY( !line->persisted && acc_idx!=UINT_MAX ) ) {
4754 : fd_accdb_accmeta_t * accmeta = &accdb->acc_pool[ acc_idx ];
4755 : ulong entry_sz = sizeof(fd_accdb_disk_meta_t)+(ulong)FD_ACCDB_SIZE_DATA( accmeta->executable_size );
4756 :
4757 : ulong old_offset = fd_accdb_acc_xchg_offset( accmeta, FD_ACCDB_OFF_INVAL );
4758 : if( FD_LIKELY( old_offset!=FD_ACCDB_OFF_INVAL ) ) {
4759 : fd_accdb_shmem_bytes_freed( shmem, old_offset, entry_sz );
4760 : FD_ATOMIC_FETCH_AND_SUB( &shmem->shmetrics->disk_used_bytes, entry_sz );
4761 : }
4762 :
4763 : fd_accdb_disk_meta_t meta;
4764 : fd_memcpy( meta.pubkey, accmeta->key.pubkey, 32UL );
4765 : meta.size = FD_ACCDB_SIZE_DATA( accmeta->executable_size );
4766 : meta.generation = accmeta->key.generation;
4767 : fd_memcpy( meta.owner, line->owner, 32UL );
4768 :
4769 : struct iovec iovs[ 2UL ] = {
4770 : { .iov_base = &meta, .iov_len = sizeof(fd_accdb_disk_meta_t) },
4771 : { .iov_base = (void *)(line+1UL), .iov_len = FD_ACCDB_SIZE_DATA( accmeta->executable_size ) }
4772 : };
4773 : ulong file_off = allocate_next_write( accdb, entry_sz );
4774 : ulong written = 0UL;
4775 : while( written<entry_sz ) {
4776 : long result = pwritev2( accdb->fd, iovs, 2, (long)(file_off+written), 0 );
4777 : if( FD_UNLIKELY( result==-1 && errno==EINTR ) ) continue;
4778 : else if( FD_UNLIKELY( result<=0 ) ) FD_LOG_ERR(( "pwritev2() failed (%d-%s)", errno, fd_io_strerror( errno ) ));
4779 : written += (ulong)result;
4780 : for( int v=0; v<2; v++ ) {
4781 : if( (ulong)result>=iovs[ v ].iov_len ) { result -= (long)iovs[ v ].iov_len; iovs[ v ].iov_len = 0UL; }
4782 : else { iovs[ v ].iov_base = (uchar *)iovs[ v ].iov_base + result; iovs[ v ].iov_len -= (ulong)result; break; }
4783 : }
4784 : }
4785 : FD_COMPILER_MFENCE();
4786 : accmeta->offset_fork = fd_accdb_acc_pack_offset_fork( file_off, fd_accdb_acc_fork_id(accmeta) );
4787 : FD_ATOMIC_FETCH_AND_ADD( &shmem->shmetrics->disk_used_bytes, entry_sz );
4788 : }
4789 :
4790 : line->persisted = 1;
4791 : line->acc_idx = UINT_MAX;
4792 : line->key.generation = UINT_MAX;
4793 : FD_COMPILER_MFENCE();
4794 : FD_VOLATILE( line->refcnt ) = 0;
4795 : cache_free_push( accdb, size_class, line );
4796 : return evicted_acc_idx;
4797 : }
4798 :
4799 : #endif
|
