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 48 : #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 1893 : off_t off ) {
22 1893 : ulong written = 0UL;
23 3786 : while( written<sz ) {
24 1893 : long res = pwrite( fd, (uchar const *)buf + written, sz-written, off+(off_t)written );
25 1893 : 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 1893 : return 0;
31 1893 : }
32 :
33 : static int
34 : store_pread_all( int fd,
35 : void * buf,
36 : ulong sz,
37 351 : off_t off ) {
38 351 : ulong read_sz = 0UL;
39 702 : while( read_sz<sz ) {
40 351 : long res = pread( fd, (uchar *)buf + read_sz, sz-read_sz, off+(off_t)read_sz );
41 351 : 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 351 : return 0;
47 351 : }
48 :
49 : static inline fd_shredb_ring_entry_t *
50 7356 : disk_ring_pool_laddr( fd_store_t const * store ) {
51 7356 : return fd_wksp_laddr_fast( fd_store_wksp( store ), store->ring_pool_gaddr );
52 7356 : }
53 :
54 : static inline fd_shredb_slot_map_t *
55 4050 : disk_slot_map_join( fd_store_t const * store ) {
56 4050 : void * shmap = fd_wksp_laddr_fast( fd_store_wksp( store ), store->shred_map_gaddr );
57 4050 : return fd_shredb_slot_map_join( shmap );
58 4050 : }
59 :
60 : static inline fd_shredb_root_map_t *
61 3795 : disk_root_map_join( fd_store_t const * store ) {
62 3795 : void * shmap = fd_wksp_laddr_fast( fd_store_wksp( store ), store->root_map_gaddr );
63 3795 : return fd_shredb_root_map_join( shmap );
64 3795 : }
65 :
66 : static inline atomic_ulong *
67 1353 : disk_slot_hint_laddr( fd_store_t const * store ) {
68 1353 : return fd_wksp_laddr_fast( fd_store_wksp( store ), store->slot_hint_gaddr );
69 1353 : }
70 :
71 : static inline uchar *
72 2250 : cache_data_laddr( fd_store_t const * store ) {
73 2250 : return fd_wksp_laddr_fast( fd_store_wksp( store ), store->cache_data_gaddr );
74 2250 : }
75 :
76 : static inline ulong *
77 2307 : cache_free_laddr( fd_store_t const * store ) {
78 2307 : return fd_wksp_laddr_fast( fd_store_wksp( store ), store->cache_free_gaddr );
79 2307 : }
80 :
81 : static inline uint *
82 1242 : spill_free_laddr( fd_store_t const * store ) {
83 1242 : return fd_wksp_laddr_fast( fd_store_wksp( store ), store->spill_free_gaddr );
84 1242 : }
85 :
86 : static inline uchar *
87 12 : spill_read_data_laddr( fd_store_t const * store ) {
88 12 : return fd_wksp_laddr_fast( fd_store_wksp( store ), store->spill_read_data_gaddr );
89 12 : }
90 :
91 : static inline fd_store_fec_t *
92 6894 : pool_ele_laddr( fd_store_t const * store ) {
93 6894 : return fd_wksp_laddr_fast( fd_store_wksp( store ), store->pool_ele_gaddr );
94 6894 : }
95 :
96 : static inline fd_store_slot_to_fecs_t *
97 : slot_to_fecs_laddr( fd_store_t const * store,
98 2229 : ulong part ) {
99 2229 : return fd_wksp_laddr_fast( fd_store_wksp( store ), store->slot_to_fecs_gaddr + part*store->slot_to_fecs_sz );
100 2229 : }
101 :
102 : static inline fd_store_pool_t
103 2219 : pool_ljoin( fd_store_t const * store ) {
104 2219 : return (fd_store_pool_t){
105 2219 : .pool = fd_wksp_laddr_fast( fd_store_wksp( store ), store->pool_mem_gaddr ),
106 2219 : .ele = fd_wksp_laddr_fast( fd_store_wksp( store ), store->pool_ele_gaddr ),
107 2219 : .ele_max = store->fec_max
108 2219 : };
109 2219 : }
110 :
111 : static fd_store_fec_t *
112 1491 : fd_store_fec_acquire( fd_store_t * store ) {
113 1491 : fd_store_pool_t pool = pool_ljoin( store );
114 1491 : fd_store_fec_t * fec = fd_store_pool_acquire( &pool );
115 1518 : if( FD_LIKELY( fec ) ) {
116 1518 : fec->data_sz = 0UL;
117 1518 : fec->data_idx = 0U;
118 1518 : fec->cache_prev = UINT_MAX;
119 1518 : fec->cache_next = UINT_MAX;
120 1518 : fec->data_pin_cnt = 0U;
121 1518 : fec->data_state = FD_STORE_FEC_DATA_EMPTY;
122 1518 : fec->data_consume_pending = 0U;
123 1518 : }
124 1491 : return fec;
125 1491 : }
126 :
127 : static void
128 : cache_slot_release_locked( fd_store_t * store,
129 765 : ulong cache_slot ) {
130 765 : FD_TEST( cache_slot<store->cache_slot_cnt );
131 765 : FD_TEST( store->cache_free_cnt<store->cache_slot_cnt );
132 765 : cache_free_laddr( store )[ store->cache_free_cnt++ ] = cache_slot;
133 765 : }
134 :
135 : static inline uint
136 : cache_fec_idx( fd_store_t const * store,
137 3039 : fd_store_fec_t const * fec ) {
138 3039 : ulong off = (ulong)fec - (ulong)pool_ele_laddr( store );
139 3039 : FD_TEST( !(off % sizeof(fd_store_fec_t)) );
140 3039 : ulong idx = off / sizeof(fd_store_fec_t);
141 3039 : FD_TEST( idx<store->fec_max );
142 3039 : return (uint)idx;
143 3039 : }
144 :
145 : static void
146 : cache_lru_remove_locked( fd_store_t * store,
147 768 : fd_store_fec_t * fec ) {
148 768 : uint idx = cache_fec_idx( store, fec );
149 768 : uint prev = fec->cache_prev;
150 768 : uint next = fec->cache_next;
151 :
152 768 : if( prev==UINT_MAX ) { FD_TEST( store->cache_lru_head==idx ); store->cache_lru_head = next; }
153 48 : else pool_ele_laddr( store )[ prev ].cache_next = next;
154 768 : if( next==UINT_MAX ) { FD_TEST( store->cache_lru_tail==idx ); store->cache_lru_tail = prev; }
155 72 : else pool_ele_laddr( store )[ next ].cache_prev = prev;
156 :
157 768 : fec->cache_prev = UINT_MAX;
158 768 : fec->cache_next = UINT_MAX;
159 768 : }
160 :
161 : static void
162 : cache_lru_push_head_locked( fd_store_t * store,
163 24 : fd_store_fec_t * fec ) {
164 24 : uint idx = cache_fec_idx( store, fec );
165 24 : FD_TEST( fec->cache_prev==UINT_MAX && fec->cache_next==UINT_MAX );
166 :
167 24 : fec->cache_next = store->cache_lru_head;
168 24 : if( store->cache_lru_head==UINT_MAX ) store->cache_lru_tail = idx;
169 18 : else pool_ele_laddr( store )[ store->cache_lru_head ].cache_prev = idx;
170 24 : store->cache_lru_head = idx;
171 24 : }
172 :
173 : static void
174 : cache_lru_push_tail_locked( fd_store_t * store,
175 1521 : fd_store_fec_t * fec ) {
176 1521 : uint idx = cache_fec_idx( store, fec );
177 1521 : FD_TEST( fec->cache_prev==UINT_MAX && fec->cache_next==UINT_MAX );
178 :
179 1521 : fec->cache_prev = store->cache_lru_tail;
180 1521 : if( store->cache_lru_tail==UINT_MAX ) store->cache_lru_head = idx;
181 813 : else pool_ele_laddr( store )[ store->cache_lru_tail ].cache_next = idx;
182 1521 : store->cache_lru_tail = idx;
183 1521 : }
184 :
185 : static void
186 : spill_free_push_locked( fd_store_t * store,
187 636 : uint spill_slot ) {
188 636 : FD_TEST( spill_slot<store->spill_slot_cnt && store->spill_free_cnt<store->spill_slot_cnt );
189 636 : spill_free_laddr( store )[ store->spill_free_cnt++ ] = spill_slot;
190 636 : }
191 :
192 : static void
193 : cache_consume_locked( fd_store_t * store,
194 756 : fd_store_fec_t * fec ) {
195 756 : if( FD_LIKELY( !fec->data_consume_pending ) ) return;
196 720 : if( FD_UNLIKELY( fec->data_pin_cnt || fec->data_state==FD_STORE_FEC_DATA_SPILLING ) ) return;
197 :
198 720 : if( FD_LIKELY( fec->data_state==FD_STORE_FEC_DATA_RAM_WRITING ||
199 720 : fec->data_state==FD_STORE_FEC_DATA_RAM_READY ) ) {
200 84 : if( fec->data_state==FD_STORE_FEC_DATA_RAM_READY ) cache_lru_remove_locked( store, fec );
201 84 : cache_slot_release_locked( store, fec->data_idx );
202 636 : } else if( fec->data_state==FD_STORE_FEC_DATA_DISK ) {
203 636 : ulong spill_slot = fec->data_idx;
204 636 : FD_TEST( spill_slot<store->spill_slot_cnt && atomic_load_explicit( &store->spill_live_cnt, memory_order_relaxed ) );
205 636 : atomic_fetch_sub_explicit( &store->spill_live_cnt, 1UL, memory_order_relaxed );
206 636 : spill_free_push_locked( store, (uint)spill_slot );
207 636 : }
208 :
209 720 : fec->data_idx = 0U;
210 720 : fec->cache_prev = UINT_MAX;
211 720 : fec->cache_next = UINT_MAX;
212 720 : fec->data_state = FD_STORE_FEC_DATA_CONSUMED;
213 720 : fec->data_consume_pending = 0U;
214 720 : }
215 :
216 : static void
217 : fd_store_fec_release( fd_store_t * store,
218 720 : fd_store_fec_t * fec ) {
219 720 : (void)cache_fec_idx( store, fec );
220 720 : for(;;) {
221 720 : fd_rwlock_write( &store->cache_lock );
222 720 : fec->data_consume_pending = 1U;
223 720 : cache_consume_locked( store, fec );
224 720 : int consumed = fec->data_state==FD_STORE_FEC_DATA_CONSUMED;
225 720 : fd_rwlock_unwrite( &store->cache_lock );
226 720 : if( FD_LIKELY( consumed ) ) break;
227 0 : FD_SPIN_PAUSE();
228 0 : }
229 720 : fd_store_pool_t pool = pool_ljoin( store );
230 720 : fd_store_pool_release( &pool, fec );
231 720 : }
232 :
233 : static int
234 : spill_one_locked( fd_store_t * store,
235 : int disk_fd,
236 666 : fd_store_fec_spill_stats_t * spill ) {
237 666 : if( FD_UNLIKELY( disk_fd<0 ) ) return FD_STORE_SPILL_NONE;
238 :
239 660 : fd_store_fec_t * fec0 = pool_ele_laddr( store );
240 660 : uint victim_idx = store->cache_lru_head;
241 660 : if( FD_UNLIKELY( victim_idx==UINT_MAX ) ) return FD_STORE_SPILL_NONE;
242 660 : fd_store_fec_t * victim = fec0 + victim_idx;
243 660 : FD_TEST( victim->data_state==FD_STORE_FEC_DATA_RAM_READY && !victim->data_pin_cnt );
244 660 : FD_TEST( victim->data_sz<=store->fec_data_max );
245 :
246 660 : uint spill_slot;
247 660 : if( FD_LIKELY( store->spill_free_cnt ) ) {
248 606 : spill_slot = spill_free_laddr( store )[ --store->spill_free_cnt ];
249 606 : } else {
250 54 : if( FD_UNLIKELY( store->spill_slot_cnt>=store->fec_max ) ) return FD_STORE_SPILL_NONE;
251 54 : spill_slot = (uint)store->spill_slot_cnt++;
252 54 : }
253 :
254 660 : ulong ram_slot = victim->data_idx;
255 660 : ulong ram_off = ram_slot * store->payload_slot_sz;
256 660 : ulong spill_off = (ulong)spill_slot * store->payload_slot_sz;
257 660 : ulong data_sz = victim->data_sz;
258 :
259 660 : cache_lru_remove_locked( store, victim );
260 660 : victim->data_state = FD_STORE_FEC_DATA_SPILLING;
261 660 : long evict_ticks = -fd_tickcount();
262 660 : fd_rwlock_unwrite( &store->cache_lock );
263 :
264 660 : 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 660 : fd_rwlock_write( &store->cache_lock );
268 660 : FD_TEST( victim->data_state==FD_STORE_FEC_DATA_SPILLING );
269 660 : int result;
270 660 : 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 660 : } else {
282 660 : victim->data_idx = spill_slot;
283 660 : victim->data_state = FD_STORE_FEC_DATA_DISK;
284 660 : atomic_fetch_add_explicit( &store->spill_live_cnt, 1UL, memory_order_relaxed );
285 660 : atomic_fetch_add_explicit( &store->fec_spill_cnt, 1UL, memory_order_relaxed );
286 660 : atomic_fetch_add_explicit( &store->fec_spill_bytes, data_sz, memory_order_relaxed );
287 660 : cache_slot_release_locked( store, ram_slot );
288 660 : result = FD_STORE_SPILL_COMPLETE;
289 660 : }
290 :
291 660 : evict_ticks += fd_tickcount();
292 660 : if( FD_LIKELY( spill ) ) {
293 12 : spill->write_cnt++;
294 12 : spill->write_bytes += data_sz;
295 12 : spill->write_ticks += (ulong)evict_ticks;
296 12 : }
297 660 : return result;
298 660 : }
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 156 : ulong shred_tile_cnt ) {
311 :
312 156 : if( FD_UNLIKELY( !shmem ) ) { FD_LOG_WARNING(( "NULL shmem" )); return NULL; }
313 156 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shmem, fd_store_align() ) ) ) { FD_LOG_WARNING(( "misaligned shmem" )); return NULL; }
314 156 : if( FD_UNLIKELY( !fec_max ) ) { FD_LOG_WARNING(( "fec_max must be non-zero" )); return NULL; }
315 156 : if( FD_UNLIKELY( fec_max>UINT_MAX ) ) { FD_LOG_WARNING(( "fec_max must fit in uint" )); return NULL; }
316 156 : if( FD_UNLIKELY( !fec_data_max ) ) { FD_LOG_WARNING(( "fec_data_max must be non-zero" )); return NULL; }
317 156 : if( FD_UNLIKELY( fec_data_max>UINT_MAX ) ) { FD_LOG_WARNING(( "fec_data_max must fit in uint" )); return NULL; }
318 156 : 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 144 : if( FD_UNLIKELY( !shred_tile_cnt ) ) { FD_LOG_WARNING(( "shred_tile_cnt must be non-zero" )); return NULL; }
320 138 : 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 138 : ulong footprint = fd_store_footprint( fec_max, fec_data_max, shred_storage_gib, shred_cache_bytes, fec_set_cnt, shred_tile_cnt );
326 138 : if( FD_UNLIKELY( !footprint ) ) { FD_LOG_WARNING(( "invalid or overflowing store footprint" )); return NULL; }
327 :
328 138 : fd_wksp_t * wksp = fd_wksp_containing( shmem );
329 138 : if( FD_UNLIKELY( !wksp ) ) { FD_LOG_WARNING(( "shmem must be part of a workspace" )); return NULL; }
330 :
331 138 : ulong chain_cnt = fd_store_map_chain_cnt_est( fec_max );
332 138 : ulong part_cnt = shred_tile_cnt;
333 138 : ulong slot_chain_cnt = fd_store_slot_to_fecs_chain_cnt_est( fec_max );
334 138 : 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 138 : ulong payload_slot_sz = fd_store_payload_slot_sz( fec_data_max );
336 138 : ulong cache_slot_cnt = shred_cache_bytes
337 138 : ? fd_ulong_min( fec_max, fd_ulong_max( 1UL, shred_cache_bytes / payload_slot_sz ) )
338 138 : : fec_max;
339 :
340 138 : FD_SCRATCH_ALLOC_INIT( l, shmem );
341 138 : fd_store_t * store = FD_SCRATCH_ALLOC_APPEND( l, fd_store_align(), sizeof(fd_store_t) );
342 138 : void * map = FD_SCRATCH_ALLOC_APPEND( l, fd_store_map_align(), fd_store_map_footprint( chain_cnt ) );
343 138 : void * shpool = FD_SCRATCH_ALLOC_APPEND( l, fd_store_pool_align(), fd_store_pool_footprint() );
344 138 : void * shele = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_store_fec_t), sizeof(fd_store_fec_t)*fec_max );
345 138 : uchar * slot_to_fecs = FD_SCRATCH_ALLOC_APPEND( l, fd_store_slot_to_fecs_align(), slot_to_fecs_sz*part_cnt );
346 138 : uchar * cache_mem = FD_SCRATCH_ALLOC_APPEND( l, FD_STORE_PAYLOAD_PAGE_SZ, payload_slot_sz*cache_slot_cnt );
347 138 : ulong * cache_free = FD_SCRATCH_ALLOC_APPEND( l, alignof(ulong), sizeof(ulong)*cache_slot_cnt );
348 138 : uint * spill_free = FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), sizeof(uint)*fec_max );
349 138 : uchar * spill_read_mem = FD_SCRATCH_ALLOC_APPEND( l, FD_STORE_PAYLOAD_PAGE_SZ, payload_slot_sz );
350 138 : fd_fec_set_t * fec_sets = fec_set_cnt
351 138 : ? FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_fec_set_t), sizeof(fd_fec_set_t)*fec_set_cnt )
352 138 : : NULL;
353 :
354 138 : void * shred_map_mem = NULL;
355 138 : fd_shredb_ring_entry_t * ring_pool_mem = NULL;
356 138 : void * root_map_mem = NULL;
357 138 : atomic_ulong * slot_hint_mem = NULL;
358 138 : ulong max_shreds = 0UL;
359 138 : ulong max_slots = 0UL;
360 138 : ulong disk_chain_cnt = 0UL;
361 :
362 138 : if( shred_storage_gib ) {
363 66 : max_shreds = fd_shredb_max_shreds( shred_storage_gib );
364 66 : max_slots = fd_shredb_max_slots( shred_storage_gib );
365 66 : disk_chain_cnt = fd_shredb_disk_chain_cnt( max_shreds );
366 :
367 66 : shred_map_mem = FD_SCRATCH_ALLOC_APPEND( l, fd_shredb_slot_map_align(), fd_shredb_slot_map_footprint( disk_chain_cnt ) );
368 66 : ring_pool_mem = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_shredb_ring_entry_t), max_shreds*sizeof(fd_shredb_ring_entry_t) );
369 66 : slot_hint_mem = FD_SCRATCH_ALLOC_APPEND( l, alignof(atomic_ulong), max_slots*sizeof(atomic_ulong) );
370 66 : root_map_mem = FD_SCRATCH_ALLOC_APPEND( l, fd_shredb_root_map_align(), fd_shredb_root_map_footprint( disk_chain_cnt ) );
371 66 : }
372 :
373 138 : FD_TEST( FD_SCRATCH_ALLOC_FINI( l, fd_store_align() )==(ulong)shmem + footprint );
374 :
375 138 : ulong payload_sz;
376 138 : ulong wire_sz;
377 138 : ulong file_sz;
378 138 : if( FD_UNLIKELY( __builtin_umull_overflow( payload_slot_sz, fec_max, &payload_sz ) ||
379 138 : __builtin_umull_overflow( max_shreds, sizeof(fd_shredb_entry_t), &wire_sz ) ||
380 138 : __builtin_uaddl_overflow( payload_sz, wire_sz, &file_sz ) ||
381 138 : 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 138 : fd_memset( store, 0, sizeof(fd_store_t) );
387 138 : store->fec_max = fec_max;
388 138 : store->fec_data_max = fec_data_max;
389 138 : store->store_gaddr = fd_wksp_gaddr_fast( wksp, store );
390 138 : store->pool_mem_gaddr = fd_wksp_gaddr_fast( wksp, shpool );
391 138 : store->pool_ele_gaddr = fd_wksp_gaddr_fast( wksp, shele );
392 138 : store->shred_tile_cnt = shred_tile_cnt;
393 138 : store->slot_to_fecs_gaddr = fd_wksp_gaddr_fast( wksp, slot_to_fecs );
394 138 : store->slot_to_fecs_sz = slot_to_fecs_sz;
395 138 : atomic_init( &store->slot_lo, ULONG_MAX );
396 138 : store->payload_slot_sz = payload_slot_sz;
397 138 : store->payload_sz = payload_sz;
398 138 : store->wire_off = payload_sz; /* wire region begins after the payload region (page-aligned) */
399 138 : store->cache_slot_cnt = cache_slot_cnt;
400 138 : store->cache_data_gaddr = fd_wksp_gaddr_fast( wksp, cache_mem );
401 138 : store->cache_free_gaddr = fd_wksp_gaddr_fast( wksp, cache_free );
402 138 : store->cache_free_cnt = cache_slot_cnt;
403 138 : ulong free_cap = fd_ulong_min( 8192UL, (64UL<<20) / payload_slot_sz );
404 138 : ulong burst_floor = fd_ulong_min( 512UL, cache_slot_cnt/2UL );
405 138 : ulong free_target = fd_ulong_max( cache_slot_cnt/10UL, burst_floor );
406 138 : store->cache_free_target = fd_ulong_max( 1UL, fd_ulong_min( free_cap, free_target ) );
407 138 : store->cache_free_low_water = fd_ulong_max( 1UL, (store->cache_free_target*3UL)/4UL );
408 138 : store->cache_lru_head = UINT_MAX;
409 138 : store->cache_lru_tail = UINT_MAX;
410 138 : store->spill_free_gaddr = fd_wksp_gaddr_fast( wksp, spill_free );
411 138 : store->spill_read_data_gaddr = fd_wksp_gaddr_fast( wksp, spill_read_mem );
412 138 : store->fec_set_cnt = fec_set_cnt;
413 138 : store->fec_sets_gaddr = fec_set_cnt ? fd_wksp_gaddr_fast( wksp, fec_sets ) : 0UL;
414 138 : store->max_shreds_per_block = max_shreds_per_block;
415 138 : fd_rwlock_new( &store->cache_lock );
416 138 : fd_rwlock_new( &store->spill_read_lock );
417 138 : fd_rwlock_new( &store->fec_lock );
418 138 : atomic_init( &store->fec_spill_cnt, 0UL );
419 138 : atomic_init( &store->spill_live_cnt, 0UL );
420 138 : atomic_init( &store->fec_spill_bytes, 0UL );
421 138 : atomic_init( &store->fec_spill_read_cnt, 0UL );
422 138 : atomic_init( &store->fec_spill_read_bytes, 0UL );
423 :
424 138 : void * shmap = fd_store_map_new( map, chain_cnt, seed );
425 138 : if( FD_UNLIKELY( !shmap ) ) { FD_LOG_WARNING(( "fd_store_map_new failed" )); return NULL; }
426 138 : store->map_gaddr = fd_wksp_gaddr_fast( wksp, shmap );
427 :
428 138 : if( FD_UNLIKELY( !fd_store_pool_new( shpool ) ) ) { FD_LOG_WARNING(( "fd_store_pool_new failed" )); return NULL; }
429 138 : fd_store_pool_t pool_ljoin;
430 138 : fd_store_pool_reset( fd_store_pool_join( &pool_ljoin, shpool, shele, fec_max ) );
431 :
432 2064 : 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 138 : fd_store_fec_t * fec0 = (fd_store_fec_t *)shele;
436 2184 : for( ulong i=0UL; i<fec_max; i++ ) {
437 2046 : fec0[ i ].data_idx = 0U;
438 2046 : fec0[ i ].cache_prev = UINT_MAX;
439 2046 : fec0[ i ].cache_next = UINT_MAX;
440 2046 : fec0[ i ].data_pin_cnt = 0U;
441 2046 : fec0[ i ].data_state = FD_STORE_FEC_DATA_EMPTY;
442 2046 : fec0[ i ].data_consume_pending = 0U;
443 2046 : }
444 :
445 291 : 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 138 : if( shred_storage_gib ) {
448 66 : void * slot_map = fd_shredb_slot_map_new( shred_map_mem, disk_chain_cnt, seed );
449 66 : FD_TEST( slot_map );
450 66 : store->shred_map_gaddr = fd_wksp_gaddr_fast( wksp, slot_map );
451 66 : store->ring_pool_gaddr = fd_wksp_gaddr_fast( wksp, ring_pool_mem );
452 66 : store->slot_hint_gaddr = fd_wksp_gaddr_fast( wksp, slot_hint_mem );
453 66 : store->disk_max_shreds = max_shreds;
454 66 : store->disk_max_slots = max_slots;
455 66 : fd_rwlock_new( &store->disk_lock );
456 :
457 66 : void * root_map = fd_shredb_root_map_new( root_map_mem, disk_chain_cnt, seed );
458 66 : FD_TEST( root_map );
459 66 : 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 66 : fd_memset( ring_pool_mem, 0, max_shreds*sizeof(fd_shredb_ring_entry_t) );
464 1730190 : for( ulong i=0UL; i<max_slots; i++ ) atomic_init( slot_hint_mem+i, 0UL );
465 :
466 66 : atomic_init( &store->disk_reservation_head, 0UL );
467 66 : atomic_init( &store->disk_cnt, 0UL );
468 66 : atomic_init( &store->disk_insert_cnt, 0UL );
469 66 : atomic_init( &store->disk_write_bytes, 0UL );
470 66 : }
471 :
472 138 : FD_COMPILER_MFENCE();
473 138 : FD_VOLATILE( store->magic ) = FD_STORE_MAGIC;
474 138 : FD_COMPILER_MFENCE();
475 :
476 138 : return shmem;
477 138 : }
478 :
479 : fd_store_t *
480 138 : fd_store_join( void * shstore ) {
481 138 : if( FD_UNLIKELY( !shstore ) ) { FD_LOG_WARNING(( "NULL store" )); return NULL; }
482 138 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shstore, fd_store_align() ) ) ) { FD_LOG_WARNING(( "misaligned store" )); return NULL; }
483 :
484 138 : fd_wksp_t * wksp = fd_wksp_containing( shstore );
485 138 : if( FD_UNLIKELY( !wksp ) ) { FD_LOG_WARNING(( "store must be part of a workspace" )); return NULL; }
486 :
487 138 : fd_store_t * store = (fd_store_t *)shstore;
488 138 : if( FD_UNLIKELY( store->magic!=FD_STORE_MAGIC ) ) { FD_LOG_WARNING(( "bad magic" )); return NULL; }
489 :
490 138 : return store;
491 138 : }
492 :
493 : int
494 : fd_store_file_create( char const * path,
495 : ulong wire_off,
496 96 : ulong disk_max_shreds ) {
497 96 : if( FD_UNLIKELY( !path ) ) {
498 0 : errno = EINVAL;
499 0 : return -1;
500 0 : }
501 :
502 96 : ulong wire_sz;
503 96 : ulong file_sz;
504 96 : if( FD_UNLIKELY( __builtin_umull_overflow( disk_max_shreds, sizeof(fd_shredb_entry_t), &wire_sz ) ||
505 96 : __builtin_uaddl_overflow( wire_off, wire_sz, &file_sz ) ||
506 96 : 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 96 : int fd = open( path, O_RDWR|O_CREAT, (mode_t)0600 );
514 96 : if( FD_UNLIKELY( fd<0 ) ) return -1;
515 :
516 96 : int err = 0;
517 96 : if( FD_UNLIKELY( ftruncate( fd, (off_t)file_sz ) ) ) err = errno;
518 96 : 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 96 : return fd;
524 96 : }
525 :
526 : void *
527 138 : fd_store_leave( fd_store_t const * store ) {
528 138 : if( FD_UNLIKELY( !store ) ) { FD_LOG_WARNING(( "NULL store" )); return NULL; }
529 138 : return (void *)store;
530 138 : }
531 :
532 : void *
533 138 : fd_store_delete( void * shstore ) {
534 138 : if( FD_UNLIKELY( !shstore ) ) { FD_LOG_WARNING(( "NULL store" )); return NULL; }
535 138 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shstore, fd_store_align() ) ) ) { FD_LOG_WARNING(( "misaligned store" )); return NULL; }
536 :
537 138 : fd_store_t * store = (fd_store_t *)shstore;
538 138 : if( FD_UNLIKELY( store->magic!=FD_STORE_MAGIC ) ) { FD_LOG_WARNING(( "bad magic" )); return NULL; }
539 :
540 138 : FD_COMPILER_MFENCE();
541 138 : FD_VOLATILE( store->magic ) = 0UL;
542 138 : FD_COMPILER_MFENCE();
543 :
544 138 : return shstore;
545 138 : }
546 :
547 :
548 : uchar *
549 : fd_store_fec_data_reserve( fd_store_t * store,
550 : int disk_fd,
551 : uint * data_idx,
552 1538 : fd_store_fec_spill_stats_t * spill ) {
553 1538 : if( FD_LIKELY( spill ) ) *spill = (fd_store_fec_spill_stats_t){0};
554 1538 : if( FD_UNLIKELY( !store || !data_idx ) ) return NULL;
555 :
556 : /* Pops a free RAM cache slot, spilling LRU payloads as needed. */
557 1538 : fd_rwlock_write( &store->cache_lock );
558 :
559 2192 : while( FD_UNLIKELY( !store->cache_free_cnt ) ) {
560 660 : 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 660 : int spill_result = spill_one_locked( store, disk_fd, spill );
567 660 : 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 660 : if( FD_UNLIKELY( !spill_result ) ) {
574 6 : fd_rwlock_unwrite( &store->cache_lock );
575 6 : return NULL;
576 6 : }
577 660 : }
578 :
579 1532 : ulong slot = cache_free_laddr( store )[ --store->cache_free_cnt ];
580 1532 : fd_rwlock_unwrite( &store->cache_lock );
581 :
582 1532 : *data_idx = (uint)slot;
583 1532 : return cache_data_laddr( store ) + slot*store->payload_slot_sz;
584 1538 : }
585 :
586 : int
587 : fd_store_fec_data_preevict( fd_store_t * store,
588 : int disk_fd,
589 6 : fd_store_fec_spill_stats_t * spill ) {
590 6 : fd_store_fec_spill_stats_t local_spill[1];
591 6 : if( FD_UNLIKELY( !spill ) ) spill = local_spill;
592 6 : *spill = (fd_store_fec_spill_stats_t){0};
593 6 : if( FD_UNLIKELY( !store || disk_fd<0 ) ) return 0;
594 :
595 : /* called every pass: don't take the write lock for nothing */
596 6 : if( FD_LIKELY( !FD_VOLATILE_CONST( store->cache_preevict_active ) &&
597 6 : FD_VOLATILE_CONST( store->cache_free_cnt )>=store->cache_free_low_water ) ) return 0;
598 :
599 6 : fd_rwlock_write( &store->cache_lock );
600 6 : if( FD_UNLIKELY( !store->cache_preevict_active && store->cache_free_cnt<store->cache_free_low_water ) )
601 6 : store->cache_preevict_active = 1U;
602 6 : if( FD_UNLIKELY( store->cache_preevict_active && store->cache_free_cnt>=store->cache_free_target ) )
603 0 : store->cache_preevict_active = 0U;
604 :
605 6 : if( FD_LIKELY( !store->cache_preevict_active ) ) {
606 0 : fd_rwlock_unwrite( &store->cache_lock );
607 0 : return 0;
608 0 : }
609 :
610 6 : spill_one_locked( store, disk_fd, spill );
611 6 : if( FD_UNLIKELY( store->cache_free_cnt>=store->cache_free_target ) )
612 6 : store->cache_preevict_active = 0U;
613 6 : fd_rwlock_unwrite( &store->cache_lock );
614 6 : return !!spill->write_cnt;
615 6 : }
616 :
617 : void
618 : fd_store_fec_cache_stats_query( fd_store_t * store,
619 6 : fd_store_fec_cache_stats_t * stats ) {
620 6 : if( FD_UNLIKELY( !store || !stats ) ) return;
621 6 : fd_rwlock_read( &store->cache_lock );
622 6 : stats->free_cnt = store->cache_free_cnt;
623 6 : stats->max = store->cache_slot_cnt;
624 6 : stats->target = store->cache_free_target;
625 6 : stats->low_water = store->cache_free_low_water;
626 6 : fd_rwlock_unread( &store->cache_lock );
627 6 : }
628 :
629 : static void
630 : cache_pin_locked( fd_store_t * store,
631 36 : fd_store_fec_t * fec ) {
632 36 : if( FD_UNLIKELY( !fec->data_pin_cnt && fec->data_state==FD_STORE_FEC_DATA_RAM_READY ) )
633 24 : cache_lru_remove_locked( store, fec );
634 36 : FD_TEST( fec->data_pin_cnt<USHORT_MAX );
635 36 : fec->data_pin_cnt++;
636 36 : store->cache_pinned_cnt++;
637 36 : }
638 :
639 : static void
640 : cache_unpin_locked( fd_store_t * store,
641 36 : fd_store_fec_t * fec ) {
642 36 : FD_TEST( fec->data_pin_cnt && store->cache_pinned_cnt );
643 36 : fec->data_pin_cnt--;
644 36 : store->cache_pinned_cnt--;
645 : /* Prefer spilling payloads that a reader has already copied while
646 : retaining their disk copy for replay backfill. */
647 36 : if( FD_UNLIKELY( !fec->data_pin_cnt && fec->data_state==FD_STORE_FEC_DATA_RAM_READY ) )
648 24 : cache_lru_push_head_locked( store, fec );
649 36 : }
650 :
651 : int
652 : fd_store_fec_data_view( fd_store_t * store,
653 : int disk_fd,
654 : fd_store_fec_t * fec,
655 42 : fd_store_fec_data_view_t * view ) {
656 42 : if( FD_UNLIKELY( !view ) ) return -1;
657 42 : view->data = NULL;
658 42 : view->fec = NULL;
659 42 : view->flags = 0U;
660 42 : if( FD_UNLIKELY( !store || !fec ) ) return -1;
661 42 : fd_store_fec_t * fec0 = pool_ele_laddr( store );
662 42 : if( FD_UNLIKELY( fec<fec0 || fec>=fec0+store->fec_max ) ) return -1;
663 :
664 42 : fd_rwlock_write( &store->cache_lock );
665 :
666 42 : 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 42 : if( FD_LIKELY( fec->data_state==FD_STORE_FEC_DATA_RAM_READY ||
672 42 : fec->data_state==FD_STORE_FEC_DATA_SPILLING ) ) {
673 24 : cache_pin_locked( store, fec );
674 24 : view->data = cache_data_laddr( store ) + (ulong)fec->data_idx*store->payload_slot_sz;
675 24 : view->fec = fec;
676 24 : fd_rwlock_unwrite( &store->cache_lock );
677 24 : return 0;
678 24 : }
679 :
680 18 : if( FD_LIKELY( fec->data_state==FD_STORE_FEC_DATA_DISK ) ) {
681 18 : if( FD_UNLIKELY( disk_fd<0 ) ) {
682 0 : fd_rwlock_unwrite( &store->cache_lock );
683 0 : return -1;
684 0 : }
685 18 : if( FD_UNLIKELY( !fd_rwlock_trywrite( &store->spill_read_lock ) ) ) {
686 6 : fd_rwlock_unwrite( &store->cache_lock );
687 6 : return -1;
688 6 : }
689 12 : ulong data_sz = fec->data_sz;
690 12 : ulong data_off = (ulong)fec->data_idx*store->payload_slot_sz;
691 12 : uchar * spill_read_data = spill_read_data_laddr( store );
692 12 : cache_pin_locked( store, fec );
693 12 : fd_rwlock_unwrite( &store->cache_lock );
694 :
695 12 : 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 12 : atomic_fetch_add_explicit( &store->fec_spill_read_cnt, 1UL, memory_order_relaxed );
699 12 : atomic_fetch_add_explicit( &store->fec_spill_read_bytes, data_sz, memory_order_relaxed );
700 :
701 12 : view->data = spill_read_data;
702 12 : view->fec = fec;
703 12 : view->flags = FD_STORE_FEC_DATA_VIEW_SPILL;
704 12 : return 0;
705 12 : }
706 :
707 0 : fd_rwlock_unwrite( &store->cache_lock );
708 0 : return -1;
709 18 : }
710 :
711 : void
712 : fd_store_fec_data_view_release( fd_store_t * store,
713 36 : fd_store_fec_data_view_t * view ) {
714 36 : if( FD_UNLIKELY( !store || !view || !view->fec ) ) return;
715 36 : if( view->flags & FD_STORE_FEC_DATA_VIEW_SPILL ) fd_rwlock_unwrite( &store->spill_read_lock );
716 :
717 36 : fd_rwlock_write( &store->cache_lock );
718 36 : cache_unpin_locked( store, view->fec );
719 36 : cache_consume_locked( store, view->fec );
720 36 : fd_rwlock_unwrite( &store->cache_lock );
721 :
722 36 : view->data = NULL;
723 36 : view->fec = NULL;
724 36 : view->flags = 0U;
725 36 : }
726 :
727 : fd_store_fec_t *
728 : fd_store_query( fd_store_map_t * map,
729 783 : fd_hash_t const * merkle_root ) {
730 783 : for(;;) {
731 783 : fd_store_map_query_t query[1];
732 783 : int err = fd_store_map_query_try( map, merkle_root, NULL, query, 0 );
733 783 : if( FD_UNLIKELY( err==FD_MAP_ERR_AGAIN ) ) continue;
734 783 : if( FD_UNLIKELY( err ) ) return NULL;
735 :
736 735 : fd_store_fec_t * fec = fd_store_map_query_ele( query );
737 735 : err = fd_store_map_query_test( query );
738 735 : if( FD_LIKELY( !err ) ) return fec;
739 0 : if( FD_UNLIKELY( err!=FD_MAP_ERR_AGAIN ) ) return NULL;
740 0 : }
741 783 : }
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 1537 : fd_store_fec_t ** fec ) {
753 1537 : FD_TEST( store && map && merkle_root && shred_sz && fec );
754 1537 : FD_TEST( data_idx<store->cache_slot_cnt && data_sz<=store->fec_data_max );
755 1537 : *fec = NULL;
756 1537 : fd_rwlock_read( &store->fec_lock );
757 :
758 1537 : struct {
759 1537 : fd_store_map_txn_t txn [1];
760 1537 : fd_store_map_txn_private_info_t info[1];
761 1537 : } map_txn_mem;
762 1537 : fd_store_map_txn_t * txn = fd_store_map_txn_init( map_txn_mem.txn, map, 1UL );
763 1537 : FD_TEST( !fd_store_map_txn_add( txn, merkle_root, 1 ) );
764 1537 : FD_TEST( !fd_store_map_txn_try( txn, FD_MAP_FLAG_BLOCKING ) );
765 :
766 1537 : fd_store_map_query_t query[1];
767 1537 : int err = fd_store_map_txn_query( map, merkle_root, NULL, query, 0 );
768 1537 : if( FD_LIKELY( err==FD_MAP_ERR_KEY ) ) {
769 1495 : fd_store_fec_t * new_fec = fd_store_fec_acquire( store );
770 1495 : FD_TEST( new_fec );
771 1495 : new_fec->key = *merkle_root;
772 1495 : new_fec->data_sz = data_sz;
773 1495 : 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 1495 : fd_rwlock_write( &store->cache_lock );
778 1495 : new_fec->data_idx = data_idx;
779 1495 : new_fec->data_state = FD_STORE_FEC_DATA_RAM_READY;
780 1495 : cache_lru_push_tail_locked( store, new_fec );
781 1495 : fd_rwlock_unwrite( &store->cache_lock );
782 :
783 1495 : FD_TEST( part<store->shred_tile_cnt );
784 1495 : new_fec->slot = (uint)fd_ulong_max( slot, store->root );
785 1495 : ulong lo = atomic_load_explicit( &store->slot_lo, memory_order_relaxed );
786 1495 : while( new_fec->slot<lo && !atomic_compare_exchange_weak_explicit( &store->slot_lo, &lo, new_fec->slot, memory_order_relaxed, memory_order_relaxed ) );
787 1495 : 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 1495 : FD_TEST( !fd_store_map_txn_insert( map, new_fec ) );
789 1495 : *fec = new_fec;
790 1495 : err = FD_MAP_SUCCESS;
791 1495 : } else {
792 42 : FD_TEST( err==FD_MAP_SUCCESS );
793 42 : err = FD_MAP_ERR_KEY;
794 42 : }
795 :
796 1537 : FD_TEST( !fd_store_map_txn_test( txn ) );
797 1537 : fd_store_map_txn_fini( txn );
798 1537 : fd_rwlock_unread( &store->fec_lock );
799 :
800 1537 : if( FD_UNLIKELY( err==FD_MAP_ERR_KEY ) ) {
801 21 : fd_rwlock_write( &store->cache_lock );
802 21 : cache_slot_release_locked( store, data_idx );
803 21 : fd_rwlock_unwrite( &store->cache_lock );
804 21 : }
805 1537 : return err;
806 1537 : }
807 :
808 : ulong
809 : fd_store_publish( fd_store_t * store,
810 : fd_store_map_t * map,
811 684 : ulong root ) {
812 684 : fd_store_fec_t * fec0 = pool_ele_laddr( store );
813 684 : ulong cnt = 0UL;
814 684 : fd_rwlock_write( &store->fec_lock );
815 684 : ulong lo = atomic_load_explicit( &store->slot_lo, memory_order_relaxed );
816 1392 : for( ulong part=0UL; part<store->shred_tile_cnt; part++ ) {
817 708 : fd_store_slot_to_fecs_t * slot_to_fecs = fd_store_slot_to_fecs_join( slot_to_fecs_laddr( store, part ) );
818 1506 : for( ulong slot=lo; slot<root; slot++ ) {
819 798 : uint key = (uint)slot;
820 798 : for( ulong idx = fd_store_slot_to_fecs_idx_remove( slot_to_fecs, &key, ULONG_MAX, fec0 );
821 1518 : idx!=ULONG_MAX;
822 798 : idx = fd_store_slot_to_fecs_idx_remove( slot_to_fecs, &key, ULONG_MAX, fec0 ) ) {
823 720 : fd_store_map_query_t query[1];
824 720 : FD_TEST( !fd_store_map_remove( map, &fec0[ idx ].key, NULL, query, FD_MAP_FLAG_BLOCKING ) );
825 720 : fd_store_fec_release( store, fd_store_map_query_ele( query ) );
826 720 : cnt++;
827 720 : }
828 798 : }
829 708 : }
830 684 : if( FD_LIKELY( lo<root ) ) atomic_store_explicit( &store->slot_lo, root, memory_order_relaxed );
831 684 : store->root = fd_ulong_max( store->root, root );
832 684 : fd_rwlock_unwrite( &store->fec_lock );
833 684 : return cnt;
834 684 : }
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 447 : #define FD_STORE_DISK_READ_RETRY_CNT (8UL)
842 :
843 : static inline ulong
844 : disk_cell_tag( ulong ticket,
845 5592 : uint state ) {
846 5592 : return ((ticket & FD_SHREDB_TAG_SEQ_MASK)<<2) | state;
847 5592 : }
848 :
849 : static inline ulong
850 4056 : disk_cell_state( ulong tag ) {
851 4056 : return tag & FD_SHREDB_CELL_STATE_MASK;
852 4056 : }
853 :
854 : static inline ulong
855 339 : disk_cell_ticket( ulong tag ) {
856 339 : return (tag>>2) & FD_SHREDB_TAG_SEQ_MASK;
857 339 : }
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 1917 : disk_cell_tag_load( fd_shredb_ring_entry_t const * cell ) {
863 1917 : return (ulong)cell->tag
864 1917 : | ((ulong)cell->tag_hi[0]<<32)
865 1917 : | ((ulong)cell->tag_hi[1]<<40)
866 1917 : | ((ulong)cell->tag_hi[2]<<48);
867 1917 : }
868 :
869 : static inline void
870 : disk_cell_tag_store( fd_shredb_ring_entry_t * cell,
871 2454 : ulong tag ) {
872 2454 : cell->tag = (uint)tag;
873 2454 : cell->tag_hi[0] = (uchar)(tag>>32);
874 2454 : cell->tag_hi[1] = (uchar)(tag>>40);
875 2454 : cell->tag_hi[2] = (uchar)(tag>>48);
876 2454 : }
877 :
878 : static void
879 : disk_slot_hint_publish_locked( fd_store_t * store,
880 : ulong slot,
881 1209 : uint shred_idx ) {
882 1209 : atomic_ulong * hint = disk_slot_hint_laddr( store ) + (slot % store->disk_max_slots);
883 1209 : ulong current = atomic_load_explicit( hint, memory_order_relaxed );
884 1209 : if( FD_LIKELY( current & FD_SHREDB_HINT_VALID ) ) {
885 810 : uint current_idx = fd_shredb_key_shred_idx( current & ~FD_SHREDB_HINT_VALID );
886 810 : if( FD_LIKELY( fd_shredb_key_slot( current )==slot ) && current_idx>=shred_idx ) return;
887 810 : }
888 1173 : ulong desired = fd_shredb_key_pack( slot, shred_idx ) | FD_SHREDB_HINT_VALID;
889 1173 : atomic_store_explicit( hint, desired, memory_order_release );
890 1173 : }
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 1614 : ulong key ) {
901 1614 : fd_shredb_slot_map_t * map = disk_slot_map_join( store );
902 1614 : return (uint)fd_shredb_slot_map_idx_query_const( map, &key, UINT_MAX, disk_ring_pool_laddr( store ) );
903 1614 : }
904 :
905 : static inline uint
906 : disk_root_lookup_locked( fd_store_t const * store,
907 1341 : fd_shredb_root_key_t const * key ) {
908 1341 : fd_shredb_root_map_t * map = disk_root_map_join( store );
909 1341 : return (uint)fd_shredb_root_map_idx_query_const( map, key, UINT_MAX, disk_ring_pool_laddr( store ) );
910 1341 : }
911 :
912 : static void
913 : disk_invalidate_locked( fd_store_t * store,
914 1227 : uint ring_idx ) {
915 1227 : fd_shredb_ring_entry_t * pool = disk_ring_pool_laddr( store );
916 1227 : fd_shredb_ring_entry_t * cell = pool + ring_idx;
917 1227 : FD_TEST( disk_cell_state( cell->tag )!=FD_SHREDB_CELL_WRITING );
918 :
919 1227 : fd_shredb_slot_map_t * slot_map = disk_slot_map_join( store );
920 1227 : if( (uint)fd_shredb_slot_map_idx_query( slot_map, &cell->key, UINT_MAX, pool )==ring_idx ) {
921 972 : fd_shredb_slot_map_idx_remove( slot_map, &cell->key, UINT_MAX, pool );
922 972 : }
923 :
924 1227 : fd_shredb_root_map_t * root_map = disk_root_map_join( store );
925 1227 : if( (uint)fd_shredb_root_map_idx_query( root_map, &cell->root_key, UINT_MAX, pool )==ring_idx ) {
926 978 : fd_shredb_root_map_idx_remove( root_map, &cell->root_key, UINT_MAX, pool );
927 978 : }
928 :
929 1227 : if( disk_cell_state( cell->tag )==FD_SHREDB_CELL_READY ) {
930 978 : ulong disk_cnt = atomic_load_explicit( &store->disk_cnt, memory_order_relaxed );
931 978 : FD_TEST( disk_cnt!=0UL );
932 978 : atomic_store_explicit( &store->disk_cnt, disk_cnt-1UL, memory_order_relaxed );
933 978 : }
934 1227 : cell->tag = (cell->tag & ~(uint)FD_SHREDB_CELL_STATE_MASK) | (uint)FD_SHREDB_CELL_INVALID;
935 1227 : }
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 1251 : disk_view_t * view ) {
951 1251 : 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 1251 : uint exist_idx = disk_root_lookup_locked( store, root_key );
956 1251 : if( exist_idx!=UINT_MAX ) {
957 18 : ulong state = disk_cell_state( pool[exist_idx].tag );
958 18 : if( FD_UNLIKELY( state==FD_SHREDB_CELL_WRITING ) ) return FD_STORE_DISK_INSERT_ERR;
959 18 : FD_TEST( state==FD_SHREDB_CELL_READY );
960 18 : return FD_STORE_DISK_INSERT_DUP;
961 18 : }
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 1233 : ulong ticket = atomic_load_explicit( &store->disk_reservation_head, memory_order_relaxed );
966 1233 : uint ring_idx = (uint)(ticket % store->disk_max_shreds);
967 1233 : atomic_store_explicit( &store->disk_reservation_head, ticket + 1UL, memory_order_relaxed );
968 1233 : 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 1227 : disk_invalidate_locked( store, ring_idx );
972 :
973 1227 : fd_shredb_ring_entry_t * cell = pool + ring_idx;
974 1227 : disk_cell_tag_store( cell, disk_cell_tag( ticket, FD_SHREDB_CELL_WRITING ) );
975 1227 : cell->key = key;
976 1227 : int linked = disk_slot_lookup_locked( store, key )==UINT_MAX;
977 1227 : if( linked ) fd_shredb_slot_map_idx_insert( disk_slot_map_join( store ), ring_idx, pool );
978 1227 : cell->root_key = *root_key;
979 1227 : fd_shredb_root_map_idx_insert( disk_root_map_join( store ), ring_idx, pool );
980 :
981 1227 : view->ticket = ticket;
982 1227 : view->ring_idx = ring_idx;
983 1227 : view->linked = linked;
984 1227 : return FD_STORE_DISK_INSERT_SUCCESS;
985 1233 : }
986 :
987 : static int
988 : disk_insert_finish_write_locked( fd_store_t * store,
989 1227 : disk_view_t view ) {
990 1227 : fd_shredb_ring_entry_t * cell = disk_ring_pool_laddr( store ) + view.ring_idx;
991 1227 : FD_TEST( disk_cell_tag_load( cell )==disk_cell_tag( view.ticket, FD_SHREDB_CELL_WRITING ) );
992 :
993 1227 : ulong disk_cnt = atomic_load_explicit( &store->disk_cnt, memory_order_relaxed );
994 1227 : atomic_store_explicit( &store->disk_cnt, disk_cnt+1UL, memory_order_relaxed );
995 1227 : disk_cell_tag_store( cell, disk_cell_tag( view.ticket, FD_SHREDB_CELL_READY ) );
996 :
997 1227 : if( view.linked ) disk_slot_hint_publish_locked( store, fd_shredb_key_slot( cell->key ), fd_shredb_key_shred_idx( cell->key ) );
998 1227 : return FD_STORE_DISK_INSERT_SUCCESS;
999 1227 : }
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 1271 : fd_hash_t const * merkle_root ) {
1008 1271 : if( FD_UNLIKELY( !store || disk_fd<0 || !shred || !merkle_root || !fd_store_has_disk( store ) ) )
1009 6 : return FD_STORE_DISK_INSERT_ERR;
1010 :
1011 1265 : ulong slot = shred->slot;
1012 1265 : uint shred_idx = shred->idx;
1013 1265 : if( FD_UNLIKELY( slot>=FD_SHREDB_KEY_SLOT_MAX || shred_idx>=store->max_shreds_per_block ) ) return FD_STORE_DISK_INSERT_ERR;
1014 1247 : ulong key = fd_shredb_key_pack( slot, shred_idx );
1015 1247 : fd_shredb_root_key_t root_key = fd_shredb_root_key( merkle_root->uc, shred_idx );
1016 :
1017 1247 : disk_view_t view;
1018 1247 : fd_rwlock_write( &store->disk_lock );
1019 1247 : int result = disk_insert_prepare_write_locked( store, key, &root_key, &view );
1020 1247 : fd_rwlock_unwrite( &store->disk_lock );
1021 1247 : if( FD_UNLIKELY( result!=FD_STORE_DISK_INSERT_SUCCESS ) ) return result;
1022 :
1023 : /* The map entries remain WRITING while the syscall runs unlocked. */
1024 1223 : fd_shredb_entry_t wr_entry[1];
1025 1223 : ulong shred_sz = fd_ulong_min( fd_shred_sz( shred ), FD_SHRED_MAX_SZ );
1026 1223 : ulong tail_off = __builtin_offsetof( fd_shredb_entry_t, shred )+shred_sz;
1027 1223 : wr_entry->tag = disk_cell_tag( view.ticket, FD_SHREDB_CELL_READY );
1028 1223 : wr_entry->key = key;
1029 1223 : wr_entry->shred_sz = (ushort)shred_sz;
1030 1223 : memcpy( wr_entry->root, merkle_root->uc, FD_SHRED_MERKLE_NODE_SZ );
1031 1223 : fd_memcpy( wr_entry->shred, shred, shred_sz );
1032 1223 : fd_memset( (uchar *)wr_entry+tail_off, 0, sizeof(wr_entry)-tail_off );
1033 :
1034 1223 : off_t off = (off_t)(store->wire_off + (ulong)view.ring_idx*sizeof(fd_shredb_entry_t));
1035 1223 : 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 1223 : atomic_fetch_add_explicit( &store->disk_write_bytes, sizeof(wr_entry), memory_order_relaxed );
1038 :
1039 1223 : fd_rwlock_write( &store->disk_lock );
1040 1223 : result = disk_insert_finish_write_locked( store, view );
1041 1223 : fd_rwlock_unwrite( &store->disk_lock );
1042 1223 : if( FD_UNLIKELY( result!=FD_STORE_DISK_INSERT_SUCCESS ) ) return result;
1043 :
1044 1223 : atomic_fetch_add_explicit( &store->disk_insert_cnt, 1UL, memory_order_relaxed );
1045 1223 : return FD_STORE_DISK_INSERT_SUCCESS;
1046 1223 : }
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 435 : disk_view_t * view ) {
1054 435 : uint ring_idx = root_key ? disk_root_lookup_locked( store, root_key ) : disk_slot_lookup_locked( store, key );
1055 435 : if( FD_UNLIKELY( ring_idx==UINT_MAX ) ) return FD_STORE_DISK_QUERY_MISS;
1056 :
1057 351 : fd_shredb_ring_entry_t const * cell = disk_ring_pool_laddr( store ) + ring_idx;
1058 351 : ulong tag = disk_cell_tag_load( cell );
1059 351 : 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 351 : if( FD_UNLIKELY( root_key && fd_shredb_key_shred_idx( cell->key )!=shred_idx ) ) {
1064 12 : return FD_STORE_DISK_QUERY_MISS;
1065 12 : }
1066 :
1067 339 : view->ticket = disk_cell_ticket( tag );
1068 339 : view->ring_idx = ring_idx;
1069 339 : view->linked = 0;
1070 339 : return 0;
1071 351 : }
1072 :
1073 : static int
1074 : disk_query_finish_read_locked( fd_store_t const * store,
1075 339 : disk_view_t view ) {
1076 339 : fd_shredb_ring_entry_t const * cell = disk_ring_pool_laddr( store ) + view.ring_idx;
1077 339 : if( FD_UNLIKELY( disk_cell_tag_load( cell )!=disk_cell_tag( view.ticket, FD_SHREDB_CELL_READY ) ) ) return FD_STORE_DISK_QUERY_BUSY;
1078 333 : return 0;
1079 339 : }
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 429 : fd_shredb_entry_t * rd_entry ) {
1091 : /* need mutable rwlock to read, even under a const ptr */
1092 429 : fd_rwlock_t * rwlock = (fd_rwlock_t *)fd_type_pun_const( &store->disk_lock );
1093 :
1094 429 : disk_view_t view;
1095 429 : fd_rwlock_read( rwlock );
1096 429 : int result = disk_query_prepare_read_locked( store, key, root_key, shred_idx, &view );
1097 429 : fd_rwlock_unread( rwlock );
1098 429 : if( FD_UNLIKELY( result ) ) return result;
1099 :
1100 333 : off_t off = (off_t)(store->wire_off + (ulong)view.ring_idx*sizeof(fd_shredb_entry_t));
1101 333 : 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 333 : fd_rwlock_read( rwlock );
1107 333 : result = disk_query_finish_read_locked( store, view );
1108 333 : fd_rwlock_unread( rwlock );
1109 333 : if( FD_UNLIKELY( result ) ) return result;
1110 :
1111 333 : ulong tag = disk_cell_tag( view.ticket, FD_SHREDB_CELL_READY );
1112 333 : if( root_key ) {
1113 54 : if( FD_UNLIKELY( rd_entry->tag!=tag ||
1114 54 : memcmp( rd_entry->root, root_key->root, FD_SHRED_MERKLE_NODE_SZ ) ||
1115 54 : fd_shredb_key_shred_idx( rd_entry->key )%FD_FEC_SHRED_CNT!=root_key->off ) ) return FD_STORE_DISK_QUERY_BUSY;
1116 54 : if( FD_UNLIKELY( fd_shredb_key_shred_idx( rd_entry->key )!=shred_idx ) ) return FD_STORE_DISK_QUERY_MISS;
1117 279 : } else {
1118 279 : if( FD_UNLIKELY( rd_entry->tag!=tag || rd_entry->key!=key ) ) return FD_STORE_DISK_QUERY_BUSY;
1119 279 : }
1120 333 : return (int)fd_ulong_min( rd_entry->shred_sz, FD_SHRED_MAX_SZ );
1121 333 : }
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 261 : uchar out[ FD_SHRED_MAX_SZ ] ) {
1129 261 : if( FD_UNLIKELY( !store || disk_fd<0 || !out || !fd_store_has_disk( store ) ||
1130 261 : slot>=FD_SHREDB_KEY_SLOT_MAX || shred_idx>=store->max_shreds_per_block ) ) return FD_STORE_DISK_QUERY_MISS;
1131 249 : ulong key = fd_shredb_key_pack( slot, shred_idx );
1132 249 : for( ulong retry=0UL; retry<FD_STORE_DISK_READ_RETRY_CNT; retry++ ) {
1133 249 : fd_shredb_entry_t entry[1];
1134 249 : int result = disk_read_once( store, disk_fd, key, NULL, shred_idx, entry );
1135 249 : if( FD_UNLIKELY( result==FD_STORE_DISK_QUERY_BUSY ) ) { FD_SPIN_PAUSE(); continue; }
1136 249 : if( FD_UNLIKELY( result<0 ) ) return result;
1137 213 : fd_memcpy( out, entry->shred, (ulong)result );
1138 213 : return result;
1139 249 : }
1140 0 : return FD_STORE_DISK_QUERY_BUSY;
1141 249 : }
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 96 : uchar out[ FD_SHRED_MAX_SZ ] ) {
1149 96 : if( FD_UNLIKELY( !store || disk_fd<0 || !merkle_root || !out || !fd_store_has_disk( store ) ||
1150 96 : shred_idx>=store->max_shreds_per_block ) ) return FD_STORE_DISK_QUERY_MISS;
1151 90 : fd_shredb_root_key_t rk = fd_shredb_root_key( merkle_root, shred_idx );
1152 90 : for( ulong retry=0UL; retry<FD_STORE_DISK_READ_RETRY_CNT; retry++ ) {
1153 90 : fd_shredb_entry_t entry[1];
1154 90 : int result = disk_read_once( store, disk_fd, 0UL, &rk, shred_idx, entry );
1155 90 : if( FD_UNLIKELY( result==FD_STORE_DISK_QUERY_BUSY ) ) { FD_SPIN_PAUSE(); continue; }
1156 90 : if( FD_UNLIKELY( result<0 ) ) return result;
1157 54 : fd_memcpy( out, entry->shred, (ulong)result );
1158 54 : return result;
1159 90 : }
1160 0 : return FD_STORE_DISK_QUERY_BUSY;
1161 90 : }
1162 :
1163 : int
1164 : fd_store_disk_probe( fd_store_t const * store,
1165 : ulong slot,
1166 96 : uint shred_idx ) {
1167 96 : if( FD_UNLIKELY( !store || !fd_store_has_disk( store ) || slot>=FD_SHREDB_KEY_SLOT_MAX ) ) return 0;
1168 84 : if( FD_UNLIKELY( shred_idx==UINT_MAX ) ) {
1169 36 : return !!(atomic_load_explicit( disk_slot_hint_laddr( store ) + (slot % store->disk_max_slots), memory_order_acquire ) & FD_SHREDB_HINT_VALID);
1170 36 : }
1171 48 : if( FD_UNLIKELY( shred_idx>=store->max_shreds_per_block ) ) return 0;
1172 :
1173 42 : fd_rwlock_t * rwlock = (fd_rwlock_t *)fd_type_pun_const( &store->disk_lock );
1174 42 : fd_rwlock_read( rwlock );
1175 42 : int present = disk_slot_lookup_locked( store, fd_shredb_key_pack( slot, shred_idx ) )!=UINT_MAX;
1176 42 : fd_rwlock_unread( rwlock );
1177 42 : return present;
1178 48 : }
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 108 : uchar out[ FD_SHRED_MAX_SZ ] ) {
1186 108 : 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 108 : for( ulong retry=0UL; retry<FD_STORE_DISK_READ_RETRY_CNT; retry++ ) {
1189 108 : atomic_ulong const * hint_ptr = disk_slot_hint_laddr( store ) + (slot % store->disk_max_slots);
1190 108 : ulong hint = atomic_load_explicit( hint_ptr, memory_order_relaxed );
1191 108 : if( FD_UNLIKELY( !(hint & FD_SHREDB_HINT_VALID) ) ) return FD_STORE_DISK_QUERY_MISS;
1192 108 : if( FD_UNLIKELY( fd_shredb_key_slot( hint )!=slot ) ) return FD_STORE_DISK_QUERY_BUSY;
1193 90 : uint idx = fd_shredb_key_shred_idx( hint & ~FD_SHREDB_HINT_VALID );
1194 90 : if( FD_UNLIKELY( idx>=store->max_shreds_per_block ) ) return FD_STORE_DISK_QUERY_MISS;
1195 :
1196 90 : fd_shredb_entry_t entry[1];
1197 90 : int result = disk_read_once( store, disk_fd, fd_shredb_key_pack( slot, idx ), NULL, idx, entry );
1198 90 : ulong hint_after = atomic_load_explicit( hint_ptr, memory_order_acquire );
1199 90 : if( FD_UNLIKELY( hint_after!=hint || result==FD_STORE_DISK_QUERY_BUSY ) ) {
1200 0 : FD_SPIN_PAUSE();
1201 0 : continue;
1202 0 : }
1203 90 : if( FD_UNLIKELY( result==FD_STORE_DISK_QUERY_MISS ) ) return FD_STORE_DISK_QUERY_SCAN_LIMIT;
1204 66 : if( FD_UNLIKELY( result<0 ) ) return result;
1205 :
1206 66 : fd_shred_t const * shred = (fd_shred_t const *)fd_type_pun_const( entry->shred );
1207 66 : if( idx<min_shred_idx && !(shred->data.flags & FD_SHRED_DATA_FLAG_SLOT_COMPLETE) ) return FD_STORE_DISK_QUERY_MISS;
1208 60 : fd_memcpy( out, entry->shred, (ulong)result );
1209 60 : return result;
1210 66 : }
1211 0 : return FD_STORE_DISK_QUERY_BUSY;
1212 108 : }
1213 :
1214 : int
1215 : fd_store_disk_stats_query( fd_store_t const * store,
1216 6 : fd_store_disk_stats_t * stats ) {
1217 6 : if( FD_UNLIKELY( !store || !stats || !fd_store_has_disk( store ) ) ) return FD_STORE_DISK_QUERY_MISS;
1218 6 : stats->shred_cnt = atomic_load_explicit( &store->disk_cnt, memory_order_relaxed );
1219 6 : stats->current_bytes = stats->shred_cnt*sizeof(fd_shredb_entry_t)
1220 6 : + atomic_load_explicit( &store->spill_live_cnt, memory_order_relaxed )*store->payload_slot_sz;
1221 6 : stats->allocated_bytes = fd_ulong_min( atomic_load_explicit( &store->disk_reservation_head, memory_order_relaxed ),
1222 6 : store->disk_max_shreds )*sizeof(fd_shredb_entry_t)
1223 6 : + FD_VOLATILE_CONST( store->spill_slot_cnt )*store->payload_slot_sz;
1224 6 : stats->insert_cnt = atomic_load_explicit( &store->disk_insert_cnt, memory_order_relaxed );
1225 : stats->write_bytes = atomic_load_explicit( &store->disk_write_bytes, memory_order_relaxed );
1226 6 : return 0;
1227 6 : }
|
