Line data Source code
1 : #define _GNU_SOURCE
2 : #include "fd_store.h"
3 :
4 : #include <stdatomic.h>
5 : #include <errno.h>
6 : #include <fcntl.h>
7 : #include <unistd.h>
8 :
9 0 : #define FD_STORE_FEC_DATA_VIEW_SPILL (1U)
10 :
11 : enum {
12 : FD_STORE_SPILL_RETRY = -1,
13 : FD_STORE_SPILL_NONE = 0,
14 : FD_STORE_SPILL_COMPLETE = 1
15 : };
16 :
17 : static int
18 : store_pwrite_all( int fd,
19 : void const * buf,
20 : ulong sz,
21 0 : off_t off ) {
22 0 : ulong written = 0UL;
23 0 : while( written<sz ) {
24 0 : long res = pwrite( fd, (uchar const *)buf + written, sz-written, off+(off_t)written );
25 0 : if( FD_LIKELY( res>0L ) ) { written += (ulong)res; continue; }
26 0 : if( FD_UNLIKELY( res<0L && errno==EINTR ) ) continue;
27 0 : if( FD_UNLIKELY( !res ) ) errno = EIO;
28 0 : return -1;
29 0 : }
30 0 : return 0;
31 0 : }
32 :
33 : static int
34 : store_pread_all( int fd,
35 : void * buf,
36 : ulong sz,
37 0 : off_t off ) {
38 0 : ulong read_sz = 0UL;
39 0 : while( read_sz<sz ) {
40 0 : long res = pread( fd, (uchar *)buf + read_sz, sz-read_sz, off+(off_t)read_sz );
41 0 : if( FD_LIKELY( res>0L ) ) { read_sz += (ulong)res; continue; }
42 0 : if( FD_UNLIKELY( res<0L && errno==EINTR ) ) continue;
43 0 : if( FD_UNLIKELY( !res ) ) errno = EIO;
44 0 : return -1;
45 0 : }
46 0 : return 0;
47 0 : }
48 :
49 : static inline fd_shredb_ring_entry_t *
50 0 : disk_ring_pool_laddr( fd_store_t const * store ) {
51 0 : return fd_wksp_laddr_fast( fd_store_wksp( store ), store->ring_pool_gaddr );
52 0 : }
53 :
54 : static inline fd_shredb_slot_map_t *
55 0 : disk_slot_map_join( fd_store_t const * store ) {
56 0 : void * shmap = fd_wksp_laddr_fast( fd_store_wksp( store ), store->shred_map_gaddr );
57 0 : return fd_shredb_slot_map_join( shmap );
58 0 : }
59 :
60 : static inline fd_shredb_root_map_t *
61 0 : disk_root_map_join( fd_store_t const * store ) {
62 0 : void * shmap = fd_wksp_laddr_fast( fd_store_wksp( store ), store->root_map_gaddr );
63 0 : return fd_shredb_root_map_join( shmap );
64 0 : }
65 :
66 : static inline atomic_ulong *
67 0 : disk_slot_hint_laddr( fd_store_t const * store ) {
68 0 : return fd_wksp_laddr_fast( fd_store_wksp( store ), store->slot_hint_gaddr );
69 0 : }
70 :
71 : static inline uchar *
72 0 : cache_data_laddr( fd_store_t const * store ) {
73 0 : return fd_wksp_laddr_fast( fd_store_wksp( store ), store->cache_data_gaddr );
74 0 : }
75 :
76 : static inline ulong *
77 0 : cache_free_laddr( fd_store_t const * store ) {
78 0 : return fd_wksp_laddr_fast( fd_store_wksp( store ), store->cache_free_gaddr );
79 0 : }
80 :
81 : static inline uint *
82 0 : spill_free_laddr( fd_store_t const * store ) {
83 0 : return fd_wksp_laddr_fast( fd_store_wksp( store ), store->spill_free_gaddr );
84 0 : }
85 :
86 : static inline uchar *
87 0 : spill_read_data_laddr( fd_store_t const * store ) {
88 0 : return fd_wksp_laddr_fast( fd_store_wksp( store ), store->spill_read_data_gaddr );
89 0 : }
90 :
91 : static inline fd_store_fec_t *
92 0 : pool_ele_laddr( fd_store_t const * store ) {
93 0 : return fd_wksp_laddr_fast( fd_store_wksp( store ), store->pool_ele_gaddr );
94 0 : }
95 :
96 : static inline fd_store_slot_to_fecs_t *
97 : slot_to_fecs_laddr( fd_store_t const * store,
98 0 : ulong part ) {
99 0 : return fd_wksp_laddr_fast( fd_store_wksp( store ), store->slot_to_fecs_gaddr + part*store->slot_to_fecs_sz );
100 0 : }
101 :
102 : static inline fd_store_pool_t
103 0 : pool_ljoin( fd_store_t const * store ) {
104 0 : return (fd_store_pool_t){
105 0 : .pool = fd_wksp_laddr_fast( fd_store_wksp( store ), store->pool_mem_gaddr ),
106 0 : .ele = fd_wksp_laddr_fast( fd_store_wksp( store ), store->pool_ele_gaddr ),
107 0 : .ele_max = store->fec_max
108 0 : };
109 0 : }
110 :
111 : static fd_store_fec_t *
112 0 : fd_store_fec_acquire( fd_store_t * store ) {
113 0 : fd_store_pool_t pool = pool_ljoin( store );
114 0 : fd_store_fec_t * fec = fd_store_pool_acquire( &pool );
115 0 : if( FD_LIKELY( fec ) ) {
116 0 : fec->data_sz = 0UL;
117 0 : fec->data_idx = 0U;
118 0 : fec->cache_prev = UINT_MAX;
119 0 : fec->cache_next = UINT_MAX;
120 0 : fec->data_pin_cnt = 0U;
121 0 : fec->data_state = FD_STORE_FEC_DATA_EMPTY;
122 0 : fec->data_consume_pending = 0U;
123 0 : }
124 0 : return fec;
125 0 : }
126 :
127 : static void
128 : cache_slot_release_locked( fd_store_t * store,
129 0 : ulong cache_slot ) {
130 0 : FD_TEST( cache_slot<store->cache_slot_cnt );
131 0 : FD_TEST( store->cache_free_cnt<store->cache_slot_cnt );
132 0 : cache_free_laddr( store )[ store->cache_free_cnt++ ] = cache_slot;
133 0 : }
134 :
135 : static inline uint
136 : cache_fec_idx( fd_store_t const * store,
137 0 : fd_store_fec_t const * fec ) {
138 0 : ulong off = (ulong)fec - (ulong)pool_ele_laddr( store );
139 0 : FD_TEST( !(off % sizeof(fd_store_fec_t)) );
140 0 : ulong idx = off / sizeof(fd_store_fec_t);
141 0 : FD_TEST( idx<store->fec_max );
142 0 : return (uint)idx;
143 0 : }
144 :
145 : static void
146 : cache_lru_remove_locked( fd_store_t * store,
147 0 : fd_store_fec_t * fec ) {
148 0 : uint idx = cache_fec_idx( store, fec );
149 0 : uint prev = fec->cache_prev;
150 0 : uint next = fec->cache_next;
151 :
152 0 : if( prev==UINT_MAX ) { FD_TEST( store->cache_lru_head==idx ); store->cache_lru_head = next; }
153 0 : else pool_ele_laddr( store )[ prev ].cache_next = next;
154 0 : if( next==UINT_MAX ) { FD_TEST( store->cache_lru_tail==idx ); store->cache_lru_tail = prev; }
155 0 : else pool_ele_laddr( store )[ next ].cache_prev = prev;
156 :
157 0 : fec->cache_prev = UINT_MAX;
158 0 : fec->cache_next = UINT_MAX;
159 0 : }
160 :
161 : static void
162 : cache_lru_push_head_locked( fd_store_t * store,
163 0 : fd_store_fec_t * fec ) {
164 0 : uint idx = cache_fec_idx( store, fec );
165 0 : FD_TEST( fec->cache_prev==UINT_MAX && fec->cache_next==UINT_MAX );
166 :
167 0 : fec->cache_next = store->cache_lru_head;
168 0 : if( store->cache_lru_head==UINT_MAX ) store->cache_lru_tail = idx;
169 0 : else pool_ele_laddr( store )[ store->cache_lru_head ].cache_prev = idx;
170 0 : store->cache_lru_head = idx;
171 0 : }
172 :
173 : static void
174 : cache_lru_push_tail_locked( fd_store_t * store,
175 0 : fd_store_fec_t * fec ) {
176 0 : uint idx = cache_fec_idx( store, fec );
177 0 : FD_TEST( fec->cache_prev==UINT_MAX && fec->cache_next==UINT_MAX );
178 :
179 0 : fec->cache_prev = store->cache_lru_tail;
180 0 : if( store->cache_lru_tail==UINT_MAX ) store->cache_lru_head = idx;
181 0 : else pool_ele_laddr( store )[ store->cache_lru_tail ].cache_next = idx;
182 0 : store->cache_lru_tail = idx;
183 0 : }
184 :
185 : static void
186 : spill_free_push_locked( fd_store_t * store,
187 0 : uint spill_slot ) {
188 0 : FD_TEST( spill_slot<store->spill_slot_cnt && store->spill_free_cnt<store->spill_slot_cnt );
189 0 : spill_free_laddr( store )[ store->spill_free_cnt++ ] = spill_slot;
190 0 : }
191 :
192 : static void
193 : cache_consume_locked( fd_store_t * store,
194 0 : fd_store_fec_t * fec ) {
195 0 : if( FD_LIKELY( !fec->data_consume_pending ) ) return;
196 0 : if( FD_UNLIKELY( fec->data_pin_cnt || fec->data_state==FD_STORE_FEC_DATA_SPILLING ) ) return;
197 :
198 0 : if( FD_LIKELY( fec->data_state==FD_STORE_FEC_DATA_RAM_WRITING ||
199 0 : fec->data_state==FD_STORE_FEC_DATA_RAM_READY ) ) {
200 0 : if( fec->data_state==FD_STORE_FEC_DATA_RAM_READY ) cache_lru_remove_locked( store, fec );
201 0 : cache_slot_release_locked( store, fec->data_idx );
202 0 : } else if( fec->data_state==FD_STORE_FEC_DATA_DISK ) {
203 0 : ulong spill_slot = fec->data_idx;
204 0 : FD_TEST( spill_slot<store->spill_slot_cnt && atomic_load_explicit( &store->spill_live_cnt, memory_order_relaxed ) );
205 0 : atomic_fetch_sub_explicit( &store->spill_live_cnt, 1UL, memory_order_relaxed );
206 0 : spill_free_push_locked( store, (uint)spill_slot );
207 0 : }
208 :
209 0 : fec->data_idx = 0U;
210 0 : fec->cache_prev = UINT_MAX;
211 0 : fec->cache_next = UINT_MAX;
212 0 : fec->data_state = FD_STORE_FEC_DATA_CONSUMED;
213 0 : fec->data_consume_pending = 0U;
214 0 : }
215 :
216 : static void
217 : fd_store_fec_release( fd_store_t * store,
218 0 : fd_store_fec_t * fec ) {
219 0 : (void)cache_fec_idx( store, fec );
220 0 : for(;;) {
221 0 : fd_rwlock_write( &store->cache_lock );
222 0 : fec->data_consume_pending = 1U;
223 0 : cache_consume_locked( store, fec );
224 0 : int consumed = fec->data_state==FD_STORE_FEC_DATA_CONSUMED;
225 0 : fd_rwlock_unwrite( &store->cache_lock );
226 0 : if( FD_LIKELY( consumed ) ) break;
227 0 : FD_SPIN_PAUSE();
228 0 : }
229 0 : fd_store_pool_t pool = pool_ljoin( store );
230 0 : fd_store_pool_release( &pool, fec );
231 0 : }
232 :
233 : static int
234 : spill_one_locked( fd_store_t * store,
235 : int disk_fd,
236 0 : fd_store_fec_spill_stats_t * spill ) {
237 0 : if( FD_UNLIKELY( disk_fd<0 ) ) return FD_STORE_SPILL_NONE;
238 :
239 0 : fd_store_fec_t * fec0 = pool_ele_laddr( store );
240 0 : uint victim_idx = store->cache_lru_head;
241 0 : if( FD_UNLIKELY( victim_idx==UINT_MAX ) ) return FD_STORE_SPILL_NONE;
242 0 : fd_store_fec_t * victim = fec0 + victim_idx;
243 0 : FD_TEST( victim->data_state==FD_STORE_FEC_DATA_RAM_READY && !victim->data_pin_cnt );
244 0 : FD_TEST( victim->data_sz<=store->fec_data_max );
245 :
246 0 : uint spill_slot;
247 0 : if( FD_LIKELY( store->spill_free_cnt ) ) {
248 0 : spill_slot = spill_free_laddr( store )[ --store->spill_free_cnt ];
249 0 : } else {
250 0 : if( FD_UNLIKELY( store->spill_slot_cnt>=store->fec_max ) ) return FD_STORE_SPILL_NONE;
251 0 : spill_slot = (uint)store->spill_slot_cnt++;
252 0 : }
253 :
254 0 : ulong ram_slot = victim->data_idx;
255 0 : ulong ram_off = ram_slot * store->payload_slot_sz;
256 0 : ulong spill_off = (ulong)spill_slot * store->payload_slot_sz;
257 0 : ulong data_sz = victim->data_sz;
258 :
259 0 : cache_lru_remove_locked( store, victim );
260 0 : victim->data_state = FD_STORE_FEC_DATA_SPILLING;
261 0 : long evict_ticks = -fd_tickcount();
262 0 : fd_rwlock_unwrite( &store->cache_lock );
263 :
264 0 : if( FD_UNLIKELY( store_pwrite_all( disk_fd, cache_data_laddr( store ) + ram_off, data_sz, (off_t)spill_off ) ) )
265 0 : FD_LOG_ERR(( "error spilling FEC payload to disk: (%d-%s)", errno, fd_io_strerror( errno ) ));
266 :
267 0 : fd_rwlock_write( &store->cache_lock );
268 0 : FD_TEST( victim->data_state==FD_STORE_FEC_DATA_SPILLING );
269 0 : int result;
270 0 : if( FD_UNLIKELY( victim->data_pin_cnt || victim->data_consume_pending ) ) {
271 0 : int try_next = !!victim->data_pin_cnt;
272 0 : spill_free_push_locked( store, spill_slot );
273 0 : victim->data_state = victim->data_consume_pending && !victim->data_pin_cnt
274 0 : ? FD_STORE_FEC_DATA_RAM_WRITING
275 0 : : FD_STORE_FEC_DATA_RAM_READY;
276 0 : if( FD_LIKELY( !victim->data_pin_cnt && !victim->data_consume_pending ) )
277 0 : cache_lru_push_tail_locked( store, victim );
278 0 : cache_consume_locked( store, victim );
279 0 : result = try_next ? FD_STORE_SPILL_RETRY :
280 0 : store->cache_free_cnt ? FD_STORE_SPILL_COMPLETE : FD_STORE_SPILL_NONE;
281 0 : } else {
282 0 : victim->data_idx = spill_slot;
283 0 : victim->data_state = FD_STORE_FEC_DATA_DISK;
284 0 : atomic_fetch_add_explicit( &store->spill_live_cnt, 1UL, memory_order_relaxed );
285 0 : atomic_fetch_add_explicit( &store->fec_spill_cnt, 1UL, memory_order_relaxed );
286 0 : atomic_fetch_add_explicit( &store->fec_spill_bytes, data_sz, memory_order_relaxed );
287 0 : cache_slot_release_locked( store, ram_slot );
288 0 : result = FD_STORE_SPILL_COMPLETE;
289 0 : }
290 :
291 0 : evict_ticks += fd_tickcount();
292 0 : if( FD_LIKELY( spill ) ) {
293 0 : spill->write_cnt++;
294 0 : spill->write_bytes += data_sz;
295 0 : spill->write_ticks += (ulong)evict_ticks;
296 0 : }
297 0 : return result;
298 0 : }
299 :
300 :
301 : void *
302 : fd_store_new( void * shmem,
303 : ulong fec_max,
304 : ulong fec_data_max,
305 : ulong shred_storage_gib,
306 : ulong shred_cache_bytes,
307 : ulong fec_set_cnt,
308 : ulong max_shreds_per_block,
309 : ulong seed,
310 0 : ulong shred_tile_cnt ) {
311 :
312 0 : if( FD_UNLIKELY( !shmem ) ) { FD_LOG_WARNING(( "NULL shmem" )); return NULL; }
313 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shmem, fd_store_align() ) ) ) { FD_LOG_WARNING(( "misaligned shmem" )); return NULL; }
314 0 : if( FD_UNLIKELY( !fec_max ) ) { FD_LOG_WARNING(( "fec_max must be non-zero" )); return NULL; }
315 0 : if( FD_UNLIKELY( fec_max>UINT_MAX ) ) { FD_LOG_WARNING(( "fec_max must fit in uint" )); return NULL; }
316 0 : if( FD_UNLIKELY( !fec_data_max ) ) { FD_LOG_WARNING(( "fec_data_max must be non-zero" )); return NULL; }
317 0 : if( FD_UNLIKELY( fec_data_max>UINT_MAX ) ) { FD_LOG_WARNING(( "fec_data_max must fit in uint" )); return NULL; }
318 0 : if( FD_UNLIKELY( !max_shreds_per_block || max_shreds_per_block>FD_SHREDB_HINT_VALID ) ) { FD_LOG_WARNING(( "bad max_shreds_per_block (%lu)", max_shreds_per_block )); return NULL; }
319 0 : if( FD_UNLIKELY( !shred_tile_cnt ) ) { FD_LOG_WARNING(( "shred_tile_cnt must be non-zero" )); return NULL; }
320 0 : if( FD_UNLIKELY( shred_storage_gib>FD_SHREDB_MAX_SIZE_GIB ) ) {
321 0 : FD_LOG_ERR(( "shred database size limit is %lu GiB, but the maximum supported size is %lu GiB",
322 0 : shred_storage_gib, FD_SHREDB_MAX_SIZE_GIB ));
323 0 : }
324 :
325 0 : ulong footprint = fd_store_footprint( fec_max, fec_data_max, shred_storage_gib, shred_cache_bytes, fec_set_cnt, shred_tile_cnt );
326 0 : if( FD_UNLIKELY( !footprint ) ) { FD_LOG_WARNING(( "invalid or overflowing store footprint" )); return NULL; }
327 :
328 0 : fd_wksp_t * wksp = fd_wksp_containing( shmem );
329 0 : if( FD_UNLIKELY( !wksp ) ) { FD_LOG_WARNING(( "shmem must be part of a workspace" )); return NULL; }
330 :
331 0 : ulong chain_cnt = fd_store_map_chain_cnt_est( fec_max );
332 0 : ulong part_cnt = shred_tile_cnt;
333 0 : ulong slot_chain_cnt = fd_store_slot_to_fecs_chain_cnt_est( fec_max );
334 0 : ulong slot_to_fecs_sz = fd_ulong_align_up( fd_store_slot_to_fecs_footprint( slot_chain_cnt ), fd_store_slot_to_fecs_align() );
335 0 : ulong payload_slot_sz = fd_store_payload_slot_sz( fec_data_max );
336 0 : ulong cache_slot_cnt = shred_cache_bytes
337 0 : ? fd_ulong_min( fec_max, fd_ulong_max( 1UL, shred_cache_bytes / payload_slot_sz ) )
338 0 : : fec_max;
339 :
340 0 : FD_SCRATCH_ALLOC_INIT( l, shmem );
341 0 : fd_store_t * store = FD_SCRATCH_ALLOC_APPEND( l, fd_store_align(), sizeof(fd_store_t) );
342 0 : void * map = FD_SCRATCH_ALLOC_APPEND( l, fd_store_map_align(), fd_store_map_footprint( chain_cnt ) );
343 0 : void * shpool = FD_SCRATCH_ALLOC_APPEND( l, fd_store_pool_align(), fd_store_pool_footprint() );
344 0 : void * shele = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_store_fec_t), sizeof(fd_store_fec_t)*fec_max );
345 0 : uchar * slot_to_fecs = FD_SCRATCH_ALLOC_APPEND( l, fd_store_slot_to_fecs_align(), slot_to_fecs_sz*part_cnt );
346 0 : uchar * cache_mem = FD_SCRATCH_ALLOC_APPEND( l, FD_STORE_PAYLOAD_PAGE_SZ, payload_slot_sz*cache_slot_cnt );
347 0 : ulong * cache_free = FD_SCRATCH_ALLOC_APPEND( l, alignof(ulong), sizeof(ulong)*cache_slot_cnt );
348 0 : uint * spill_free = FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), sizeof(uint)*fec_max );
349 0 : uchar * spill_read_mem = FD_SCRATCH_ALLOC_APPEND( l, FD_STORE_PAYLOAD_PAGE_SZ, payload_slot_sz );
350 0 : fd_fec_set_t * fec_sets = fec_set_cnt
351 0 : ? FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_fec_set_t), sizeof(fd_fec_set_t)*fec_set_cnt )
352 0 : : NULL;
353 :
354 0 : void * shred_map_mem = NULL;
355 0 : fd_shredb_ring_entry_t * ring_pool_mem = NULL;
356 0 : void * root_map_mem = NULL;
357 0 : atomic_ulong * slot_hint_mem = NULL;
358 0 : ulong max_shreds = 0UL;
359 0 : ulong max_slots = 0UL;
360 0 : ulong disk_chain_cnt = 0UL;
361 :
362 0 : if( shred_storage_gib ) {
363 0 : max_shreds = fd_shredb_max_shreds( shred_storage_gib );
364 0 : max_slots = fd_shredb_max_slots( shred_storage_gib );
365 0 : disk_chain_cnt = fd_shredb_disk_chain_cnt( max_shreds );
366 :
367 0 : shred_map_mem = FD_SCRATCH_ALLOC_APPEND( l, fd_shredb_slot_map_align(), fd_shredb_slot_map_footprint( disk_chain_cnt ) );
368 0 : ring_pool_mem = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_shredb_ring_entry_t), max_shreds*sizeof(fd_shredb_ring_entry_t) );
369 0 : slot_hint_mem = FD_SCRATCH_ALLOC_APPEND( l, alignof(atomic_ulong), max_slots*sizeof(atomic_ulong) );
370 0 : root_map_mem = FD_SCRATCH_ALLOC_APPEND( l, fd_shredb_root_map_align(), fd_shredb_root_map_footprint( disk_chain_cnt ) );
371 0 : }
372 :
373 0 : FD_TEST( FD_SCRATCH_ALLOC_FINI( l, fd_store_align() )==(ulong)shmem + footprint );
374 :
375 0 : ulong payload_sz;
376 0 : ulong wire_sz;
377 0 : ulong file_sz;
378 0 : if( FD_UNLIKELY( __builtin_umull_overflow( payload_slot_sz, fec_max, &payload_sz ) ||
379 0 : __builtin_umull_overflow( max_shreds, sizeof(fd_shredb_entry_t), &wire_sz ) ||
380 0 : __builtin_uaddl_overflow( payload_sz, wire_sz, &file_sz ) ||
381 0 : file_sz>(ulong)LONG_MAX ) ) {
382 0 : FD_LOG_WARNING(( "store backing file size overflows off_t" ));
383 0 : return NULL;
384 0 : }
385 :
386 0 : fd_memset( store, 0, sizeof(fd_store_t) );
387 0 : store->fec_max = fec_max;
388 0 : store->fec_data_max = fec_data_max;
389 0 : store->store_gaddr = fd_wksp_gaddr_fast( wksp, store );
390 0 : store->pool_mem_gaddr = fd_wksp_gaddr_fast( wksp, shpool );
391 0 : store->pool_ele_gaddr = fd_wksp_gaddr_fast( wksp, shele );
392 0 : store->shred_tile_cnt = shred_tile_cnt;
393 0 : store->slot_to_fecs_gaddr = fd_wksp_gaddr_fast( wksp, slot_to_fecs );
394 0 : store->slot_to_fecs_sz = slot_to_fecs_sz;
395 0 : atomic_init( &store->slot_lo, ULONG_MAX );
396 0 : store->payload_slot_sz = payload_slot_sz;
397 0 : store->payload_sz = payload_sz;
398 0 : store->wire_off = payload_sz; /* wire region begins after the payload region (page-aligned) */
399 0 : store->cache_slot_cnt = cache_slot_cnt;
400 0 : store->cache_data_gaddr = fd_wksp_gaddr_fast( wksp, cache_mem );
401 0 : store->cache_free_gaddr = fd_wksp_gaddr_fast( wksp, cache_free );
402 0 : store->cache_free_cnt = cache_slot_cnt;
403 0 : ulong free_cap = fd_ulong_min( 8192UL, (64UL<<20) / payload_slot_sz );
404 0 : ulong burst_floor = fd_ulong_min( 512UL, cache_slot_cnt/2UL );
405 0 : ulong free_target = fd_ulong_max( cache_slot_cnt/10UL, burst_floor );
406 0 : store->cache_free_target = fd_ulong_max( 1UL, fd_ulong_min( free_cap, free_target ) );
407 0 : store->cache_free_low_water = fd_ulong_max( 1UL, (store->cache_free_target*3UL)/4UL );
408 0 : store->cache_lru_head = UINT_MAX;
409 0 : store->cache_lru_tail = UINT_MAX;
410 0 : store->spill_free_gaddr = fd_wksp_gaddr_fast( wksp, spill_free );
411 0 : store->spill_read_data_gaddr = fd_wksp_gaddr_fast( wksp, spill_read_mem );
412 0 : store->fec_set_cnt = fec_set_cnt;
413 0 : store->fec_sets_gaddr = fec_set_cnt ? fd_wksp_gaddr_fast( wksp, fec_sets ) : 0UL;
414 0 : store->max_shreds_per_block = max_shreds_per_block;
415 0 : fd_rwlock_new( &store->cache_lock );
416 0 : fd_rwlock_new( &store->spill_read_lock );
417 0 : fd_rwlock_new( &store->fec_lock );
418 0 : atomic_init( &store->fec_spill_cnt, 0UL );
419 0 : atomic_init( &store->spill_live_cnt, 0UL );
420 0 : atomic_init( &store->fec_spill_bytes, 0UL );
421 0 : atomic_init( &store->fec_spill_read_cnt, 0UL );
422 0 : atomic_init( &store->fec_spill_read_bytes, 0UL );
423 :
424 0 : void * shmap = fd_store_map_new( map, chain_cnt, seed );
425 0 : if( FD_UNLIKELY( !shmap ) ) { FD_LOG_WARNING(( "fd_store_map_new failed" )); return NULL; }
426 0 : store->map_gaddr = fd_wksp_gaddr_fast( wksp, shmap );
427 :
428 0 : if( FD_UNLIKELY( !fd_store_pool_new( shpool ) ) ) { FD_LOG_WARNING(( "fd_store_pool_new failed" )); return NULL; }
429 0 : fd_store_pool_t pool_ljoin;
430 0 : fd_store_pool_reset( fd_store_pool_join( &pool_ljoin, shpool, shele, fec_max ) );
431 :
432 0 : for( ulong i=0UL; i<cache_slot_cnt; i++ ) cache_free[ i ] = cache_slot_cnt - 1UL - i;
433 :
434 : /* FEC metadata starts without a payload location. */
435 0 : fd_store_fec_t * fec0 = (fd_store_fec_t *)shele;
436 0 : for( ulong i=0UL; i<fec_max; i++ ) {
437 0 : fec0[ i ].data_idx = 0U;
438 0 : fec0[ i ].cache_prev = UINT_MAX;
439 0 : fec0[ i ].cache_next = UINT_MAX;
440 0 : fec0[ i ].data_pin_cnt = 0U;
441 0 : fec0[ i ].data_state = FD_STORE_FEC_DATA_EMPTY;
442 0 : fec0[ i ].data_consume_pending = 0U;
443 0 : }
444 :
445 0 : for( ulong i=0UL; i<part_cnt; i++ ) FD_TEST( fd_store_slot_to_fecs_new( slot_to_fecs + i*slot_to_fecs_sz, slot_chain_cnt, seed ) );
446 :
447 0 : if( shred_storage_gib ) {
448 0 : void * slot_map = fd_shredb_slot_map_new( shred_map_mem, disk_chain_cnt, seed );
449 0 : FD_TEST( slot_map );
450 0 : store->shred_map_gaddr = fd_wksp_gaddr_fast( wksp, slot_map );
451 0 : store->ring_pool_gaddr = fd_wksp_gaddr_fast( wksp, ring_pool_mem );
452 0 : store->slot_hint_gaddr = fd_wksp_gaddr_fast( wksp, slot_hint_mem );
453 0 : store->disk_max_shreds = max_shreds;
454 0 : store->disk_max_slots = max_slots;
455 0 : fd_rwlock_new( &store->disk_lock );
456 :
457 0 : void * root_map = fd_shredb_root_map_new( root_map_mem, disk_chain_cnt, seed );
458 0 : FD_TEST( root_map );
459 0 : store->root_map_gaddr = fd_wksp_gaddr_fast( wksp, root_map );
460 :
461 : /* Unlink examines the old keys even for a never-used cell. Initialize
462 : both keys and the INVALID tag; chain links are set on insertion. */
463 0 : fd_memset( ring_pool_mem, 0, max_shreds*sizeof(fd_shredb_ring_entry_t) );
464 0 : for( ulong i=0UL; i<max_slots; i++ ) atomic_init( slot_hint_mem+i, 0UL );
465 :
466 0 : atomic_init( &store->disk_reservation_head, 0UL );
467 0 : atomic_init( &store->disk_cnt, 0UL );
468 0 : atomic_init( &store->disk_insert_cnt, 0UL );
469 0 : atomic_init( &store->disk_write_bytes, 0UL );
470 0 : }
471 :
472 0 : FD_COMPILER_MFENCE();
473 0 : FD_VOLATILE( store->magic ) = FD_STORE_MAGIC;
474 0 : FD_COMPILER_MFENCE();
475 :
476 0 : return shmem;
477 0 : }
478 :
479 : fd_store_t *
480 0 : fd_store_join( void * shstore ) {
481 0 : if( FD_UNLIKELY( !shstore ) ) { FD_LOG_WARNING(( "NULL store" )); return NULL; }
482 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shstore, fd_store_align() ) ) ) { FD_LOG_WARNING(( "misaligned store" )); return NULL; }
483 :
484 0 : fd_wksp_t * wksp = fd_wksp_containing( shstore );
485 0 : if( FD_UNLIKELY( !wksp ) ) { FD_LOG_WARNING(( "store must be part of a workspace" )); return NULL; }
486 :
487 0 : fd_store_t * store = (fd_store_t *)shstore;
488 0 : if( FD_UNLIKELY( store->magic!=FD_STORE_MAGIC ) ) { FD_LOG_WARNING(( "bad magic" )); return NULL; }
489 :
490 0 : return store;
491 0 : }
492 :
493 : int
494 : fd_store_file_create( char const * path,
495 : ulong wire_off,
496 0 : ulong disk_max_shreds ) {
497 0 : if( FD_UNLIKELY( !path ) ) {
498 0 : errno = EINVAL;
499 0 : return -1;
500 0 : }
501 :
502 0 : ulong wire_sz;
503 0 : ulong file_sz;
504 0 : if( FD_UNLIKELY( __builtin_umull_overflow( disk_max_shreds, sizeof(fd_shredb_entry_t), &wire_sz ) ||
505 0 : __builtin_uaddl_overflow( wire_off, wire_sz, &file_sz ) ||
506 0 : file_sz>(ulong)LONG_MAX ) ) {
507 0 : errno = EOVERFLOW;
508 0 : return -1;
509 0 : }
510 :
511 : /* No O_TRUNC it frees a previous run's extents and page cache
512 : synchronously. */
513 0 : int fd = open( path, O_RDWR|O_CREAT, (mode_t)0600 );
514 0 : if( FD_UNLIKELY( fd<0 ) ) return -1;
515 :
516 0 : int err = 0;
517 0 : if( FD_UNLIKELY( ftruncate( fd, (off_t)file_sz ) ) ) err = errno;
518 0 : if( FD_UNLIKELY( err ) ) {
519 0 : if( FD_UNLIKELY( close( fd ) ) ) FD_LOG_WARNING(( "close(%s) failed (%i-%s)", path, errno, fd_io_strerror( errno ) ));
520 0 : errno = err;
521 0 : return -1;
522 0 : }
523 0 : return fd;
524 0 : }
525 :
526 : void *
527 0 : fd_store_leave( fd_store_t const * store ) {
528 0 : if( FD_UNLIKELY( !store ) ) { FD_LOG_WARNING(( "NULL store" )); return NULL; }
529 0 : return (void *)store;
530 0 : }
531 :
532 : void *
533 0 : fd_store_delete( void * shstore ) {
534 0 : if( FD_UNLIKELY( !shstore ) ) { FD_LOG_WARNING(( "NULL store" )); return NULL; }
535 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shstore, fd_store_align() ) ) ) { FD_LOG_WARNING(( "misaligned store" )); return NULL; }
536 :
537 0 : fd_store_t * store = (fd_store_t *)shstore;
538 0 : if( FD_UNLIKELY( store->magic!=FD_STORE_MAGIC ) ) { FD_LOG_WARNING(( "bad magic" )); return NULL; }
539 :
540 0 : FD_COMPILER_MFENCE();
541 0 : FD_VOLATILE( store->magic ) = 0UL;
542 0 : FD_COMPILER_MFENCE();
543 :
544 0 : return shstore;
545 0 : }
546 :
547 :
548 : uchar *
549 : fd_store_fec_data_reserve( fd_store_t * store,
550 : int disk_fd,
551 : uint * data_idx,
552 0 : fd_store_fec_spill_stats_t * spill ) {
553 0 : if( FD_LIKELY( spill ) ) *spill = (fd_store_fec_spill_stats_t){0};
554 0 : if( FD_UNLIKELY( !store || !data_idx ) ) return NULL;
555 :
556 : /* Pops a free RAM cache slot, spilling LRU payloads as needed. */
557 0 : fd_rwlock_write( &store->cache_lock );
558 :
559 0 : while( FD_UNLIKELY( !store->cache_free_cnt ) ) {
560 0 : if( FD_UNLIKELY( store->cache_lru_head==UINT_MAX ) ) {
561 0 : fd_rwlock_unwrite( &store->cache_lock );
562 0 : FD_SPIN_PAUSE();
563 0 : fd_rwlock_write( &store->cache_lock );
564 0 : continue;
565 0 : }
566 0 : int spill_result = spill_one_locked( store, disk_fd, spill );
567 0 : if( FD_UNLIKELY( spill_result==FD_STORE_SPILL_RETRY ) ) {
568 0 : fd_rwlock_unwrite( &store->cache_lock );
569 0 : FD_SPIN_PAUSE();
570 0 : fd_rwlock_write( &store->cache_lock );
571 0 : continue;
572 0 : }
573 0 : if( FD_UNLIKELY( !spill_result ) ) {
574 0 : fd_rwlock_unwrite( &store->cache_lock );
575 0 : return NULL;
576 0 : }
577 0 : }
578 :
579 0 : ulong slot = cache_free_laddr( store )[ --store->cache_free_cnt ];
580 0 : fd_rwlock_unwrite( &store->cache_lock );
581 :
582 0 : *data_idx = (uint)slot;
583 0 : return cache_data_laddr( store ) + slot*store->payload_slot_sz;
584 0 : }
585 :
586 : int
587 : fd_store_fec_data_preevict( fd_store_t * store,
588 : int disk_fd,
589 0 : fd_store_fec_spill_stats_t * spill ) {
590 0 : fd_store_fec_spill_stats_t local_spill[1];
591 0 : if( FD_UNLIKELY( !spill ) ) spill = local_spill;
592 0 : *spill = (fd_store_fec_spill_stats_t){0};
593 0 : if( FD_UNLIKELY( !store || disk_fd<0 ) ) return 0;
594 :
595 : /* called every pass: don't take the write lock for nothing */
596 0 : if( FD_LIKELY( !FD_VOLATILE_CONST( store->cache_preevict_active ) &&
597 0 : FD_VOLATILE_CONST( store->cache_free_cnt )>=store->cache_free_low_water ) ) return 0;
598 :
599 0 : fd_rwlock_write( &store->cache_lock );
600 0 : if( FD_UNLIKELY( !store->cache_preevict_active && store->cache_free_cnt<store->cache_free_low_water ) )
601 0 : store->cache_preevict_active = 1U;
602 0 : if( FD_UNLIKELY( store->cache_preevict_active && store->cache_free_cnt>=store->cache_free_target ) )
603 0 : store->cache_preevict_active = 0U;
604 :
605 0 : if( FD_LIKELY( !store->cache_preevict_active ) ) {
606 0 : fd_rwlock_unwrite( &store->cache_lock );
607 0 : return 0;
608 0 : }
609 :
610 0 : spill_one_locked( store, disk_fd, spill );
611 0 : if( FD_UNLIKELY( store->cache_free_cnt>=store->cache_free_target ) )
612 0 : store->cache_preevict_active = 0U;
613 0 : fd_rwlock_unwrite( &store->cache_lock );
614 0 : return !!spill->write_cnt;
615 0 : }
616 :
617 : void
618 : fd_store_fec_cache_stats_query( fd_store_t * store,
619 0 : fd_store_fec_cache_stats_t * stats ) {
620 0 : if( FD_UNLIKELY( !store || !stats ) ) return;
621 0 : fd_rwlock_read( &store->cache_lock );
622 0 : stats->free_cnt = store->cache_free_cnt;
623 0 : stats->max = store->cache_slot_cnt;
624 0 : stats->target = store->cache_free_target;
625 0 : stats->low_water = store->cache_free_low_water;
626 0 : fd_rwlock_unread( &store->cache_lock );
627 0 : }
628 :
629 : static void
630 : cache_pin_locked( fd_store_t * store,
631 0 : fd_store_fec_t * fec ) {
632 0 : if( FD_UNLIKELY( !fec->data_pin_cnt && fec->data_state==FD_STORE_FEC_DATA_RAM_READY ) )
633 0 : cache_lru_remove_locked( store, fec );
634 0 : FD_TEST( fec->data_pin_cnt<USHORT_MAX );
635 0 : fec->data_pin_cnt++;
636 0 : store->cache_pinned_cnt++;
637 0 : }
638 :
639 : static void
640 : cache_unpin_locked( fd_store_t * store,
641 0 : fd_store_fec_t * fec ) {
642 0 : FD_TEST( fec->data_pin_cnt && store->cache_pinned_cnt );
643 0 : fec->data_pin_cnt--;
644 0 : store->cache_pinned_cnt--;
645 : /* Prefer spilling payloads that a reader has already copied while
646 : retaining their disk copy for replay backfill. */
647 0 : if( FD_UNLIKELY( !fec->data_pin_cnt && fec->data_state==FD_STORE_FEC_DATA_RAM_READY ) )
648 0 : cache_lru_push_head_locked( store, fec );
649 0 : }
650 :
651 : int
652 : fd_store_fec_data_view( fd_store_t * store,
653 : int disk_fd,
654 : fd_store_fec_t * fec,
655 0 : fd_store_fec_data_view_t * view ) {
656 0 : if( FD_UNLIKELY( !view ) ) return -1;
657 0 : view->data = NULL;
658 0 : view->fec = NULL;
659 0 : view->flags = 0U;
660 0 : if( FD_UNLIKELY( !store || !fec ) ) return -1;
661 0 : fd_store_fec_t * fec0 = pool_ele_laddr( store );
662 0 : if( FD_UNLIKELY( fec<fec0 || fec>=fec0+store->fec_max ) ) return -1;
663 :
664 0 : fd_rwlock_write( &store->cache_lock );
665 :
666 0 : if( FD_UNLIKELY( fec->data_sz>store->fec_data_max ) ) {
667 0 : fd_rwlock_unwrite( &store->cache_lock );
668 0 : return -1;
669 0 : }
670 :
671 0 : if( FD_LIKELY( fec->data_state==FD_STORE_FEC_DATA_RAM_READY ||
672 0 : fec->data_state==FD_STORE_FEC_DATA_SPILLING ) ) {
673 0 : cache_pin_locked( store, fec );
674 0 : view->data = cache_data_laddr( store ) + (ulong)fec->data_idx*store->payload_slot_sz;
675 0 : view->fec = fec;
676 0 : fd_rwlock_unwrite( &store->cache_lock );
677 0 : return 0;
678 0 : }
679 :
680 0 : if( FD_LIKELY( fec->data_state==FD_STORE_FEC_DATA_DISK ) ) {
681 0 : if( FD_UNLIKELY( disk_fd<0 ) ) {
682 0 : fd_rwlock_unwrite( &store->cache_lock );
683 0 : return -1;
684 0 : }
685 0 : if( FD_UNLIKELY( !fd_rwlock_trywrite( &store->spill_read_lock ) ) ) {
686 0 : fd_rwlock_unwrite( &store->cache_lock );
687 0 : return -1;
688 0 : }
689 0 : ulong data_sz = fec->data_sz;
690 0 : ulong data_off = (ulong)fec->data_idx*store->payload_slot_sz;
691 0 : uchar * spill_read_data = spill_read_data_laddr( store );
692 0 : cache_pin_locked( store, fec );
693 0 : fd_rwlock_unwrite( &store->cache_lock );
694 :
695 0 : if( FD_UNLIKELY( store_pread_all( disk_fd, spill_read_data, data_sz, (off_t)data_off ) ) )
696 0 : FD_LOG_ERR(( "error reading spilled FEC payload: (%d-%s)", errno, fd_io_strerror( errno ) ));
697 :
698 0 : atomic_fetch_add_explicit( &store->fec_spill_read_cnt, 1UL, memory_order_relaxed );
699 0 : atomic_fetch_add_explicit( &store->fec_spill_read_bytes, data_sz, memory_order_relaxed );
700 :
701 0 : view->data = spill_read_data;
702 0 : view->fec = fec;
703 0 : view->flags = FD_STORE_FEC_DATA_VIEW_SPILL;
704 0 : return 0;
705 0 : }
706 :
707 0 : fd_rwlock_unwrite( &store->cache_lock );
708 0 : return -1;
709 0 : }
710 :
711 : void
712 : fd_store_fec_data_view_release( fd_store_t * store,
713 0 : fd_store_fec_data_view_t * view ) {
714 0 : if( FD_UNLIKELY( !store || !view || !view->fec ) ) return;
715 0 : if( view->flags & FD_STORE_FEC_DATA_VIEW_SPILL ) fd_rwlock_unwrite( &store->spill_read_lock );
716 :
717 0 : fd_rwlock_write( &store->cache_lock );
718 0 : cache_unpin_locked( store, view->fec );
719 0 : cache_consume_locked( store, view->fec );
720 0 : fd_rwlock_unwrite( &store->cache_lock );
721 :
722 0 : view->data = NULL;
723 0 : view->fec = NULL;
724 0 : view->flags = 0U;
725 0 : }
726 :
727 : fd_store_fec_t *
728 : fd_store_query( fd_store_map_t * map,
729 0 : fd_hash_t const * merkle_root ) {
730 0 : for(;;) {
731 0 : fd_store_map_query_t query[1];
732 0 : int err = fd_store_map_query_try( map, merkle_root, NULL, query, 0 );
733 0 : if( FD_UNLIKELY( err==FD_MAP_ERR_AGAIN ) ) continue;
734 0 : if( FD_UNLIKELY( err ) ) return NULL;
735 :
736 0 : fd_store_fec_t * fec = fd_store_map_query_ele( query );
737 0 : err = fd_store_map_query_test( query );
738 0 : if( FD_LIKELY( !err ) ) return fec;
739 0 : if( FD_UNLIKELY( err!=FD_MAP_ERR_AGAIN ) ) return NULL;
740 0 : }
741 0 : }
742 :
743 : int
744 : fd_store_insert( fd_store_t * store,
745 : fd_store_map_t * map,
746 : fd_hash_t const * merkle_root,
747 : ulong slot,
748 : ulong part,
749 : uint data_idx,
750 : uint data_sz,
751 : ushort const shred_sz[ FD_FEC_SHRED_CNT ],
752 0 : fd_store_fec_t ** fec ) {
753 0 : FD_TEST( store && map && merkle_root && shred_sz && fec );
754 0 : FD_TEST( data_idx<store->cache_slot_cnt && data_sz<=store->fec_data_max );
755 0 : *fec = NULL;
756 0 : fd_rwlock_read( &store->fec_lock );
757 :
758 0 : struct {
759 0 : fd_store_map_txn_t txn [1];
760 0 : fd_store_map_txn_private_info_t info[1];
761 0 : } map_txn_mem;
762 0 : fd_store_map_txn_t * txn = fd_store_map_txn_init( map_txn_mem.txn, map, 1UL );
763 0 : FD_TEST( !fd_store_map_txn_add( txn, merkle_root, 1 ) );
764 0 : FD_TEST( !fd_store_map_txn_try( txn, FD_MAP_FLAG_BLOCKING ) );
765 :
766 0 : fd_store_map_query_t query[1];
767 0 : int err = fd_store_map_txn_query( map, merkle_root, NULL, query, 0 );
768 0 : if( FD_LIKELY( err==FD_MAP_ERR_KEY ) ) {
769 0 : fd_store_fec_t * new_fec = fd_store_fec_acquire( store );
770 0 : FD_TEST( new_fec );
771 0 : new_fec->key = *merkle_root;
772 0 : new_fec->data_sz = data_sz;
773 0 : memcpy( new_fec->shred_sz, shred_sz, sizeof(new_fec->shred_sz) );
774 :
775 : /* RAM_READY before it is linked: fd_store_publish (exclusive
776 : fec_lock) never sees it mid-write. */
777 0 : fd_rwlock_write( &store->cache_lock );
778 0 : new_fec->data_idx = data_idx;
779 0 : new_fec->data_state = FD_STORE_FEC_DATA_RAM_READY;
780 0 : cache_lru_push_tail_locked( store, new_fec );
781 0 : fd_rwlock_unwrite( &store->cache_lock );
782 :
783 0 : FD_TEST( part<store->shred_tile_cnt );
784 0 : new_fec->slot = (uint)fd_ulong_max( slot, store->root );
785 0 : ulong lo = atomic_load_explicit( &store->slot_lo, memory_order_relaxed );
786 0 : while( new_fec->slot<lo && !atomic_compare_exchange_weak_explicit( &store->slot_lo, &lo, new_fec->slot, memory_order_relaxed, memory_order_relaxed ) );
787 0 : fd_store_slot_to_fecs_ele_insert( fd_store_slot_to_fecs_join( slot_to_fecs_laddr( store, part ) ), new_fec, pool_ele_laddr( store ) );
788 0 : FD_TEST( !fd_store_map_txn_insert( map, new_fec ) );
789 0 : *fec = new_fec;
790 0 : err = FD_MAP_SUCCESS;
791 0 : } else {
792 0 : FD_TEST( err==FD_MAP_SUCCESS );
793 0 : err = FD_MAP_ERR_KEY;
794 0 : }
795 :
796 0 : FD_TEST( !fd_store_map_txn_test( txn ) );
797 0 : fd_store_map_txn_fini( txn );
798 0 : fd_rwlock_unread( &store->fec_lock );
799 :
800 0 : if( FD_UNLIKELY( err==FD_MAP_ERR_KEY ) ) {
801 0 : fd_rwlock_write( &store->cache_lock );
802 0 : cache_slot_release_locked( store, data_idx );
803 0 : fd_rwlock_unwrite( &store->cache_lock );
804 0 : }
805 0 : return err;
806 0 : }
807 :
808 : ulong
809 : fd_store_publish( fd_store_t * store,
810 : fd_store_map_t * map,
811 0 : ulong root ) {
812 0 : fd_store_fec_t * fec0 = pool_ele_laddr( store );
813 0 : ulong cnt = 0UL;
814 0 : fd_rwlock_write( &store->fec_lock );
815 0 : ulong lo = atomic_load_explicit( &store->slot_lo, memory_order_relaxed );
816 0 : for( ulong part=0UL; part<store->shred_tile_cnt; part++ ) {
817 0 : fd_store_slot_to_fecs_t * slot_to_fecs = fd_store_slot_to_fecs_join( slot_to_fecs_laddr( store, part ) );
818 0 : for( ulong slot=lo; slot<root; slot++ ) {
819 0 : uint key = (uint)slot;
820 0 : for( ulong idx = fd_store_slot_to_fecs_idx_remove( slot_to_fecs, &key, ULONG_MAX, fec0 );
821 0 : idx!=ULONG_MAX;
822 0 : idx = fd_store_slot_to_fecs_idx_remove( slot_to_fecs, &key, ULONG_MAX, fec0 ) ) {
823 0 : fd_store_map_query_t query[1];
824 0 : FD_TEST( !fd_store_map_remove( map, &fec0[ idx ].key, NULL, query, FD_MAP_FLAG_BLOCKING ) );
825 0 : fd_store_fec_release( store, fd_store_map_query_ele( query ) );
826 0 : cnt++;
827 0 : }
828 0 : }
829 0 : }
830 0 : if( FD_LIKELY( lo<root ) ) atomic_store_explicit( &store->slot_lo, root, memory_order_relaxed );
831 0 : store->root = fd_ulong_max( store->root, root );
832 0 : fd_rwlock_unwrite( &store->fec_lock );
833 0 : return cnt;
834 0 : }
835 :
836 :
837 : /* Both chain maps, their keys and links, tags, and reservation decisions
838 : use disk_lock. Production has one rserve tile, the sole disk_* caller,
839 : so map access has little contention. Disk I/O runs without the lock;
840 : WRITING entries reserve keys and prevent reuse until the write finishes. */
841 0 : #define FD_STORE_DISK_READ_RETRY_CNT (8UL)
842 :
843 : static inline ulong
844 : disk_cell_tag( ulong ticket,
845 0 : uint state ) {
846 0 : return ((ticket & FD_SHREDB_TAG_SEQ_MASK)<<2) | state;
847 0 : }
848 :
849 : static inline ulong
850 0 : disk_cell_state( ulong tag ) {
851 0 : return tag & FD_SHREDB_CELL_STATE_MASK;
852 0 : }
853 :
854 : static inline ulong
855 0 : disk_cell_ticket( ulong tag ) {
856 0 : return (tag>>2) & FD_SHREDB_TAG_SEQ_MASK;
857 0 : }
858 :
859 : /* The tag is split across seven bytes in the ring, but is a ulong on
860 : disk. All ring accesses to these fields hold disk_lock. */
861 : static inline ulong
862 0 : disk_cell_tag_load( fd_shredb_ring_entry_t const * cell ) {
863 0 : return (ulong)cell->tag
864 0 : | ((ulong)cell->tag_hi[0]<<32)
865 0 : | ((ulong)cell->tag_hi[1]<<40)
866 0 : | ((ulong)cell->tag_hi[2]<<48);
867 0 : }
868 :
869 : static inline void
870 : disk_cell_tag_store( fd_shredb_ring_entry_t * cell,
871 0 : ulong tag ) {
872 0 : cell->tag = (uint)tag;
873 0 : cell->tag_hi[0] = (uchar)(tag>>32);
874 0 : cell->tag_hi[1] = (uchar)(tag>>40);
875 0 : cell->tag_hi[2] = (uchar)(tag>>48);
876 0 : }
877 :
878 : static void
879 : disk_slot_hint_publish_locked( fd_store_t * store,
880 : ulong slot,
881 0 : uint shred_idx ) {
882 0 : atomic_ulong * hint = disk_slot_hint_laddr( store ) + (slot % store->disk_max_slots);
883 0 : ulong current = atomic_load_explicit( hint, memory_order_relaxed );
884 0 : if( FD_LIKELY( current & FD_SHREDB_HINT_VALID ) ) {
885 0 : uint current_idx = fd_shredb_key_shred_idx( current & ~FD_SHREDB_HINT_VALID );
886 0 : if( FD_LIKELY( fd_shredb_key_slot( current )==slot ) && current_idx>=shred_idx ) return;
887 0 : }
888 0 : ulong desired = fd_shredb_key_pack( slot, shred_idx ) | FD_SHREDB_HINT_VALID;
889 0 : atomic_store_explicit( hint, desired, memory_order_release );
890 0 : }
891 :
892 : /* The slot and root maps are fd_map_chain instances over the shared ring
893 : pool; all helpers require disk_lock. Lookups use the _const query so
894 : concurrent readers (shared disk_lock) are safe; link/unlink mutate and
895 : run only under the exclusive write lock. UINT_MAX is the miss sentinel
896 : (no ring index can equal it since disk_max_shreds < UINT_MAX). The root
897 : comparison uses the complete 21-byte key. */
898 : static inline uint
899 : disk_slot_lookup_locked( fd_store_t const * store,
900 0 : ulong key ) {
901 0 : fd_shredb_slot_map_t * map = disk_slot_map_join( store );
902 0 : return (uint)fd_shredb_slot_map_idx_query_const( map, &key, UINT_MAX, disk_ring_pool_laddr( store ) );
903 0 : }
904 :
905 : static inline uint
906 : disk_root_lookup_locked( fd_store_t const * store,
907 0 : fd_shredb_root_key_t const * key ) {
908 0 : fd_shredb_root_map_t * map = disk_root_map_join( store );
909 0 : return (uint)fd_shredb_root_map_idx_query_const( map, key, UINT_MAX, disk_ring_pool_laddr( store ) );
910 0 : }
911 :
912 : static void
913 : disk_invalidate_locked( fd_store_t * store,
914 0 : uint ring_idx ) {
915 0 : fd_shredb_ring_entry_t * pool = disk_ring_pool_laddr( store );
916 0 : fd_shredb_ring_entry_t * cell = pool + ring_idx;
917 0 : FD_TEST( disk_cell_state( cell->tag )!=FD_SHREDB_CELL_WRITING );
918 :
919 0 : fd_shredb_slot_map_t * slot_map = disk_slot_map_join( store );
920 0 : if( (uint)fd_shredb_slot_map_idx_query( slot_map, &cell->key, UINT_MAX, pool )==ring_idx ) {
921 0 : fd_shredb_slot_map_idx_remove( slot_map, &cell->key, UINT_MAX, pool );
922 0 : }
923 :
924 0 : fd_shredb_root_map_t * root_map = disk_root_map_join( store );
925 0 : if( (uint)fd_shredb_root_map_idx_query( root_map, &cell->root_key, UINT_MAX, pool )==ring_idx ) {
926 0 : fd_shredb_root_map_idx_remove( root_map, &cell->root_key, UINT_MAX, pool );
927 0 : }
928 :
929 0 : if( disk_cell_state( cell->tag )==FD_SHREDB_CELL_READY ) {
930 0 : ulong disk_cnt = atomic_load_explicit( &store->disk_cnt, memory_order_relaxed );
931 0 : FD_TEST( disk_cnt!=0UL );
932 0 : atomic_store_explicit( &store->disk_cnt, disk_cnt-1UL, memory_order_relaxed );
933 0 : }
934 0 : cell->tag = (cell->tag & ~(uint)FD_SHREDB_CELL_STATE_MASK) | (uint)FD_SHREDB_CELL_INVALID;
935 0 : }
936 :
937 : /* A snapshot of a cell's tag kept after releasing the rwlock.
938 : This view is only held on the caller's stack across disk I/O. */
939 : struct disk_view {
940 : ulong ticket;
941 : uint ring_idx;
942 : int linked;
943 : };
944 : typedef struct disk_view disk_view_t;
945 :
946 : static int
947 : disk_insert_prepare_write_locked( fd_store_t * store,
948 : ulong key,
949 : fd_shredb_root_key_t const * root_key,
950 0 : disk_view_t * view ) {
951 0 : fd_shredb_ring_entry_t * pool = disk_ring_pool_laddr( store );
952 :
953 : /* A READY root-map key wins even when the new shred has a different
954 : full index. */
955 0 : uint exist_idx = disk_root_lookup_locked( store, root_key );
956 0 : if( exist_idx!=UINT_MAX ) {
957 0 : ulong state = disk_cell_state( pool[exist_idx].tag );
958 0 : if( FD_UNLIKELY( state==FD_SHREDB_CELL_WRITING ) ) return FD_STORE_DISK_INSERT_ERR;
959 0 : FD_TEST( state==FD_SHREDB_CELL_READY );
960 0 : return FD_STORE_DISK_INSERT_DUP;
961 0 : }
962 :
963 : /* Grab a ticket & ring_idx to write to.
964 : May wrap around and overwrite an existing cell that's not mid-write. */
965 0 : ulong ticket = atomic_load_explicit( &store->disk_reservation_head, memory_order_relaxed );
966 0 : uint ring_idx = (uint)(ticket % store->disk_max_shreds);
967 0 : atomic_store_explicit( &store->disk_reservation_head, ticket + 1UL, memory_order_relaxed );
968 0 : if( FD_UNLIKELY( disk_cell_state( pool[ring_idx].tag )==FD_SHREDB_CELL_WRITING ) ) return FD_STORE_DISK_INSERT_ERR;
969 :
970 : /* Retire the ring cell's old keys before changing either key field. */
971 0 : disk_invalidate_locked( store, ring_idx );
972 :
973 0 : fd_shredb_ring_entry_t * cell = pool + ring_idx;
974 0 : disk_cell_tag_store( cell, disk_cell_tag( ticket, FD_SHREDB_CELL_WRITING ) );
975 0 : cell->key = key;
976 0 : int linked = disk_slot_lookup_locked( store, key )==UINT_MAX;
977 0 : if( linked ) fd_shredb_slot_map_idx_insert( disk_slot_map_join( store ), ring_idx, pool );
978 0 : cell->root_key = *root_key;
979 0 : fd_shredb_root_map_idx_insert( disk_root_map_join( store ), ring_idx, pool );
980 :
981 0 : view->ticket = ticket;
982 0 : view->ring_idx = ring_idx;
983 0 : view->linked = linked;
984 0 : return FD_STORE_DISK_INSERT_SUCCESS;
985 0 : }
986 :
987 : static int
988 : disk_insert_finish_write_locked( fd_store_t * store,
989 0 : disk_view_t view ) {
990 0 : fd_shredb_ring_entry_t * cell = disk_ring_pool_laddr( store ) + view.ring_idx;
991 0 : FD_TEST( disk_cell_tag_load( cell )==disk_cell_tag( view.ticket, FD_SHREDB_CELL_WRITING ) );
992 :
993 0 : ulong disk_cnt = atomic_load_explicit( &store->disk_cnt, memory_order_relaxed );
994 0 : atomic_store_explicit( &store->disk_cnt, disk_cnt+1UL, memory_order_relaxed );
995 0 : disk_cell_tag_store( cell, disk_cell_tag( view.ticket, FD_SHREDB_CELL_READY ) );
996 :
997 0 : if( view.linked ) disk_slot_hint_publish_locked( store, fd_shredb_key_slot( cell->key ), fd_shredb_key_shred_idx( cell->key ) );
998 0 : return FD_STORE_DISK_INSERT_SUCCESS;
999 0 : }
1000 :
1001 : FD_STATIC_ASSERT( __builtin_offsetof( fd_shredb_entry_t, shred )==__builtin_offsetof( fd_shredb_entry_t, shred_sz )+sizeof(ushort), shredb_entry_hdr_packed );
1002 :
1003 : int
1004 : fd_store_disk_insert( fd_store_t * store,
1005 : int disk_fd,
1006 : fd_shred_t const * shred,
1007 0 : fd_hash_t const * merkle_root ) {
1008 0 : if( FD_UNLIKELY( !store || disk_fd<0 || !shred || !merkle_root || !fd_store_has_disk( store ) ) )
1009 0 : return FD_STORE_DISK_INSERT_ERR;
1010 :
1011 0 : ulong slot = shred->slot;
1012 0 : uint shred_idx = shred->idx;
1013 0 : if( FD_UNLIKELY( slot>=FD_SHREDB_KEY_SLOT_MAX || shred_idx>=store->max_shreds_per_block ) ) return FD_STORE_DISK_INSERT_ERR;
1014 0 : ulong key = fd_shredb_key_pack( slot, shred_idx );
1015 0 : fd_shredb_root_key_t root_key = fd_shredb_root_key( merkle_root->uc, shred_idx );
1016 :
1017 0 : disk_view_t view;
1018 0 : fd_rwlock_write( &store->disk_lock );
1019 0 : int result = disk_insert_prepare_write_locked( store, key, &root_key, &view );
1020 0 : fd_rwlock_unwrite( &store->disk_lock );
1021 0 : if( FD_UNLIKELY( result!=FD_STORE_DISK_INSERT_SUCCESS ) ) return result;
1022 :
1023 : /* The map entries remain WRITING while the syscall runs unlocked. */
1024 0 : fd_shredb_entry_t wr_entry[1];
1025 0 : ulong shred_sz = fd_ulong_min( fd_shred_sz( shred ), FD_SHRED_MAX_SZ );
1026 0 : ulong tail_off = __builtin_offsetof( fd_shredb_entry_t, shred )+shred_sz;
1027 0 : wr_entry->tag = disk_cell_tag( view.ticket, FD_SHREDB_CELL_READY );
1028 0 : wr_entry->key = key;
1029 0 : wr_entry->shred_sz = (ushort)shred_sz;
1030 0 : memcpy( wr_entry->root, merkle_root->uc, FD_SHRED_MERKLE_NODE_SZ );
1031 0 : fd_memcpy( wr_entry->shred, shred, shred_sz );
1032 0 : fd_memset( (uchar *)wr_entry+tail_off, 0, sizeof(wr_entry)-tail_off );
1033 :
1034 0 : off_t off = (off_t)(store->wire_off + (ulong)view.ring_idx*sizeof(fd_shredb_entry_t));
1035 0 : if( FD_UNLIKELY( store_pwrite_all( disk_fd, wr_entry, sizeof(wr_entry), off ) ) )
1036 0 : FD_LOG_ERR(( "error writing to disk store: (%d-%s)", errno, fd_io_strerror( errno ) ));
1037 0 : atomic_fetch_add_explicit( &store->disk_write_bytes, sizeof(wr_entry), memory_order_relaxed );
1038 :
1039 0 : fd_rwlock_write( &store->disk_lock );
1040 0 : result = disk_insert_finish_write_locked( store, view );
1041 0 : fd_rwlock_unwrite( &store->disk_lock );
1042 0 : if( FD_UNLIKELY( result!=FD_STORE_DISK_INSERT_SUCCESS ) ) return result;
1043 :
1044 0 : atomic_fetch_add_explicit( &store->disk_insert_cnt, 1UL, memory_order_relaxed );
1045 0 : return FD_STORE_DISK_INSERT_SUCCESS;
1046 0 : }
1047 :
1048 : static int
1049 : disk_query_prepare_read_locked( fd_store_t const * store,
1050 : ulong key,
1051 : fd_shredb_root_key_t const * root_key,
1052 : uint shred_idx,
1053 0 : disk_view_t * view ) {
1054 0 : uint ring_idx = root_key ? disk_root_lookup_locked( store, root_key ) : disk_slot_lookup_locked( store, key );
1055 0 : if( FD_UNLIKELY( ring_idx==UINT_MAX ) ) return FD_STORE_DISK_QUERY_MISS;
1056 :
1057 0 : fd_shredb_ring_entry_t const * cell = disk_ring_pool_laddr( store ) + ring_idx;
1058 0 : ulong tag = disk_cell_tag_load( cell );
1059 0 : if( FD_UNLIKELY( disk_cell_state( tag )!=FD_SHREDB_CELL_READY ) ) {
1060 0 : if( disk_cell_state( tag )==FD_SHREDB_CELL_WRITING ) return FD_STORE_DISK_QUERY_BUSY;
1061 0 : return FD_STORE_DISK_QUERY_MISS;
1062 0 : }
1063 0 : if( FD_UNLIKELY( root_key && fd_shredb_key_shred_idx( cell->key )!=shred_idx ) ) {
1064 0 : return FD_STORE_DISK_QUERY_MISS;
1065 0 : }
1066 :
1067 0 : view->ticket = disk_cell_ticket( tag );
1068 0 : view->ring_idx = ring_idx;
1069 0 : view->linked = 0;
1070 0 : return 0;
1071 0 : }
1072 :
1073 : static int
1074 : disk_query_finish_read_locked( fd_store_t const * store,
1075 0 : disk_view_t view ) {
1076 0 : fd_shredb_ring_entry_t const * cell = disk_ring_pool_laddr( store ) + view.ring_idx;
1077 0 : if( FD_UNLIKELY( disk_cell_tag_load( cell )!=disk_cell_tag( view.ticket, FD_SHREDB_CELL_READY ) ) ) return FD_STORE_DISK_QUERY_BUSY;
1078 0 : return 0;
1079 0 : }
1080 :
1081 : /* disk_read_once snapshots and validates under disk_lock, releasing it
1082 : only for pread. The tag and mapping checks reject a torn read if the
1083 : cell was overwritten during I/O. */
1084 : static int
1085 : disk_read_once( fd_store_t const * store,
1086 : int disk_fd,
1087 : ulong key,
1088 : fd_shredb_root_key_t const * root_key,
1089 : uint shred_idx,
1090 0 : fd_shredb_entry_t * rd_entry ) {
1091 : /* need mutable rwlock to read, even under a const ptr */
1092 0 : fd_rwlock_t * rwlock = (fd_rwlock_t *)fd_type_pun_const( &store->disk_lock );
1093 :
1094 0 : disk_view_t view;
1095 0 : fd_rwlock_read( rwlock );
1096 0 : int result = disk_query_prepare_read_locked( store, key, root_key, shred_idx, &view );
1097 0 : fd_rwlock_unread( rwlock );
1098 0 : if( FD_UNLIKELY( result ) ) return result;
1099 :
1100 0 : off_t off = (off_t)(store->wire_off + (ulong)view.ring_idx*sizeof(fd_shredb_entry_t));
1101 0 : if( FD_UNLIKELY( store_pread_all( disk_fd, rd_entry, sizeof(*rd_entry), off ) ) ) {
1102 0 : FD_LOG_ERR(( "error reading disk store: (%d-%s)", errno, fd_io_strerror( errno ) ));
1103 0 : return FD_STORE_DISK_QUERY_BUSY;
1104 0 : }
1105 :
1106 0 : fd_rwlock_read( rwlock );
1107 0 : result = disk_query_finish_read_locked( store, view );
1108 0 : fd_rwlock_unread( rwlock );
1109 0 : if( FD_UNLIKELY( result ) ) return result;
1110 :
1111 0 : ulong tag = disk_cell_tag( view.ticket, FD_SHREDB_CELL_READY );
1112 0 : if( root_key ) {
1113 0 : if( FD_UNLIKELY( rd_entry->tag!=tag ||
1114 0 : memcmp( rd_entry->root, root_key->root, FD_SHRED_MERKLE_NODE_SZ ) ||
1115 0 : fd_shredb_key_shred_idx( rd_entry->key )%FD_FEC_SHRED_CNT!=root_key->off ) ) return FD_STORE_DISK_QUERY_BUSY;
1116 0 : if( FD_UNLIKELY( fd_shredb_key_shred_idx( rd_entry->key )!=shred_idx ) ) return FD_STORE_DISK_QUERY_MISS;
1117 0 : } else {
1118 0 : if( FD_UNLIKELY( rd_entry->tag!=tag || rd_entry->key!=key ) ) return FD_STORE_DISK_QUERY_BUSY;
1119 0 : }
1120 0 : return (int)fd_ulong_min( rd_entry->shred_sz, FD_SHRED_MAX_SZ );
1121 0 : }
1122 :
1123 : int
1124 : fd_store_disk_query( fd_store_t const * store,
1125 : int disk_fd,
1126 : ulong slot,
1127 : uint shred_idx,
1128 0 : uchar out[ FD_SHRED_MAX_SZ ] ) {
1129 0 : if( FD_UNLIKELY( !store || disk_fd<0 || !out || !fd_store_has_disk( store ) ||
1130 0 : slot>=FD_SHREDB_KEY_SLOT_MAX || shred_idx>=store->max_shreds_per_block ) ) return FD_STORE_DISK_QUERY_MISS;
1131 0 : ulong key = fd_shredb_key_pack( slot, shred_idx );
1132 0 : for( ulong retry=0UL; retry<FD_STORE_DISK_READ_RETRY_CNT; retry++ ) {
1133 0 : fd_shredb_entry_t entry[1];
1134 0 : int result = disk_read_once( store, disk_fd, key, NULL, shred_idx, entry );
1135 0 : if( FD_UNLIKELY( result==FD_STORE_DISK_QUERY_BUSY ) ) { FD_SPIN_PAUSE(); continue; }
1136 0 : if( FD_UNLIKELY( result<0 ) ) return result;
1137 0 : fd_memcpy( out, entry->shred, (ulong)result );
1138 0 : return result;
1139 0 : }
1140 0 : return FD_STORE_DISK_QUERY_BUSY;
1141 0 : }
1142 :
1143 : int
1144 : fd_store_disk_query_root( fd_store_t const * store,
1145 : int disk_fd,
1146 : uchar const * merkle_root,
1147 : uint shred_idx,
1148 0 : uchar out[ FD_SHRED_MAX_SZ ] ) {
1149 0 : if( FD_UNLIKELY( !store || disk_fd<0 || !merkle_root || !out || !fd_store_has_disk( store ) ||
1150 0 : shred_idx>=store->max_shreds_per_block ) ) return FD_STORE_DISK_QUERY_MISS;
1151 0 : fd_shredb_root_key_t rk = fd_shredb_root_key( merkle_root, shred_idx );
1152 0 : for( ulong retry=0UL; retry<FD_STORE_DISK_READ_RETRY_CNT; retry++ ) {
1153 0 : fd_shredb_entry_t entry[1];
1154 0 : int result = disk_read_once( store, disk_fd, 0UL, &rk, shred_idx, entry );
1155 0 : if( FD_UNLIKELY( result==FD_STORE_DISK_QUERY_BUSY ) ) { FD_SPIN_PAUSE(); continue; }
1156 0 : if( FD_UNLIKELY( result<0 ) ) return result;
1157 0 : fd_memcpy( out, entry->shred, (ulong)result );
1158 0 : return result;
1159 0 : }
1160 0 : return FD_STORE_DISK_QUERY_BUSY;
1161 0 : }
1162 :
1163 : int
1164 : fd_store_disk_probe( fd_store_t const * store,
1165 : ulong slot,
1166 0 : uint shred_idx ) {
1167 0 : if( FD_UNLIKELY( !store || !fd_store_has_disk( store ) || slot>=FD_SHREDB_KEY_SLOT_MAX ) ) return 0;
1168 0 : if( FD_UNLIKELY( shred_idx==UINT_MAX ) ) {
1169 0 : return !!(atomic_load_explicit( disk_slot_hint_laddr( store ) + (slot % store->disk_max_slots), memory_order_acquire ) & FD_SHREDB_HINT_VALID);
1170 0 : }
1171 0 : if( FD_UNLIKELY( shred_idx>=store->max_shreds_per_block ) ) return 0;
1172 :
1173 0 : fd_rwlock_t * rwlock = (fd_rwlock_t *)fd_type_pun_const( &store->disk_lock );
1174 0 : fd_rwlock_read( rwlock );
1175 0 : int present = disk_slot_lookup_locked( store, fd_shredb_key_pack( slot, shred_idx ) )!=UINT_MAX;
1176 0 : fd_rwlock_unread( rwlock );
1177 0 : return present;
1178 0 : }
1179 :
1180 : int
1181 : fd_store_disk_query_highest( fd_store_t const * store,
1182 : int disk_fd,
1183 : ulong slot,
1184 : uint min_shred_idx,
1185 0 : uchar out[ FD_SHRED_MAX_SZ ] ) {
1186 0 : if( FD_UNLIKELY( !store || disk_fd<0 || !out || !fd_store_has_disk( store ) || slot>=FD_SHREDB_KEY_SLOT_MAX ) )
1187 0 : return FD_STORE_DISK_QUERY_MISS;
1188 0 : for( ulong retry=0UL; retry<FD_STORE_DISK_READ_RETRY_CNT; retry++ ) {
1189 0 : atomic_ulong const * hint_ptr = disk_slot_hint_laddr( store ) + (slot % store->disk_max_slots);
1190 0 : ulong hint = atomic_load_explicit( hint_ptr, memory_order_relaxed );
1191 0 : if( FD_UNLIKELY( !(hint & FD_SHREDB_HINT_VALID) ) ) return FD_STORE_DISK_QUERY_MISS;
1192 0 : if( FD_UNLIKELY( fd_shredb_key_slot( hint )!=slot ) ) return FD_STORE_DISK_QUERY_BUSY;
1193 0 : uint idx = fd_shredb_key_shred_idx( hint & ~FD_SHREDB_HINT_VALID );
1194 0 : if( FD_UNLIKELY( idx>=store->max_shreds_per_block ) ) return FD_STORE_DISK_QUERY_MISS;
1195 :
1196 0 : fd_shredb_entry_t entry[1];
1197 0 : int result = disk_read_once( store, disk_fd, fd_shredb_key_pack( slot, idx ), NULL, idx, entry );
1198 0 : ulong hint_after = atomic_load_explicit( hint_ptr, memory_order_acquire );
1199 0 : if( FD_UNLIKELY( hint_after!=hint || result==FD_STORE_DISK_QUERY_BUSY ) ) {
1200 0 : FD_SPIN_PAUSE();
1201 0 : continue;
1202 0 : }
1203 0 : if( FD_UNLIKELY( result==FD_STORE_DISK_QUERY_MISS ) ) return FD_STORE_DISK_QUERY_SCAN_LIMIT;
1204 0 : if( FD_UNLIKELY( result<0 ) ) return result;
1205 :
1206 0 : fd_shred_t const * shred = (fd_shred_t const *)fd_type_pun_const( entry->shred );
1207 0 : if( idx<min_shred_idx && !(shred->data.flags & FD_SHRED_DATA_FLAG_SLOT_COMPLETE) ) return FD_STORE_DISK_QUERY_MISS;
1208 0 : fd_memcpy( out, entry->shred, (ulong)result );
1209 0 : return result;
1210 0 : }
1211 0 : return FD_STORE_DISK_QUERY_BUSY;
1212 0 : }
1213 :
1214 : int
1215 : fd_store_disk_stats_query( fd_store_t const * store,
1216 0 : fd_store_disk_stats_t * stats ) {
1217 0 : if( FD_UNLIKELY( !store || !stats || !fd_store_has_disk( store ) ) ) return FD_STORE_DISK_QUERY_MISS;
1218 0 : stats->shred_cnt = atomic_load_explicit( &store->disk_cnt, memory_order_relaxed );
1219 0 : stats->current_bytes = stats->shred_cnt*sizeof(fd_shredb_entry_t)
1220 0 : + atomic_load_explicit( &store->spill_live_cnt, memory_order_relaxed )*store->payload_slot_sz;
1221 0 : stats->allocated_bytes = fd_ulong_min( atomic_load_explicit( &store->disk_reservation_head, memory_order_relaxed ),
1222 0 : store->disk_max_shreds )*sizeof(fd_shredb_entry_t)
1223 0 : + FD_VOLATILE_CONST( store->spill_slot_cnt )*store->payload_slot_sz;
1224 0 : stats->insert_cnt = atomic_load_explicit( &store->disk_insert_cnt, memory_order_relaxed );
1225 0 : stats->write_bytes = atomic_load_explicit( &store->disk_write_bytes, memory_order_relaxed );
1226 0 : return 0;
1227 0 : }
|
