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 : #include <linux/falloc.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 1160 : off_t off ) {
22 1160 : ulong written = 0UL;
23 2321 : while( written<sz ) {
24 1159 : long res = pwrite( fd, (uchar const *)buf + written, sz-written, off+(off_t)written );
25 1161 : if( FD_LIKELY( res>0L ) ) { written += (ulong)res; continue; }
26 >1844*10^16 : if( FD_UNLIKELY( res<0L && errno==EINTR ) ) continue;
27 >1844*10^16 : if( FD_UNLIKELY( !res ) ) errno = EIO;
28 >1844*10^16 : return -1;
29 >1844*10^16 : }
30 1162 : return 0;
31 1160 : }
32 :
33 : static int
34 : store_pread_all( int fd,
35 : void * buf,
36 : ulong sz,
37 240 : off_t off ) {
38 240 : ulong read_sz = 0UL;
39 480 : while( read_sz<sz ) {
40 240 : long res = pread( fd, (uchar *)buf + read_sz, sz-read_sz, off+(off_t)read_sz );
41 240 : 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 240 : return 0;
47 240 : }
48 :
49 : static inline fd_shredb_shred_entry_t *
50 2688 : disk_shred_pool_laddr( fd_store_t const * store ) {
51 2688 : return fd_wksp_laddr_fast( fd_store_wksp( store ), store->shred_pool_gaddr );
52 2688 : }
53 :
54 : static inline fd_shredb_shred_map_t *
55 : disk_shred_map_ljoin( fd_store_t const * store,
56 1362 : fd_shredb_shred_map_t * join ) {
57 1362 : fd_wksp_t * wksp = fd_store_wksp( store );
58 1362 : return fd_shredb_shred_map_join( join,
59 1362 : fd_wksp_laddr_fast( wksp, store->shred_map_gaddr ),
60 1362 : disk_shred_pool_laddr( store ),
61 1362 : store->disk_max_shreds );
62 1362 : }
63 :
64 : static inline atomic_ulong *
65 1191 : disk_slot_hint_laddr( fd_store_t const * store ) {
66 1191 : return fd_wksp_laddr_fast( fd_store_wksp( store ), store->slot_hint_gaddr );
67 1191 : }
68 :
69 : static inline uchar *
70 186 : cache_data_laddr( fd_store_t const * store ) {
71 186 : return fd_wksp_laddr_fast( fd_store_wksp( store ), store->cache_data_gaddr );
72 186 : }
73 :
74 : static inline ulong *
75 180 : cache_free_laddr( fd_store_t const * store ) {
76 180 : return fd_wksp_laddr_fast( fd_store_wksp( store ), store->cache_free_gaddr );
77 180 : }
78 :
79 : static inline uint *
80 18 : spill_free_laddr( fd_store_t const * store ) {
81 18 : return fd_wksp_laddr_fast( fd_store_wksp( store ), store->spill_free_gaddr );
82 18 : }
83 :
84 : static inline uint *
85 42 : spill_reclaim_laddr( fd_store_t const * store ) {
86 42 : return fd_wksp_laddr_fast( fd_store_wksp( store ), store->spill_reclaim_gaddr );
87 42 : }
88 :
89 : static inline uchar *
90 15 : spill_read_data_laddr( fd_store_t const * store ) {
91 15 : return fd_wksp_laddr_fast( fd_store_wksp( store ), store->spill_read_data_gaddr );
92 15 : }
93 :
94 : static inline fd_store_fec_t *
95 579 : pool_ele_laddr( fd_store_t const * store ) {
96 579 : return fd_wksp_laddr_fast( fd_store_wksp( store ), store->pool_ele_gaddr );
97 579 : }
98 :
99 : static inline fd_store_pool_t
100 832 : pool_ljoin( fd_store_t const * store ) {
101 832 : return (fd_store_pool_t){
102 832 : .pool = fd_wksp_laddr_fast( fd_store_wksp( store ), store->pool_mem_gaddr ),
103 832 : .ele = fd_wksp_laddr_fast( fd_store_wksp( store ), store->pool_ele_gaddr ),
104 832 : .ele_max = store->fec_max
105 832 : };
106 832 : }
107 :
108 : static fd_store_fec_t *
109 787 : fd_store_fec_acquire( fd_store_t * store ) {
110 787 : fd_store_pool_t pool = pool_ljoin( store );
111 787 : fd_store_fec_t * fec = fd_store_pool_acquire( &pool );
112 840 : if( FD_LIKELY( fec ) ) {
113 840 : fec->data_sz = 0UL;
114 840 : fec->data_off = 0UL;
115 840 : fec->cache_prev = UINT_MAX;
116 840 : fec->cache_next = UINT_MAX;
117 840 : fec->data_pin_cnt = 0U;
118 840 : fec->data_state = FD_STORE_FEC_DATA_EMPTY;
119 840 : fec->data_consume_pending = 0U;
120 840 : }
121 787 : return fec;
122 787 : }
123 :
124 : static void
125 : cache_slot_release_locked( fd_store_t * store,
126 60 : ulong data_off ) {
127 60 : FD_TEST( data_off<store->cache_slot_cnt*store->payload_slot_sz );
128 60 : FD_TEST( !(data_off % store->payload_slot_sz) );
129 60 : FD_TEST( store->cache_free_cnt<store->cache_slot_cnt );
130 60 : cache_free_laddr( store )[ store->cache_free_cnt++ ] = data_off / store->payload_slot_sz;
131 60 : }
132 :
133 : static inline uint
134 : cache_fec_idx( fd_store_t const * store,
135 252 : fd_store_fec_t const * fec ) {
136 252 : ulong off = (ulong)fec - (ulong)pool_ele_laddr( store );
137 252 : FD_TEST( !(off % sizeof(fd_store_fec_t)) );
138 252 : ulong idx = off / sizeof(fd_store_fec_t);
139 252 : FD_TEST( idx<store->fec_max );
140 252 : return (uint)idx;
141 252 : }
142 :
143 : static void
144 : cache_lru_remove_locked( fd_store_t * store,
145 78 : fd_store_fec_t * fec ) {
146 78 : uint idx = cache_fec_idx( store, fec );
147 78 : uint prev = fec->cache_prev;
148 78 : uint next = fec->cache_next;
149 :
150 78 : if( prev==UINT_MAX ) { FD_TEST( store->cache_lru_head==idx ); store->cache_lru_head = next; }
151 6 : else pool_ele_laddr( store )[ prev ].cache_next = next;
152 78 : if( next==UINT_MAX ) { FD_TEST( store->cache_lru_tail==idx ); store->cache_lru_tail = prev; }
153 42 : else pool_ele_laddr( store )[ next ].cache_prev = prev;
154 :
155 78 : fec->cache_prev = UINT_MAX;
156 78 : fec->cache_next = UINT_MAX;
157 78 : }
158 :
159 : static void
160 : cache_lru_push_head_locked( fd_store_t * store,
161 18 : fd_store_fec_t * fec ) {
162 18 : uint idx = cache_fec_idx( store, fec );
163 18 : FD_TEST( fec->cache_prev==UINT_MAX && fec->cache_next==UINT_MAX );
164 :
165 18 : fec->cache_next = store->cache_lru_head;
166 18 : if( store->cache_lru_head==UINT_MAX ) store->cache_lru_tail = idx;
167 18 : else pool_ele_laddr( store )[ store->cache_lru_head ].cache_prev = idx;
168 18 : store->cache_lru_head = idx;
169 18 : }
170 :
171 : static void
172 : cache_lru_push_tail_locked( fd_store_t * store,
173 114 : fd_store_fec_t * fec ) {
174 114 : uint idx = cache_fec_idx( store, fec );
175 114 : FD_TEST( fec->cache_prev==UINT_MAX && fec->cache_next==UINT_MAX );
176 :
177 114 : fec->cache_prev = store->cache_lru_tail;
178 114 : if( store->cache_lru_tail==UINT_MAX ) store->cache_lru_head = idx;
179 54 : else pool_ele_laddr( store )[ store->cache_lru_tail ].cache_next = idx;
180 114 : store->cache_lru_tail = idx;
181 114 : }
182 :
183 : static void
184 : spill_reclaim_push_locked( fd_store_t * store,
185 24 : uint spill_slot ) {
186 24 : FD_TEST( spill_slot<store->spill_slot_cnt );
187 24 : FD_TEST( store->spill_reclaim_cnt+store->spill_reclaiming_cnt+store->spill_reuse_cnt<store->fec_max );
188 24 : spill_reclaim_laddr( store )[ store->spill_reclaim_cnt++ ] = spill_slot;
189 24 : }
190 :
191 : static void
192 : spill_reuse_push_locked( fd_store_t * store,
193 0 : uint spill_slot ) {
194 0 : FD_TEST( spill_slot<store->spill_slot_cnt );
195 0 : FD_TEST( store->spill_reclaim_cnt+store->spill_reclaiming_cnt+store->spill_reuse_cnt<store->fec_max );
196 0 : spill_reclaim_laddr( store )[ store->fec_max-1UL-store->spill_reuse_cnt++ ] = spill_slot;
197 0 : }
198 :
199 : static uint
200 0 : spill_reuse_pop_locked( fd_store_t * store ) {
201 0 : FD_TEST( store->spill_reuse_cnt );
202 0 : return spill_reclaim_laddr( store )[ store->fec_max-store->spill_reuse_cnt-- ];
203 0 : }
204 :
205 : static void
206 : cache_consume_locked( fd_store_t * store,
207 96 : fd_store_fec_t * fec ) {
208 96 : if( FD_LIKELY( !fec->data_consume_pending ) ) return;
209 66 : if( FD_UNLIKELY( fec->data_pin_cnt || fec->data_state==FD_STORE_FEC_DATA_SPILLING ) ) return;
210 :
211 45 : if( FD_LIKELY( fec->data_state==FD_STORE_FEC_DATA_RAM_WRITING ||
212 45 : fec->data_state==FD_STORE_FEC_DATA_RAM_READY ) ) {
213 12 : if( fec->data_state==FD_STORE_FEC_DATA_RAM_READY ) cache_lru_remove_locked( store, fec );
214 12 : cache_slot_release_locked( store, fec->data_off );
215 33 : } else if( fec->data_state==FD_STORE_FEC_DATA_DISK ) {
216 24 : FD_TEST( fd_ulong_is_aligned( fec->data_off, store->payload_slot_sz ) );
217 24 : ulong spill_slot = fec->data_off / store->payload_slot_sz;
218 24 : FD_TEST( spill_slot<store->spill_slot_cnt && atomic_load_explicit( &store->spill_live_cnt, memory_order_relaxed ) );
219 24 : atomic_fetch_sub_explicit( &store->spill_live_cnt, 1UL, memory_order_relaxed );
220 24 : spill_reclaim_push_locked( store, (uint)spill_slot );
221 24 : }
222 :
223 45 : fec->data_off = 0UL;
224 45 : fec->cache_prev = UINT_MAX;
225 45 : fec->cache_next = UINT_MAX;
226 45 : fec->data_state = FD_STORE_FEC_DATA_CONSUMED;
227 45 : fec->data_consume_pending = 0U;
228 45 : }
229 :
230 : static void
231 : fd_store_fec_release( fd_store_t * store,
232 42 : fd_store_fec_t * fec ) {
233 42 : (void)cache_fec_idx( store, fec );
234 63 : for(;;) {
235 63 : fd_rwlock_write( &store->cache_lock );
236 63 : fec->data_consume_pending = 1U;
237 63 : cache_consume_locked( store, fec );
238 63 : int consumed = fec->data_state==FD_STORE_FEC_DATA_CONSUMED;
239 63 : fd_rwlock_unwrite( &store->cache_lock );
240 63 : if( FD_LIKELY( consumed ) ) break;
241 21 : FD_SPIN_PAUSE();
242 21 : }
243 42 : fd_store_pool_t pool = pool_ljoin( store );
244 42 : fd_store_pool_release( &pool, fec );
245 42 : }
246 :
247 : static int
248 : spill_one_locked( fd_store_t * store,
249 : int disk_fd,
250 48 : fd_store_fec_spill_stats_t * spill ) {
251 48 : if( FD_UNLIKELY( disk_fd<0 ) ) return FD_STORE_SPILL_NONE;
252 :
253 48 : fd_store_fec_t * fec0 = pool_ele_laddr( store );
254 48 : uint victim_idx = store->cache_lru_head;
255 48 : if( FD_UNLIKELY( victim_idx==UINT_MAX ) ) return FD_STORE_SPILL_NONE;
256 48 : fd_store_fec_t * victim = fec0 + victim_idx;
257 48 : FD_TEST( victim->data_state==FD_STORE_FEC_DATA_RAM_READY && !victim->data_pin_cnt );
258 48 : FD_TEST( victim->data_sz<=store->fec_data_max );
259 :
260 48 : uint spill_slot;
261 48 : int spill_slot_allocated;
262 48 : if( FD_LIKELY( store->spill_reuse_cnt ) ) {
263 0 : spill_slot = spill_reuse_pop_locked( store );
264 0 : spill_slot_allocated = 1;
265 48 : } else if( FD_LIKELY( store->spill_reclaim_cnt ) ) {
266 6 : spill_slot = spill_reclaim_laddr( store )[ --store->spill_reclaim_cnt ];
267 6 : spill_slot_allocated = 1;
268 42 : } else if( FD_LIKELY( store->spill_free_cnt ) ) {
269 6 : spill_slot = spill_free_laddr( store )[ --store->spill_free_cnt ];
270 6 : spill_slot_allocated = 0;
271 36 : } else {
272 36 : if( FD_UNLIKELY( store->spill_slot_cnt>=store->fec_max ) )
273 0 : return store->spill_reclaiming_cnt ? FD_STORE_SPILL_RETRY : FD_STORE_SPILL_NONE;
274 36 : spill_slot = (uint)store->spill_slot_cnt++;
275 36 : spill_slot_allocated = 0;
276 36 : }
277 :
278 48 : ulong ram_off = victim->data_off;
279 48 : ulong spill_off = (ulong)spill_slot * store->payload_slot_sz;
280 48 : ulong data_sz = victim->data_sz;
281 :
282 48 : cache_lru_remove_locked( store, victim );
283 48 : victim->data_state = FD_STORE_FEC_DATA_SPILLING;
284 48 : long evict_ticks = -fd_tickcount();
285 48 : fd_rwlock_unwrite( &store->cache_lock );
286 :
287 48 : if( FD_UNLIKELY( store_pwrite_all( disk_fd, cache_data_laddr( store ) + ram_off, data_sz, (off_t)spill_off ) ) )
288 0 : FD_LOG_ERR(( "error spilling FEC payload to disk: (%d-%s)", errno, fd_io_strerror( errno ) ));
289 :
290 48 : fd_rwlock_write( &store->cache_lock );
291 48 : FD_TEST( victim->data_state==FD_STORE_FEC_DATA_SPILLING );
292 48 : int result;
293 48 : if( FD_UNLIKELY( victim->data_pin_cnt || victim->data_consume_pending ) ) {
294 0 : int try_next = !!victim->data_pin_cnt;
295 0 : atomic_fetch_add_explicit( &store->spill_allocated_cnt, (ulong)!spill_slot_allocated, memory_order_relaxed );
296 0 : spill_reclaim_push_locked( store, spill_slot );
297 0 : victim->data_state = victim->data_consume_pending && !victim->data_pin_cnt
298 0 : ? FD_STORE_FEC_DATA_RAM_WRITING
299 0 : : FD_STORE_FEC_DATA_RAM_READY;
300 0 : if( FD_LIKELY( !victim->data_pin_cnt && !victim->data_consume_pending ) )
301 0 : cache_lru_push_tail_locked( store, victim );
302 0 : cache_consume_locked( store, victim );
303 0 : result = try_next ? FD_STORE_SPILL_RETRY :
304 0 : store->cache_free_cnt ? FD_STORE_SPILL_COMPLETE : FD_STORE_SPILL_NONE;
305 48 : } else {
306 48 : victim->data_off = spill_off;
307 48 : victim->data_state = FD_STORE_FEC_DATA_DISK;
308 48 : atomic_fetch_add_explicit( &store->spill_live_cnt, 1UL, memory_order_relaxed );
309 48 : atomic_fetch_add_explicit( &store->spill_allocated_cnt, (ulong)!spill_slot_allocated, memory_order_relaxed );
310 48 : atomic_fetch_add_explicit( &store->fec_spill_cnt, 1UL, memory_order_relaxed );
311 48 : atomic_fetch_add_explicit( &store->fec_spill_bytes, data_sz, memory_order_relaxed );
312 48 : cache_slot_release_locked( store, ram_off );
313 48 : result = FD_STORE_SPILL_COMPLETE;
314 48 : }
315 :
316 48 : evict_ticks += fd_tickcount();
317 48 : if( FD_LIKELY( spill ) ) {
318 12 : spill->write_cnt++;
319 12 : spill->write_bytes += data_sz;
320 12 : spill->write_ticks += (ulong)evict_ticks;
321 12 : }
322 48 : return result;
323 48 : }
324 :
325 :
326 : void *
327 : fd_store_new( void * shmem,
328 : ulong fec_max,
329 : ulong fec_data_max,
330 : ulong shred_storage_gib,
331 : ulong shred_cache_bytes,
332 : ulong fec_set_cnt,
333 : ulong max_shreds_per_block,
334 120 : ulong seed ) {
335 :
336 120 : if( FD_UNLIKELY( !shmem ) ) { FD_LOG_WARNING(( "NULL shmem" )); return NULL; }
337 120 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shmem, fd_store_align() ) ) ) { FD_LOG_WARNING(( "misaligned shmem" )); return NULL; }
338 120 : if( FD_UNLIKELY( !fec_max ) ) { FD_LOG_WARNING(( "fec_max must be non-zero" )); return NULL; }
339 120 : if( FD_UNLIKELY( fec_max>UINT_MAX ) ) { FD_LOG_WARNING(( "fec_max must fit in uint" )); return NULL; }
340 120 : if( FD_UNLIKELY( !fec_data_max ) ) { FD_LOG_WARNING(( "fec_data_max must be non-zero" )); return NULL; }
341 120 : 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; }
342 108 : if( FD_UNLIKELY( shred_storage_gib>FD_SHREDB_MAX_SIZE_GIB ) ) {
343 0 : FD_LOG_ERR(( "shred database size limit is %lu GiB, but the maximum supported size is %lu GiB",
344 0 : shred_storage_gib, FD_SHREDB_MAX_SIZE_GIB ));
345 0 : }
346 :
347 108 : ulong footprint = fd_store_footprint( fec_max, fec_data_max, shred_storage_gib, shred_cache_bytes, fec_set_cnt );
348 108 : if( FD_UNLIKELY( !footprint ) ) { FD_LOG_WARNING(( "invalid or overflowing store footprint" )); return NULL; }
349 :
350 108 : fd_wksp_t * wksp = fd_wksp_containing( shmem );
351 108 : if( FD_UNLIKELY( !wksp ) ) { FD_LOG_WARNING(( "shmem must be part of a workspace" )); return NULL; }
352 :
353 108 : ulong chain_cnt = fd_store_map_chain_cnt_est( fec_max );
354 108 : ulong payload_slot_sz = fd_store_payload_slot_sz( fec_data_max );
355 108 : ulong cache_slot_cnt = shred_cache_bytes
356 108 : ? fd_ulong_min( fec_max, fd_ulong_max( 1UL, shred_cache_bytes / payload_slot_sz ) )
357 108 : : fec_max;
358 :
359 108 : FD_SCRATCH_ALLOC_INIT( l, shmem );
360 108 : fd_store_t * store = FD_SCRATCH_ALLOC_APPEND( l, fd_store_align(), sizeof(fd_store_t) );
361 108 : void * map = FD_SCRATCH_ALLOC_APPEND( l, fd_store_map_align(), fd_store_map_footprint( chain_cnt ) );
362 108 : void * shpool = FD_SCRATCH_ALLOC_APPEND( l, fd_store_pool_align(), fd_store_pool_footprint() );
363 108 : void * shele = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_store_fec_t), sizeof(fd_store_fec_t)*fec_max );
364 108 : uchar * cache_mem = FD_SCRATCH_ALLOC_APPEND( l, FD_STORE_PAYLOAD_PAGE_SZ, payload_slot_sz*cache_slot_cnt );
365 108 : ulong * cache_free = FD_SCRATCH_ALLOC_APPEND( l, alignof(ulong), sizeof(ulong)*cache_slot_cnt );
366 108 : uint * spill_free = FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), sizeof(uint)*fec_max );
367 108 : uint * spill_reclaim = FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), sizeof(uint)*fec_max );
368 108 : uchar * spill_read_mem = FD_SCRATCH_ALLOC_APPEND( l, FD_STORE_PAYLOAD_PAGE_SZ, payload_slot_sz );
369 108 : fd_fec_set_t * fec_sets = fec_set_cnt
370 108 : ? FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_fec_set_t), sizeof(fd_fec_set_t)*fec_set_cnt )
371 108 : : NULL;
372 :
373 108 : void * shred_map_mem = NULL;
374 108 : void * shred_pool_mem = NULL;
375 108 : atomic_ulong * slot_hint_mem = NULL;
376 108 : ulong max_shreds = 0UL;
377 108 : ulong max_slots = 0UL;
378 108 : ulong disk_chain_cnt = 0UL;
379 :
380 108 : if( shred_storage_gib ) {
381 48 : max_shreds = fd_shredb_max_shreds( shred_storage_gib );
382 48 : max_slots = fd_shredb_max_slots( shred_storage_gib );
383 48 : disk_chain_cnt = fd_shredb_shred_map_chain_cnt_est( max_shreds );
384 :
385 48 : shred_map_mem = FD_SCRATCH_ALLOC_APPEND( l, fd_shredb_shred_map_align(), fd_shredb_shred_map_footprint( disk_chain_cnt ) );
386 48 : shred_pool_mem = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_shredb_shred_entry_t), max_shreds * sizeof(fd_shredb_shred_entry_t) );
387 48 : slot_hint_mem = FD_SCRATCH_ALLOC_APPEND( l, alignof(atomic_ulong), max_slots * sizeof(atomic_ulong) );
388 48 : }
389 :
390 108 : FD_TEST( FD_SCRATCH_ALLOC_FINI( l, fd_store_align() )==(ulong)shmem + footprint );
391 :
392 108 : ulong payload_sz;
393 108 : ulong wire_sz;
394 108 : ulong file_sz;
395 108 : if( FD_UNLIKELY( __builtin_umull_overflow( payload_slot_sz, fec_max, &payload_sz ) ||
396 108 : __builtin_umull_overflow( max_shreds, sizeof(fd_shredb_entry_t), &wire_sz ) ||
397 108 : __builtin_uaddl_overflow( payload_sz, wire_sz, &file_sz ) ||
398 108 : file_sz>(ulong)LONG_MAX ) ) {
399 0 : FD_LOG_WARNING(( "store backing file size overflows off_t" ));
400 0 : return NULL;
401 0 : }
402 :
403 108 : fd_memset( store, 0, sizeof(fd_store_t) );
404 108 : store->fec_max = fec_max;
405 108 : store->fec_data_max = fec_data_max;
406 108 : store->store_gaddr = fd_wksp_gaddr_fast( wksp, store );
407 108 : store->pool_mem_gaddr = fd_wksp_gaddr_fast( wksp, shpool );
408 108 : store->pool_ele_gaddr = fd_wksp_gaddr_fast( wksp, shele );
409 108 : store->payload_slot_sz = payload_slot_sz;
410 108 : store->payload_sz = payload_sz;
411 108 : store->wire_off = payload_sz; /* wire region begins after the payload region (page-aligned) */
412 108 : store->cache_slot_cnt = cache_slot_cnt;
413 108 : store->cache_data_gaddr = fd_wksp_gaddr_fast( wksp, cache_mem );
414 108 : store->cache_free_gaddr = fd_wksp_gaddr_fast( wksp, cache_free );
415 108 : store->cache_free_cnt = cache_slot_cnt;
416 108 : ulong free_cap = fd_ulong_min( 8192UL, (64UL<<20) / payload_slot_sz );
417 108 : ulong burst_floor = fd_ulong_min( 512UL, cache_slot_cnt/2UL );
418 108 : ulong free_target = fd_ulong_max( cache_slot_cnt/10UL, burst_floor );
419 108 : store->cache_free_target = fd_ulong_max( 1UL, fd_ulong_min( free_cap, free_target ) );
420 108 : store->cache_free_low_water = fd_ulong_max( 1UL, (store->cache_free_target*3UL)/4UL );
421 108 : store->cache_lru_head = UINT_MAX;
422 108 : store->cache_lru_tail = UINT_MAX;
423 108 : store->spill_free_gaddr = fd_wksp_gaddr_fast( wksp, spill_free );
424 108 : store->spill_reclaim_gaddr = fd_wksp_gaddr_fast( wksp, spill_reclaim );
425 108 : store->spill_read_data_gaddr = fd_wksp_gaddr_fast( wksp, spill_read_mem );
426 108 : store->fec_set_cnt = fec_set_cnt;
427 108 : store->fec_sets_gaddr = fec_set_cnt ? fd_wksp_gaddr_fast( wksp, fec_sets ) : 0UL;
428 108 : store->max_shreds_per_block = max_shreds_per_block;
429 108 : fd_rwlock_new( &store->cache_lock );
430 108 : fd_rwlock_new( &store->spill_read_lock );
431 108 : fd_rwlock_new( &store->fec_lock );
432 108 : atomic_init( &store->fec_spill_cnt, 0UL );
433 108 : atomic_init( &store->spill_live_cnt, 0UL );
434 108 : atomic_init( &store->spill_allocated_cnt, 0UL );
435 108 : atomic_init( &store->fec_spill_bytes, 0UL );
436 108 : atomic_init( &store->fec_spill_read_cnt, 0UL );
437 108 : atomic_init( &store->fec_spill_read_bytes, 0UL );
438 :
439 108 : void * shmap = fd_store_map_new( map, chain_cnt, seed );
440 108 : if( FD_UNLIKELY( !shmap ) ) { FD_LOG_WARNING(( "fd_store_map_new failed" )); return NULL; }
441 108 : store->map_gaddr = fd_wksp_gaddr_fast( wksp, shmap );
442 :
443 108 : if( FD_UNLIKELY( !fd_store_pool_new( shpool ) ) ) { FD_LOG_WARNING(( "fd_store_pool_new failed" )); return NULL; }
444 108 : fd_store_pool_t pool_ljoin;
445 108 : fd_store_pool_reset( fd_store_pool_join( &pool_ljoin, shpool, shele, fec_max ) );
446 :
447 1818 : for( ulong i=0UL; i<cache_slot_cnt; i++ ) cache_free[ i ] = cache_slot_cnt - 1UL - i;
448 :
449 : /* FEC metadata starts without a payload location. */
450 108 : fd_store_fec_t * fec0 = (fd_store_fec_t *)shele;
451 1866 : for( ulong i=0UL; i<fec_max; i++ ) {
452 1758 : fec0[ i ].data_off = 0UL;
453 1758 : fec0[ i ].cache_prev = UINT_MAX;
454 1758 : fec0[ i ].cache_next = UINT_MAX;
455 1758 : fec0[ i ].data_pin_cnt = 0U;
456 1758 : fec0[ i ].data_state = FD_STORE_FEC_DATA_EMPTY;
457 1758 : fec0[ i ].data_consume_pending = 0U;
458 1758 : }
459 :
460 108 : if( shred_storage_gib ) {
461 48 : void * shred_shmap = fd_shredb_shred_map_new( shred_map_mem, disk_chain_cnt, seed );
462 48 : FD_TEST( shred_shmap );
463 48 : store->shred_map_gaddr = fd_wksp_gaddr_fast( wksp, shred_shmap );
464 48 : store->shred_pool_gaddr = fd_wksp_gaddr_fast( wksp, shred_pool_mem );
465 48 : store->slot_hint_gaddr = fd_wksp_gaddr_fast( wksp, slot_hint_mem );
466 48 : store->disk_max_shreds = max_shreds;
467 48 : store->disk_max_slots = max_slots;
468 :
469 48 : fd_shredb_shred_entry_t * cell = (fd_shredb_shred_entry_t *)shred_pool_mem;
470 40265328 : for( ulong i=0UL; i<max_shreds; i++ ) {
471 40265280 : cell[ i ].key = 0UL;
472 40265280 : cell[ i ].next = UINT_MAX;
473 40265280 : atomic_init( &cell[ i ].tag, 0UL );
474 40265280 : }
475 1258320 : for( ulong i=0UL; i<max_slots; i++ ) {
476 1258272 : atomic_init( slot_hint_mem+i, 0UL );
477 1258272 : }
478 :
479 48 : atomic_init( &store->disk_reservation_head, 0UL );
480 48 : atomic_init( &store->disk_cnt, 0UL );
481 48 : atomic_init( &store->disk_insert_cnt, 0UL );
482 48 : atomic_init( &store->disk_write_bytes, 0UL );
483 48 : }
484 :
485 108 : FD_COMPILER_MFENCE();
486 108 : FD_VOLATILE( store->magic ) = FD_STORE_MAGIC;
487 108 : FD_COMPILER_MFENCE();
488 :
489 108 : return shmem;
490 108 : }
491 :
492 : fd_store_t *
493 108 : fd_store_join( void * shstore ) {
494 108 : if( FD_UNLIKELY( !shstore ) ) { FD_LOG_WARNING(( "NULL store" )); return NULL; }
495 108 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shstore, fd_store_align() ) ) ) { FD_LOG_WARNING(( "misaligned store" )); return NULL; }
496 :
497 108 : fd_wksp_t * wksp = fd_wksp_containing( shstore );
498 108 : if( FD_UNLIKELY( !wksp ) ) { FD_LOG_WARNING(( "store must be part of a workspace" )); return NULL; }
499 :
500 108 : fd_store_t * store = (fd_store_t *)shstore;
501 108 : if( FD_UNLIKELY( store->magic!=FD_STORE_MAGIC ) ) { FD_LOG_WARNING(( "bad magic" )); return NULL; }
502 :
503 108 : return store;
504 108 : }
505 :
506 : int
507 : fd_store_file_create( char const * path,
508 : ulong wire_off,
509 72 : ulong disk_max_shreds ) {
510 72 : if( FD_UNLIKELY( !path ) ) {
511 0 : errno = EINVAL;
512 0 : return -1;
513 0 : }
514 :
515 72 : ulong wire_sz;
516 72 : ulong file_sz;
517 72 : if( FD_UNLIKELY( __builtin_umull_overflow( disk_max_shreds, sizeof(fd_shredb_entry_t), &wire_sz ) ||
518 72 : __builtin_uaddl_overflow( wire_off, wire_sz, &file_sz ) ||
519 72 : file_sz>(ulong)LONG_MAX ) ) {
520 0 : errno = EOVERFLOW;
521 0 : return -1;
522 0 : }
523 :
524 72 : int fd = open( path, O_RDWR|O_CREAT|O_TRUNC, (mode_t)0600 );
525 72 : if( FD_UNLIKELY( fd<0 ) ) return -1;
526 :
527 72 : int err = 0;
528 72 : if( FD_UNLIKELY( ftruncate( fd, (off_t)file_sz ) ) ) err = errno;
529 72 : if( FD_UNLIKELY( err ) ) {
530 0 : if( FD_UNLIKELY( close( fd ) ) ) FD_LOG_WARNING(( "close(%s) failed (%i-%s)", path, errno, fd_io_strerror( errno ) ));
531 0 : errno = err;
532 0 : return -1;
533 0 : }
534 72 : return fd;
535 72 : }
536 :
537 : void *
538 108 : fd_store_leave( fd_store_t const * store ) {
539 108 : if( FD_UNLIKELY( !store ) ) { FD_LOG_WARNING(( "NULL store" )); return NULL; }
540 108 : return (void *)store;
541 108 : }
542 :
543 : void *
544 108 : fd_store_delete( void * shstore ) {
545 108 : if( FD_UNLIKELY( !shstore ) ) { FD_LOG_WARNING(( "NULL store" )); return NULL; }
546 108 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shstore, fd_store_align() ) ) ) { FD_LOG_WARNING(( "misaligned store" )); return NULL; }
547 :
548 108 : fd_store_t * store = (fd_store_t *)shstore;
549 108 : if( FD_UNLIKELY( store->magic!=FD_STORE_MAGIC ) ) { FD_LOG_WARNING(( "bad magic" )); return NULL; }
550 :
551 108 : FD_COMPILER_MFENCE();
552 108 : FD_VOLATILE( store->magic ) = 0UL;
553 108 : FD_COMPILER_MFENCE();
554 :
555 108 : return shstore;
556 108 : }
557 :
558 :
559 : uchar *
560 : fd_store_fec_data_acquire( fd_store_t * store,
561 : int disk_fd,
562 108 : fd_store_fec_t * fec ) {
563 108 : return fd_store_fec_data_acquire_ex( store, disk_fd, fec, NULL );
564 108 : }
565 :
566 : uchar *
567 : fd_store_fec_data_acquire_ex( fd_store_t * store,
568 : int disk_fd,
569 : fd_store_fec_t * fec,
570 120 : fd_store_fec_spill_stats_t * spill ) {
571 120 : if( FD_LIKELY( spill ) ) *spill = (fd_store_fec_spill_stats_t){0};
572 120 : if( FD_UNLIKELY( !store || !fec ) ) return NULL;
573 :
574 120 : fd_store_fec_t * fec0 = pool_ele_laddr( store );
575 120 : if( FD_UNLIKELY( fec<fec0 || fec>=fec0+store->fec_max ) ) return NULL;
576 :
577 120 : fd_rwlock_write( &store->cache_lock );
578 :
579 120 : if( FD_UNLIKELY( fec->data_state==FD_STORE_FEC_DATA_RAM_WRITING ) ) {
580 0 : uchar * data = cache_data_laddr( store ) + fec->data_off;
581 0 : fd_rwlock_unwrite( &store->cache_lock );
582 0 : return data;
583 0 : }
584 :
585 120 : if( FD_UNLIKELY( fec->data_state!=FD_STORE_FEC_DATA_EMPTY &&
586 120 : fec->data_state!=FD_STORE_FEC_DATA_CONSUMED ) ) {
587 0 : fd_rwlock_unwrite( &store->cache_lock );
588 0 : return NULL;
589 0 : }
590 :
591 162 : while( FD_UNLIKELY( !store->cache_free_cnt ) ) {
592 42 : if( FD_UNLIKELY( store->cache_lru_head==UINT_MAX ) ) {
593 0 : fd_rwlock_unwrite( &store->cache_lock );
594 0 : FD_SPIN_PAUSE();
595 0 : fd_rwlock_write( &store->cache_lock );
596 0 : continue;
597 0 : }
598 42 : int spill_result = spill_one_locked( store, disk_fd, spill );
599 42 : if( FD_UNLIKELY( spill_result==FD_STORE_SPILL_RETRY ) ) {
600 0 : fd_rwlock_unwrite( &store->cache_lock );
601 0 : FD_SPIN_PAUSE();
602 0 : fd_rwlock_write( &store->cache_lock );
603 0 : continue;
604 0 : }
605 42 : if( FD_UNLIKELY( !spill_result ) ) {
606 0 : fd_rwlock_unwrite( &store->cache_lock );
607 0 : return NULL;
608 0 : }
609 42 : }
610 :
611 120 : ulong * free = cache_free_laddr( store );
612 120 : ulong slot = free[ --store->cache_free_cnt ];
613 120 : fec->data_off = slot * store->payload_slot_sz;
614 120 : fec->cache_prev = UINT_MAX;
615 120 : fec->cache_next = UINT_MAX;
616 120 : fec->data_pin_cnt = 0U;
617 120 : fec->data_state = FD_STORE_FEC_DATA_RAM_WRITING;
618 :
619 120 : uchar * data = cache_data_laddr( store ) + fec->data_off;
620 120 : fd_rwlock_unwrite( &store->cache_lock );
621 120 : return data;
622 120 : }
623 :
624 : int
625 : fd_store_fec_data_preevict( fd_store_t * store,
626 : int disk_fd,
627 6 : fd_store_fec_spill_stats_t * spill ) {
628 6 : fd_store_fec_spill_stats_t local_spill[1];
629 6 : if( FD_UNLIKELY( !spill ) ) spill = local_spill;
630 6 : *spill = (fd_store_fec_spill_stats_t){0};
631 6 : if( FD_UNLIKELY( !store || disk_fd<0 ) ) return 0;
632 :
633 6 : fd_rwlock_write( &store->cache_lock );
634 6 : if( FD_UNLIKELY( !store->cache_preevict_active && store->cache_free_cnt<store->cache_free_low_water ) )
635 6 : store->cache_preevict_active = 1U;
636 6 : if( FD_UNLIKELY( store->cache_preevict_active && store->cache_free_cnt>=store->cache_free_target ) )
637 0 : store->cache_preevict_active = 0U;
638 :
639 6 : if( FD_LIKELY( !store->cache_preevict_active ) ) {
640 0 : fd_rwlock_unwrite( &store->cache_lock );
641 0 : return 0;
642 0 : }
643 :
644 6 : spill_one_locked( store, disk_fd, spill );
645 6 : if( FD_UNLIKELY( store->cache_free_cnt>=store->cache_free_target ) )
646 6 : store->cache_preevict_active = 0U;
647 6 : fd_rwlock_unwrite( &store->cache_lock );
648 6 : return !!spill->write_cnt;
649 6 : }
650 :
651 : void
652 : fd_store_fec_cache_stats_query( fd_store_t * store,
653 6 : fd_store_fec_cache_stats_t * stats ) {
654 6 : if( FD_UNLIKELY( !store || !stats ) ) return;
655 6 : fd_rwlock_read( &store->cache_lock );
656 6 : stats->free_cnt = store->cache_free_cnt;
657 6 : stats->max = store->cache_slot_cnt;
658 6 : stats->target = store->cache_free_target;
659 6 : stats->low_water = store->cache_free_low_water;
660 6 : fd_rwlock_unread( &store->cache_lock );
661 6 : }
662 :
663 : void
664 : fd_store_fec_data_publish( fd_store_t * store,
665 114 : fd_store_fec_t * fec ) {
666 114 : fd_rwlock_write( &store->cache_lock );
667 114 : FD_TEST( fec->data_state==FD_STORE_FEC_DATA_RAM_WRITING );
668 114 : FD_TEST( fec->data_sz<=store->fec_data_max );
669 114 : fec->data_state = FD_STORE_FEC_DATA_RAM_READY;
670 114 : cache_lru_push_tail_locked( store, fec );
671 114 : fd_rwlock_unwrite( &store->cache_lock );
672 114 : }
673 :
674 : static void
675 : cache_pin_locked( fd_store_t * store,
676 33 : fd_store_fec_t * fec ) {
677 33 : if( FD_UNLIKELY( !fec->data_pin_cnt && fec->data_state==FD_STORE_FEC_DATA_RAM_READY ) )
678 18 : cache_lru_remove_locked( store, fec );
679 33 : FD_TEST( fec->data_pin_cnt<UINT_MAX );
680 33 : fec->data_pin_cnt++;
681 33 : store->cache_pinned_cnt++;
682 33 : }
683 :
684 : static void
685 : cache_unpin_locked( fd_store_t * store,
686 33 : fd_store_fec_t * fec ) {
687 33 : FD_TEST( fec->data_pin_cnt && store->cache_pinned_cnt );
688 33 : fec->data_pin_cnt--;
689 33 : store->cache_pinned_cnt--;
690 : /* Prefer spilling payloads that a reader has already copied while
691 : retaining their disk copy for replay backfill. */
692 33 : if( FD_UNLIKELY( !fec->data_pin_cnt && fec->data_state==FD_STORE_FEC_DATA_RAM_READY ) )
693 18 : cache_lru_push_head_locked( store, fec );
694 33 : }
695 :
696 : int
697 : fd_store_fec_data_view( fd_store_t * store,
698 : int disk_fd,
699 : fd_store_fec_t * fec,
700 39 : fd_store_fec_data_view_t * view ) {
701 39 : if( FD_UNLIKELY( !view ) ) return -1;
702 39 : view->data = NULL;
703 39 : view->fec = NULL;
704 39 : view->flags = 0U;
705 39 : if( FD_UNLIKELY( !store || !fec ) ) return -1;
706 39 : fd_store_fec_t * fec0 = pool_ele_laddr( store );
707 39 : if( FD_UNLIKELY( fec<fec0 || fec>=fec0+store->fec_max ) ) return -1;
708 :
709 39 : fd_rwlock_write( &store->cache_lock );
710 :
711 39 : if( FD_UNLIKELY( fec->data_sz>store->fec_data_max ) ) {
712 0 : fd_rwlock_unwrite( &store->cache_lock );
713 0 : return -1;
714 0 : }
715 :
716 39 : if( FD_LIKELY( fec->data_state==FD_STORE_FEC_DATA_RAM_READY ||
717 39 : fec->data_state==FD_STORE_FEC_DATA_SPILLING ) ) {
718 18 : cache_pin_locked( store, fec );
719 18 : view->data = cache_data_laddr( store ) + fec->data_off;
720 18 : view->fec = fec;
721 18 : fd_rwlock_unwrite( &store->cache_lock );
722 18 : return 0;
723 18 : }
724 :
725 21 : if( FD_LIKELY( fec->data_state==FD_STORE_FEC_DATA_DISK ) ) {
726 21 : if( FD_UNLIKELY( disk_fd<0 ) ) {
727 0 : fd_rwlock_unwrite( &store->cache_lock );
728 0 : return -1;
729 0 : }
730 21 : if( FD_UNLIKELY( !fd_rwlock_trywrite( &store->spill_read_lock ) ) ) {
731 6 : fd_rwlock_unwrite( &store->cache_lock );
732 6 : return -1;
733 6 : }
734 15 : ulong data_sz = fec->data_sz;
735 15 : ulong data_off = fec->data_off;
736 15 : uchar * spill_read_data = spill_read_data_laddr( store );
737 15 : cache_pin_locked( store, fec );
738 15 : fd_rwlock_unwrite( &store->cache_lock );
739 :
740 15 : if( FD_UNLIKELY( store_pread_all( disk_fd, spill_read_data, data_sz, (off_t)data_off ) ) )
741 0 : FD_LOG_ERR(( "error reading spilled FEC payload: (%d-%s)", errno, fd_io_strerror( errno ) ));
742 :
743 15 : atomic_fetch_add_explicit( &store->fec_spill_read_cnt, 1UL, memory_order_relaxed );
744 15 : atomic_fetch_add_explicit( &store->fec_spill_read_bytes, data_sz, memory_order_relaxed );
745 :
746 15 : view->data = spill_read_data;
747 15 : view->fec = fec;
748 15 : view->flags = FD_STORE_FEC_DATA_VIEW_SPILL;
749 15 : return 0;
750 15 : }
751 :
752 0 : fd_rwlock_unwrite( &store->cache_lock );
753 0 : return -1;
754 21 : }
755 :
756 : void
757 : fd_store_fec_data_view_release( fd_store_t * store,
758 33 : fd_store_fec_data_view_t * view ) {
759 33 : if( FD_UNLIKELY( !store || !view || !view->fec ) ) return;
760 33 : if( view->flags & FD_STORE_FEC_DATA_VIEW_SPILL ) fd_rwlock_unwrite( &store->spill_read_lock );
761 :
762 33 : fd_rwlock_write( &store->cache_lock );
763 33 : cache_unpin_locked( store, view->fec );
764 33 : cache_consume_locked( store, view->fec );
765 33 : fd_rwlock_unwrite( &store->cache_lock );
766 :
767 33 : view->data = NULL;
768 33 : view->fec = NULL;
769 33 : view->flags = 0U;
770 33 : }
771 :
772 : fd_store_fec_t *
773 : fd_store_query( fd_store_map_t * map,
774 768 : fd_hash_t const * merkle_root ) {
775 768 : for(;;) {
776 768 : fd_store_map_query_t query[1];
777 768 : int err = fd_store_map_query_try( map, merkle_root, NULL, query, 0 );
778 768 : if( FD_UNLIKELY( err==FD_MAP_ERR_AGAIN ) ) continue;
779 768 : if( FD_UNLIKELY( err ) ) return NULL;
780 :
781 729 : fd_store_fec_t * fec = fd_store_map_query_ele( query );
782 729 : err = fd_store_map_query_test( query );
783 729 : if( FD_LIKELY( !err ) ) return fec;
784 0 : if( FD_UNLIKELY( err!=FD_MAP_ERR_AGAIN ) ) return NULL;
785 0 : }
786 768 : }
787 :
788 : int
789 : fd_store_insert( fd_store_t * store,
790 : fd_store_map_t * map,
791 : fd_hash_t const * merkle_root,
792 839 : fd_store_fec_t ** fec ) {
793 839 : FD_TEST( store && map && merkle_root && fec );
794 839 : *fec = NULL;
795 839 : fd_rwlock_read( &store->fec_lock );
796 :
797 839 : struct {
798 839 : fd_store_map_txn_t txn [1];
799 839 : fd_store_map_txn_private_info_t info[1];
800 839 : } map_txn_mem;
801 839 : fd_store_map_txn_t * txn = fd_store_map_txn_init( map_txn_mem.txn, map, 1UL );
802 839 : FD_TEST( !fd_store_map_txn_add( txn, merkle_root, 1 ) );
803 839 : FD_TEST( !fd_store_map_txn_try( txn, FD_MAP_FLAG_BLOCKING ) );
804 :
805 839 : fd_store_map_query_t query[1];
806 839 : int err = fd_store_map_txn_query( map, merkle_root, NULL, query, 0 );
807 839 : if( FD_LIKELY( err==FD_MAP_ERR_KEY ) ) {
808 784 : fd_store_fec_t * new_fec = fd_store_fec_acquire( store );
809 784 : FD_TEST( new_fec );
810 784 : new_fec->key = *merkle_root;
811 784 : FD_TEST( !fd_store_map_txn_insert( map, new_fec ) );
812 784 : *fec = new_fec;
813 784 : err = FD_MAP_SUCCESS;
814 784 : } else {
815 55 : FD_TEST( err==FD_MAP_SUCCESS );
816 55 : err = FD_MAP_ERR_KEY;
817 55 : }
818 :
819 839 : FD_TEST( !fd_store_map_txn_test( txn ) );
820 839 : fd_store_map_txn_fini( txn );
821 839 : fd_rwlock_unread( &store->fec_lock );
822 839 : return err;
823 839 : }
824 :
825 : int
826 : fd_store_remove( fd_store_t * store,
827 : fd_store_map_t * map,
828 54 : fd_hash_t const * merkle_root ) {
829 54 : fd_rwlock_write( &store->fec_lock );
830 54 : fd_store_map_query_t query[1];
831 54 : int err = fd_store_map_remove( map, merkle_root, NULL, query, FD_MAP_FLAG_BLOCKING );
832 54 : if( FD_LIKELY( !err ) ) fd_store_fec_release( store, fd_store_map_query_ele( query ) );
833 12 : else FD_TEST( err==FD_MAP_ERR_KEY );
834 54 : fd_rwlock_unwrite( &store->fec_lock );
835 54 : return !err;
836 54 : }
837 :
838 :
839 : static inline ulong
840 : disk_cell_tag( ulong ticket,
841 2208 : ulong state ) {
842 2208 : return (ticket<<2) | state;
843 2208 : }
844 :
845 : static inline ulong
846 2433 : disk_cell_state( ulong tag ) {
847 2433 : return tag & FD_SHREDB_CELL_STATE_MASK;
848 2433 : }
849 :
850 : static inline ulong
851 1107 : disk_cell_ticket( ulong tag ) {
852 1107 : return tag >> 2;
853 1107 : }
854 :
855 : enum {
856 : DISK_PUBLISH_ERR = -1,
857 : DISK_PUBLISH_NOOP = 0,
858 : DISK_PUBLISH_SUCCESS = 1
859 : };
860 :
861 : static void
862 : disk_slot_hint_publish( fd_store_t * store,
863 : ulong slot,
864 1107 : uint shred_idx ) {
865 1107 : atomic_ulong * hint = disk_slot_hint_laddr( store ) + (slot % store->disk_max_slots);
866 1107 : ulong desired = fd_shredb_key_pack( slot, shred_idx ) | FD_SHREDB_HINT_VALID;
867 1107 : ulong current = atomic_load_explicit( hint, memory_order_acquire );
868 1107 : for(;;) {
869 1107 : if( FD_LIKELY( current & FD_SHREDB_HINT_VALID ) ) {
870 768 : uint current_idx = fd_shredb_key_shred_idx( current & ~FD_SHREDB_HINT_VALID );
871 768 : if( FD_LIKELY( fd_shredb_key_slot( current )==slot ) && current_idx>=shred_idx ) return;
872 768 : }
873 1083 : if( atomic_compare_exchange_strong_explicit( hint, ¤t, desired,
874 1083 : memory_order_release, memory_order_acquire ) ) return;
875 1083 : }
876 1107 : }
877 :
878 : static int
879 : disk_exact_publish( fd_store_t * store,
880 : fd_shredb_shred_entry_t * cell,
881 : ulong old_key,
882 1107 : ulong new_key ) {
883 1107 : fd_shredb_shred_map_t map[1];
884 1107 : FD_TEST( disk_shred_map_ljoin( store, map ) );
885 1107 : struct {
886 1107 : fd_shredb_shred_map_txn_t txn [1];
887 1107 : fd_shredb_shred_map_txn_private_info_t info[2];
888 1107 : } txn_mem;
889 1107 : fd_shredb_shred_map_txn_t * txn = fd_shredb_shred_map_txn_init( txn_mem.txn, map, 2UL );
890 1107 : FD_TEST( !fd_shredb_shred_map_txn_add( txn, &old_key, 1 ) );
891 1107 : FD_TEST( !fd_shredb_shred_map_txn_add( txn, &new_key, 1 ) );
892 1107 : FD_TEST( !fd_shredb_shred_map_txn_try( txn, FD_MAP_FLAG_BLOCKING ) );
893 :
894 1107 : fd_shredb_shred_map_query_t old_query[1];
895 1107 : fd_shredb_shred_map_query_t new_query[1];
896 1107 : int old_err = fd_shredb_shred_map_txn_query( map, &old_key, NULL, old_query, 0 );
897 1107 : int new_err = fd_shredb_shred_map_txn_query( map, &new_key, NULL, new_query, 0 );
898 1107 : FD_TEST( old_err==FD_MAP_SUCCESS || old_err==FD_MAP_ERR_KEY );
899 1107 : FD_TEST( new_err==FD_MAP_SUCCESS || new_err==FD_MAP_ERR_KEY );
900 1107 : fd_shredb_shred_entry_t * old_ele = old_err ? NULL : fd_shredb_shred_map_query_ele( old_query );
901 :
902 1107 : int result = DISK_PUBLISH_SUCCESS;
903 1107 : if( FD_UNLIKELY( !new_err ) ) {
904 0 : fd_shredb_shred_entry_t * new_ele = fd_shredb_shred_map_query_ele( new_query );
905 0 : if( new_ele!=cell ) {
906 0 : ulong state = disk_cell_state( atomic_load_explicit( &new_ele->tag, memory_order_acquire ) );
907 0 : if( FD_LIKELY( state==FD_SHREDB_CELL_READY ) ) result = DISK_PUBLISH_NOOP;
908 0 : else if( FD_UNLIKELY( state==FD_SHREDB_CELL_WRITING ) ) result = DISK_PUBLISH_ERR;
909 0 : else FD_TEST( !fd_shredb_shred_map_txn_remove( map, &new_key, NULL, new_query, 0 ) );
910 0 : }
911 0 : }
912 :
913 1107 : if( old_ele==cell ) {
914 918 : fd_shredb_shred_map_query_t remove_query[1];
915 918 : FD_TEST( !fd_shredb_shred_map_txn_remove( map, &old_key, NULL, remove_query, 0 ) );
916 918 : }
917 :
918 1107 : if( FD_LIKELY( result==DISK_PUBLISH_SUCCESS ) ) {
919 1107 : cell->key = new_key;
920 1107 : FD_TEST( !fd_shredb_shred_map_txn_insert( map, cell ) );
921 1107 : }
922 1107 : FD_TEST( !fd_shredb_shred_map_txn_test( txn ) );
923 1107 : fd_shredb_shred_map_txn_fini( txn );
924 1107 : return result;
925 1107 : }
926 :
927 : int
928 : fd_store_disk_insert( fd_store_t * store,
929 : int disk_fd,
930 1124 : fd_shred_t const * shred ) {
931 1124 : if( FD_UNLIKELY( !store || disk_fd<0 || !shred || !fd_store_has_disk( store ) ) )
932 0 : return FD_STORE_DISK_INSERT_ERR;
933 :
934 1124 : ulong slot = shred->slot;
935 1124 : uint shred_idx = shred->idx;
936 1124 : if( FD_UNLIKELY( slot>=FD_SHREDB_KEY_SLOT_MAX || shred_idx>=store->max_shreds_per_block ) ) return FD_STORE_DISK_INSERT_ERR;
937 1106 : ulong key = fd_shredb_key_pack( slot, shred_idx );
938 :
939 1106 : ulong ticket = atomic_fetch_add_explicit( &store->disk_reservation_head, 1UL, memory_order_relaxed ) + 1UL;
940 1106 : if( FD_UNLIKELY( !ticket || ticket>(ULONG_MAX>>2) ) ) return FD_STORE_DISK_INSERT_ERR;
941 1106 : ulong ring_idx = (ticket-1UL) % store->disk_max_shreds;
942 :
943 1106 : fd_shredb_shred_entry_t * cell = disk_shred_pool_laddr( store ) + ring_idx;
944 1106 : ulong old_tag = atomic_load_explicit( &cell->tag, memory_order_acquire );
945 1106 : ulong old_state = disk_cell_state( old_tag );
946 1106 : if( FD_UNLIKELY( old_state==FD_SHREDB_CELL_WRITING || disk_cell_ticket( old_tag )>=ticket ) )
947 0 : return FD_STORE_DISK_INSERT_ERR;
948 1106 : ulong writing_tag = disk_cell_tag( ticket, FD_SHREDB_CELL_WRITING );
949 1106 : if( FD_UNLIKELY( !atomic_compare_exchange_strong_explicit( &cell->tag, &old_tag, writing_tag,
950 1106 : memory_order_acq_rel, memory_order_acquire ) ) )
951 0 : return FD_STORE_DISK_INSERT_ERR;
952 1106 : old_state = disk_cell_state( old_tag );
953 1106 : ulong old_key = cell->key;
954 :
955 1106 : fd_shredb_entry_t wr_entry[1];
956 1106 : ulong shred_sz = fd_ulong_min( fd_shred_sz( shred ), FD_SHRED_MAX_SZ );
957 1106 : fd_memset( wr_entry, 0, sizeof(wr_entry) );
958 1106 : wr_entry->tag = disk_cell_tag( ticket, FD_SHREDB_CELL_READY );
959 1106 : wr_entry->key = key;
960 1106 : wr_entry->shred_sz = (ushort)shred_sz;
961 1106 : fd_memcpy( wr_entry->shred, shred, shred_sz );
962 :
963 1106 : off_t off = (off_t)(store->wire_off + ring_idx*sizeof(fd_shredb_entry_t));
964 1106 : if( FD_UNLIKELY( store_pwrite_all( disk_fd, wr_entry, sizeof(wr_entry), off ) ) )
965 0 : FD_LOG_ERR(( "error writing to disk store: (%d-%s)", errno, fd_io_strerror( errno ) ));
966 1106 : atomic_fetch_add_explicit( &store->disk_write_bytes, sizeof(wr_entry), memory_order_relaxed );
967 :
968 1106 : int publish_result = disk_exact_publish( store, cell, old_key, key );
969 1107 : if( FD_LIKELY( publish_result==DISK_PUBLISH_SUCCESS ) ) {
970 1107 : disk_slot_hint_publish( store, slot, shred_idx );
971 1107 : atomic_store_explicit( &cell->tag, wr_entry->tag, memory_order_release );
972 1107 : if( old_state!=FD_SHREDB_CELL_READY )
973 1107 : atomic_fetch_add_explicit( &store->disk_cnt, 1UL, memory_order_relaxed );
974 1107 : atomic_fetch_add_explicit( &store->disk_insert_cnt, 1UL, memory_order_relaxed );
975 1107 : return FD_STORE_DISK_INSERT_SUCCESS;
976 1107 : }
977 :
978 1 : atomic_store_explicit( &cell->tag, disk_cell_tag( ticket, FD_SHREDB_CELL_INVALID ), memory_order_release );
979 >1844*10^16 : if( old_state==FD_SHREDB_CELL_READY )
980 >1844*10^16 : atomic_fetch_sub_explicit( &store->disk_cnt, 1UL, memory_order_relaxed );
981 >1844*10^16 : if( publish_result==DISK_PUBLISH_NOOP ) return FD_STORE_DISK_INSERT_SUCCESS;
982 >1844*10^16 : return FD_STORE_DISK_INSERT_ERR;
983 >1844*10^16 : }
984 :
985 267 : #define FD_STORE_DISK_READ_RETRY_CNT (8UL)
986 :
987 : static int
988 : disk_read_key_once( fd_store_t const * store,
989 : int disk_fd,
990 : ulong key,
991 255 : fd_shredb_entry_t * rd_entry ) {
992 255 : fd_shredb_shred_map_t map[1];
993 255 : FD_TEST( disk_shred_map_ljoin( store, map ) );
994 255 : fd_shredb_shred_map_query_t query[1];
995 255 : int err = fd_shredb_shred_map_query_try( map, &key, NULL, query, 0 );
996 255 : if( FD_UNLIKELY( err==FD_MAP_ERR_AGAIN || err==FD_MAP_ERR_CORRUPT ) ) return FD_STORE_DISK_QUERY_BUSY;
997 255 : if( FD_UNLIKELY( err==FD_MAP_ERR_KEY ) ) return FD_STORE_DISK_QUERY_MISS;
998 219 : FD_TEST( !err );
999 :
1000 219 : fd_shredb_shred_entry_t const * cell = fd_shredb_shred_map_query_ele_const( query );
1001 219 : ulong ring_idx = (ulong)(cell-disk_shred_pool_laddr( store ));
1002 219 : ulong tag = atomic_load_explicit( &cell->tag, memory_order_acquire );
1003 219 : if( FD_UNLIKELY( disk_cell_state( tag )!=FD_SHREDB_CELL_READY ) ) {
1004 0 : int query_err = fd_shredb_shred_map_query_test( query );
1005 0 : if( FD_UNLIKELY( query_err || disk_cell_state( tag )==FD_SHREDB_CELL_WRITING ) ) return FD_STORE_DISK_QUERY_BUSY;
1006 0 : return FD_STORE_DISK_QUERY_MISS;
1007 0 : }
1008 :
1009 219 : off_t off = (off_t)(store->wire_off + ring_idx*sizeof(fd_shredb_entry_t));
1010 219 : if( FD_UNLIKELY( store_pread_all( disk_fd, rd_entry, sizeof(fd_shredb_entry_t), off ) ) )
1011 0 : FD_LOG_ERR(( "error reading disk store: (%d-%s)", errno, fd_io_strerror( errno ) ));
1012 219 : ulong tag_after = atomic_load_explicit( &cell->tag, memory_order_acquire );
1013 219 : int query_err = fd_shredb_shred_map_query_test( query );
1014 219 : if( FD_UNLIKELY( query_err || tag_after!=tag ) ) return FD_STORE_DISK_QUERY_BUSY;
1015 219 : if( FD_UNLIKELY( rd_entry->tag!=tag || rd_entry->key!=key ) ) return FD_STORE_DISK_QUERY_BUSY;
1016 219 : return (int)fd_ulong_min( rd_entry->shred_sz, FD_SHRED_MAX_SZ );
1017 219 : }
1018 :
1019 : int
1020 : fd_store_disk_query( fd_store_t const * store,
1021 : int disk_fd,
1022 : ulong slot,
1023 : uint shred_idx,
1024 195 : uchar out[ FD_SHRED_MAX_SZ ] ) {
1025 195 : if( FD_UNLIKELY( !store || disk_fd<0 || !out || !fd_store_has_disk( store ) ||
1026 195 : slot>=FD_SHREDB_KEY_SLOT_MAX || shred_idx>=store->max_shreds_per_block ) ) return FD_STORE_DISK_QUERY_MISS;
1027 183 : ulong key = fd_shredb_key_pack( slot, shred_idx );
1028 183 : for( ulong retry=0UL; retry<FD_STORE_DISK_READ_RETRY_CNT; retry++ ) {
1029 183 : fd_shredb_entry_t rd_entry[1];
1030 183 : int result = disk_read_key_once( store, disk_fd, key, rd_entry );
1031 183 : if( FD_UNLIKELY( result==FD_STORE_DISK_QUERY_BUSY ) ) { FD_SPIN_PAUSE(); continue; }
1032 183 : if( FD_UNLIKELY( result<0 ) ) return result;
1033 165 : fd_memcpy( out, rd_entry->shred, (ulong)result );
1034 165 : return result;
1035 183 : }
1036 0 : return FD_STORE_DISK_QUERY_BUSY;
1037 183 : }
1038 :
1039 : int
1040 : fd_store_disk_query_highest( fd_store_t const * store,
1041 : int disk_fd,
1042 : ulong slot,
1043 : uint min_shred_idx,
1044 84 : uchar out[ FD_SHRED_MAX_SZ ] ) {
1045 84 : if( FD_UNLIKELY( !store || disk_fd<0 || !out || !fd_store_has_disk( store ) || slot>=FD_SHREDB_KEY_SLOT_MAX ) )
1046 0 : return FD_STORE_DISK_QUERY_MISS;
1047 84 : for( ulong retry=0UL; retry<FD_STORE_DISK_READ_RETRY_CNT; retry++ ) {
1048 84 : atomic_ulong const * hint_ptr = disk_slot_hint_laddr( store ) + (slot % store->disk_max_slots);
1049 84 : ulong hint = atomic_load_explicit( hint_ptr, memory_order_acquire );
1050 84 : if( FD_UNLIKELY( !(hint & FD_SHREDB_HINT_VALID) ) ) return FD_STORE_DISK_QUERY_MISS;
1051 84 : if( FD_UNLIKELY( fd_shredb_key_slot( hint )!=slot ) ) return FD_STORE_DISK_QUERY_BUSY;
1052 72 : uint idx = fd_shredb_key_shred_idx( hint & ~FD_SHREDB_HINT_VALID );
1053 72 : if( FD_UNLIKELY( idx>=store->max_shreds_per_block ) ) return FD_STORE_DISK_QUERY_MISS;
1054 :
1055 72 : fd_shredb_entry_t rd_entry[1];
1056 72 : int result = disk_read_key_once( store, disk_fd, fd_shredb_key_pack( slot, idx ), rd_entry );
1057 72 : ulong hint_after = atomic_load_explicit( hint_ptr, memory_order_acquire );
1058 72 : if( FD_UNLIKELY( hint_after!=hint || result==FD_STORE_DISK_QUERY_BUSY ) ) {
1059 0 : FD_SPIN_PAUSE();
1060 0 : continue;
1061 0 : }
1062 72 : if( FD_UNLIKELY( result==FD_STORE_DISK_QUERY_MISS ) ) return FD_STORE_DISK_QUERY_SCAN_LIMIT;
1063 54 : if( FD_UNLIKELY( result<0 ) ) return result;
1064 :
1065 54 : fd_shred_t const * shred = (fd_shred_t const *)fd_type_pun_const( rd_entry->shred );
1066 54 : if( FD_UNLIKELY( idx < min_shred_idx && !(shred->data.flags & FD_SHRED_DATA_FLAG_SLOT_COMPLETE) ) )
1067 6 : return FD_STORE_DISK_QUERY_MISS;
1068 48 : fd_memcpy( out, rd_entry->shred, (ulong)result );
1069 48 : return result;
1070 54 : }
1071 0 : return FD_STORE_DISK_QUERY_BUSY;
1072 84 : }
1073 :
1074 : static int
1075 : spill_reclaim_one( fd_store_t * store,
1076 12 : int disk_fd ) {
1077 12 : if( FD_LIKELY( !FD_VOLATILE_CONST( store->spill_reclaim_cnt ) ) ) return 0;
1078 :
1079 12 : fd_rwlock_write( &store->cache_lock );
1080 12 : if( FD_LIKELY( !store->spill_reclaim_cnt ) ) {
1081 0 : fd_rwlock_unwrite( &store->cache_lock );
1082 0 : return 0;
1083 0 : }
1084 12 : uint spill_slot = spill_reclaim_laddr( store )[ --store->spill_reclaim_cnt ];
1085 12 : store->spill_reclaiming_cnt++;
1086 12 : fd_rwlock_unwrite( &store->cache_lock );
1087 :
1088 12 : off_t off = (off_t)((ulong)spill_slot*store->payload_slot_sz);
1089 12 : int err = fallocate( disk_fd, FALLOC_FL_PUNCH_HOLE | FALLOC_FL_KEEP_SIZE, off, (off_t)store->payload_slot_sz );
1090 12 : int err_no = errno;
1091 :
1092 12 : fd_rwlock_write( &store->cache_lock );
1093 12 : FD_TEST( store->spill_reclaiming_cnt );
1094 12 : store->spill_reclaiming_cnt--;
1095 12 : if( FD_LIKELY( !err ) ) {
1096 12 : FD_TEST( atomic_load_explicit( &store->spill_allocated_cnt, memory_order_relaxed ) );
1097 12 : atomic_fetch_sub_explicit( &store->spill_allocated_cnt, 1UL, memory_order_relaxed );
1098 12 : FD_TEST( store->spill_free_cnt<store->spill_slot_cnt );
1099 12 : spill_free_laddr( store )[ store->spill_free_cnt++ ] = spill_slot;
1100 12 : } else {
1101 0 : spill_reuse_push_locked( store, spill_slot );
1102 0 : }
1103 12 : fd_rwlock_unwrite( &store->cache_lock );
1104 12 : if( FD_UNLIKELY( err ) )
1105 0 : FD_LOG_WARNING(( "error reclaiming spilled FEC page: (%d-%s)", err_no, fd_io_strerror( err_no ) ));
1106 12 : return 1;
1107 12 : }
1108 :
1109 : int
1110 : fd_store_disk_maintain( fd_store_t * store,
1111 12 : int disk_fd ) {
1112 12 : if( FD_UNLIKELY( !store || disk_fd<0 ) ) return 0;
1113 12 : return spill_reclaim_one( store, disk_fd );
1114 12 : }
1115 :
1116 : int
1117 : fd_store_disk_stats_query( fd_store_t const * store,
1118 6 : fd_store_disk_stats_t * stats ) {
1119 6 : if( FD_UNLIKELY( !store || !stats || !fd_store_has_disk( store ) ) ) return FD_STORE_DISK_QUERY_MISS;
1120 6 : stats->shred_cnt = atomic_load_explicit( &store->disk_cnt, memory_order_relaxed );
1121 6 : stats->current_bytes = stats->shred_cnt*sizeof(fd_shredb_entry_t)
1122 6 : + atomic_load_explicit( &store->spill_live_cnt, memory_order_relaxed )*store->payload_slot_sz;
1123 6 : stats->allocated_bytes = fd_ulong_min( atomic_load_explicit( &store->disk_reservation_head, memory_order_relaxed ),
1124 6 : store->disk_max_shreds )*sizeof(fd_shredb_entry_t)
1125 6 : + atomic_load_explicit( &store->spill_allocated_cnt, memory_order_relaxed )*store->payload_slot_sz;
1126 6 : stats->insert_cnt = atomic_load_explicit( &store->disk_insert_cnt, memory_order_relaxed );
1127 : stats->write_bytes = atomic_load_explicit( &store->disk_write_bytes, memory_order_relaxed );
1128 6 : return 0;
1129 6 : }
|
