Line data Source code
1 : #include "fd_chainer.h"
2 : #include "../../disco/shred/fd_fec_set.h"
3 : #include "../../ballet/bmtree/fd_bmtree.h"
4 : #include "../../ballet/sha256/fd_sha256.h"
5 :
6 : #include <stdio.h>
7 :
8 : void *
9 : fd_chainer_new( void * shmem,
10 : ulong ele_max,
11 : ulong max_shreds_per_block,
12 96 : ulong seed ) {
13 96 : ulong footprint = fd_chainer_footprint( ele_max, max_shreds_per_block );
14 96 : if( FD_UNLIKELY( !footprint ) ) {
15 0 : FD_LOG_WARNING(( "bad footprint: %lu %lu", ele_max, max_shreds_per_block ));
16 0 : return NULL;
17 0 : }
18 :
19 96 : fd_wksp_t * wksp = fd_wksp_containing( shmem );
20 96 : if( FD_UNLIKELY( !wksp ) ) {
21 0 : FD_LOG_WARNING(( "shmem must be part of a workspace" ));
22 0 : return NULL;
23 0 : }
24 :
25 96 : fd_memset( shmem, 0, footprint );
26 96 : fd_chainer_t * chainer;
27 :
28 96 : ulong blk_max = ele_max * FD_CHAINER_SLOT_VER_MAX;
29 96 : ulong fec_blk_max = max_shreds_per_block / FD_FEC_SHRED_CNT;
30 96 : ulong fec_max = blk_max * fec_blk_max;
31 96 : ulong fec_chain_cnt = fd_fec_map_chain_cnt_est( fec_max );
32 96 : ulong blk_chain_cnt = fd_slotv_map_chain_cnt_est( blk_max );
33 :
34 96 : FD_SCRATCH_ALLOC_INIT( l, shmem );
35 96 : chainer = FD_SCRATCH_ALLOC_APPEND( l, fd_chainer_align(), sizeof(fd_chainer_t) );
36 96 : void * fec_pool = FD_SCRATCH_ALLOC_APPEND( l, fd_fec_pool_align(), fd_fec_pool_footprint ( fec_max ) );
37 96 : void * fec_map = FD_SCRATCH_ALLOC_APPEND( l, fd_fec_map_align(), fd_fec_map_footprint ( fec_chain_cnt ) );
38 96 : void * slotv_pool = FD_SCRATCH_ALLOC_APPEND( l, fd_slotv_pool_align(), fd_slotv_pool_footprint ( blk_max ) );
39 96 : void * fec_tbl = FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), fec_max*sizeof(uint) );
40 96 : void * slotv_map = FD_SCRATCH_ALLOC_APPEND( l, fd_slotv_map_align(), fd_slotv_map_footprint ( blk_chain_cnt ) );
41 96 : void * work_pool = FD_SCRATCH_ALLOC_APPEND( l, fd_work_pool_align(), fd_work_pool_footprint ( blk_max ) );
42 96 : void * work_map = FD_SCRATCH_ALLOC_APPEND( l, fd_work_map_align(), fd_work_map_footprint ( blk_chain_cnt ) );
43 96 : void * repair_treap = FD_SCRATCH_ALLOC_APPEND( l, fd_work_repair_align(), fd_work_repair_footprint ( blk_max ) );
44 96 : void * orphan_treap = FD_SCRATCH_ALLOC_APPEND( l, fd_work_orphan_align(), fd_work_orphan_footprint ( blk_max ) );
45 96 : void * bfs = FD_SCRATCH_ALLOC_APPEND( l, bfs_align(), bfs_footprint ( blk_max ) );
46 96 : void * out_queue = FD_SCRATCH_ALLOC_APPEND( l, out_queue_align(), out_queue_footprint ( fec_max ) );
47 96 : FD_TEST( FD_SCRATCH_ALLOC_FINI( l, fd_chainer_align() ) == (ulong)shmem + footprint );
48 :
49 96 : chainer->root = ULONG_MAX;
50 96 : chainer->highest_repaired = 0UL;
51 96 : chainer->wksp_gaddr = fd_wksp_gaddr_fast( wksp, chainer );
52 96 : chainer->fec_pool = fd_fec_pool_join ( fd_fec_pool_new ( fec_pool, fec_max ) );
53 96 : chainer->fec_map = fd_fec_map_join ( fd_fec_map_new ( fec_map, fec_chain_cnt, seed ) );
54 96 : chainer->slotv_pool = fd_slotv_pool_join ( fd_slotv_pool_new ( slotv_pool, blk_max ) );
55 96 : chainer->fec_tbl = fec_tbl;
56 96 : chainer->fec_blk_max = fec_blk_max;
57 96 : chainer->slotv_map = fd_slotv_map_join ( fd_slotv_map_new ( slotv_map, blk_chain_cnt, seed ) );
58 96 : chainer->work_pool = fd_work_pool_join ( fd_work_pool_new ( work_pool, blk_max ) );
59 96 : chainer->work_map = fd_work_map_join ( fd_work_map_new ( work_map, blk_chain_cnt, seed ) );
60 96 : chainer->repair_treap = fd_work_repair_join ( fd_work_repair_new ( repair_treap, blk_max ) );
61 96 : chainer->orphan_treap = fd_work_orphan_join ( fd_work_orphan_new ( orphan_treap, blk_max ) );
62 96 : chainer->bfs = bfs_join ( bfs_new ( bfs, blk_max ) );
63 96 : chainer->out_queue = out_queue_join ( out_queue_new ( out_queue, fec_max ) );
64 :
65 96 : fd_work_repair_seed( chainer->work_pool, blk_max, seed );
66 96 : fd_work_orphan_seed( chainer->work_pool, blk_max, seed ^ 0x9eUL );
67 :
68 96 : FD_COMPILER_MFENCE();
69 96 : FD_VOLATILE( chainer->magic ) = FD_CHAINER_MAGIC;
70 96 : FD_COMPILER_MFENCE();
71 :
72 96 : return shmem;
73 96 : }
74 :
75 : fd_chainer_t *
76 96 : fd_chainer_join( void * shchainer ) {
77 96 : fd_chainer_t * chainer = (fd_chainer_t *)shchainer;
78 96 : if( FD_UNLIKELY( chainer->magic!=FD_CHAINER_MAGIC ) ) {
79 0 : FD_LOG_WARNING(( "bad magic" ));
80 0 : return NULL;
81 0 : }
82 96 : return chainer;
83 96 : }
84 :
85 : /* slotv_iter_{init,next} iterate the versions of a slot via the
86 : MAP_MULTI chain. Usage:
87 : for( ulong i=slotv_iter_init(chainer,slot); i!=ULONG_MAX; i=slotv_iter_next(chainer,i) ) {
88 : fd_chainer_slotv_t * slotv = slotv_iter_ele( chainer, i );
89 : ...
90 : } */
91 :
92 : static inline ulong
93 94287 : slotv_iter_init( fd_chainer_t * chainer, ulong slot ) {
94 94287 : return fd_slotv_map_idx_query_const( chainer->slotv_map, &slot, ULONG_MAX, chainer->slotv_pool );
95 94287 : }
96 :
97 : static inline ulong
98 98670 : slotv_iter_next( fd_chainer_t * chainer, ulong idx ) {
99 98670 : return fd_slotv_map_idx_next_const( idx, ULONG_MAX, chainer->slotv_pool );
100 98670 : }
101 :
102 : static inline fd_chainer_slotv_t *
103 105645 : slotv_iter_ele( fd_chainer_t * chainer, ulong idx ) {
104 105645 : return fd_slotv_pool_ele( chainer->slotv_pool, idx );
105 105645 : }
106 :
107 : /* acquire_slotv allocates, initializes, and map-inserts a fresh
108 : (turbine, i.e. all-zero block_id) version of slot. Callers that know
109 : the version's block_id (notar-fallback, parent discovery) set it after. */
110 :
111 : static fd_chainer_slotv_t *
112 327 : acquire_slotv( fd_chainer_t * chainer, ulong slot ) {
113 327 : fd_slotv_map_t * slotv_map = chainer->slotv_map;
114 327 : fd_chainer_slotv_t * slotv_pool = chainer->slotv_pool;
115 327 : FD_TEST( fd_slotv_pool_free( slotv_pool ) );
116 :
117 327 : ulong slotv_cnt = 0UL;
118 516 : for( ulong i=slotv_iter_init( chainer, slot ); i!=ULONG_MAX; i=slotv_iter_next( chainer, i ) ) {
119 189 : slotv_cnt++;
120 189 : }
121 327 : if( FD_UNLIKELY( slotv_cnt>=FD_CHAINER_SLOT_VER_MAX ) ) FD_LOG_CRIT(( "slots stored exceeds protocol limits, %lu versions of slot %lu already stored", slotv_cnt, slot ));
122 :
123 327 : fd_chainer_slotv_t * slotv = fd_slotv_pool_ele_acquire( slotv_pool );
124 327 : slotv->slot = slot;
125 327 : slotv->turbine = 0;
126 327 : slotv->abandoned = 0;
127 327 : slotv->parent_slot = AG_UNKNOWN_SLOT;
128 327 : slotv->parent_slot_batch = UINT_MAX;
129 327 : slotv->complete_idx = UINT_MAX;
130 327 : slotv->buffered_idx = UINT_MAX;
131 327 : slotv->buffered_fec_idx = UINT_MAX;
132 327 : slotv->delivered_idx = UINT_MAX;
133 327 : slotv->connected = 0;
134 327 : slotv->highest_requested = UINT_MAX;
135 :
136 327 : fd_memset( &slotv->block_id, 0, sizeof(fd_hash_t) );
137 327 : fd_memset( &slotv->parent_block_id, 0, sizeof(fd_hash_t) );
138 327 : fd_memset( fd_chainer_slotv_fecs( chainer, slotv ), 0xff, chainer->fec_blk_max*sizeof(uint) ); /* UINT_MAX pool_idx sentinel */
139 :
140 327 : fd_slotv_map_ele_insert( slotv_map, slotv, slotv_pool );
141 327 : fd_chainer_repair_add( chainer, slotv ); /* new slotv -> has un-requested shreds */
142 327 : fd_chainer_orphan_add( chainer, slotv ); /* new slotv -> ancestry unknown until parent confirmed present */
143 327 : return slotv;
144 327 : }
145 :
146 : /* orphans_resolve releases every orphan whose ancestry is now settled */
147 : static void
148 219 : orphans_resolve( fd_chainer_t * chainer ) {
149 219 : ulong next;
150 399 : for( ulong it=fd_chainer_orphan_iter_init( chainer ); !fd_chainer_work_iter_done( it ); it=next ) {
151 180 : next = fd_chainer_orphan_iter_next( chainer, it );
152 180 : fd_chainer_slotv_t * o = fd_chainer_work_iter_ele( chainer, it );
153 :
154 180 : int parent_resolved = o->parent_slot!=AG_UNKNOWN_SLOT && ( o->parent_slot<=chainer->root || fd_chainer_slot_version_query( chainer, o->parent_slot, &o->parent_block_id ) );
155 180 : if( FD_LIKELY( parent_resolved ) ) fd_chainer_orphan_remove( chainer, o );
156 180 : }
157 219 : }
158 :
159 : void
160 : fd_chainer_init( fd_chainer_t * chainer,
161 : ulong slot,
162 96 : fd_hash_t const * block_id ) {
163 96 : fd_chainer_slotv_t * slotv = acquire_slotv( chainer, slot );
164 96 : slotv->parent_slot = slot;
165 96 : slotv->complete_idx = 0;
166 96 : slotv->buffered_idx = 0;
167 96 : slotv->connected = 1;
168 96 : slotv->delivered_idx = 0; /* must equal complete_idx at init */
169 96 : slotv->highest_requested = 0;
170 96 : slotv->buffered_fec_idx = UINT_MAX; /* no complete FEC set buffered; must
171 : be one-below a FD_FEC_SHRED_CNT
172 : multiple, which UINT_MAX satisfies */
173 96 : slotv->block_id = *block_id;
174 96 : fd_chainer_repair_remove( chainer, slotv );
175 96 : fd_chainer_orphan_remove( chainer, slotv );
176 :
177 96 : chainer->root = slot;
178 96 : chainer->highest_repaired = slot;
179 96 : }
180 :
181 : /* slotv_fec returns the FEC that slotv owns at fec_set_idx, or NULL if
182 : it holds none there (or fec_set_idx is beyond max_shreds_per_block). */
183 :
184 : static fd_chainer_fec_t *
185 240339 : slotv_fec( fd_chainer_t * chainer, fd_chainer_slotv_t const * slotv, uint fec_set_idx ) {
186 240339 : ulong k = fec_set_idx / FD_FEC_SHRED_CNT;
187 240339 : if( FD_UNLIKELY( k>=chainer->fec_blk_max ) ) return NULL;
188 240327 : uint idx = fd_chainer_slotv_fecs( chainer, slotv )[ k ];
189 240327 : if( FD_UNLIKELY( idx==UINT_MAX ) ) return NULL;
190 78879 : return fd_fec_pool_ele( chainer->fec_pool, (ulong)idx );
191 240327 : }
192 :
193 : fd_chainer_fec_t *
194 : fd_chainer_fec_query( fd_chainer_t * chainer,
195 : ulong slot,
196 : uint fec_set_idx,
197 2094 : fd_hash_t const * block_id ) {
198 2094 : fd_chainer_slotv_t * slotv = fd_chainer_slot_version_query( chainer, slot, block_id );
199 2094 : if( FD_UNLIKELY( !slotv ) ) return NULL;
200 2091 : return slotv_fec( chainer, slotv, fec_set_idx );
201 2094 : }
202 :
203 : /* fec_query returns the FEC whose merkle_root matches mr, or NULL. */
204 :
205 : static fd_chainer_fec_t *
206 34824 : fec_query( fd_chainer_t * chainer, fd_hash_t const * mr ) {
207 34824 : fd_fec_map_t * fec_map = chainer->fec_map;
208 34824 : fd_chainer_fec_t * fec_pool = chainer->fec_pool;
209 34824 : return fd_fec_map_ele_query( fec_map, mr, NULL, fec_pool );
210 34824 : }
211 :
212 : /* fec_join records that slotv includes the FEC at (slot, fec_set_idx)
213 : with root mr, creating the entry if this root has not been seen yet. */
214 :
215 : static fd_chainer_fec_t *
216 : fec_join( fd_chainer_t * chainer,
217 : ulong slot,
218 : uint fec_set_idx,
219 : fd_chainer_slotv_t * slotv,
220 582 : fd_hash_t const * mr ) {
221 582 : if( FD_UNLIKELY( slot>(ulong)UINT_MAX ) ) FD_LOG_CRIT(( "slot %lu exceeds uint range", slot ));
222 582 : ulong k = fec_set_idx / FD_FEC_SHRED_CNT;
223 582 : FD_TEST( k<chainer->fec_blk_max );
224 :
225 582 : fd_fec_map_t * fec_map = chainer->fec_map;
226 582 : fd_chainer_fec_t * fec_pool = chainer->fec_pool;
227 582 : fd_chainer_fec_t * fec = fd_fec_map_ele_query( fec_map, mr, NULL, fec_pool );
228 582 : if( FD_UNLIKELY( !fec ) ) {
229 507 : if( FD_UNLIKELY( !fd_fec_pool_free( fec_pool ) ) ) FD_LOG_CRIT(( "fec_pool is full" ));
230 507 : fec = fd_fec_pool_ele_acquire( fec_pool );
231 507 : fec->merkle_root = *mr;
232 507 : fec->slot = (uint)slot;
233 507 : fec->fec_set_idx = fec_set_idx & ((1U<<28)-1U);
234 507 : fec->data_idxs = 0U;
235 507 : fec->complete = 0;
236 507 : fec->slot_complete = 0;
237 507 : fec->data_complete = 0;
238 507 : fec->is_leader = 0;
239 507 : fd_fec_map_ele_insert( fec_map, fec, fec_pool );
240 507 : }
241 582 : fd_chainer_slotv_fecs( chainer, slotv )[ k ] = (uint)fd_fec_pool_idx( fec_pool, fec );
242 582 : return fec;
243 582 : }
244 :
245 : int
246 : fd_chainer_shred_test( fd_chainer_t * chainer,
247 : fd_chainer_slotv_t const * slotv,
248 63654 : uint shred_idx ) {
249 63654 : fd_chainer_fec_t * fec = slotv_fec( chainer, slotv, shred_idx & ~( (uint)FD_FEC_SHRED_CNT - 1U ) );
250 63654 : if( FD_UNLIKELY( !fec ) ) return 0;
251 41871 : return !!( fec->data_idxs & ( 1U << ( shred_idx & ( (uint)FD_FEC_SHRED_CNT - 1U ) ) ) );
252 63654 : }
253 :
254 : /* slotv_abandon freezes a turbine slotv. Removed from the repair
255 : worklists, and (via the abandoned flag) excluded from delivery and
256 : block_id finalization. */
257 :
258 : static void
259 54 : slotv_abandon( fd_chainer_t * chainer, fd_chainer_slotv_t * slotv ) {
260 54 : FD_TEST( slotv->turbine );
261 54 : fd_chainer_repair_remove( chainer, slotv );
262 54 : fd_chainer_orphan_remove( chainer, slotv );
263 54 : slotv->abandoned = 1;
264 54 : }
265 :
266 : /* abandon_turbine abandons slot's turbine version, if one exists. */
267 :
268 : static void
269 3 : abandon_turbine( fd_chainer_t * chainer, ulong slot ) {
270 6 : for( ulong i=slotv_iter_init( chainer, slot ); i!=ULONG_MAX; i=slotv_iter_next( chainer, i ) ) {
271 6 : fd_chainer_slotv_t * slotv = slotv_iter_ele( chainer, i );
272 6 : if( FD_LIKELY( !slotv->turbine || slotv->abandoned ) ) continue;
273 3 : if( FD_LIKELY( fd_hash_check_zero( &slotv->block_id ) ) ) slotv_abandon( chainer, slotv );
274 3 : return;
275 6 : }
276 3 : }
277 :
278 : /* turbine_slotv_query returns the turbine version of slot -- creating
279 : it if none exists. */
280 :
281 : static fd_chainer_slotv_t *
282 33618 : turbine_slotv_query( fd_chainer_t * chainer, ulong slot ) {
283 52044 : for( ulong i=slotv_iter_init( chainer, slot ); i!=ULONG_MAX; i=slotv_iter_next( chainer, i ) ) {
284 51906 : fd_chainer_slotv_t * slotv = slotv_iter_ele( chainer, i );
285 51906 : if( FD_LIKELY( slotv->turbine ) ) return slotv;
286 51906 : }
287 138 : fd_chainer_slotv_t * slotv = acquire_slotv( chainer, slot );
288 138 : slotv->turbine = 1;
289 138 : return slotv;
290 33618 : }
291 :
292 : /* finalize_block_id computes the slotv's double-merkle block_id and
293 : writes it to slotv->block_id. Returns 1 on success, 0 on failure. */
294 :
295 : static int
296 96 : finalize_block_id( fd_chainer_t * chainer, fd_chainer_slotv_t * slotv ) {
297 96 : if( FD_UNLIKELY( slotv->complete_idx==UINT_MAX ) ) return 0;
298 96 : if( FD_UNLIKELY( slotv->parent_slot==AG_UNKNOWN_SLOT ) ) return 0;
299 96 : if( FD_UNLIKELY( fd_hash_check_zero( &slotv->parent_block_id ) ) ) return 0;
300 :
301 96 : uint fec_set_cnt = ( slotv->complete_idx + 1U ) / FD_FEC_SHRED_CNT;
302 96 : uchar tree_mem[ FD_BMTREE_COMMIT_FOOTPRINT( 0UL ) ] __attribute__((aligned(FD_BMTREE_COMMIT_ALIGN)));
303 96 : fd_bmtree_commit_t * tree = fd_bmtree_commit_init( tree_mem, 20UL, FD_BMTREE_LONG_PREFIX_SZ, 0UL );
304 :
305 465 : for( uint i=0U; i<fec_set_cnt; i++ ) {
306 369 : fd_chainer_fec_t * fec = slotv_fec( chainer, slotv, i*FD_FEC_SHRED_CNT );
307 369 : if( FD_UNLIKELY( !fec ) ) return 0;
308 :
309 369 : fd_bmtree_node_t leaf[1];
310 369 : memcpy( leaf->hash, fec->merkle_root.uc, sizeof(fd_hash_t) );
311 369 : fd_bmtree_commit_append( tree, leaf, 1UL );
312 369 : }
313 :
314 : /* final parent-info leaf */
315 96 : fd_bmtree_node_t parent_info[1];
316 96 : fd_sha256_t sha[1];
317 96 : fd_sha256_init ( sha );
318 96 : fd_sha256_append( sha, &slotv->parent_slot, sizeof(ulong) );
319 96 : fd_sha256_append( sha, slotv->parent_block_id.uc, sizeof(fd_hash_t) );
320 96 : fd_sha256_append( sha, &fec_set_cnt, sizeof(uint) );
321 96 : fd_sha256_fini ( sha, parent_info->hash );
322 96 : fd_bmtree_commit_append( tree, parent_info, 1UL );
323 :
324 96 : uchar * root = fd_bmtree_commit_fini( tree );
325 96 : memcpy( slotv->block_id.uc, root, sizeof(fd_hash_t) );
326 96 : return 1;
327 96 : }
328 :
329 : static void
330 : fec_rekey( fd_chainer_t * chainer,
331 : fd_chainer_fec_t * sentinel,
332 : fd_hash_t const * full_mr );
333 :
334 : void
335 : fd_chainer_shred_insert( fd_chainer_t * chainer,
336 : ulong slot,
337 : uint shred_idx,
338 : int slot_complete,
339 : fd_hash_t const * mr,
340 : ulong parent_slot,
341 33528 : fd_hash_t const * parent_block_id ) {
342 33528 : FD_TEST( slot>chainer->root );
343 33528 : uint fec_set_idx = shred_idx & ~( (uint)FD_FEC_SHRED_CNT - 1U );
344 33528 : ulong k = fec_set_idx / FD_FEC_SHRED_CNT;
345 33528 : uint shred_max = (uint)( chainer->fec_blk_max*FD_FEC_SHRED_CNT );
346 33528 : FD_TEST( k<chainer->fec_blk_max ); /* guaranteed by fec_resolver */
347 :
348 : /* Identify the slot versions this shred belongs to. */
349 :
350 33528 : fd_chainer_slotv_t * turbine = turbine_slotv_query( chainer, slot );
351 :
352 : /* If a votor-driven version of the slot already exists (block-id
353 : repair started before this turbine shred arrived), abandon the
354 : turbine version. */
355 33528 : if( FD_UNLIKELY( !turbine->abandoned && fd_hash_check_zero( &turbine->block_id ) ) ) {
356 52638 : for( ulong i=slotv_iter_init( chainer, slot ); i!=ULONG_MAX; i=slotv_iter_next( chainer, i ) ) {
357 26319 : if( FD_UNLIKELY( i!=fd_slotv_pool_idx( chainer->slotv_pool, turbine ) ) ) {
358 0 : slotv_abandon( chainer, turbine );
359 0 : break;
360 0 : }
361 26319 : }
362 26319 : }
363 :
364 : /* A getFecRoot response carries only the 20-byte root prefix, so the
365 : FEC it created (the sentinel) is keyed by the zero-padded prefix.
366 : If a sentinel for mr exists, re-key it now so this and later shreds
367 : find it by full root. */
368 :
369 33528 : fd_hash_t prefix = {0};
370 33528 : memcpy( prefix.uc, mr->uc, FD_SHRED_MERKLE_NODE_SZ );
371 33528 : if( FD_LIKELY( !fd_hash_eq( &prefix, mr ) ) ) {
372 585 : fd_chainer_fec_t * sentinel = fec_query( chainer, &prefix );
373 585 : if( FD_UNLIKELY( sentinel && sentinel->slot==(uint)slot && sentinel->fec_set_idx==fec_set_idx ) ) {
374 12 : fec_rekey( chainer, sentinel, mr );
375 12 : }
376 585 : }
377 :
378 : /* Find or create the FEC for this shred's root
379 :
380 : If the turbine version holds no root at this position it adopts
381 : this one, whether it is newly seen FEC or an entry a getFecRoot
382 : sentinel already created. If turbine already holds a *different*
383 : root here and nothing authorized this one, the shred is an
384 : unauthorized equivocation and is dropped. */
385 :
386 33528 : fd_chainer_fec_t * fec = fec_query( chainer, mr );
387 33528 : fd_chainer_fec_t * turbine_fec = slotv_fec( chainer, turbine, fec_set_idx );
388 33528 : if( FD_LIKELY( !turbine_fec ) ) {
389 438 : fec = fec_join( chainer, slot, fec_set_idx, turbine, mr );
390 33090 : } else if( FD_UNLIKELY( !fec ) ) {
391 195 : return;
392 195 : }
393 :
394 33333 : fec->data_idxs |= 1U << ( shred_idx - fec_set_idx );
395 33333 : if( FD_UNLIKELY( slot_complete ) ) fec->slot_complete = 1;
396 :
397 : /* Update every version that owns this FEC root at this position. */
398 :
399 33333 : uint fec_idx = (uint)fd_fec_pool_idx( chainer->fec_pool, fec );
400 33333 : for( ulong _i =slotv_iter_init( chainer, slot );
401 85971 : _i!=ULONG_MAX;
402 52638 : _i =slotv_iter_next( chainer, _i ) ) {
403 52638 : fd_chainer_slotv_t * slotv = slotv_iter_ele( chainer, _i );
404 52638 : if( FD_UNLIKELY( fd_chainer_slotv_fecs( chainer, slotv )[ k ]!=fec_idx ) ) continue;
405 :
406 : /* update slot-level shred indexing */
407 38145 : if( FD_UNLIKELY( slot_complete ) ) slotv->complete_idx = shred_idx;
408 59016 : while( slotv->buffered_idx + 1 < shred_max && fd_chainer_shred_test( chainer, slotv, slotv->buffered_idx + 1U ) ) {
409 20871 : slotv->buffered_idx++;
410 20871 : }
411 :
412 : /* If equivocating, buffered_idx needs to be clamped to complete_idx */
413 38145 : if( FD_UNLIKELY( slotv->buffered_idx != UINT_MAX && slotv->complete_idx != UINT_MAX && slotv->buffered_idx > slotv->complete_idx ) ) slotv->buffered_idx = slotv->complete_idx;
414 :
415 : /* parent_slot_batch tracks which batch the information came from
416 : so a later UpdateParent supersedes the header (it may only move
417 : forward). UINT_MAX means "nothing known yet", so it is not a
418 : batch index to compare against. */
419 38145 : if( FD_UNLIKELY( parent_slot!=AG_UNKNOWN_SLOT && ( slotv->parent_slot_batch==UINT_MAX || shred_idx>slotv->parent_slot_batch ) ) ) {
420 150 : FD_TEST( parent_block_id ); /* TODO handholding check */
421 :
422 150 : slotv->parent_slot = parent_slot;
423 150 : slotv->parent_slot_batch = shred_idx;
424 150 : slotv->parent_block_id = *parent_block_id;
425 150 : if( parent_slot < chainer->root || fd_chainer_slot_version_query( chainer, slotv->parent_slot, &slotv->parent_block_id ) ) {
426 147 : fd_chainer_orphan_remove( chainer, slotv ); /* TODO check safe in FLH case */
427 147 : }
428 :
429 150 : fd_chainer_slotv_t * parent = fd_chainer_slot_version_query( chainer, parent_slot, parent_block_id );
430 150 : if( FD_LIKELY( parent && parent->connected ) ) slotv->connected = 1;
431 150 : }
432 38145 : }
433 33333 : }
434 :
435 : /* chainer_deliver queues a delivered FEC for publish to replay. The
436 : rotor tile drains the out_queue in after_credit. */
437 :
438 : static void
439 : chainer_deliver( fd_chainer_t * chainer,
440 : fd_chainer_slotv_t * slotv,
441 519 : fd_chainer_fec_t * fec ) {
442 519 : out_ele_t * out_queue = chainer->out_queue;
443 519 : if( FD_UNLIKELY( out_queue_full( out_queue ) ) ) FD_LOG_CRIT(( "chainer out_queue full" ));
444 519 : out_queue_push_tail( out_queue, (out_ele_t){ .slotv_idx = (uint)fd_slotv_pool_idx( chainer->slotv_pool, slotv ),
445 519 : .fec_idx = (uint)fd_fec_pool_idx ( chainer->fec_pool, fec ) } );
446 519 : }
447 :
448 : /* chainer_advance delivers as many contiguous completed FEC sets as
449 : possible from `root` slotv, then cascades: when an slotv's
450 : slot_complete FEC is delivered, every child slotv (parent_block_id ==
451 : this slotv's block_id) becomes connected and is drained in turn. */
452 :
453 : static void
454 783 : chainer_advance( fd_chainer_t * chainer, fd_chainer_slotv_t * root ) {
455 783 : fd_slotv_map_t * slotv_map = chainer->slotv_map;
456 783 : fd_chainer_slotv_t * slotv_pool = chainer->slotv_pool;
457 783 : ulong * bfs = chainer->bfs;
458 :
459 783 : bfs_push_tail( bfs, fd_slotv_pool_idx( slotv_pool, root ) );
460 :
461 1569 : while( FD_LIKELY( !bfs_empty( bfs ) ) ) {
462 786 : fd_chainer_slotv_t * slotv = fd_slotv_pool_ele( slotv_pool, bfs_pop_head( bfs ) );
463 786 : if( FD_UNLIKELY( !slotv->connected ) ) continue;
464 774 : if( FD_UNLIKELY( slotv->abandoned ) ) continue;
465 :
466 774 : fd_chainer_slotv_t * parent = fd_chainer_slot_version_query( chainer, slotv->parent_slot, &slotv->parent_block_id );
467 774 : if( FD_UNLIKELY( !parent || parent->complete_idx==UINT_MAX || parent->delivered_idx!=parent->complete_idx ) ) continue;
468 :
469 1158 : for(;;) {
470 1158 : uint next = slotv->delivered_idx==UINT_MAX ? 0U
471 1158 : : slotv->delivered_idx + 1;
472 1158 : fd_chainer_fec_t * fec = slotv_fec( chainer, slotv, next );
473 1158 : if( FD_LIKELY( !fec || !fec->complete ) ) break; /* next FEC not completed yet */
474 :
475 519 : chainer_deliver( chainer, slotv, fec );
476 519 : slotv->delivered_idx = next + (FD_FEC_SHRED_CNT - 1);
477 :
478 519 : if( FD_UNLIKELY( fec->slot_complete ) ) {
479 132 : chainer->highest_repaired = fd_ulong_max( chainer->highest_repaired, slotv->slot );
480 132 : fd_chainer_repair_remove( chainer, slotv ); /* nothing left to repair */
481 132 : FD_TEST( !fd_hash_check_zero( &slotv->block_id ) );
482 :
483 : /* Scan for children. TODO could index children by
484 : parent_block_id; O(n) scan for now. */
485 132 : for( fd_slotv_map_iter_t it = fd_slotv_map_iter_init( slotv_map, slotv_pool );
486 555 : !fd_slotv_map_iter_done( it, slotv_map, slotv_pool );
487 423 : it = fd_slotv_map_iter_next( it, slotv_map, slotv_pool ) ) {
488 423 : fd_chainer_slotv_t * child = fd_slotv_map_iter_ele( it, slotv_map, slotv_pool );
489 423 : if( FD_UNLIKELY( fd_hash_eq( &child->parent_block_id, &slotv->block_id ) ) ) {
490 3 : child->connected = 1;
491 3 : bfs_push_tail( bfs, fd_slotv_pool_idx( slotv_pool, child ) );
492 3 : }
493 423 : }
494 132 : break;
495 132 : }
496 519 : }
497 771 : }
498 783 : }
499 :
500 : int
501 : fd_chainer_fec_complete( fd_chainer_t * chainer,
502 : ulong slot,
503 : uint fec_set_idx_,
504 : int slot_complete,
505 : int data_complete,
506 : int is_leader,
507 549 : fd_hash_t * mr ) {
508 549 : FD_TEST( slot>chainer->root );
509 549 : uint fec_set_idx = (uint)fec_set_idx_;
510 549 : ulong k = fec_set_idx / FD_FEC_SHRED_CNT;
511 549 : FD_TEST( k<chainer->fec_blk_max ); /* guaranteed by fec_resolver */
512 :
513 18117 : for( uint i=0U; i<FD_FEC_SHRED_CNT; i++ ) {
514 17568 : fd_chainer_shred_insert( chainer, slot, fec_set_idx_ + i, slot_complete && ( i==FD_FEC_SHRED_CNT-1 ), mr, AG_UNKNOWN_SLOT, NULL );
515 17568 : }
516 :
517 : /* By the time we get here the FEC exists unless turbine refused an
518 : unauthorized equivocating root -- in which case it was dropped and
519 : there is nothing to complete. */
520 :
521 549 : fd_chainer_fec_t * fec = fec_query( chainer, mr );
522 549 : if( FD_UNLIKELY( !fec ) ) return 1;
523 :
524 546 : fec->complete = 1; /* set is now reconstructable -> deliverable */
525 546 : if( FD_UNLIKELY( slot_complete ) ) fec->slot_complete = 1;
526 546 : if( FD_UNLIKELY( data_complete ) ) fec->data_complete = 1;
527 546 : if( FD_UNLIKELY( is_leader ) ) fec->is_leader = 1;
528 :
529 546 : uint fec_idx = (uint)fd_fec_pool_idx( chainer->fec_pool, fec );
530 :
531 1458 : for( ulong _i=slotv_iter_init( chainer, slot ); _i!=ULONG_MAX; _i=slotv_iter_next( chainer, _i ) ) {
532 912 : fd_chainer_slotv_t * slotv = slotv_iter_ele( chainer, _i );
533 912 : if( FD_UNLIKELY( fd_chainer_slotv_fecs( chainer, slotv )[ k ]!=fec_idx || slotv->abandoned ) ) continue;
534 :
535 1155 : for(;;) {
536 1155 : fd_chainer_fec_t * next = slotv_fec( chainer, slotv, slotv->buffered_fec_idx + 1U );
537 1155 : if( !next || !next->complete ) break;
538 525 : slotv->buffered_fec_idx += FD_FEC_SHRED_CNT;
539 525 : }
540 :
541 : /* clamp buffered_fec_idx to complete_idx always. should never happen for non-turbine versions */
542 630 : if( FD_UNLIKELY( slotv->complete_idx!=UINT_MAX && slotv->buffered_fec_idx!=UINT_MAX &&
543 630 : slotv->buffered_fec_idx>slotv->complete_idx ) ) slotv->buffered_fec_idx = slotv->complete_idx;
544 :
545 630 : if( FD_LIKELY( slotv->turbine ) ) {
546 : /* slot is complete implies we can record the block_id. Only the
547 : turbine version needs its block_id computed. */
548 447 : fd_chainer_slotv_t * turbine = slotv;
549 447 : if( FD_UNLIKELY( turbine->complete_idx!=UINT_MAX &&
550 447 : turbine->buffered_fec_idx==turbine->complete_idx &&
551 447 : fd_hash_check_zero( &turbine->block_id ) ) ) {
552 96 : if( FD_LIKELY( finalize_block_id( chainer, turbine ) ) ) orphans_resolve( chainer ); /* children waiting on this block_id */
553 0 : else FD_LOG_WARNING(( "failed to finalize block_id for slot %lu, parent_slot %lu parent_bid is zero %d", slot, turbine->parent_slot, fd_hash_check_zero( &turbine->parent_block_id ) ));
554 96 : }
555 447 : }
556 :
557 630 : chainer_advance( chainer, slotv );
558 630 : }
559 546 : return 0;
560 549 : }
561 :
562 : void
563 : fd_chainer_fec_evicted( fd_chainer_t * chainer,
564 : ulong slot,
565 : uint fec_set_idx,
566 6 : fd_hash_t * merkle_root ) {
567 6 : fd_chainer_fec_t * fec = fec_query( chainer, merkle_root );
568 6 : if( FD_UNLIKELY( !fec ) ) return;
569 6 : ulong k = fec_set_idx / FD_FEC_SHRED_CNT;
570 6 : FD_TEST( k<chainer->fec_blk_max ); /* guaranteed by fec_resolver */
571 :
572 : /* We choose not to remove the FEC from the chainer. If this FEC
573 : belongs to a turbine slot and we are having trouble completing it
574 : (the leader gave up on disseminating the shreds), then eventually
575 : this slot will get skipped or we will repair a different version
576 : through a votor repair block id event. If this FEC is part of a
577 : votor cert, then we should keep it in the chainer because the
578 : merkle root is verified and we definitely want to continue
579 : repairing it; it is getting evicted only because fec_resolver is
580 : under pressure. */
581 :
582 6 : fec->data_idxs = 0U;
583 6 : uint fec_idx = (uint)fd_fec_pool_idx( chainer->fec_pool, fec );
584 :
585 : /* find slots that have this FEC root */
586 15 : for( ulong _i=slotv_iter_init( chainer, slot ); _i!=ULONG_MAX; _i=slotv_iter_next( chainer, _i ) ) {
587 9 : fd_chainer_slotv_t * slotv = slotv_iter_ele( chainer, _i );
588 9 : if( FD_UNLIKELY( fd_chainer_slotv_fecs( chainer, slotv )[ k ]!=fec_idx ) ) continue;
589 :
590 : /* rederive buffered_idx */
591 6 : if( FD_UNLIKELY( slotv->buffered_idx!=UINT_MAX && slotv->buffered_idx>=fec_set_idx ) ) {
592 6 : slotv->buffered_idx = fec_set_idx - 1U;
593 6 : }
594 6 : if( FD_UNLIKELY( slotv->highest_requested!=UINT_MAX && slotv->highest_requested>=fec_set_idx ) ) {
595 3 : slotv->highest_requested = fec_set_idx - 1U;
596 3 : }
597 6 : if( FD_LIKELY( !slotv->abandoned ) ) fd_chainer_repair_add( chainer, slotv ); /* abandoned versions stay off the worklists */
598 6 : }
599 6 : }
600 :
601 : fd_chainer_slotv_t *
602 : fd_chainer_verified_parent_fec_count( fd_chainer_t * chainer,
603 : ulong slot,
604 : fd_hash_t * block_id,
605 : uint fec_set_cnt,
606 : ulong parent_slot,
607 63 : fd_hash_t * parent_block_id ) {
608 63 : fd_chainer_slotv_t * slotv = fd_chainer_slot_version_query( chainer, slot, block_id );
609 63 : if( FD_UNLIKELY( !slotv ) ) FD_LOG_CRIT(( "slotv not found for slot %lu", slot ));
610 :
611 63 : FD_TEST( fec_set_cnt>0U && fec_set_cnt<=chainer->fec_blk_max );
612 63 : slotv->complete_idx = ( fec_set_cnt*FD_FEC_SHRED_CNT ) - 1;
613 63 : slotv->parent_slot = parent_slot;
614 63 : slotv->parent_block_id = *parent_block_id;
615 :
616 63 : fd_chainer_slotv_t * parent_slotv = fd_chainer_slot_version_query( chainer, parent_slot, parent_block_id );
617 63 : if( FD_UNLIKELY( !parent_slotv ) ) {
618 3 : if( FD_UNLIKELY( parent_slot<=chainer->root ) ) {
619 : /* Names a parent that is a dead fork. */
620 0 : fd_chainer_orphan_remove( chainer, slotv );
621 0 : fd_chainer_repair_remove( chainer, slotv );
622 0 : return NULL;
623 0 : }
624 3 : parent_slotv = acquire_slotv( chainer, parent_slot );
625 3 : parent_slotv->block_id = *parent_block_id;
626 3 : abandon_turbine( chainer, parent_slot );
627 3 : orphans_resolve( chainer ); /* siblings waiting on this parent */
628 3 : }
629 :
630 63 : fd_chainer_orphan_remove( chainer, slotv );
631 63 : fd_chainer_repair_add( chainer, slotv );
632 :
633 : /* parent now identified, connect this slotv if the parent is. */
634 63 : if( FD_UNLIKELY( parent_slotv->connected ) ) slotv->connected = 1;
635 63 : return parent_slotv;
636 63 : }
637 :
638 : void
639 : fd_chainer_verified_hash_insert( fd_chainer_t * chainer,
640 : ulong slot,
641 : fd_hash_t * block_id,
642 : uint fec_set_idx,
643 144 : fd_hash_t * mr ) {
644 144 : fd_chainer_slotv_t * slotv = fd_chainer_slot_version_query( chainer, slot, block_id );
645 144 : if( FD_UNLIKELY( !slotv ) ) FD_LOG_CRIT(( "slotv not found for slot %lu - verify this is a CRIT", slot ));
646 :
647 : /* Already have this version's FEC entry -> nothing to fetch. */
648 144 : if( FD_UNLIKELY( slotv_fec( chainer, slotv, fec_set_idx ) ) ) return;
649 :
650 : /* The same root may have already started progress through repairing
651 : another slot version. If so, create this version's entry
652 : already-complete and replay the completion through
653 : fd_chainer_fec_complete. Otherwise create an incomplete entry that
654 : is awaiting shreds. */
655 144 : fd_chainer_fec_t * shared = fec_query( chainer, mr );
656 144 : int shared_complete = shared && shared->complete;
657 :
658 144 : fd_chainer_fec_t * fec = fec_join( chainer, slot, fec_set_idx, slotv, mr );
659 144 : if( FD_UNLIKELY( fec_set_idx==slotv->complete_idx - ( FD_FEC_SHRED_CNT-1 ) ) ) {
660 51 : fec->slot_complete = 1;
661 51 : }
662 :
663 144 : if( FD_LIKELY( shared_complete ) ) {
664 : /* TODO double check we don't need to be updating slotv buffered_idx
665 : when FEC is not complete as well */
666 48 : fd_chainer_fec_complete( chainer, slot, fec_set_idx, shared->slot_complete, shared->data_complete, 0, mr );
667 48 : }
668 144 : fd_chainer_repair_add( chainer, slotv ); /* new sentinel -> re-add for shred fill */
669 144 : chainer_advance( chainer, slotv );
670 144 : }
671 :
672 : /* fec_rekey re-keys sentinel, a FEC created from a getFecRoot response
673 : and keyed by only its zero-padded 20-byte root prefix, to the
674 : full merkle root full_mr that a shred just delivered. If a FEC keyed
675 : by full_mr already exists (e.g. turbine saw the set first), the
676 : sentinel is merged into it instead: every version pointing at the
677 : sentinel is repointed and, if the existing FEC is already complete,
678 : its completion is replayed so those versions deliver it. */
679 :
680 : static void
681 : fec_rekey( fd_chainer_t * chainer,
682 : fd_chainer_fec_t * sentinel,
683 12 : fd_hash_t const * full_mr ) {
684 12 : fd_chainer_fec_t * existing = fec_query( chainer, full_mr );
685 12 : if( FD_LIKELY( !existing ) ) {
686 6 : fd_fec_map_ele_remove_fast( chainer->fec_map, sentinel, chainer->fec_pool );
687 6 : sentinel->merkle_root = *full_mr;
688 6 : fd_fec_map_ele_insert( chainer->fec_map, sentinel, chainer->fec_pool );
689 6 : return;
690 6 : }
691 :
692 6 : ulong slot = sentinel->slot;
693 6 : uint fec_set_idx = sentinel->fec_set_idx;
694 6 : uint k = fec_set_idx / FD_FEC_SHRED_CNT;
695 6 : uint sentinel_idx = (uint)fd_fec_pool_idx( chainer->fec_pool, sentinel );
696 6 : uint existing_idx = (uint)fd_fec_pool_idx( chainer->fec_pool, existing );
697 :
698 6 : fd_chainer_slotv_t * repointed[ FD_CHAINER_SLOT_VER_MAX ];
699 6 : ulong repointed_cnt = 0UL;
700 21 : for( ulong i=slotv_iter_init( chainer, slot ); i!=ULONG_MAX; i=slotv_iter_next( chainer, i ) ) {
701 15 : fd_chainer_slotv_t * slotv = slotv_iter_ele( chainer, i );
702 15 : uint * fecs = fd_chainer_slotv_fecs( chainer, slotv );
703 15 : if( FD_LIKELY( fecs[ k ]!=sentinel_idx ) ) continue;
704 6 : fecs[ k ] = existing_idx;
705 6 : repointed[ repointed_cnt++ ] = slotv;
706 6 : }
707 6 : fd_fec_map_ele_remove_fast( chainer->fec_map, sentinel, chainer->fec_pool );
708 6 : fd_fec_pool_ele_release ( chainer->fec_pool, sentinel );
709 :
710 6 : if( FD_LIKELY( existing->complete ) ) {
711 6 : fd_hash_t full = *full_mr;
712 6 : fd_chainer_fec_complete( chainer, slot, fec_set_idx, existing->slot_complete, existing->data_complete, 0, &full );
713 6 : }
714 12 : for( ulong i=0UL; i<repointed_cnt; i++ ) {
715 6 : fd_chainer_repair_add( chainer, repointed[ i ] ); /* re-add for remaining shred fill */
716 6 : chainer_advance( chainer, repointed[ i ] );
717 6 : }
718 6 : }
719 :
720 : void
721 : fd_chainer_verified_block_insert( fd_chainer_t * chainer,
722 : ulong slot,
723 93 : fd_hash_t block_id ) {
724 93 : FD_TEST( slot>chainer->root );
725 :
726 93 : if( FD_LIKELY( fd_chainer_slot_version_query( chainer, slot, &block_id ) ) ) return;
727 :
728 90 : fd_chainer_slotv_t * slotv = acquire_slotv( chainer, slot );
729 90 : slotv->block_id = block_id;
730 90 : orphans_resolve( chainer );
731 :
732 90 : fd_chainer_slotv_t * turbine = turbine_slotv_query( chainer, slot );
733 90 : if( FD_UNLIKELY( turbine && fd_hash_check_zero( &turbine->block_id ) ) ) {
734 : /* Turbine slotv is not yet complete, but votor repair events for
735 : this slot have already started arriving, suggesting we are way
736 : behind on repairing this slot. At this point just abandon the
737 : turbine version and only deliver votor verified versions. */
738 51 : slotv_abandon( chainer, turbine );
739 51 : }
740 90 : }
741 :
742 : void
743 : fd_chainer_publish( fd_chainer_t * chainer,
744 : ulong new_root,
745 : fd_hash_t const * new_root_block_id,
746 30 : fd_store_t * store ) {
747 30 : fd_store_map_t store_map[1];
748 30 : if( store ) FD_TEST( fd_store_map_ljoin( store, store_map ) );
749 :
750 30 : fd_slotv_map_t * slotv_map = chainer->slotv_map;
751 30 : fd_chainer_slotv_t * slotv_pool = chainer->slotv_pool;
752 30 : fd_chainer_fec_t * fec_pool = chainer->fec_pool;
753 30 : fd_fec_map_t * fec_map = chainer->fec_map;
754 :
755 30 : out_ele_t * out_queue = chainer->out_queue;
756 30 : if( FD_UNLIKELY( !out_queue_empty( out_queue ) ) ) FD_LOG_CRIT(( "chainer out_queue not empty before publish" ));
757 :
758 30 : ulong root = chainer->root;
759 30 : if( FD_UNLIKELY( root==ULONG_MAX ) ) return;
760 30 : FD_TEST( root<new_root );
761 30 : FD_TEST( fd_chainer_slot_query( chainer, new_root ) );
762 :
763 : /* Identify the canonical (rooted) version of new_root. Every other
764 : version of it is an equivocating sibling that is now dead. If no
765 : version matches, keep them all rather than guess wrong and prune
766 : the version we are actually rooted on. TODO: block_id is now
767 : always wired through from replay, so this could be a CRIT. */
768 30 : fd_chainer_slotv_t * canonical = new_root_block_id ? fd_chainer_slot_version_query( chainer, new_root, new_root_block_id ) : NULL;
769 30 : if( FD_UNLIKELY( !canonical ) ) {
770 12 : FD_LOG_DEBUG(( "chainer publish %lu: no version matches the rooted block_id; keeping all versions", new_root ));
771 12 : }
772 :
773 : /* Prune every version of every slot in [root, new_root]: release the
774 : FECs it owns, drop it from the worklists, and free it. Only the
775 : canonical version of new_root survives (all of them if canonical is
776 : unknown) and its FEC list is cleared, since a rooted slot's FEC
777 : data is never needed again. */
778 99 : for( ulong slot=root; slot<=new_root; slot++ ) {
779 168 : for( ulong i=slotv_iter_init( chainer, slot ); i!=ULONG_MAX; ) {
780 99 : fd_chainer_slotv_t * s = slotv_iter_ele ( chainer, i );
781 99 : ulong next = slotv_iter_next( chainer, i );
782 :
783 138339 : for( uint k=0U; k<chainer->fec_blk_max; k++ ) {
784 138240 : fd_chainer_fec_t * fec = slotv_fec( chainer, s, k * FD_FEC_SHRED_CNT );
785 : /* FEC pool eles can be shared across versions of an equivocating
786 : slot, so check the map to avoid double-freeing one a sibling
787 : already released. */
788 138240 : if( FD_UNLIKELY( fec && fd_fec_map_ele_query_const( fec_map, &fec->merkle_root, NULL, fec_pool ) ) ) {
789 297 : if( FD_LIKELY( store && fec->complete ) ) fd_store_remove( store, store_map, &fec->merkle_root ); /* only complete FECs are in the store */
790 297 : fd_fec_map_ele_remove_fast( fec_map, fec, fec_pool );
791 297 : fd_fec_pool_ele_release( fec_pool, fec );
792 297 : }
793 138240 : }
794 99 : fd_memset( fd_chainer_slotv_fecs( chainer, s ), 0xff, chainer->fec_blk_max*sizeof(uint) );
795 :
796 99 : fd_chainer_orphan_remove( chainer, s );
797 99 : fd_chainer_repair_remove( chainer, s );
798 :
799 99 : int survives = slot==new_root && ( !canonical || s==canonical );
800 99 : if( FD_LIKELY( !survives ) ) {
801 69 : fd_slotv_map_ele_remove_fast( slotv_map, s, slotv_pool );
802 69 : fd_slotv_pool_ele_release( slotv_pool, s );
803 69 : }
804 99 : i = next;
805 99 : }
806 69 : }
807 :
808 30 : chainer->root = new_root;
809 30 : orphans_resolve( chainer ); /* parents now at or below the root are settled */
810 :
811 : /* Connect the surviving version(s) of the new root. */
812 60 : for( ulong i=slotv_iter_init( chainer, new_root ); i!=ULONG_MAX; i=slotv_iter_next( chainer, i ) ) {
813 30 : fd_chainer_slotv_t * s = slotv_iter_ele( chainer, i );
814 30 : if( FD_UNLIKELY( s->parent_slot==AG_UNKNOWN_SLOT ) ) s->parent_slot = new_root;
815 30 : s->connected = 1;
816 30 : s->complete_idx = 0U;
817 30 : s->buffered_idx = 0U;
818 30 : s->delivered_idx = 0U;
819 30 : s->buffered_fec_idx = UINT_MAX; /* rooted slot has no buffered FEC set */
820 30 : }
821 :
822 : /* The new root becomes a delivered anchor here WITHOUT going through
823 : chainer_advance's slot_complete cascade, so children that already
824 : completed while waiting on it were never connected/delivered.
825 : Cascade to them now, mirroring chainer_advance's child scan. */
826 60 : for( ulong i=slotv_iter_init( chainer, new_root ); i!=ULONG_MAX; i=slotv_iter_next( chainer, i ) ) {
827 30 : fd_chainer_slotv_t * s = slotv_iter_ele( chainer, i );
828 30 : if( FD_UNLIKELY( fd_hash_check_zero( &s->block_id ) ) ) continue;
829 30 : for( fd_slotv_map_iter_t it = fd_slotv_map_iter_init( slotv_map, slotv_pool );
830 63 : !fd_slotv_map_iter_done( it, slotv_map, slotv_pool );
831 33 : it = fd_slotv_map_iter_next( it, slotv_map, slotv_pool ) ) {
832 33 : fd_chainer_slotv_t * child = fd_slotv_map_iter_ele( it, slotv_map, slotv_pool );
833 33 : if( FD_UNLIKELY( fd_hash_eq( &child->parent_block_id, &s->block_id ) ) ) {
834 3 : child->connected = 1;
835 3 : chainer_advance( chainer, child );
836 3 : }
837 33 : }
838 30 : }
839 30 : }
840 :
841 : void
842 0 : fd_chainer_print( fd_chainer_t * chainer ) {
843 0 : if( FD_UNLIKELY( chainer->root==ULONG_MAX ) ) return;
844 :
845 0 : fd_chainer_slotv_t * slotv_pool = chainer->slotv_pool;
846 0 : fd_slotv_map_t * slotv_map = chainer->slotv_map;
847 :
848 0 : printf( "\n[Chainer] root: %lu, highest repaired: %lu\n", chainer->root, chainer->highest_repaired );
849 :
850 0 : ulong cnt = 0UL;
851 0 : for( fd_slotv_map_iter_t it = fd_slotv_map_iter_init( slotv_map, slotv_pool );
852 0 : !fd_slotv_map_iter_done( it, slotv_map, slotv_pool );
853 0 : it = fd_slotv_map_iter_next( it, slotv_map, slotv_pool ) ) {
854 0 : fd_chainer_slotv_t * o = fd_slotv_map_iter_ele( it, slotv_map, slotv_pool );
855 :
856 0 : ulong slot = o->slot;
857 :
858 0 : FD_BASE58_ENCODE_32_BYTES( o->block_id.uc, out )
859 0 : if( FD_UNLIKELY( o->parent_slot==AG_UNKNOWN_SLOT ) ) {
860 0 : printf( "%lu - ???: shreds: (%u/%u) turb: %d block_id: %s \n", slot, o->buffered_idx+1U, o->complete_idx+1U, o->turbine, out );
861 0 : }
862 0 : else {
863 0 : printf( "%lu - %lu: shreds: (%u/%u) turb: %d block_id: %s connected: %d\n ", slot, o->parent_slot, o->buffered_idx+1U, o->complete_idx+1U, o->turbine, out, o->connected );
864 0 : }
865 0 : cnt++;
866 0 : }
867 0 : printf( "(%lu total slotvs)\n", cnt );
868 0 : fflush( stdout );
869 0 : }
870 :
871 : /* Worklist helpers */
872 :
873 : /* work_ele returns the worklist ele shadowing slotv, or NULL if the
874 : slotv is in neither treap. */
875 :
876 : static inline fd_chainer_work_t *
877 2994 : work_ele( fd_chainer_t * chainer, fd_chainer_slotv_t const * slotv ) {
878 2994 : ulong slotv_idx = fd_slotv_pool_idx( chainer->slotv_pool, slotv );
879 2994 : return fd_work_map_ele_query( chainer->work_map, &slotv_idx, NULL, chainer->work_pool );
880 2994 : }
881 :
882 : /* work_ele_acquire returns the ele shadowing slotv, creating (and
883 : map-inserting) it if none exists yet. */
884 :
885 : static inline fd_chainer_work_t *
886 882 : work_ele_acquire( fd_chainer_t * chainer, fd_chainer_slotv_t * slotv ) {
887 882 : fd_chainer_work_t * ele = work_ele( chainer, slotv );
888 882 : if( FD_LIKELY( ele ) ) return ele;
889 339 : ele = fd_work_pool_ele_acquire( chainer->work_pool );
890 339 : ele->slotv_idx = fd_slotv_pool_idx( chainer->slotv_pool, slotv );
891 339 : ele->slot = slotv->slot;
892 339 : ele->in_repair = 0;
893 339 : ele->in_orphan = 0;
894 339 : fd_work_map_ele_insert( chainer->work_map, ele, chainer->work_pool );
895 339 : return ele;
896 882 : }
897 :
898 : void
899 : fd_chainer_repair_add( fd_chainer_t * chainer,
900 552 : fd_chainer_slotv_t * slotv ) {
901 552 : fd_chainer_work_t * ele = work_ele_acquire( chainer, slotv );
902 552 : if( FD_UNLIKELY( ele->in_repair ) ) return;
903 339 : fd_work_repair_ele_insert( chainer->repair_treap, ele, chainer->work_pool );
904 339 : ele->in_repair = 1;
905 339 : }
906 :
907 : void
908 : fd_chainer_repair_remove( fd_chainer_t * chainer,
909 426 : fd_chainer_slotv_t * slotv ) {
910 426 : fd_chainer_work_t * ele = work_ele( chainer, slotv );
911 426 : if( FD_UNLIKELY( !ele || !ele->in_repair ) ) return;
912 285 : fd_work_repair_ele_remove( chainer->repair_treap, ele, chainer->work_pool );
913 285 : ele->in_repair = 0;
914 :
915 285 : if( FD_LIKELY( ele->in_orphan ) ) return; /* still on the orphan worklist */
916 168 : fd_work_map_ele_remove_fast( chainer->work_map, ele, chainer->work_pool );
917 168 : fd_work_pool_ele_release( chainer->work_pool, ele );
918 168 : }
919 :
920 : void
921 : fd_chainer_orphan_add( fd_chainer_t * chainer,
922 330 : fd_chainer_slotv_t * slotv ) {
923 330 : fd_chainer_work_t * ele = work_ele_acquire( chainer, slotv );
924 330 : if( FD_UNLIKELY( ele->in_orphan ) ) return;
925 330 : fd_work_orphan_ele_insert( chainer->orphan_treap, ele, chainer->work_pool );
926 330 : ele->in_orphan = 1;
927 330 : }
928 :
929 : void
930 : fd_chainer_orphan_remove( fd_chainer_t * chainer,
931 465 : fd_chainer_slotv_t * slotv ) {
932 465 : fd_chainer_work_t * ele = work_ele( chainer, slotv );
933 465 : if( FD_UNLIKELY( !ele || !ele->in_orphan ) ) return;
934 300 : fd_work_orphan_ele_remove( chainer->orphan_treap, ele, chainer->work_pool );
935 300 : ele->in_orphan = 0;
936 :
937 300 : if( FD_LIKELY( ele->in_repair ) ) return; /* still on the repair worklist */
938 117 : fd_work_map_ele_remove_fast( chainer->work_map, ele, chainer->work_pool );
939 117 : fd_work_pool_ele_release( chainer->work_pool, ele );
940 117 : }
941 :
942 : ulong
943 2184 : fd_chainer_repair_iter_init( fd_chainer_t * chainer ) {
944 2184 : return fd_work_repair_fwd_iter_init( chainer->repair_treap, chainer->work_pool );
945 2184 : }
946 :
947 : ulong
948 : fd_chainer_repair_iter_next( fd_chainer_t * chainer,
949 2094 : ulong iter ) {
950 2094 : return fd_work_repair_fwd_iter_next( iter, chainer->work_pool );
951 2094 : }
952 :
953 : ulong
954 2430 : fd_chainer_orphan_iter_init( fd_chainer_t * chainer ) {
955 2430 : return fd_work_orphan_fwd_iter_init( chainer->orphan_treap, chainer->work_pool );
956 2430 : }
957 :
958 : ulong
959 : fd_chainer_orphan_iter_next( fd_chainer_t * chainer,
960 270 : ulong iter ) {
961 270 : return fd_work_orphan_fwd_iter_next( iter, chainer->work_pool );
962 270 : }
963 :
964 : int
965 5004 : fd_chainer_work_iter_done( ulong iter ) {
966 5004 : return iter==ULONG_MAX;
967 5004 : }
968 :
969 : fd_chainer_slotv_t *
970 : fd_chainer_work_iter_ele( fd_chainer_t * chainer,
971 2364 : ulong iter ) {
972 2364 : return fd_slotv_pool_ele( chainer->slotv_pool, fd_work_pool_ele( chainer->work_pool, iter )->slotv_idx );
973 2364 : }
974 :
975 : int
976 : fd_chainer_in_repair( fd_chainer_t * chainer,
977 618 : fd_chainer_slotv_t const * slotv ) {
978 618 : fd_chainer_work_t * ele = work_ele( chainer, slotv );
979 618 : return ele && ele->in_repair;
980 618 : }
981 :
982 : int
983 : fd_chainer_in_orphan( fd_chainer_t * chainer,
984 603 : fd_chainer_slotv_t const * slotv ) {
985 603 : fd_chainer_work_t * ele = work_ele( chainer, slotv );
986 603 : return ele && ele->in_orphan;
987 603 : }
988 :
989 : int
990 11361 : fd_chainer_verify( fd_chainer_t const * chainer ) {
991 11361 : # define FAIL( msg ) do { FD_LOG_WARNING(( "fd_chainer_verify: %s", msg )); return -1; } while(0)
992 :
993 11361 : if( FD_UNLIKELY( !chainer ) ) FAIL( "NULL chainer" );
994 11358 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)chainer, fd_chainer_align() ) ) ) FAIL( "misaligned chainer" );
995 11358 : if( FD_UNLIKELY( !fd_wksp_containing( chainer ) ) ) FAIL( "chainer must be part of a workspace" );
996 11358 : if( FD_UNLIKELY( chainer->magic!=FD_CHAINER_MAGIC ) ) FAIL( "bad magic" );
997 :
998 11355 : fd_chainer_t * chainer_ = (fd_chainer_t *)chainer;
999 :
1000 11355 : fd_chainer_slotv_t const * slotv_pool = chainer_->slotv_pool;
1001 11355 : fd_slotv_map_t const * slotv_map = chainer_->slotv_map;
1002 11355 : fd_chainer_work_t const * work_pool = chainer_->work_pool;
1003 11355 : fd_work_map_t const * work_map = chainer_->work_map;
1004 11355 : fd_work_repair_t const * rtreap = chainer_->repair_treap;
1005 11355 : fd_work_orphan_t const * otreap = chainer_->orphan_treap;
1006 11355 : fd_chainer_fec_t const * fec_pool = chainer_->fec_pool;
1007 11355 : fd_fec_map_t const * fec_map = chainer_->fec_map;
1008 :
1009 11355 : if( FD_UNLIKELY( fd_slotv_map_verify( slotv_map, fd_slotv_pool_max( slotv_pool ), slotv_pool )==-1 ) ) FAIL( "slotv map corrupted" );
1010 11355 : if( FD_UNLIKELY( fd_work_map_verify ( work_map, fd_work_pool_max ( work_pool ), work_pool )==-1 ) ) FAIL( "work map corrupted" );
1011 11355 : if( FD_UNLIKELY( fd_fec_map_verify ( fec_map, fd_fec_pool_max ( fec_pool ), fec_pool )==-1 ) ) FAIL( "fec map corrupted" );
1012 11355 : if( FD_UNLIKELY( fd_work_repair_verify( rtreap, work_pool )==-1 ) ) FAIL( "repair treap corrupted" );
1013 11355 : if( FD_UNLIKELY( fd_work_orphan_verify( otreap, work_pool )==-1 ) ) FAIL( "orphan treap corrupted" );
1014 :
1015 : /* The root, if set, must have at least one connected version -- it is
1016 : by definition the start of every ancestry chain. Uniquely among
1017 : slots, the root need not have a version 0: publish prunes the
1018 : non-canonical versions of the new root, and version 0 may have been
1019 : one of them. Its FEC list is released along with it, which is fine
1020 : because a rooted slot's FEC data is never needed again. */
1021 :
1022 11355 : if( FD_LIKELY( chainer->root!=ULONG_MAX ) ) {
1023 11304 : int root_present = 0;
1024 11304 : int root_connected = 0;
1025 11304 : ulong root = chainer->root;
1026 11304 : for( ulong i = fd_slotv_map_idx_query_const( slotv_map, &root, ULONG_MAX, slotv_pool );
1027 22608 : i != ULONG_MAX;
1028 11304 : i = fd_slotv_map_idx_next_const( i, ULONG_MAX, slotv_pool ) ) {
1029 11304 : fd_chainer_slotv_t const * root_slotv = fd_slotv_pool_ele_const( slotv_pool, i );
1030 11304 : root_present = 1;
1031 11304 : root_connected |= !!root_slotv->connected;
1032 11304 : }
1033 11304 : if( FD_UNLIKELY( !root_present ) ) FAIL( "root has no slotv" );
1034 11304 : if( FD_UNLIKELY( !root_connected ) ) FAIL( "no root slotv is connected" );
1035 11304 : }
1036 :
1037 11355 : for( fd_slotv_map_iter_t it = fd_slotv_map_iter_init( slotv_map, slotv_pool );
1038 35649 : !fd_slotv_map_iter_done( it, slotv_map, slotv_pool );
1039 24303 : it = fd_slotv_map_iter_next( it, slotv_map, slotv_pool ) ) {
1040 24303 : fd_chainer_slotv_t const * slotv = fd_slotv_map_iter_ele_const( it, slotv_map, slotv_pool );
1041 :
1042 24303 : ulong slot = slotv->slot;
1043 :
1044 : /* Nothing below the root may survive a publish. */
1045 :
1046 24303 : if( FD_UNLIKELY( chainer->root!=ULONG_MAX && slot<chainer->root ) ) FAIL( "slotv below the root" );
1047 :
1048 : /* Shred index bookkeeping */
1049 :
1050 24300 : if( FD_UNLIKELY( slotv->complete_idx!=UINT_MAX && slotv->buffered_idx !=UINT_MAX &&
1051 24300 : slotv->buffered_idx >slotv->complete_idx ) ) FAIL( "buffered_idx > complete_idx" );
1052 24297 : if( FD_UNLIKELY( slotv->complete_idx!=UINT_MAX && slotv->delivered_idx!=UINT_MAX &&
1053 24297 : slotv->delivered_idx>slotv->complete_idx ) ) FAIL( "delivered_idx > complete_idx" );
1054 :
1055 : /* buffered_fec_idx is the last shred idx of a FEC set, so it is
1056 : always one below a multiple of FD_FEC_SHRED_CNT (UINT_MAX, the
1057 : "none" sentinel, satisfies this too). */
1058 :
1059 24294 : if( FD_UNLIKELY( ( slotv->buffered_fec_idx + 1U ) % FD_FEC_SHRED_CNT ) ) FAIL( "buffered_fec_idx is not the last idx of a FEC set" );
1060 :
1061 : /* A buffered FEC set means all of its shreds are in hand, so the
1062 : contiguous FEC prefix can never run ahead of the contiguous shred
1063 : prefix. */
1064 :
1065 24294 : if( FD_UNLIKELY( slotv->buffered_fec_idx!=UINT_MAX &&
1066 24294 : ( slotv->buffered_idx==UINT_MAX ||
1067 24294 : slotv->buffered_idx<slotv->buffered_fec_idx ) ) ) FAIL( "buffered_fec_idx runs ahead of buffered_idx" );
1068 :
1069 : /* An abandoned version is always a turbine version and never on a
1070 : worklist (see slotv_abandon). */
1071 :
1072 24294 : if( FD_UNLIKELY( slotv->abandoned && !slotv->turbine ) ) FAIL( "abandoned non-turbine slotv" );
1073 24294 : if( FD_UNLIKELY( slotv->abandoned && ( fd_chainer_in_repair( chainer_, slotv ) || fd_chainer_in_orphan( chainer_, slotv ) ) ) ) FAIL( "abandoned slotv on a worklist" );
1074 24294 : }
1075 :
1076 : /* Worklist consistency. Every work ele must shadow a live slotv, be
1077 : in at least one treap (else it should have been gc'd), and carry the
1078 : slot of its slotv. The per-treap membership counts must match the
1079 : treap element counts. */
1080 :
1081 11346 : ulong ele_max = fd_slotv_pool_max( slotv_pool );
1082 11346 : ulong in_treap_cnt = 0UL;
1083 11346 : ulong in_orphan_cnt = 0UL;
1084 11346 : for( fd_work_map_iter_t it = fd_work_map_iter_init( work_map, work_pool );
1085 22635 : !fd_work_map_iter_done( it, work_map, work_pool );
1086 11346 : it = fd_work_map_iter_next( it, work_map, work_pool ) ) {
1087 11292 : fd_chainer_work_t const * ele = fd_work_map_iter_ele_const( it, work_map, work_pool );
1088 11292 : if( FD_UNLIKELY( ele->slotv_idx>=ele_max ) ) FAIL( "work ele slotv_idx out of range" );
1089 11292 : if( FD_UNLIKELY( !ele->in_repair && !ele->in_orphan ) ) FAIL( "work ele in neither treap (should be gc'd)" );
1090 11289 : if( FD_UNLIKELY( ele->slot!=fd_slotv_pool_ele_const( slotv_pool, ele->slotv_idx )->slot ) ) FAIL( "work ele slot mismatches slotv" );
1091 11289 : in_treap_cnt += !!ele->in_repair;
1092 11289 : in_orphan_cnt += !!ele->in_orphan;
1093 11289 : }
1094 :
1095 : /* No treap may hold an ele not accounted for in the work map. */
1096 :
1097 11343 : if( FD_UNLIKELY( in_treap_cnt !=fd_work_repair_ele_cnt( rtreap ) ) ) FAIL( "repair treap holds eles that are not in the work map" );
1098 11343 : if( FD_UNLIKELY( in_orphan_cnt!=fd_work_orphan_ele_cnt( otreap ) ) ) FAIL( "orphan treap holds eles that are not in the work map" );
1099 :
1100 11340 : for( fd_fec_map_iter_t it = fd_fec_map_iter_init( fec_map, fec_pool );
1101 235947 : !fd_fec_map_iter_done( it, fec_map, fec_pool );
1102 224610 : it = fd_fec_map_iter_next( it, fec_map, fec_pool ) ) {
1103 224610 : fd_chainer_fec_t const * fec = fd_fec_map_iter_ele_const( it, fec_map, fec_pool );
1104 :
1105 224610 : ulong slot = fec->slot;
1106 224610 : uint fec_set_idx = fec->fec_set_idx;
1107 224610 : uint fec_idx = (uint)fd_fec_pool_idx( fec_pool, fec );
1108 :
1109 224610 : if( FD_UNLIKELY( fec_set_idx % FD_FEC_SHRED_CNT ) ) FAIL( "fec_set_idx is not a multiple of FD_FEC_SHRED_CNT" );
1110 224610 : if( FD_UNLIKELY( fec_set_idx / FD_FEC_SHRED_CNT >= chainer->fec_blk_max ) ) FAIL( "fec_set_idx out of range" );
1111 :
1112 224610 : if( FD_UNLIKELY( chainer->root!=ULONG_MAX && slot<chainer->root ) ) FAIL( "fec below the root" );
1113 :
1114 : /* A slot with a FEC must have at least one version to anchor the list
1115 : and own the FEC -- without one publish could never reach it. */
1116 :
1117 224610 : if( FD_UNLIKELY( !fd_chainer_slot_query( chainer_, slot ) ) ) FAIL( "slot has a fec but no version to anchor the list" );
1118 :
1119 : /* A FEC is owned by every version whose fec_tbl row points at it,
1120 : and at least one must -- otherwise it is unreachable garbage that
1121 : publish would leak. */
1122 :
1123 224610 : int owned = 0;
1124 224610 : for( ulong i = fd_slotv_map_idx_query_const( slotv_map, &slot, ULONG_MAX, slotv_pool );
1125 455193 : i != ULONG_MAX;
1126 230583 : i = fd_slotv_map_idx_next_const( i, ULONG_MAX, slotv_pool ) ) {
1127 230583 : fd_chainer_slotv_t const * slotv = fd_slotv_pool_ele_const( slotv_pool, i );
1128 230583 : if( FD_LIKELY( fd_chainer_slotv_fecs( chainer, slotv )[ fec_set_idx / FD_FEC_SHRED_CNT ]==fec_idx ) ) owned = 1;
1129 230583 : }
1130 224610 : if( FD_UNLIKELY( !owned ) ) FAIL( "fec claimed by no version" );
1131 224610 : }
1132 :
1133 11337 : return 0;
1134 11340 : }
1135 : #undef FAIL
|
