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