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