Line data Source code
1 : #define _GNU_SOURCE
2 : #include "fd_shredb.h"
3 :
4 : #include <errno.h>
5 : #include <fcntl.h>
6 : #include <unistd.h>
7 :
8 : static inline ulong
9 180 : fd_shredb_max_shreds_for_gib( ulong max_size_gib ) {
10 180 : return (max_size_gib*1024UL*1024UL*1024UL) / sizeof(fd_shredb_entry_t);
11 180 : }
12 :
13 : /* We size the slot map such that it will never fill before we start
14 : evicting from the shred_map/ring buffer. The minimum number of shreds
15 : per slot is 32 (one FEC set), so it is guaranteed that in the worst case
16 : we will be able to represent every FEC set inserted into the database.
17 :
18 : Remember that we will always be inserting complete sets, consisting of
19 : 32 data shreds at a time. */
20 : static inline ulong
21 90 : fd_shredb_max_slots_for_gib( ulong max_size_gib ) {
22 90 : return fd_shredb_max_shreds_for_gib( max_size_gib ) / 32UL;
23 90 : }
24 :
25 : FD_FN_CONST ulong
26 66 : fd_shredb_footprint( ulong max_size_gib ) {
27 66 : if( FD_UNLIKELY( !max_size_gib ) ) return 0UL;
28 :
29 63 : ulong max_shreds = fd_shredb_max_shreds_for_gib( max_size_gib );
30 63 : ulong max_slots = fd_shredb_max_slots_for_gib ( max_size_gib );
31 :
32 63 : int lg_shred_cnt = fd_ulong_find_msb( fd_ulong_pow2_up( max_shreds ) );
33 63 : int lg_slot_cnt = fd_ulong_find_msb( fd_ulong_pow2_up( max_slots ) );
34 :
35 63 : ulong l = FD_LAYOUT_INIT;
36 63 : l = FD_LAYOUT_APPEND( l, alignof(fd_shredb_t), sizeof(fd_shredb_t) );
37 63 : l = FD_LAYOUT_APPEND( l, fd_shredb_shred_map_align(), fd_shredb_shred_map_footprint( lg_shred_cnt ) );
38 63 : l = FD_LAYOUT_APPEND( l, fd_shredb_slot_map_align(), fd_shredb_slot_map_footprint ( lg_slot_cnt ) );
39 63 : ulong bitset_words = (max_shreds + 63UL) / 64UL;
40 63 : l = FD_LAYOUT_APPEND( l, alignof(ulong), max_shreds * sizeof(ulong) ); /* evict_keys */
41 63 : l = FD_LAYOUT_APPEND( l, alignof(ulong), bitset_words * sizeof(ulong) ); /* evict_occupied */
42 63 : return FD_LAYOUT_FINI( l, fd_shredb_align() );
43 66 : }
44 :
45 : void *
46 : fd_shredb_new( void * shmem,
47 : ulong max_size_gib,
48 : char const * file_path,
49 27 : ulong seed ) {
50 27 : if( FD_UNLIKELY( !shmem ) ) {
51 0 : FD_LOG_WARNING(( "NULL shmem" ));
52 0 : return NULL;
53 0 : }
54 :
55 27 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shmem, fd_shredb_align() ) ) ) {
56 0 : FD_LOG_WARNING(( "misaligned shmem" ));
57 0 : return NULL;
58 0 : }
59 :
60 27 : if( FD_UNLIKELY( !file_path ) ) {
61 0 : FD_LOG_WARNING(( "NULL file_path" ));
62 0 : return NULL;
63 0 : }
64 :
65 27 : ulong footprint = fd_shredb_footprint( max_size_gib );
66 27 : if( FD_UNLIKELY( !footprint ) ) {
67 0 : FD_LOG_WARNING(( "bad max_size_gib (%lu)", max_size_gib ));
68 0 : return NULL;
69 0 : }
70 :
71 27 : ulong max_shreds = fd_shredb_max_shreds_for_gib( max_size_gib );
72 27 : ulong max_slots = fd_shredb_max_slots_for_gib ( max_size_gib );
73 :
74 27 : int lg_shred_cnt = fd_ulong_find_msb( fd_ulong_pow2_up( max_shreds ) );
75 27 : int lg_slot_cnt = fd_ulong_find_msb( fd_ulong_pow2_up( max_slots ) );
76 :
77 27 : FD_SCRATCH_ALLOC_INIT( l, shmem );
78 27 : /**/ FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_shredb_t), sizeof(fd_shredb_t) );
79 27 : void * shred_map_mem = FD_SCRATCH_ALLOC_APPEND( l, fd_shredb_shred_map_align(), fd_shredb_shred_map_footprint( lg_shred_cnt ) );
80 27 : void * slot_map_mem = FD_SCRATCH_ALLOC_APPEND( l, fd_shredb_slot_map_align(), fd_shredb_slot_map_footprint ( lg_slot_cnt ) );
81 27 : ulong bitset_words = (max_shreds + 63UL) / 64UL;
82 27 : void * evict_k_mem = FD_SCRATCH_ALLOC_APPEND( l, alignof(ulong), max_shreds * sizeof(ulong) );
83 27 : void * evict_o_mem = FD_SCRATCH_ALLOC_APPEND( l, alignof(ulong), bitset_words * sizeof(ulong) );
84 :
85 27 : fd_shredb_t * store = (fd_shredb_t *)shmem;
86 27 : store->shred_map = fd_shredb_shred_map_new( shred_map_mem, lg_shred_cnt, seed );
87 27 : store->slot_map = fd_shredb_slot_map_new ( slot_map_mem, lg_slot_cnt, seed );
88 27 : store->evict_keys = (ulong *)evict_k_mem;
89 27 : store->evict_occupied = (ulong *)evict_o_mem;
90 27 : fd_memset( store->evict_occupied, 0, bitset_words * sizeof(ulong) );
91 :
92 27 : int fd = open( file_path, O_RDWR | O_CREAT | O_TRUNC, (mode_t)0600 );
93 27 : if( FD_UNLIKELY( fd<0 ) ) {
94 0 : FD_LOG_WARNING(( "open(%s) failed (%i-%s)", file_path, errno, fd_io_strerror( errno ) ));
95 0 : return NULL;
96 0 : }
97 :
98 27 : ulong initial_shreds = 128UL;
99 27 : ulong initial_sz = initial_shreds * sizeof(fd_shredb_entry_t);
100 27 : if( FD_UNLIKELY( fallocate( fd, 0, 0, (off_t)initial_sz ) ) ) {
101 0 : FD_LOG_WARNING(( "fallocate failed (%i-%s)", errno, fd_io_strerror( errno ) ));
102 0 : close( fd );
103 0 : return NULL;
104 0 : }
105 :
106 27 : store->max_shreds = max_shreds;
107 27 : store->write_head = 0UL;
108 27 : store->cnt = 0UL;
109 27 : store->fd = fd;
110 27 : store->file_shreds = initial_shreds;
111 :
112 27 : FD_TEST( FD_SCRATCH_ALLOC_FINI( l, fd_shredb_align() )==(ulong)shmem + footprint );
113 :
114 27 : return shmem;
115 27 : }
116 :
117 : fd_shredb_t *
118 27 : fd_shredb_join( void * shstore ) {
119 27 : if( FD_UNLIKELY( !shstore ) ) {
120 0 : FD_LOG_WARNING(( "NULL shstore" ));
121 0 : return NULL;
122 0 : }
123 :
124 27 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shstore, fd_shredb_align() ) ) ) {
125 0 : FD_LOG_WARNING(( "misaligned shstore" ));
126 0 : return NULL;
127 0 : }
128 :
129 27 : fd_shredb_t * store = (fd_shredb_t *)shstore;
130 27 : store->shred_map = fd_shredb_shred_map_join( store->shred_map );
131 27 : store->slot_map = fd_shredb_slot_map_join ( store->slot_map );
132 :
133 27 : return (fd_shredb_t *)shstore;
134 27 : }
135 :
136 : void *
137 27 : fd_shredb_leave( fd_shredb_t const * store ) {
138 27 : if( FD_UNLIKELY( !store ) ) {
139 0 : FD_LOG_WARNING(( "NULL store" ));
140 0 : return NULL;
141 0 : }
142 :
143 27 : return (void *)store;
144 27 : }
145 :
146 : void *
147 27 : fd_shredb_delete( void * shstore ) {
148 27 : if( FD_UNLIKELY( !shstore ) ) {
149 0 : FD_LOG_WARNING(( "NULL shstore" ));
150 0 : return NULL;
151 0 : }
152 :
153 27 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shstore, fd_shredb_align() ) ) ) {
154 0 : FD_LOG_WARNING(( "misaligned shstore" ));
155 0 : return NULL;
156 0 : }
157 :
158 27 : fd_shredb_t * store = (fd_shredb_t *)shstore;
159 27 : close( store->fd );
160 :
161 27 : return shstore;
162 27 : }
163 :
164 : static void
165 : fd_shredb_slot_evict( fd_shredb_t * store,
166 : ulong slot,
167 483852 : uint evicted_shred_idx ) {
168 483852 : fd_shredb_slot_entry_t * se = fd_shredb_slot_map_query( store->slot_map, slot, NULL );
169 483852 : FD_TEST( se );
170 :
171 483852 : se->cnt--;
172 483852 : if( FD_UNLIKELY( se->cnt==0UL ) ) {
173 15213 : fd_shredb_slot_map_remove( store->slot_map, se );
174 15213 : return;
175 15213 : }
176 :
177 : /* If the shred evicted was the highest in that slot, walk down and
178 : find the new highest that still exists in the per-shred map. */
179 468639 : if( evicted_shred_idx==se->highest_shred_idx ) {
180 0 : for( uint idx = evicted_shred_idx; ; idx-- ) {
181 0 : ulong key = fd_shredb_key_pack( slot, idx );
182 0 : if( fd_shredb_shred_map_query( store->shred_map, key, NULL ) ) {
183 0 : se->highest_shred_idx = idx;
184 0 : return;
185 0 : }
186 0 : if( FD_UNLIKELY( idx==0U ) ) break;
187 0 : }
188 0 : FD_LOG_ERR(( "corrupt store state" ));
189 0 : }
190 468639 : }
191 :
192 : void
193 : fd_shredb_insert( fd_shredb_t * store,
194 3060699 : fd_shred_t const * shred ) {
195 3060699 : ulong shred_sz = fd_shred_sz( shred );
196 3060699 : ulong slot = shred->slot;
197 3060699 : uint shred_idx = shred->idx;
198 :
199 3060699 : ulong key = fd_shredb_key_pack( slot, shred_idx );
200 3060699 : if( fd_shredb_shred_map_query( store->shred_map, key, NULL ) ) return;
201 :
202 : /* Grow the backing file if the write head has reached the current
203 : file capacity. Double the file size each time (superlinear growth)
204 : until we hit max_shreds, after which the ring simply evicts. */
205 3060699 : if( FD_UNLIKELY( store->write_head>=store->file_shreds ) ) {
206 81 : ulong old_file_sz = store->file_shreds * sizeof(fd_shredb_entry_t);
207 81 : ulong new_file_shreds = fd_ulong_min( store->file_shreds * 2UL, store->max_shreds );
208 81 : ulong new_file_sz = new_file_shreds * sizeof(fd_shredb_entry_t);
209 81 : if( FD_UNLIKELY( fallocate( store->fd, 0, (off_t)old_file_sz, (off_t)(new_file_sz - old_file_sz) ) ) ) {
210 0 : FD_LOG_ERR(( "fallocate failed (%i-%s)", errno, fd_io_strerror( errno ) ));
211 0 : }
212 81 : store->file_shreds = new_file_shreds;
213 81 : }
214 :
215 3060699 : ulong wh_word = store->write_head / 64UL;
216 3060699 : ulong wh_bit = store->write_head % 64UL;
217 3060699 : if( FD_LIKELY( store->evict_occupied[ wh_word ] & (1UL << wh_bit) ) ) {
218 483852 : ulong old_key = store->evict_keys[ store->write_head ];
219 483852 : ulong old_slot = fd_shredb_key_slot( old_key );
220 483852 : uint old_idx = fd_shredb_key_shred_idx( old_key );
221 :
222 483852 : fd_shredb_shred_entry_t * old = fd_shredb_shred_map_query( store->shred_map, old_key, NULL );
223 483852 : if( FD_LIKELY( old ) ) fd_shredb_shred_map_remove( store->shred_map, old );
224 :
225 483852 : fd_shredb_slot_evict( store, old_slot, old_idx );
226 483852 : store->cnt--;
227 483852 : }
228 :
229 3060699 : fd_shredb_entry_t wr_entry[1];
230 3060699 : wr_entry->shred_sz = (ushort)shred_sz;
231 3060699 : fd_memcpy( wr_entry->shred, shred, shred_sz );
232 :
233 3060699 : off_t off = (off_t)(store->write_head * sizeof(fd_shredb_entry_t));
234 3060699 : long res = pwrite( store->fd, wr_entry, sizeof(fd_shredb_entry_t), off );
235 3060699 : if( FD_UNLIKELY( res!=(long)sizeof(fd_shredb_entry_t) ) ) FD_LOG_ERR(( "error writing to shredb: (%d-%s)", errno, fd_io_strerror( errno ) ));
236 :
237 3060699 : store->evict_keys [ store->write_head ] = key;
238 3060699 : store->evict_occupied[ wh_word ] |= (1UL << wh_bit);
239 :
240 3060699 : fd_shredb_shred_entry_t * map_entry = fd_shredb_shred_map_insert( store->shred_map, key );
241 3060699 : FD_TEST( map_entry );
242 3060699 : map_entry->ring_idx = store->write_head;
243 :
244 3060699 : fd_shredb_slot_entry_t * se = fd_shredb_slot_map_query( store->slot_map, slot, NULL );
245 3060699 : if( FD_LIKELY( se ) ) {
246 2964915 : se->cnt++;
247 2964915 : se->highest_shred_idx = fd_uint_max( se->highest_shred_idx, shred_idx );
248 2964915 : } else {
249 95784 : se = fd_shredb_slot_map_insert( store->slot_map, slot );
250 95784 : FD_TEST( se );
251 95784 : se->highest_shred_idx = shred_idx;
252 95784 : se->cnt = 1UL;
253 95784 : }
254 :
255 3060699 : store->cnt++;
256 3060699 : store->write_head = (store->write_head + 1UL) % store->max_shreds;
257 3060699 : }
258 :
259 : int
260 : fd_shredb_query( fd_shredb_t * store,
261 : ulong slot,
262 : uint shred_idx,
263 90279 : uchar out[ FD_SHRED_MAX_SZ ] ) {
264 : /* Fast-fail, if we have never heard of this slot, we must have no shreds for it. */
265 90279 : if( !fd_shredb_slot_map_query( store->slot_map, slot, NULL ) ) return -1;
266 :
267 60267 : ulong key = fd_shredb_key_pack( slot, shred_idx );
268 60267 : fd_shredb_shred_entry_t const * map_entry = fd_shredb_shred_map_query( store->shred_map, key, NULL );
269 60267 : if( FD_UNLIKELY( !map_entry ) ) return -1; /* No such shred. */
270 :
271 60264 : fd_shredb_entry_t rd_entry[1];
272 60264 : off_t off = (off_t)(map_entry->ring_idx * sizeof(fd_shredb_entry_t));
273 60264 : long res = pread( store->fd, rd_entry, sizeof(fd_shredb_entry_t), off );
274 60264 : if( FD_UNLIKELY( res!=(long)sizeof(fd_shredb_entry_t) ) ) FD_LOG_ERR(( "error reading from shredb: (%d-%s)", errno, fd_io_strerror( errno ) ));
275 :
276 60264 : fd_memcpy( out, rd_entry->shred, rd_entry->shred_sz );
277 60264 : return rd_entry->shred_sz;
278 60264 : }
279 :
280 : int fd_shredb_query_highest( fd_shredb_t * store,
281 : ulong slot,
282 : uint min_shred_idx,
283 0 : uchar out[ FD_SHRED_MAX_SZ ] ) {
284 0 : fd_shredb_slot_entry_t * se = fd_shredb_slot_map_query( store->slot_map, slot, NULL );
285 0 : if( FD_UNLIKELY( !se ) ) return -1;
286 :
287 : /* Check if the highest known index meets the threshold. */
288 0 : if( se->highest_shred_idx < min_shred_idx ) return -1;
289 :
290 0 : ulong key = fd_shredb_key_pack( slot, se->highest_shred_idx );
291 0 : fd_shredb_shred_entry_t const * map_entry = fd_shredb_shred_map_query( store->shred_map, key, NULL );
292 0 : FD_TEST( map_entry );
293 :
294 0 : fd_shredb_entry_t rd_entry[1];
295 0 : off_t off = (off_t)(map_entry->ring_idx * sizeof(fd_shredb_entry_t));
296 0 : long res = pread( store->fd, rd_entry, sizeof(fd_shredb_entry_t), off );
297 0 : if( FD_UNLIKELY( res!=(long)sizeof(fd_shredb_entry_t) ) ) FD_LOG_ERR(( "error reading from shredb: (%d-%s)", errno, fd_io_strerror( errno ) ));
298 :
299 0 : fd_memcpy( out, rd_entry->shred, rd_entry->shred_sz );
300 0 : return rd_entry->shred_sz;
301 0 : }
|
