Line data Source code
1 0 : case FD_VINYL_REQ_TYPE_ACQUIRE: {
2 0 : ulong req_flags = (ulong)req->flags;
3 0 : ulong req_val_max = (ulong)req->val_max;
4 0 : fd_vinyl_key_t const * req_key = MAP_REQ_GADDR( req->key_gaddr, fd_vinyl_key_t, batch_cnt );
5 0 : ulong * req_val_gaddr = MAP_REQ_GADDR( req->val_gaddr_gaddr, ulong, batch_cnt );
6 0 : schar * req_err = MAP_REQ_GADDR( req->err_gaddr, schar, batch_cnt );
7 :
8 0 : int req_flag_modify = fd_vinyl_req_flag_modify( req_flags );
9 0 : int req_flag_ignore = fd_vinyl_req_flag_ignore( req_flags );
10 0 : int req_flag_create = fd_vinyl_req_flag_create( req_flags );
11 0 : int req_flag_excl = fd_vinyl_req_flag_excl ( req_flags );
12 0 : int req_evict_prio = fd_vinyl_req_evict_prio ( req_flags );
13 :
14 0 : int bad_gaddr = (!!batch_cnt) & ((!req_key) | (!req_val_gaddr) | (!req_err));
15 0 : int bad_val_max = req_flag_modify & (req_val_max>FD_VINYL_VAL_MAX);
16 0 : int bad_quota = quota_rem<batch_cnt;
17 :
18 0 : if( FD_UNLIKELY( bad_gaddr | bad_val_max | bad_quota ) ) {
19 0 : comp_err = (bad_gaddr | bad_val_max) ? FD_VINYL_ERR_INVAL : FD_VINYL_ERR_FULL;
20 0 : break;
21 0 : }
22 :
23 0 : for( ulong batch_idx=0UL; batch_idx<batch_cnt; batch_idx++ ) {
24 :
25 0 : # define DONE(err) do { \
26 0 : int _err = (err); \
27 0 : FD_COMPILER_MFENCE(); \
28 0 : req_err[ batch_idx ] = (schar)_err; \
29 0 : FD_COMPILER_MFENCE(); \
30 0 : quota_rem -= (ulong) !_err; \
31 0 : fail_cnt += (ulong)!!_err; \
32 0 : goto next_acquire; /* sigh ... can't use continue */ \
33 0 : } while(0)
34 :
35 : /* Query vinyl meta for key */
36 :
37 0 : fd_vinyl_key_t const * key = req_key + batch_idx;
38 :
39 0 : ulong memo = fd_vinyl_key_memo( meta_seed, key );
40 :
41 0 : ulong _ele_idx; /* avoid pointer escape */
42 0 : int err = fd_vinyl_meta_query_fast( ele0, ele_max, key, memo, &_ele_idx );
43 0 : ulong ele_idx = _ele_idx; /* In [0,ele_max) */
44 :
45 0 : if( FD_LIKELY( !err ) ) { /* pair key meta cached */
46 :
47 : /* At this point, pair key either exists at bstream seq_present
48 : or is in the process of being created. If pair key is being
49 : created, fail with AGAIN (it must be acquired for modify). */
50 :
51 0 : ulong pair_ctl = ele0[ ele_idx ].phdr.ctl;
52 :
53 0 : FD_CRIT( (fd_vinyl_bstream_ctl_type( pair_ctl )==FD_VINYL_BSTREAM_CTL_TYPE_PAIR) | (pair_ctl==ULONG_MAX),
54 0 : "corruption detected" );
55 :
56 0 : if( FD_UNLIKELY( pair_ctl==ULONG_MAX ) ) DONE( FD_VINYL_ERR_AGAIN );
57 :
58 : /* At this point, pair key exists at bstream seq_present. */
59 :
60 0 : ulong val_sz = (ulong)ele0[ ele_idx ].phdr.info.val_sz;
61 0 : ulong line_idx = ele0[ ele_idx ].line_idx;
62 :
63 0 : FD_CRIT( val_sz<=FD_VINYL_VAL_MAX, "corruption detected" );
64 0 : FD_CRIT( (line_idx<line_cnt) | (line_idx==ULONG_MAX), "corruption detected" );
65 :
66 0 : if( FD_LIKELY( line_idx<line_cnt ) ) {
67 :
68 : /* At this point, pair key is cached. Get the cache info for
69 : line line_idx. */
70 :
71 0 : accum_cache_hit++;
72 :
73 0 : FD_CRIT( line[ line_idx ].ele_idx==ele_idx, "corruption detected" );
74 :
75 0 : fd_vinyl_data_obj_t * obj = line[ line_idx ].obj;
76 :
77 0 : FD_ALERT( fd_vinyl_data_is_valid_obj( obj, vol, vol_cnt ), "corruption detected" );
78 0 : FD_CRIT ( obj->line_idx==line_idx, "corruption detected" );
79 :
80 0 : ulong line_ctl = line[ line_idx ].ctl;
81 :
82 0 : ulong ver = fd_vinyl_line_ctl_ver( line_ctl );
83 0 : long ref = fd_vinyl_line_ctl_ref( line_ctl );
84 :
85 0 : if( FD_LIKELY( !req_flag_modify ) ) {
86 :
87 : /* At this point, we are acquiring a cached pair for read.
88 : If the line is acquired for modify, fail with AGAIN. If
89 : there are too many acquires for read on this pair, CRIT
90 : (could consider AGAIN here). Otherwise, we update the
91 : ref count (don't change the ver), point the client at the
92 : line caching pair key to finish the acquire. Note that
93 : we don't validate the pair header if we detect that an
94 : earlier acquire in this batch started fetching the pair
95 : because the read might still be in progress (see note
96 : below for more details). */
97 :
98 0 : if( FD_UNLIKELY( ref<0L ) ) DONE( FD_VINYL_ERR_AGAIN );
99 0 : if( FD_UNLIKELY( ref>=FD_VINYL_LINE_REF_MAX ) ) FD_LOG_CRIT(( "too many acquires for read on this pair" ));
100 :
101 0 : if( FD_LIKELY( !obj->rd_active ) ) {
102 0 : fd_vinyl_bstream_phdr_t * phdr = fd_vinyl_data_obj_phdr( obj );
103 :
104 0 : FD_CRIT( fd_vinyl_data_obj_val_max( obj ) >= val_sz, "corruption detected" );
105 0 : FD_CRIT( phdr->ctl==fd_vinyl_bstream_ctl( FD_VINYL_BSTREAM_CTL_TYPE_PAIR,
106 0 : FD_VINYL_BSTREAM_CTL_STYLE_RAW, val_sz ), "corruption detected" );
107 0 : FD_CRIT( fd_vinyl_key_eq( &phdr->key, key ), "corruption detected" );
108 0 : FD_CRIT( !memcmp( &phdr->info, &ele0[ ele_idx ].phdr.info, sizeof(fd_vinyl_info_t) ), "corruption detected" );
109 0 : }
110 :
111 0 : line[ line_idx ].ctl = fd_vinyl_line_ctl( ver, ref+1L ); /* don't bump ver */
112 :
113 0 : req_val_gaddr[ batch_idx ] = (ulong)fd_vinyl_data_obj_val( obj ) - data_laddr0;
114 :
115 0 : DONE( FD_VINYL_SUCCESS );
116 :
117 0 : }
118 :
119 : /* At this point, we are acquiring a cached pair for modify.
120 : If we are not allowed to acquire an existing pair for
121 : modify (INVAL) or if the line line_idx is already acquired
122 : for anything (AGAIN), fail. */
123 :
124 0 : if( FD_UNLIKELY( ref ) ) DONE( FD_VINYL_ERR_AGAIN );
125 0 : if( FD_UNLIKELY( req_flag_excl ) ) DONE( FD_VINYL_ERR_INVAL );
126 :
127 0 : fd_vinyl_bstream_phdr_t * phdr = fd_vinyl_data_obj_phdr( obj );
128 :
129 0 : FD_CRIT( !obj->rd_active, "corruption detected" );
130 0 : FD_CRIT( fd_vinyl_data_obj_val_max( obj ) >= val_sz, "corruption detected" );
131 0 : FD_CRIT( phdr->ctl==fd_vinyl_bstream_ctl( FD_VINYL_BSTREAM_CTL_TYPE_PAIR,
132 0 : FD_VINYL_BSTREAM_CTL_STYLE_RAW, val_sz ), "corruption detected" );
133 0 : FD_CRIT( fd_vinyl_key_eq( &phdr->key, key ), "corruption detected" );
134 0 : FD_CRIT( !memcmp( &phdr->info, &ele0[ ele_idx ].phdr.info, sizeof(fd_vinyl_info_t) ), "corruption detected" );
135 :
136 : /* If the ignore flag is set, set the cached value size to 0. */
137 :
138 0 : if( req_flag_ignore ) {
139 0 : phdr->info.val_sz = 0U;
140 0 : val_sz = 0UL;
141 0 : }
142 :
143 : /* If the current location for the pair key's data isn't
144 : sufficient to hold the worst case val_sz that the client
145 : might modify the pair's value into, adjust the space
146 : available for the pair to the user's val_max. Because we
147 : might be ignoring the existing value, this could be smaller
148 : than the current object. (We could chose to not trim in
149 : this case because it will get trimmed again on release.
150 : But doing so makes a more consistent guarantee to the
151 : client and makes testing easier.) */
152 :
153 0 : ulong csz = sizeof(fd_vinyl_bstream_phdr_t) + val_sz;
154 :
155 0 : ulong szc_new = fd_vinyl_data_szc( fd_ulong_max( val_sz, req_val_max ) );
156 0 : ulong szc_old = (ulong)obj->szc;
157 :
158 0 : if( FD_UNLIKELY( szc_new != szc_old ) ) {
159 :
160 0 : fd_vinyl_data_obj_t * obj_new = fd_vinyl_data_alloc( data, szc_new );
161 0 : if( FD_UNLIKELY( !obj_new ) ) FD_LOG_CRIT(( "increase data cache size" ));
162 :
163 0 : fd_vinyl_bstream_phdr_t * phdr_new = fd_vinyl_data_obj_phdr( obj_new );
164 :
165 0 : memcpy( phdr_new, phdr, csz );
166 :
167 0 : fd_vinyl_data_free( data, obj );
168 :
169 0 : phdr = phdr_new;
170 0 : obj = obj_new;
171 :
172 0 : line[ line_idx ].obj = obj; obj->line_idx = line_idx; obj->rd_active = (short)0;
173 0 : }
174 :
175 : /* Zero out any remaining space in the pair. */
176 :
177 0 : ulong zsz = fd_vinyl_bstream_pair_sz( fd_vinyl_data_szc_val_max( szc_new ) ) - csz;
178 0 : memset( ((uchar *)phdr) + csz, 0, zsz );
179 :
180 : /* Finish up acquiring for modify */
181 :
182 : //line[ line_idx ].obj = ... already init;
183 : //line[ line_idx ].ele_idx = ... already init;
184 0 : line[ line_idx ].ctl = fd_vinyl_line_ctl( ver+1UL, -1L ); /* bump ver */
185 :
186 0 : fd_vinyl_line_evict_prio( &vinyl->line_idx_lru, line, line_cnt, line_idx, req_evict_prio );
187 :
188 : //phdr->ctl = ... already init
189 : //phdr->key = ... already init
190 : //phdr->info = ... already init
191 :
192 : //ele0[ ele_idx ] = ... already init
193 :
194 0 : req_val_gaddr[ batch_idx ] = (ulong)fd_vinyl_data_obj_val( obj ) - data_laddr0;
195 :
196 0 : DONE( FD_VINYL_SUCCESS );
197 :
198 0 : } /* pair key data cached */
199 :
200 : /* At this point, pair key is not cached. If we are not allowed
201 : to acquire this pair, fail. Otherwise, evict the least
202 : recently used evictable line (this should always be possible
203 : if quotas are confiured correctly) to make room to cache this
204 : pair. Connect this line to meta element ele_idx, set the
205 : line's reference count appropriately, bump the line's version
206 : and move the line to the desired location in the eviction
207 : sequence. We don't modify any shared fields in meta element
208 : ele_idx so we can do the modification fast.
209 :
210 : We do this upfront to free data cache for the alloc if the
211 : LRU line is in use and to handle the same pair appearing
212 : multiple times in an acquire.
213 :
214 : That is, if req_key appears multiple times in an acquire to
215 : modify, the trailing redundant acquires will see the object
216 : as cached with ref==-1 and fail with AGAIN. If the key
217 : appears multiple times in an acquire for read, the trailing
218 : redundant acquires will see the object as cached with ref>0
219 : and rd_active==1, conclude that the first redundant acquire
220 : is in the process of reading the pair into cache, skip any
221 : racy metadata checks, increase the ref count and succeed.
222 :
223 : IMPORTANT SAFETY TIP! Note that this implies that client
224 : doing an acquire-for-read with redundant keys and with
225 : speculative processing will see req_err transition to success
226 : for the trailing redundant items for a key before the leading
227 : item of that key transitions to success (and thus before the
228 : object is fully read / verified and/or decoded). It is up to
229 : the client doing speculative cut through processing to avoid
230 : redundant keys or react accordingly. */
231 :
232 0 : if( FD_UNLIKELY( req_flag_modify & req_flag_excl ) ) DONE( FD_VINYL_ERR_INVAL );
233 :
234 0 : line_idx = fd_vinyl_line_evict_lru( &vinyl->line_idx_lru, line, line_cnt, ele0, ele_max, data );
235 :
236 0 : ulong line_ctl = line[ line_idx ].ctl;
237 :
238 0 : ulong ver = fd_vinyl_line_ctl_ver( line_ctl );
239 :
240 0 : line[ line_idx ].ele_idx = ele_idx; ele0[ ele_idx ].line_idx = line_idx;
241 0 : line[ line_idx ].ctl = fd_vinyl_line_ctl( ver+1UL, req_flag_modify ? -1L : 1L );
242 :
243 0 : fd_vinyl_line_evict_prio( &vinyl->line_idx_lru, line, line_cnt, line_idx, req_evict_prio );
244 :
245 : /* Allocate an appropriately sized object to hold this pair,
246 : connect it to this line and report the location to the client. */
247 :
248 0 : ulong val_max = fd_ulong_if( !req_flag_modify, val_sz,
249 0 : fd_ulong_if( !req_flag_ignore, fd_ulong_max( val_sz, req_val_max ),
250 0 : req_val_max ) );
251 :
252 0 : ulong szc = fd_vinyl_data_szc( val_max );
253 :
254 0 : fd_vinyl_data_obj_t * obj = fd_vinyl_data_alloc( data, szc );
255 0 : if( FD_UNLIKELY( !obj ) ) FD_LOG_CRIT(( "increase data cache size" ));
256 :
257 0 : line[ line_idx ].obj = obj; obj->line_idx = line_idx;
258 :
259 0 : void * val = fd_vinyl_data_obj_val( obj );
260 :
261 0 : req_val_gaddr[ batch_idx ] = (ulong)val - data_laddr0;
262 :
263 : /* If we need to do I/O, start reading encoded pair data and
264 : defer the data integrity and decoding to later (and then in
265 : whatever order the I/O layer sees fit). */
266 :
267 0 : if( FD_LIKELY( !(req_flag_modify & req_flag_ignore) ) ) {
268 0 : obj->rd_active = (short)1;
269 :
270 0 : int style = fd_vinyl_bstream_ctl_style( pair_ctl );
271 0 : ulong val_esz = fd_vinyl_bstream_ctl_sz ( pair_ctl );
272 :
273 0 : FD_CRIT( val_esz<=FD_VINYL_VAL_MAX, "corruption detected" );
274 0 : FD_CRIT( (style!=FD_VINYL_BSTREAM_CTL_STYLE_RAW) | (val_sz==val_esz), "corruption detected" );
275 :
276 0 : fd_vinyl_data_obj_t * cobj;
277 :
278 0 : if( FD_LIKELY( style==FD_VINYL_BSTREAM_CTL_STYLE_RAW ) ) cobj = obj;
279 0 : else {
280 0 : cobj = fd_vinyl_data_alloc( data, fd_vinyl_data_szc( val_esz ) );
281 0 : if( FD_UNLIKELY( !cobj ) ) FD_LOG_CRIT(( "increase data cache size" ));
282 0 : }
283 :
284 0 : cobj->rd->ctx = (ulong)obj;
285 0 : cobj->rd->seq = ele0[ ele_idx ].seq;
286 0 : cobj->rd->dst = fd_vinyl_data_obj_phdr( cobj );
287 0 : cobj->rd->sz = fd_vinyl_bstream_pair_sz( val_esz );
288 :
289 0 : cobj->rd_err = req_err + batch_idx;
290 :
291 0 : fd_vinyl_io_read( io, cobj->rd );
292 0 : read_cnt++;
293 :
294 0 : quota_rem--;
295 0 : goto next_acquire;
296 0 : }
297 :
298 : /* At this point, we are acquiring to modify but we don't need
299 : the existing value. We populate the cached pair header
300 : appropriately for the modify and zero the rest to complete
301 : this request immediately. */
302 :
303 0 : obj->rd_active = (short)0;
304 :
305 0 : fd_vinyl_bstream_phdr_t * phdr = fd_vinyl_data_obj_phdr( obj );
306 :
307 0 : phdr->ctl = fd_vinyl_bstream_ctl( FD_VINYL_BSTREAM_CTL_TYPE_PAIR, FD_VINYL_BSTREAM_CTL_STYLE_RAW, val_sz );
308 0 : phdr->key = *key;
309 0 : phdr->info = ele0[ ele_idx ].phdr.info;
310 :
311 0 : phdr->info.val_sz = 0U;
312 :
313 0 : memset( val, 0, fd_vinyl_data_szc_obj_footprint( szc ) - sizeof(fd_vinyl_data_obj_t) - sizeof(fd_vinyl_bstream_phdr_t) );
314 :
315 0 : DONE( FD_VINYL_SUCCESS );
316 :
317 0 : } /* pair key meta cached */
318 :
319 : /* At this point, pair key does not exist at bstream seq_present
320 : and is not in the process of being created. If we aren't
321 : allowed to create pair key, fail. Otherwise, evict the least
322 : recently used evictable line (this should always be possible if
323 : quotas are confiured correctly) to make room to cache this
324 : pair, set the line's reference count appropriately, bump the
325 : version and move the line to the desired location in the
326 : eviction sequence. We do this upfront to free data cache for
327 : the alloc if the LRU line is in use. */
328 :
329 0 : if( FD_UNLIKELY( !(req_flag_modify & req_flag_create) ) ) DONE( FD_VINYL_ERR_KEY );
330 :
331 0 : ulong line_idx = fd_vinyl_line_evict_lru( &vinyl->line_idx_lru, line, line_cnt, ele0, ele_max, data );
332 :
333 0 : ulong line_ctl = line[ line_idx ].ctl;
334 :
335 0 : ulong ver = fd_vinyl_line_ctl_ver( line_ctl );
336 :
337 0 : line[ line_idx ].ctl = fd_vinyl_line_ctl( ver+1UL, -1L );
338 :
339 0 : fd_vinyl_line_evict_prio( &vinyl->line_idx_lru, line, line_cnt, line_idx, req_evict_prio );
340 :
341 : /* Allocate an appropriately sized object to hold this pair and
342 : connect it to this line. */
343 :
344 0 : ulong szc = fd_vinyl_data_szc( req_val_max );
345 :
346 0 : fd_vinyl_data_obj_t * obj = fd_vinyl_data_alloc( data, szc );
347 0 : if( FD_UNLIKELY( !obj ) ) FD_LOG_CRIT(( "increase data cache size" ));
348 :
349 0 : line[ line_idx ].obj = obj; obj->line_idx = line_idx; obj->rd_active = (short)0;
350 :
351 : /* Allocate a meta element to hold metadata for this pair and
352 : connect it to this line. Since we are inserting at meta
353 : element ele_idx, we don't need to lock anything so long as we
354 : mark the element as in use very last. */
355 :
356 0 : ulong pair_cnt = vinyl->pair_cnt;
357 0 : if( FD_UNLIKELY( pair_cnt>=pair_max ) ) FD_LOG_CRIT(( "increase meta cache size" ));
358 0 : vinyl->pair_cnt = pair_cnt + 1UL;
359 :
360 0 : ele0[ ele_idx ].memo = memo;
361 : //ele0[ ele_idx ].phdr.ctl init below
362 0 : ele0[ ele_idx ].phdr.key = *key;
363 0 : memset( &ele0[ ele_idx ].phdr.info, 0, sizeof(fd_vinyl_info_t) ); /* sets val_sz to 0 */
364 0 : ele0[ ele_idx ].line_idx = line_idx;
365 0 : ele0[ ele_idx ].seq = 0UL; /* Will be init on release */
366 :
367 0 : FD_COMPILER_MFENCE();
368 0 : ele0[ ele_idx ].phdr.ctl = ULONG_MAX; /* Mark as being created */
369 0 : FD_COMPILER_MFENCE();
370 :
371 0 : line[ line_idx ].ele_idx = ele_idx;
372 :
373 : /* Initialize the data region for a new pair */
374 :
375 0 : *fd_vinyl_data_obj_phdr( obj ) = ele0[ ele_idx ].phdr;
376 :
377 0 : uchar * val = (uchar *)fd_vinyl_data_obj_val( obj );
378 :
379 0 : memset( val, 0, fd_vinyl_data_szc_obj_footprint( szc ) - sizeof(fd_vinyl_data_obj_t) - sizeof(fd_vinyl_bstream_phdr_t) );
380 :
381 0 : req_val_gaddr[ batch_idx ] = (ulong)val - data_laddr0;
382 :
383 0 : DONE( FD_VINYL_SUCCESS );
384 :
385 0 : next_acquire: /* silly language restriction */;
386 :
387 0 : # undef DONE
388 :
389 0 : } /* for batch_idx */
390 :
391 0 : FD_CRIT( (!read_cnt) | (!(req_flag_modify & req_flag_ignore)), "corruption detected" );
392 :
393 0 : comp_err = FD_VINYL_SUCCESS;
394 0 : break;
395 0 : }
|
