Line data Source code
1 : #include "fd_crds.h"
2 : #include "fd_crds_contact_info.c"
3 : #include "../fd_gossip_types.h"
4 :
5 : #include "../../../ballet/sha256/fd_sha256.h"
6 : #include "../../../funk/fd_funk_base.h" /* no link dependency, only using hash */
7 :
8 : #include <string.h>
9 :
10 : FD_STATIC_ASSERT( CRDS_MAX_CONTACT_INFO==FD_CONTACT_INFO_TABLE_SIZE,
11 : "CRDS_MAX_CONTACT_INFO must match FD_CONTACT_INFO_TABLE_SIZE" );
12 :
13 : struct fd_crds_key {
14 : uchar tag;
15 : uchar pubkey[ 32UL ];
16 : union {
17 : uchar vote_index;
18 : uchar epoch_slots_index;
19 : ushort duplicate_shred_index;
20 : };
21 : };
22 :
23 : typedef struct fd_crds_key fd_crds_key_t;
24 :
25 : /* The CRDS at a high level is just a list of all the messages we have
26 : received over gossip. These are called the CRDS values. Values
27 : are not arbitrary, and must conform to a strictly typed schema of
28 : around 10 different messages. */
29 :
30 : struct fd_crds_entry_private {
31 : /* The core operation of the CRDS is to "upsert" a value. Basically,
32 : all of the message types are keyed by the originators public key,
33 : and we only want to store the most recent message of each type.
34 :
35 : This key field is the key for the hash table. */
36 : fd_crds_key_t key;
37 :
38 : union {
39 : struct {
40 : fd_crds_contact_info_entry_t * ci;
41 : long instance_creation_wallclock_nanos;
42 : uchar is_active;
43 : ulong sampler_idx;
44 :
45 : /* A list of "fresh" contact info entries is maintained, holding
46 : entries that have been refreshed/inserted in the last 60s in
47 : upsertion order (oldest first).
48 :
49 : fd_crds_advance periodically checks for and removes peers from
50 : this list if they exceed the threshold. Peers removed in this
51 : loop are also re-scored in the peer sampler. This is different
52 : from dropping the CRDS entry entirely, which also removes the
53 : entry from this list. To avoid double-popping an entry we use
54 : in_list as a presence check prior to removing */
55 : struct {
56 : ulong prev;
57 : ulong next;
58 : uchar in_list; /* 1 if in the fresh list, 0 otherwise */
59 : } fresh_dlist;
60 :
61 : /* The contact info side table has a separate size limit, so
62 : we maintain a separate evict list to make space for new
63 : entries */
64 : struct {
65 : ulong prev;
66 : ulong next;
67 : } evict_dlist;
68 :
69 : /* TODO: stake-ordered treap/pq? */
70 : } contact_info;
71 : struct {
72 : ulong token;
73 : } node_instance;
74 : };
75 :
76 : /* When an originator creates a CRDS message, they attach their local
77 : wallclock time to it. This time is used to determine when a
78 : message should be upserted. If messages have the same key, the
79 : newer one (as created by the originator) is used.
80 :
81 : Messages encode wallclock in millis, firedancer converts
82 : them into nanos internally. */
83 : long wallclock_nanos;
84 :
85 : uchar value_bytes[ FD_GOSSIP_CRDS_MAX_SZ ];
86 : ushort value_sz;
87 :
88 : /* The value hash is the sha256 of the value_bytes. It is used in
89 : bloom filter generation and as a tiebreaker when a
90 : fd_crds_checks_fast call returns CHECK_UNDETERMINED. */
91 : uchar value_hash[ 32UL ];
92 : ulong num_duplicates;
93 : ulong stake;
94 :
95 : struct {
96 : ulong next;
97 : } pool;
98 :
99 : /* The CRDS needs to perform a variety of actions on the message table
100 : quickly, so there are various indexes woven through them values to
101 : support these actions. They are ...
102 :
103 : lookup is used to enable the core map<key, value> functionality
104 : described for upserts defined by value->key. */
105 : struct {
106 : ulong next;
107 : ulong prev;
108 : } lookup;
109 :
110 : /* The table has a fixed size message capacity, and supports eviction
111 : so insertion never fails. If the table is full and we wish to
112 : insert a new value, the "lowest priority" message is evicted to
113 : make room. This is accomplished with a treap sorted by stake, so
114 : the lowest stake message is removed. */
115 : struct {
116 : ulong parent;
117 : ulong left;
118 : ulong right;
119 : ulong prio;
120 : ulong next;
121 : ulong prev;
122 : } evict;
123 :
124 : /* Values in the table expire after a pre-determined amount of time,
125 : so we also keep a linked list of values sorted by creation time.
126 : The time used here is our nodes wallclock when we received the
127 : CRDS, not the originators local wallclock, which they could skew
128 : to cause their values to live longer.
129 :
130 : There are actually two lists that reuse the same pointers here,
131 : and a value will be in exactly one of the lists. One is for staked
132 : nodes, which values expire after 48 hours, and one is for unstaked
133 : nodes, which expire after 15 seconds. */
134 : struct {
135 : long wallclock_nanos;
136 : ulong prev;
137 : ulong next;
138 : } expire;
139 :
140 : /* Finally, a core operation on the CRDS is to to query for values by
141 : hash, to respond to pull requests. This is done with a treap
142 : sorted by hash, which is just the first 8 bytes value_hash. */
143 : struct {
144 : ulong hash;
145 : ulong parent;
146 : ulong left;
147 : ulong right;
148 : ulong next;
149 : ulong prev;
150 : ulong prio;
151 : } hash;
152 : };
153 :
154 : #define POOL_NAME crds_pool
155 18 : #define POOL_T fd_crds_entry_t
156 9366 : #define POOL_NEXT pool.next
157 :
158 : #include "../../../util/tmpl/fd_pool.c"
159 :
160 : #define TREAP_NAME evict_treap
161 : #define TREAP_T fd_crds_entry_t
162 : #define TREAP_QUERY_T void * /* We don't use query ... */
163 : #define TREAP_CMP(q,e) (__extension__({ (void)(q); (void)(e); -1; })) /* which means we don't need to give a real
164 : implementation to cmp either */
165 873 : #define TREAP_IDX_T ulong
166 687 : #define TREAP_LT(e0,e1) ((e0)->stake<(e1)->stake)
167 573 : #define TREAP_PARENT evict.parent
168 291 : #define TREAP_LEFT evict.left
169 150 : #define TREAP_RIGHT evict.right
170 9648 : #define TREAP_PRIO evict.prio
171 : #define TREAP_OPTIMIZE_ITERATION 1
172 150 : #define TREAP_NEXT evict.next
173 150 : #define TREAP_PREV evict.prev
174 :
175 : #include "../../../util/tmpl/fd_treap.c"
176 :
177 : #define DLIST_NAME staked_expire_dlist
178 : #define DLIST_ELE_T fd_crds_entry_t
179 150 : #define DLIST_PREV expire.prev
180 150 : #define DLIST_NEXT expire.next
181 :
182 : #include "../../../util/tmpl/fd_dlist.c"
183 :
184 : #define DLIST_NAME unstaked_expire_dlist
185 : #define DLIST_ELE_T fd_crds_entry_t
186 0 : #define DLIST_PREV expire.prev
187 0 : #define DLIST_NEXT expire.next
188 :
189 : #include "../../../util/tmpl/fd_dlist.c"
190 :
191 :
192 : #define DLIST_NAME crds_contact_info_fresh_list
193 : #define DLIST_ELE_T fd_crds_entry_t
194 150 : #define DLIST_PREV contact_info.fresh_dlist.prev
195 150 : #define DLIST_NEXT contact_info.fresh_dlist.next
196 : #include "../../../util/tmpl/fd_dlist.c"
197 :
198 : #define DLIST_NAME crds_contact_info_evict_dlist
199 : #define DLIST_ELE_T fd_crds_entry_t
200 150 : #define DLIST_PREV contact_info.evict_dlist.prev
201 150 : #define DLIST_NEXT contact_info.evict_dlist.next
202 : #include "../../../util/tmpl/fd_dlist.c"
203 :
204 : #define TREAP_NAME hash_treap
205 : #define TREAP_T fd_crds_entry_t
206 : #define TREAP_QUERY_T ulong
207 0 : #define TREAP_CMP(q,e) ((q>e->hash.hash)-(q<e->hash.hash))
208 1305 : #define TREAP_IDX_T ulong
209 : #define TREAP_OPTIMIZE_ITERATION 1
210 150 : #define TREAP_NEXT hash.next
211 150 : #define TREAP_PREV hash.prev
212 834 : #define TREAP_LT(e0,e1) ((e0)->hash.hash<(e1)->hash.hash)
213 :
214 1005 : #define TREAP_PARENT hash.parent
215 435 : #define TREAP_LEFT hash.left
216 150 : #define TREAP_RIGHT hash.right
217 9792 : #define TREAP_PRIO hash.prio
218 :
219 : #include "../../../util/tmpl/fd_treap.c"
220 :
221 : static inline ulong
222 : lookup_hash( fd_crds_key_t const * key,
223 450 : ulong seed ) {
224 450 : ulong hash_fn = ((ulong)key->tag)<<16;
225 450 : switch( key->tag ) {
226 0 : case FD_GOSSIP_VALUE_VOTE:
227 0 : hash_fn ^= key->vote_index;
228 0 : break;
229 0 : case FD_GOSSIP_VALUE_EPOCH_SLOTS:
230 0 : hash_fn ^= key->epoch_slots_index;
231 0 : break;
232 0 : case FD_GOSSIP_VALUE_DUPLICATE_SHRED:
233 0 : hash_fn ^= key->duplicate_shred_index;
234 0 : break;
235 450 : default:
236 450 : break;
237 450 : }
238 450 : return fd_funk_rec_key_hash1( key->pubkey, hash_fn, seed );
239 450 : }
240 :
241 : static inline int
242 : lookup_eq( fd_crds_key_t const * key0,
243 153 : fd_crds_key_t const * key1 ) {
244 153 : if( FD_UNLIKELY( key0->tag!=key1->tag ) ) return 0;
245 153 : if( FD_UNLIKELY( !!memcmp( key0->pubkey, key1->pubkey, 32UL ) ) ) return 0;
246 150 : switch( key0->tag ) {
247 0 : case FD_GOSSIP_VALUE_VOTE:
248 0 : return key0->vote_index==key1->vote_index;
249 0 : case FD_GOSSIP_VALUE_EPOCH_SLOTS:
250 0 : return key0->epoch_slots_index==key1->epoch_slots_index;
251 0 : case FD_GOSSIP_VALUE_DUPLICATE_SHRED:
252 0 : return key0->duplicate_shred_index==key1->duplicate_shred_index;
253 150 : default:
254 150 : break;
255 150 : }
256 150 : return 1;
257 150 : }
258 :
259 : #define MAP_NAME lookup_map
260 : #define MAP_ELE_T fd_crds_entry_t
261 : #define MAP_KEY_T fd_crds_key_t
262 150 : #define MAP_KEY key
263 759 : #define MAP_IDX_T ulong
264 153 : #define MAP_NEXT lookup.next
265 153 : #define MAP_PREV lookup.prev
266 450 : #define MAP_KEY_HASH(k,s) (lookup_hash( k, s ))
267 153 : #define MAP_KEY_EQ(k0,k1) (lookup_eq( k0, k1 ))
268 : #define MAP_OPTIMIZE_RANDOM_ACCESS_REMOVAL 1
269 :
270 : #include "../../../util/tmpl/fd_map_chain.c"
271 :
272 : #include "fd_crds_peer_samplers.c"
273 :
274 : struct fd_crds_purged {
275 : uchar hash[ 32UL ];
276 : struct {
277 : ulong next;
278 : } pool;
279 :
280 : /* Similar to fd_crds_entry, we want the ability to query and iterate
281 : through value by hash[:8] to generate pull requests. */
282 : struct {
283 : ulong parent;
284 : ulong left;
285 : ulong right;
286 : ulong next;
287 : ulong prev;
288 : ulong prio;
289 : } treap;
290 :
291 : /* Similar to fd_crds_entry, we keep a linked list of purged values sorted
292 : by insertion time. The time used here is our node's wallclock.
293 :
294 : There are actually two (mutually exclusive) lists that reuse the same
295 : pointers here: one for "purged" entries that expire in 60s and one for
296 : "failed_inserts" that expire after 20s. */
297 : struct {
298 : long wallclock_nanos;
299 : ulong next;
300 : ulong prev;
301 : } expire;
302 : };
303 : typedef struct fd_crds_purged fd_crds_purged_t;
304 :
305 : #define POOL_NAME purged_pool
306 18 : #define POOL_T fd_crds_purged_t
307 6144 : #define POOL_NEXT pool.next
308 :
309 : #include "../../../util/tmpl/fd_pool.c"
310 :
311 : #define TREAP_NAME purged_treap
312 : #define TREAP_T fd_crds_purged_t
313 : #define TREAP_QUERY_T ulong
314 0 : #define TREAP_CMP(q,e) ((q>*(ulong *)(e->hash))-(q<*(ulong *)(e->hash)))
315 : #define TREAP_OPTIMIZE_ITERATION 1
316 0 : #define TREAP_NEXT treap.next
317 0 : #define TREAP_PREV treap.prev
318 0 : #define TREAP_LT(e0,e1) (*(ulong *)((e0)->hash)<*(ulong *)((e1)->hash))
319 :
320 0 : #define TREAP_PARENT treap.parent
321 0 : #define TREAP_LEFT treap.left
322 0 : #define TREAP_RIGHT treap.right
323 6144 : #define TREAP_PRIO treap.prio
324 :
325 : #include "../../../util/tmpl/fd_treap.c"
326 :
327 : #define DLIST_NAME failed_inserts_dlist
328 : #define DLIST_ELE_T fd_crds_purged_t
329 0 : #define DLIST_PREV expire.prev
330 0 : #define DLIST_NEXT expire.next
331 :
332 : #include "../../../util/tmpl/fd_dlist.c"
333 :
334 : #define DLIST_NAME purged_dlist
335 : #define DLIST_ELE_T fd_crds_purged_t
336 0 : #define DLIST_PREV expire.prev
337 0 : #define DLIST_NEXT expire.next
338 :
339 : #include "../../../util/tmpl/fd_dlist.c"
340 :
341 : struct fd_crds_private {
342 : fd_gossip_out_ctx_t * gossip_update;
343 :
344 : fd_sha256_t sha256[1];
345 :
346 : int has_staked_node;
347 :
348 : fd_crds_entry_t * pool;
349 :
350 : evict_treap_t * evict_treap;
351 : staked_expire_dlist_t * staked_expire_dlist;
352 : unstaked_expire_dlist_t * unstaked_expire_dlist;
353 : hash_treap_t * hash_treap;
354 : lookup_map_t * lookup_map;
355 :
356 : struct {
357 : fd_crds_purged_t * pool;
358 : purged_treap_t * treap;
359 : purged_dlist_t * purged_dlist;
360 : failed_inserts_dlist_t * failed_inserts_dlist;
361 : } purged;
362 :
363 : struct {
364 : fd_crds_contact_info_entry_t * pool;
365 : crds_contact_info_fresh_list_t * fresh_dlist;
366 : crds_contact_info_evict_dlist_t * evict_dlist;
367 : } contact_info;
368 :
369 : crds_samplers_t samplers[1];
370 :
371 : fd_crds_metrics_t metrics[1];
372 :
373 : ulong magic;
374 : };
375 :
376 : FD_FN_CONST ulong
377 42 : fd_crds_align( void ) {
378 42 : return FD_CRDS_ALIGN;
379 42 : }
380 :
381 : FD_FN_CONST ulong
382 : fd_crds_footprint( ulong ele_max,
383 24 : ulong purged_max ) {
384 24 : ulong l;
385 24 : l = FD_LAYOUT_INIT;
386 24 : l = FD_LAYOUT_APPEND( l, FD_CRDS_ALIGN, sizeof(fd_crds_t) );
387 24 : l = FD_LAYOUT_APPEND( l, crds_pool_align(), crds_pool_footprint( ele_max ) );
388 24 : l = FD_LAYOUT_APPEND( l, evict_treap_align(), evict_treap_footprint( ele_max ) );
389 24 : l = FD_LAYOUT_APPEND( l, staked_expire_dlist_align(), staked_expire_dlist_footprint() );
390 24 : l = FD_LAYOUT_APPEND( l, unstaked_expire_dlist_align(), unstaked_expire_dlist_footprint() );
391 24 : l = FD_LAYOUT_APPEND( l, hash_treap_align(), hash_treap_footprint( ele_max ) );
392 24 : l = FD_LAYOUT_APPEND( l, lookup_map_align(), lookup_map_footprint( ele_max ) );
393 24 : l = FD_LAYOUT_APPEND( l, purged_pool_align(), purged_pool_footprint( purged_max ) );
394 24 : l = FD_LAYOUT_APPEND( l, purged_treap_align(), purged_treap_footprint( purged_max ) );
395 24 : l = FD_LAYOUT_APPEND( l, purged_dlist_align(), purged_dlist_footprint() );
396 24 : l = FD_LAYOUT_APPEND( l, failed_inserts_dlist_align(), failed_inserts_dlist_footprint() );
397 24 : l = FD_LAYOUT_APPEND( l, crds_contact_info_pool_align(), crds_contact_info_pool_footprint( CRDS_MAX_CONTACT_INFO ) );
398 24 : l = FD_LAYOUT_APPEND( l, crds_contact_info_fresh_list_align(), crds_contact_info_fresh_list_footprint() );
399 24 : l = FD_LAYOUT_APPEND( l, crds_contact_info_evict_dlist_align(), crds_contact_info_evict_dlist_footprint() );
400 24 : return FD_LAYOUT_FINI( l, FD_CRDS_ALIGN );
401 24 : }
402 :
403 : void *
404 : fd_crds_new( void * shmem,
405 : fd_rng_t * rng,
406 : ulong ele_max,
407 : ulong purged_max,
408 9 : fd_gossip_out_ctx_t * gossip_update_out ) {
409 9 : if( FD_UNLIKELY( !shmem ) ) {
410 0 : FD_LOG_WARNING(( "NULL shmem" ));
411 0 : return NULL;
412 0 : }
413 :
414 9 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shmem, fd_crds_align() ) ) ) {
415 0 : FD_LOG_WARNING(( "misaligned shmem" ));
416 0 : return NULL;
417 0 : }
418 :
419 9 : if( FD_UNLIKELY( !fd_ulong_is_pow2( ele_max ) ) ) {
420 0 : FD_LOG_WARNING(( "ele_max must be a power of 2" ));
421 0 : return NULL;
422 0 : }
423 :
424 9 : if( FD_UNLIKELY( !fd_ulong_is_pow2( purged_max ) ) ) {
425 0 : FD_LOG_WARNING(( "purged_max must be a power of 2" ));
426 0 : return NULL;
427 0 : }
428 :
429 9 : if( FD_UNLIKELY( !rng ) ) {
430 0 : FD_LOG_WARNING(( "NULL rng" ));
431 0 : return NULL;
432 0 : }
433 :
434 9 : if( FD_UNLIKELY( !gossip_update_out ) ) {
435 0 : FD_LOG_WARNING(( "NULL gossip_out" ));
436 0 : return NULL;
437 0 : }
438 :
439 9 : FD_SCRATCH_ALLOC_INIT( l, shmem );
440 9 : fd_crds_t * crds = FD_SCRATCH_ALLOC_APPEND( l, FD_CRDS_ALIGN, sizeof(fd_crds_t) );
441 9 : void * _pool = FD_SCRATCH_ALLOC_APPEND( l, crds_pool_align(), crds_pool_footprint( ele_max ) );
442 9 : void * _evict_treap = FD_SCRATCH_ALLOC_APPEND( l, evict_treap_align(), evict_treap_footprint( ele_max ) );
443 9 : void * _staked_expire_dlist = FD_SCRATCH_ALLOC_APPEND( l, staked_expire_dlist_align(), staked_expire_dlist_footprint() );
444 9 : void * _unstaked_expire_dlist = FD_SCRATCH_ALLOC_APPEND( l, unstaked_expire_dlist_align(), unstaked_expire_dlist_footprint() );
445 9 : void * _hash_treap = FD_SCRATCH_ALLOC_APPEND( l, hash_treap_align(), hash_treap_footprint( ele_max ) );
446 9 : void * _lookup_map = FD_SCRATCH_ALLOC_APPEND( l, lookup_map_align(), lookup_map_footprint( ele_max ) );
447 9 : void * _purged_pool = FD_SCRATCH_ALLOC_APPEND( l, purged_pool_align(), purged_pool_footprint( purged_max ) );
448 9 : void * _purged_treap = FD_SCRATCH_ALLOC_APPEND( l, purged_treap_align(), purged_treap_footprint( purged_max ) );
449 9 : void * _purged_dlist = FD_SCRATCH_ALLOC_APPEND( l, purged_dlist_align(), purged_dlist_footprint() );
450 9 : void * _failed_inserts_dlist = FD_SCRATCH_ALLOC_APPEND( l, failed_inserts_dlist_align(), failed_inserts_dlist_footprint() );
451 9 : void * _ci_pool = FD_SCRATCH_ALLOC_APPEND( l, crds_contact_info_pool_align(), crds_contact_info_pool_footprint( CRDS_MAX_CONTACT_INFO ) );
452 9 : void * _ci_dlist = FD_SCRATCH_ALLOC_APPEND( l, crds_contact_info_fresh_list_align(), crds_contact_info_fresh_list_footprint() );
453 9 : void * _ci_evict_dlist = FD_SCRATCH_ALLOC_APPEND( l, crds_contact_info_evict_dlist_align(), crds_contact_info_evict_dlist_footprint() );
454 9 : FD_TEST( FD_SCRATCH_ALLOC_FINI( l, FD_CRDS_ALIGN ) == (ulong)shmem + fd_crds_footprint( ele_max, purged_max ) );
455 :
456 9 : crds->pool = crds_pool_join( crds_pool_new( _pool, ele_max ) );
457 9 : FD_TEST( crds->pool );
458 :
459 9 : crds->evict_treap = evict_treap_join( evict_treap_new( _evict_treap, ele_max ) );
460 9 : FD_TEST( crds->evict_treap );
461 9 : evict_treap_seed( crds->pool, ele_max, fd_rng_ulong( rng ) );
462 :
463 9 : crds->staked_expire_dlist = staked_expire_dlist_join( staked_expire_dlist_new( _staked_expire_dlist ) );
464 9 : FD_TEST( crds->staked_expire_dlist );
465 :
466 9 : crds->unstaked_expire_dlist = unstaked_expire_dlist_join( unstaked_expire_dlist_new( _unstaked_expire_dlist ) );
467 9 : FD_TEST( crds->unstaked_expire_dlist );
468 :
469 9 : crds->hash_treap = hash_treap_join( hash_treap_new( _hash_treap, ele_max ) );
470 9 : FD_TEST( crds->hash_treap );
471 9 : hash_treap_seed( crds->pool, ele_max, fd_rng_ulong( rng ) );
472 :
473 9 : crds->lookup_map = lookup_map_join( lookup_map_new( _lookup_map, ele_max, fd_rng_ulong( rng ) ) );
474 9 : FD_TEST( crds->lookup_map );
475 :
476 9 : crds->purged.pool = purged_pool_join( purged_pool_new( _purged_pool, purged_max ) );
477 9 : FD_TEST( crds->purged.pool );
478 :
479 9 : crds->purged.treap = purged_treap_join( purged_treap_new( _purged_treap, purged_max ) );
480 9 : FD_TEST( crds->purged.treap );
481 9 : purged_treap_seed( crds->purged.pool, purged_max, fd_rng_ulong( rng ) );
482 :
483 9 : crds->purged.purged_dlist = purged_dlist_join( purged_dlist_new( _purged_dlist ) );
484 9 : FD_TEST( crds->purged.purged_dlist );
485 :
486 9 : crds->purged.failed_inserts_dlist = failed_inserts_dlist_join( failed_inserts_dlist_new( _failed_inserts_dlist ) );
487 9 : FD_TEST( crds->purged.failed_inserts_dlist );
488 :
489 9 : crds->contact_info.pool = crds_contact_info_pool_join( crds_contact_info_pool_new( _ci_pool, CRDS_MAX_CONTACT_INFO ) );
490 9 : FD_TEST( crds->contact_info.pool );
491 :
492 9 : crds->contact_info.fresh_dlist = crds_contact_info_fresh_list_join( crds_contact_info_fresh_list_new( _ci_dlist ) );
493 9 : FD_TEST( crds->contact_info.fresh_dlist );
494 :
495 9 : crds->contact_info.evict_dlist = crds_contact_info_evict_dlist_join( crds_contact_info_evict_dlist_new( _ci_evict_dlist ) );
496 9 : FD_TEST( crds->contact_info.evict_dlist );
497 :
498 9 : FD_TEST( fd_sha256_join( fd_sha256_new( crds->sha256 ) ) );
499 :
500 9 : crds_samplers_new( crds->samplers );
501 :
502 9 : memset( crds->metrics, 0, sizeof(fd_crds_metrics_t) );
503 :
504 9 : crds->gossip_update = gossip_update_out;
505 9 : crds->has_staked_node = 0;
506 :
507 9 : FD_COMPILER_MFENCE();
508 9 : FD_VOLATILE( crds->magic ) = FD_CRDS_MAGIC;
509 9 : FD_COMPILER_MFENCE();
510 :
511 9 : return (void *)crds;
512 9 : }
513 :
514 : fd_crds_t *
515 9 : fd_crds_join( void * shcrds ) {
516 9 : if( FD_UNLIKELY( !shcrds ) ) {
517 0 : FD_LOG_WARNING(( "NULL shcrds" ));
518 0 : return NULL;
519 0 : }
520 :
521 9 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shcrds, fd_crds_align() ) ) ) {
522 0 : FD_LOG_WARNING(( "misaligned shcrds" ));
523 0 : return NULL;
524 0 : }
525 :
526 9 : fd_crds_t * crds = (fd_crds_t *)shcrds;
527 :
528 9 : if( FD_UNLIKELY( crds->magic!=FD_CRDS_MAGIC ) ) {
529 0 : FD_LOG_WARNING(( "bad magic" ));
530 0 : return NULL;
531 0 : }
532 :
533 9 : return crds;
534 9 : }
535 :
536 : fd_crds_metrics_t const *
537 0 : fd_crds_metrics( fd_crds_t const * crds ) {
538 0 : return crds->metrics;
539 0 : }
540 :
541 : static inline void
542 : remove_contact_info( fd_crds_t * crds,
543 : fd_crds_entry_t * ci,
544 : long now,
545 0 : fd_stem_context_t * stem ) {
546 0 : if( FD_UNLIKELY( !stem ) ) return;
547 0 : fd_gossip_update_message_t * msg = fd_gossip_out_get_chunk( crds->gossip_update );
548 0 : msg->tag = FD_GOSSIP_UPDATE_TAG_CONTACT_INFO_REMOVE;
549 0 : msg->wallclock_nanos = now;
550 0 : fd_memcpy( msg->origin_pubkey, ci->key.pubkey, 32UL );
551 0 : msg->contact_info_remove.idx = crds_contact_info_pool_idx( crds->contact_info.pool, ci->contact_info.ci );
552 0 : fd_gossip_tx_publish_chunk( crds->gossip_update, stem, (ulong)msg->tag, FD_GOSSIP_UPDATE_SZ_CONTACT_INFO_REMOVE, now );
553 :
554 0 : if( FD_LIKELY( ci->stake ) ) crds->metrics->peer_staked_cnt--;
555 0 : else crds->metrics->peer_unstaked_cnt--;
556 :
557 0 : crds->metrics->peer_visible_stake -= ci->stake;
558 :
559 0 : if( FD_LIKELY( !!ci->contact_info.fresh_dlist.in_list ) ) {
560 0 : crds_contact_info_fresh_list_ele_remove( crds->contact_info.fresh_dlist, ci, crds->pool );
561 0 : }
562 0 : crds_contact_info_evict_dlist_ele_remove( crds->contact_info.evict_dlist, ci, crds->pool );
563 0 : crds_contact_info_pool_ele_release( crds->contact_info.pool, ci->contact_info.ci );
564 :
565 : /* FIXME: If the peer is in any active set bucket, it is NOT removed
566 : here. If the peer is re-inserted into the CRDS table in the future,
567 : it is added back into the bucket's sampler. This means a peer can
568 : be sampled in a bucket (at least) twice during
569 : fd_active_set_rotate. */
570 0 : crds_samplers_rem_peer( crds->samplers, ci );
571 0 : }
572 :
573 : ulong
574 0 : fd_crds_len( fd_crds_t const * crds ) {
575 0 : return crds_pool_used( crds->pool );
576 0 : }
577 :
578 : ulong
579 0 : fd_crds_purged_len( fd_crds_t const * crds ) {
580 0 : return purged_pool_used( crds->purged.pool );
581 0 : }
582 :
583 : void
584 : fd_crds_release( fd_crds_t * crds,
585 0 : fd_crds_entry_t * value ) {
586 0 : crds_pool_ele_release( crds->pool, value );
587 0 : crds->metrics->count[ value->key.tag ]--;
588 0 : }
589 :
590 : static inline void
591 : expire( fd_crds_t * crds,
592 : long now,
593 0 : fd_stem_context_t * stem ){
594 0 : static const long SLOT_DURATION_NANOS = 400L*1000L*1000L;
595 0 : static const long STAKED_EXPIRE_DURATION_NANOS = 432000L*SLOT_DURATION_NANOS;
596 0 : static const long UNSTAKED_EXPIRE_DURATION_NANOS = 15L*1000L*1000L*1000L;
597 :
598 0 : while( !staked_expire_dlist_is_empty( crds->staked_expire_dlist, crds->pool ) ) {
599 0 : fd_crds_entry_t * head = staked_expire_dlist_ele_peek_head( crds->staked_expire_dlist, crds->pool );
600 :
601 0 : if( FD_LIKELY( head->expire.wallclock_nanos>now-STAKED_EXPIRE_DURATION_NANOS ) ) break;
602 :
603 0 : staked_expire_dlist_ele_pop_head( crds->staked_expire_dlist, crds->pool );
604 0 : hash_treap_ele_remove( crds->hash_treap, head, crds->pool );
605 0 : lookup_map_ele_remove( crds->lookup_map, &head->key, NULL, crds->pool );
606 0 : evict_treap_ele_remove( crds->evict_treap, head, crds->pool );
607 :
608 0 : if( FD_UNLIKELY( head->key.tag==FD_GOSSIP_VALUE_CONTACT_INFO ) ) remove_contact_info( crds, head, now, stem );
609 0 : fd_crds_release( crds, head );
610 :
611 0 : crds->metrics->expired_cnt++;
612 0 : }
613 :
614 0 : long unstaked_expire_duration_nanos = fd_long_if( crds->has_staked_node,
615 0 : UNSTAKED_EXPIRE_DURATION_NANOS,
616 0 : STAKED_EXPIRE_DURATION_NANOS );
617 :
618 0 : while( !unstaked_expire_dlist_is_empty( crds->unstaked_expire_dlist, crds->pool ) ) {
619 0 : fd_crds_entry_t * head = unstaked_expire_dlist_ele_peek_head( crds->unstaked_expire_dlist, crds->pool );
620 :
621 0 : if( FD_LIKELY( head->expire.wallclock_nanos>now-unstaked_expire_duration_nanos ) ) break;
622 :
623 0 : unstaked_expire_dlist_ele_pop_head( crds->unstaked_expire_dlist, crds->pool );
624 0 : hash_treap_ele_remove( crds->hash_treap, head, crds->pool );
625 0 : lookup_map_ele_remove( crds->lookup_map, &head->key, NULL, crds->pool );
626 0 : evict_treap_ele_remove( crds->evict_treap, head, crds->pool );
627 :
628 0 : if( FD_UNLIKELY( head->key.tag==FD_GOSSIP_VALUE_CONTACT_INFO ) ) remove_contact_info( crds, head, now, stem );
629 0 : fd_crds_release( crds, head );
630 :
631 0 : crds->metrics->expired_cnt++;
632 0 : }
633 :
634 0 : while( !purged_dlist_is_empty( crds->purged.purged_dlist, crds->purged.pool ) ) {
635 0 : fd_crds_purged_t * head = purged_dlist_ele_peek_head( crds->purged.purged_dlist, crds->purged.pool );
636 :
637 0 : if( FD_LIKELY( head->expire.wallclock_nanos>now-60L*1000L*1000L*1000L ) ) break;
638 :
639 0 : purged_dlist_ele_pop_head( crds->purged.purged_dlist, crds->purged.pool );
640 0 : purged_treap_ele_remove( crds->purged.treap, head, crds->purged.pool );
641 0 : purged_pool_ele_release( crds->purged.pool, head );
642 :
643 0 : crds->metrics->purged_cnt--;
644 0 : crds->metrics->purged_expired_cnt++;
645 0 : }
646 :
647 0 : while( !failed_inserts_dlist_is_empty( crds->purged.failed_inserts_dlist, crds->purged.pool ) ) {
648 0 : fd_crds_purged_t * head = failed_inserts_dlist_ele_peek_head( crds->purged.failed_inserts_dlist, crds->purged.pool );
649 :
650 0 : if( FD_LIKELY( head->expire.wallclock_nanos>now-20L*1000L*1000L*1000L ) ) break;
651 :
652 0 : failed_inserts_dlist_ele_pop_head( crds->purged.failed_inserts_dlist, crds->purged.pool );
653 0 : purged_treap_ele_remove( crds->purged.treap, head, crds->purged.pool );
654 0 : purged_pool_ele_release( crds->purged.pool, head );
655 :
656 0 : crds->metrics->purged_cnt--;
657 0 : crds->metrics->purged_expired_cnt++;
658 0 : }
659 0 : }
660 :
661 : void
662 : unfresh( fd_crds_t * crds,
663 0 : long now ) {
664 0 : while( !crds_contact_info_fresh_list_is_empty( crds->contact_info.fresh_dlist, crds->pool ) ) {
665 0 : fd_crds_entry_t * head = crds_contact_info_fresh_list_ele_peek_head( crds->contact_info.fresh_dlist, crds->pool );
666 :
667 0 : if( FD_LIKELY( head->expire.wallclock_nanos>now-60L*1000L*1000L*1000L ) ) break;
668 :
669 0 : head = crds_contact_info_fresh_list_ele_pop_head( crds->contact_info.fresh_dlist, crds->pool );
670 0 : head->contact_info.fresh_dlist.in_list = 0;
671 0 : crds_samplers_upd_peer_at_idx( crds->samplers, head, head->contact_info.sampler_idx, now );
672 0 : }
673 0 : }
674 :
675 : void
676 : fd_crds_advance( fd_crds_t * crds,
677 : long now,
678 0 : fd_stem_context_t * stem ) {
679 0 : expire( crds, now, stem );
680 0 : unfresh( crds, now );
681 0 : }
682 :
683 : fd_crds_entry_t *
684 : fd_crds_acquire( fd_crds_t * crds,
685 : long now,
686 150 : fd_stem_context_t * stem ) {
687 150 : if( FD_UNLIKELY( !crds_pool_free( crds->pool ) ) ) {
688 0 : evict_treap_fwd_iter_t head = evict_treap_fwd_iter_init( crds->evict_treap, crds->pool );
689 0 : FD_TEST( !evict_treap_fwd_iter_done( head ) );
690 0 : fd_crds_entry_t * evict = evict_treap_fwd_iter_ele( head, crds->pool );
691 :
692 0 : if( FD_LIKELY( !evict->stake ) ) unstaked_expire_dlist_ele_remove( crds->unstaked_expire_dlist, evict, crds->pool );
693 0 : else staked_expire_dlist_ele_remove( crds->staked_expire_dlist, evict, crds->pool );
694 :
695 0 : hash_treap_ele_remove( crds->hash_treap, evict, crds->pool );
696 0 : lookup_map_ele_remove( crds->lookup_map, &evict->key, NULL, crds->pool );
697 0 : evict_treap_ele_remove( crds->evict_treap, evict, crds->pool );
698 0 : if( FD_UNLIKELY( evict->key.tag==FD_GOSSIP_VALUE_CONTACT_INFO ) ) remove_contact_info( crds, evict, now, stem );
699 :
700 0 : crds->metrics->evicted_cnt++;
701 :
702 0 : return evict;
703 150 : } else {
704 150 : return crds_pool_ele_acquire( crds->pool );
705 150 : }
706 150 : }
707 :
708 : int
709 0 : fd_crds_has_staked_node( fd_crds_t const * crds ) {
710 0 : return crds->has_staked_node;
711 0 : }
712 :
713 : static inline void
714 : generate_key( fd_gossip_view_crds_value_t const * view,
715 : uchar const * payload,
716 150 : fd_crds_key_t * out_key ) {
717 150 : out_key->tag = view->tag;
718 150 : fd_memcpy( out_key->pubkey, payload+view->pubkey_off, 32UL );
719 :
720 150 : switch( out_key->tag ) {
721 0 : case FD_GOSSIP_VALUE_VOTE:
722 0 : out_key->vote_index = view->vote->index;
723 0 : break;
724 0 : case FD_GOSSIP_VALUE_EPOCH_SLOTS:
725 0 : out_key->epoch_slots_index = view->epoch_slots->index;
726 0 : break;
727 0 : case FD_GOSSIP_VALUE_DUPLICATE_SHRED:
728 0 : out_key->duplicate_shred_index = view->duplicate_shred->index;
729 0 : break;
730 150 : default:
731 150 : break;
732 150 : }
733 150 : }
734 :
735 : void
736 : fd_crds_generate_hash( fd_sha256_t * sha,
737 : uchar const * crds_value,
738 : ulong crds_value_sz,
739 150 : uchar out_hash[ static 32UL ] ){
740 150 : fd_sha256_init( sha );
741 150 : fd_sha256_append( sha, crds_value, crds_value_sz );
742 150 : fd_sha256_fini( sha, out_hash );
743 150 : }
744 :
745 : static inline void
746 : crds_entry_init( fd_gossip_view_crds_value_t const * view,
747 : fd_sha256_t * sha,
748 : uchar const * payload,
749 : ulong stake,
750 150 : fd_crds_entry_t * out_value ) {
751 : /* Construct key */
752 150 : fd_crds_key_t * key = &out_value->key;
753 150 : generate_key( view, payload, key );
754 :
755 150 : out_value->wallclock_nanos = view->wallclock_nanos;
756 150 : out_value->stake = stake;
757 :
758 150 : fd_crds_generate_hash( sha, payload+view->value_off, view->length, out_value->value_hash );
759 150 : out_value->hash.hash = fd_ulong_load_8( out_value->value_hash );
760 :
761 150 : if( FD_UNLIKELY( view->tag==FD_GOSSIP_VALUE_NODE_INSTANCE ) ) {
762 0 : out_value->node_instance.token = view->node_instance->token;
763 150 : } else if( FD_UNLIKELY( key->tag==FD_GOSSIP_VALUE_CONTACT_INFO ) ) {
764 150 : out_value->contact_info.instance_creation_wallclock_nanos = view->ci_view->contact_info->instance_creation_wallclock_nanos;
765 : /* Contact Info entry will be added to sampler upon successful insertion */
766 150 : out_value->contact_info.sampler_idx = SAMPLE_IDX_SENTINEL;
767 150 : }
768 150 : }
769 :
770 : static inline void
771 : purged_init( fd_crds_purged_t * purged,
772 : uchar const * hash,
773 0 : long now ) {
774 0 : fd_memcpy( purged->hash, hash, 32UL );
775 0 : purged->expire.wallclock_nanos = now;
776 0 : }
777 :
778 : void
779 : insert_purged( fd_crds_t * crds,
780 : uchar const * hash,
781 0 : long now ) {
782 0 : if( purged_treap_ele_query( crds->purged.treap, *(ulong *)hash, crds->purged.pool ) ) {
783 0 : return;
784 0 : }
785 0 : fd_crds_purged_t * purged;
786 0 : if( FD_UNLIKELY( !purged_pool_free( crds->purged.pool ) ) ) {
787 0 : purged = purged_dlist_ele_pop_head( crds->purged.purged_dlist, crds->purged.pool );
788 0 : purged_treap_ele_remove( crds->purged.treap, purged, crds->purged.pool );
789 0 : if( FD_LIKELY( crds->metrics ) ) {
790 0 : crds->metrics->purged_evicted_cnt++;
791 0 : }
792 0 : } else {
793 0 : purged = purged_pool_ele_acquire( crds->purged.pool );
794 0 : if( FD_LIKELY( crds->metrics ) ) {
795 0 : crds->metrics->purged_cnt++;
796 0 : }
797 0 : }
798 0 : purged_init( purged, hash, now );
799 0 : purged_treap_ele_insert( crds->purged.treap, purged, crds->purged.pool );
800 0 : purged_dlist_ele_push_tail( crds->purged.purged_dlist, purged, crds->purged.pool );
801 0 : }
802 :
803 : /* overrides_fast
804 : - returns 1 if candidate overrides existing (incumbent) CRDS value
805 : - returns 0 if candidate does not override existing CRDS value
806 : - return -1 if further checks are needed (e.g. hash comparison) */
807 : int
808 : overrides_fast( fd_crds_entry_t const * incumbent,
809 : fd_gossip_view_crds_value_t const * candidate,
810 0 : uchar const * payload ){
811 0 : long existing_wc = incumbent->wallclock_nanos;
812 0 : long candidate_wc = candidate->wallclock_nanos;
813 0 : long existing_ci_onset = incumbent->contact_info.instance_creation_wallclock_nanos;
814 0 : long candidate_ci_onset = candidate->ci_view->contact_info->instance_creation_wallclock_nanos;
815 :
816 0 : switch( candidate->tag ) {
817 0 : case FD_GOSSIP_VALUE_CONTACT_INFO:
818 0 : if( FD_UNLIKELY( candidate_ci_onset>existing_ci_onset ) ) return 1;
819 0 : else if( FD_UNLIKELY( candidate_ci_onset<existing_ci_onset ) ) return 0;
820 0 : else if( FD_UNLIKELY( candidate_wc>existing_wc ) ) return 1;
821 0 : else if( FD_UNLIKELY( candidate_wc<existing_wc ) ) return 0;
822 0 : break;
823 0 : case FD_GOSSIP_VALUE_NODE_INSTANCE:
824 0 : if( FD_LIKELY( candidate->node_instance->token==incumbent->node_instance.token ) ) break;
825 0 : else if( FD_LIKELY( memcmp( payload+candidate->pubkey_off, incumbent->key.pubkey, 32UL ) ) ) break;
826 0 : else if( FD_UNLIKELY( candidate_wc>existing_wc ) ) return 1;
827 0 : else if( FD_UNLIKELY( candidate_wc<existing_wc ) ) return 0;
828 0 : else if( candidate->node_instance->token<incumbent->node_instance.token ) return 0;;
829 0 : break;
830 0 : default:
831 0 : break;
832 0 : }
833 :
834 0 : if( FD_UNLIKELY( candidate_wc>existing_wc ) ) return 1;
835 0 : else if( FD_UNLIKELY( candidate_wc<existing_wc ) ) return 0;
836 0 : return -1;
837 0 : }
838 :
839 :
840 : void
841 : fd_crds_insert_failed_insert( fd_crds_t * crds,
842 : uchar const * hash,
843 0 : long now ) {
844 0 : if( purged_treap_ele_query( crds->purged.treap, *(ulong *)hash, crds->purged.pool ) ) {
845 0 : return;
846 0 : }
847 0 : fd_crds_purged_t * failed;
848 0 : if( FD_UNLIKELY( !purged_pool_free( crds->purged.pool ) ) ) {
849 0 : failed = failed_inserts_dlist_ele_pop_head( crds->purged.failed_inserts_dlist, crds->purged.pool );
850 0 : purged_treap_ele_remove( crds->purged.treap, failed, crds->purged.pool );
851 0 : if( FD_LIKELY( crds->metrics ) ) {
852 0 : crds->metrics->purged_evicted_cnt++;
853 0 : }
854 0 : } else {
855 0 : failed = purged_pool_ele_acquire( crds->purged.pool );
856 0 : if( FD_LIKELY( crds->metrics ) ) {
857 0 : crds->metrics->purged_cnt++;
858 0 : }
859 0 : }
860 0 : purged_init( failed, hash, now );
861 0 : purged_treap_ele_insert( crds->purged.treap, failed, crds->purged.pool );
862 0 : failed_inserts_dlist_ele_push_tail( crds->purged.failed_inserts_dlist, failed, crds->purged.pool );
863 0 : }
864 :
865 : int
866 : fd_crds_checks_fast( fd_crds_t * crds,
867 : fd_gossip_view_crds_value_t const * candidate,
868 : uchar const * payload,
869 0 : uchar from_push_msg ) {
870 0 : fd_crds_key_t candidate_key;
871 0 : generate_key( candidate, payload, &candidate_key );
872 0 : fd_crds_entry_t * incumbent = lookup_map_ele_query( crds->lookup_map, &candidate_key, NULL, crds->pool );
873 :
874 0 : if( FD_UNLIKELY( !incumbent ) ) return FD_CRDS_UPSERT_CHECK_UPSERTS;
875 :
876 0 : if( FD_UNLIKELY( *(ulong *)incumbent->value_bytes==(*(ulong *)(payload+candidate->value_off)) ) ) {
877 : /* We have a duplicate, so we return the number of duplicates */
878 0 : return (int)(++incumbent->num_duplicates);
879 0 : }
880 0 : int overrides = overrides_fast( incumbent, candidate, payload );
881 0 : if( FD_LIKELY( overrides==1 ) ) return FD_CRDS_UPSERT_CHECK_UPSERTS;
882 :
883 0 : uchar cand_hash[ 32UL ];
884 0 : fd_crds_generate_hash( crds->sha256, payload+candidate->value_off, candidate->length, cand_hash );
885 :
886 0 : if( FD_UNLIKELY( overrides==-1 ) ) {
887 : /* Tiebreaker case, we compare hash values */
888 0 : int res = memcmp( cand_hash, incumbent->value_hash, 32UL );
889 0 : if( FD_UNLIKELY( !res ) ) {
890 : /* Hashes match, so we treat this as a duplicate */
891 0 : return (int)(++incumbent->num_duplicates);
892 0 : } else if( res>0 ) {
893 : /* Candidate hash is greater than incumbent hash, so we treat
894 : this as an upsert */
895 0 : return FD_CRDS_UPSERT_CHECK_UPSERTS;
896 0 : }
897 0 : }
898 :
899 0 : from_push_msg ? insert_purged( crds, cand_hash, candidate->wallclock_nanos ) :
900 0 : fd_crds_insert_failed_insert( crds, cand_hash, candidate->wallclock_nanos );
901 :
902 0 : return FD_CRDS_UPSERT_CHECK_FAILS;
903 0 : }
904 :
905 : static inline void
906 : publish_update_msg( fd_crds_t * crds,
907 : fd_crds_entry_t * entry,
908 : fd_gossip_view_crds_value_t const * entry_view,
909 : uchar const * payload,
910 : long now,
911 150 : fd_stem_context_t * stem ) {
912 150 : if( FD_UNLIKELY( !stem ) ) return;
913 0 : if( FD_LIKELY( entry->key.tag!=FD_GOSSIP_VALUE_CONTACT_INFO &&
914 0 : entry->key.tag!=FD_GOSSIP_VALUE_LOWEST_SLOT &&
915 0 : entry->key.tag!=FD_GOSSIP_VALUE_VOTE &&
916 0 : entry->key.tag!=FD_GOSSIP_VALUE_DUPLICATE_SHRED &&
917 0 : entry->key.tag!=FD_GOSSIP_VALUE_INC_SNAPSHOT_HASHES ) ) {
918 0 : return;
919 0 : }
920 :
921 0 : fd_gossip_update_message_t * msg = fd_gossip_out_get_chunk( crds->gossip_update );
922 0 : msg->wallclock_nanos = now;
923 0 : fd_memcpy( msg->origin_pubkey, entry->key.pubkey, 32UL );
924 0 : ulong sz;
925 0 : switch( entry->key.tag ) {
926 0 : case FD_GOSSIP_VALUE_CONTACT_INFO:
927 0 : msg->tag = FD_GOSSIP_UPDATE_TAG_CONTACT_INFO;
928 0 : *msg->contact_info.contact_info = *entry->contact_info.ci->contact_info;
929 0 : msg->contact_info.idx = crds_contact_info_pool_idx( crds->contact_info.pool, entry->contact_info.ci );
930 0 : sz = FD_GOSSIP_UPDATE_SZ_CONTACT_INFO;
931 0 : break;
932 0 : case FD_GOSSIP_VALUE_LOWEST_SLOT:
933 0 : msg->tag = FD_GOSSIP_UPDATE_TAG_LOWEST_SLOT;
934 0 : sz = FD_GOSSIP_UPDATE_SZ_LOWEST_SLOT;
935 0 : msg->lowest_slot = entry_view->lowest_slot;
936 0 : break;
937 0 : case FD_GOSSIP_VALUE_VOTE:
938 0 : msg->tag = FD_GOSSIP_UPDATE_TAG_VOTE;
939 : /* TODO: dynamic sizing */
940 0 : sz = FD_GOSSIP_UPDATE_SZ_VOTE;
941 0 : msg->vote.vote_tower_index = entry->key.vote_index;
942 0 : msg->vote.txn_sz = entry_view->vote->txn_sz;
943 0 : fd_memcpy( msg->vote.txn, payload+entry_view->vote->txn_off, entry_view->vote->txn_sz );
944 0 : break;
945 0 : case FD_GOSSIP_VALUE_DUPLICATE_SHRED:
946 0 : msg->tag = FD_GOSSIP_UPDATE_TAG_DUPLICATE_SHRED;
947 : /* TODO: dynamic sizing */
948 0 : sz = FD_GOSSIP_UPDATE_SZ_DUPLICATE_SHRED;
949 0 : {
950 0 : fd_gossip_view_duplicate_shred_t const * ds = entry_view->duplicate_shred;
951 0 : fd_gossip_duplicate_shred_t * ds_msg = &msg->duplicate_shred;
952 :
953 0 : ds_msg->index = ds->index;
954 0 : ds_msg->slot = ds->slot;
955 0 : ds_msg->num_chunks = ds->num_chunks;
956 0 : ds_msg->chunk_index = ds->chunk_index;
957 0 : ds_msg->wallclock = entry->wallclock_nanos;
958 0 : ds_msg->chunk_len = ds->chunk_len;
959 0 : fd_memcpy( ds_msg->chunk, payload+ds->chunk_off, ds->chunk_len );
960 0 : }
961 0 : break;
962 0 : case FD_GOSSIP_VALUE_INC_SNAPSHOT_HASHES:
963 0 : msg->tag = FD_GOSSIP_UPDATE_TAG_SNAPSHOT_HASHES;
964 : /* TODO: dynamic sizing */
965 0 : sz = FD_GOSSIP_UPDATE_SZ_SNAPSHOT_HASHES;
966 0 : {
967 0 : fd_gossip_view_snapshot_hashes_t const * sh = entry_view->snapshot_hashes;
968 0 : fd_gossip_snapshot_hashes_t * sh_msg = &msg->snapshot_hashes;
969 :
970 0 : sh_msg->incremental_len = sh->inc_len;
971 0 : fd_memcpy( sh_msg->full, payload+sh->full_off, sizeof(fd_gossip_snapshot_hash_pair_t) );
972 0 : fd_memcpy( sh_msg->incremental, payload+sh->inc_off, sh->inc_len*sizeof(fd_gossip_snapshot_hash_pair_t) );
973 0 : }
974 0 : break;
975 0 : default:
976 0 : FD_LOG_ERR(( "impossible" ));
977 0 : }
978 0 : fd_gossip_tx_publish_chunk( crds->gossip_update,
979 0 : stem,
980 0 : (ulong)msg->tag,
981 0 : sz,
982 0 : now );
983 0 : }
984 :
985 : fd_crds_entry_t const *
986 : fd_crds_insert( fd_crds_t * crds,
987 : fd_gossip_view_crds_value_t const * candidate_view,
988 : uchar const * payload,
989 : ulong origin_stake,
990 : uchar is_from_me,
991 : long now ,
992 150 : fd_stem_context_t * stem ) {
993 : /* Update table count metrics at the end to avoid early return
994 : handling */
995 150 : fd_crds_entry_t * candidate = fd_crds_acquire( crds, now, stem );
996 150 : crds_entry_init( candidate_view, crds->sha256, payload, origin_stake, candidate );
997 :
998 150 : crds->metrics->count[ candidate->key.tag ]++;
999 :
1000 150 : fd_crds_entry_t * incumbent = lookup_map_ele_query( crds->lookup_map, &candidate->key, NULL, crds->pool );
1001 150 : uchar is_replacing = incumbent!=NULL;
1002 150 : if( FD_LIKELY( is_replacing ) ) {
1003 0 : insert_purged( crds, incumbent->value_hash, now );
1004 :
1005 0 : if( FD_UNLIKELY( incumbent->key.tag==FD_GOSSIP_VALUE_CONTACT_INFO ) ) {
1006 0 : if( FD_LIKELY( !!incumbent->contact_info.fresh_dlist.in_list ) ) crds_contact_info_fresh_list_ele_remove( crds->contact_info.fresh_dlist, incumbent, crds->pool );
1007 0 : crds_contact_info_evict_dlist_ele_remove( crds->contact_info.evict_dlist, incumbent, crds->pool );
1008 0 : candidate->contact_info.ci = incumbent->contact_info.ci;
1009 :
1010 : /* is_active is user controlled (specifically by ping_tracker),
1011 : and is used in sampler score calculations. So we inherit the
1012 : incumbent's setting. */
1013 0 : candidate->contact_info.is_active = incumbent->contact_info.is_active;
1014 0 : if( FD_LIKELY( !is_from_me ) ) {
1015 0 : if( FD_UNLIKELY( candidate->stake!=incumbent->stake ) ) {
1016 : /* Perform a rescore here (expensive) */
1017 0 : crds_samplers_upd_peer_at_idx( crds->samplers, candidate, incumbent->contact_info.sampler_idx, now );
1018 0 : } else {
1019 0 : crds_samplers_swap_peer_at_idx( crds->samplers, candidate, incumbent->contact_info.sampler_idx );
1020 0 : }
1021 0 : }
1022 :
1023 0 : if( FD_LIKELY( incumbent->stake ) ) crds->metrics->peer_staked_cnt--;
1024 0 : else crds->metrics->peer_unstaked_cnt--;
1025 0 : crds->metrics->peer_visible_stake -= incumbent->stake;
1026 0 : }
1027 :
1028 0 : if( FD_LIKELY( incumbent->stake ) ) {
1029 0 : staked_expire_dlist_ele_remove( crds->staked_expire_dlist, incumbent, crds->pool );
1030 0 : } else {
1031 0 : unstaked_expire_dlist_ele_remove( crds->unstaked_expire_dlist, incumbent, crds->pool );
1032 0 : }
1033 0 : evict_treap_ele_remove( crds->evict_treap, incumbent, crds->pool );
1034 0 : hash_treap_ele_remove( crds->hash_treap, incumbent, crds->pool );
1035 0 : lookup_map_ele_remove( crds->lookup_map, &incumbent->key, NULL, crds->pool );
1036 0 : fd_crds_release( crds, incumbent );
1037 150 : } else if( candidate->key.tag==FD_GOSSIP_VALUE_CONTACT_INFO ) {
1038 150 : if( FD_UNLIKELY( !crds_contact_info_pool_free( crds->contact_info.pool ) ) ) {
1039 0 : fd_crds_entry_t * evict = crds_contact_info_evict_dlist_ele_peek_head( crds->contact_info.evict_dlist, crds->pool );
1040 0 : remove_contact_info( crds, evict, now, stem );
1041 0 : if( FD_LIKELY( evict->stake ) ) {
1042 0 : staked_expire_dlist_ele_remove( crds->staked_expire_dlist, evict, crds->pool );
1043 0 : } else {
1044 0 : unstaked_expire_dlist_ele_remove( crds->unstaked_expire_dlist, evict, crds->pool );
1045 0 : }
1046 0 : evict_treap_ele_remove( crds->evict_treap, evict, crds->pool );
1047 0 : hash_treap_ele_remove( crds->hash_treap, evict, crds->pool );
1048 0 : lookup_map_ele_remove( crds->lookup_map, &evict->key, NULL, crds->pool );
1049 0 : fd_crds_release( crds, evict );
1050 0 : crds->metrics->peer_evicted_cnt++;
1051 0 : crds->metrics->evicted_cnt++;
1052 0 : }
1053 :
1054 150 : candidate->contact_info.ci = crds_contact_info_pool_ele_acquire( crds->contact_info.pool );
1055 150 : }
1056 :
1057 150 : candidate->num_duplicates = 0UL;
1058 150 : candidate->expire.wallclock_nanos = now;
1059 150 : candidate->value_sz = candidate_view->length;
1060 150 : fd_memcpy( candidate->value_bytes, payload+candidate_view->value_off, candidate_view->length );
1061 :
1062 150 : crds->has_staked_node |= candidate->stake ? 1 : 0;
1063 :
1064 150 : evict_treap_ele_insert( crds->evict_treap, candidate, crds->pool );
1065 150 : if( FD_LIKELY( candidate->stake ) ) {
1066 150 : staked_expire_dlist_ele_push_tail( crds->staked_expire_dlist, candidate, crds->pool );
1067 150 : } else {
1068 0 : unstaked_expire_dlist_ele_push_tail( crds->unstaked_expire_dlist, candidate, crds->pool );
1069 0 : }
1070 150 : hash_treap_ele_insert( crds->hash_treap, candidate, crds->pool );
1071 150 : lookup_map_ele_insert( crds->lookup_map, candidate, crds->pool );
1072 :
1073 150 : if( FD_UNLIKELY( candidate->key.tag==FD_GOSSIP_VALUE_CONTACT_INFO ) ) {
1074 150 : fd_memcpy( candidate->contact_info.ci->contact_info, candidate_view->ci_view->contact_info, sizeof(fd_contact_info_t) );
1075 : /* Default to active, since we filter inactive entries prior to insertion */
1076 150 : candidate->contact_info.is_active = 1;
1077 :
1078 150 : crds_contact_info_evict_dlist_ele_push_tail( crds->contact_info.evict_dlist, candidate, crds->pool );
1079 :
1080 150 : if( FD_LIKELY( !is_from_me ) ){
1081 150 : crds_contact_info_fresh_list_ele_push_tail( crds->contact_info.fresh_dlist, candidate, crds->pool );
1082 150 : candidate->contact_info.fresh_dlist.in_list = 1;
1083 150 : } else {
1084 0 : candidate->contact_info.fresh_dlist.in_list = 0;
1085 0 : }
1086 :
1087 150 : if( FD_UNLIKELY( !is_replacing && !is_from_me ) ) {
1088 150 : crds_samplers_add_peer( crds->samplers, candidate, now);
1089 150 : }
1090 :
1091 150 : if( FD_LIKELY( candidate->stake ) ) crds->metrics->peer_staked_cnt++;
1092 0 : else crds->metrics->peer_unstaked_cnt++;
1093 150 : crds->metrics->peer_visible_stake += candidate->stake;
1094 150 : }
1095 :
1096 150 : publish_update_msg( crds, candidate, candidate_view, payload, now, stem );
1097 150 : return candidate;
1098 150 : }
1099 :
1100 : void
1101 : fd_crds_entry_value( fd_crds_entry_t const * entry,
1102 : uchar const ** value_bytes,
1103 0 : ulong * value_sz ) {
1104 0 : *value_bytes = entry->value_bytes;
1105 0 : *value_sz = entry->value_sz;
1106 0 : }
1107 :
1108 : uchar const *
1109 0 : fd_crds_entry_hash( fd_crds_entry_t const * entry ) {
1110 0 : return entry->value_hash;
1111 0 : }
1112 :
1113 : inline static void
1114 : make_contact_info_key( uchar const * pubkey,
1115 150 : fd_crds_key_t * key_out ) {
1116 150 : key_out->tag = FD_GOSSIP_VALUE_CONTACT_INFO;
1117 150 : fd_memcpy( key_out->pubkey, pubkey, 32UL );
1118 150 : }
1119 :
1120 : int
1121 0 : fd_crds_entry_is_contact_info( fd_crds_entry_t const * entry ) {
1122 0 : return entry->key.tag==FD_GOSSIP_VALUE_CONTACT_INFO;
1123 0 : }
1124 :
1125 : fd_contact_info_t *
1126 3 : fd_crds_entry_contact_info( fd_crds_entry_t const * entry ) {
1127 3 : return entry->contact_info.ci->contact_info;
1128 3 : }
1129 :
1130 :
1131 : fd_contact_info_t const *
1132 : fd_crds_contact_info_lookup( fd_crds_t const * crds,
1133 0 : uchar const * pubkey ) {
1134 :
1135 0 : fd_crds_key_t key[1];
1136 0 : make_contact_info_key( pubkey, key );
1137 0 : fd_crds_entry_t * peer_ci = lookup_map_ele_query( crds->lookup_map, key, NULL, crds->pool );
1138 0 : if( FD_UNLIKELY( !peer_ci ) ) {
1139 0 : return NULL;
1140 0 : }
1141 :
1142 0 : return peer_ci->contact_info.ci->contact_info;
1143 0 : }
1144 :
1145 : ulong
1146 3 : fd_crds_peer_count( fd_crds_t const * crds ){
1147 3 : return crds_contact_info_pool_used( crds->contact_info.pool );
1148 3 : }
1149 :
1150 : static inline void
1151 : set_peer_active_status( fd_crds_t * crds,
1152 : uchar const * peer_pubkey,
1153 : uchar status,
1154 150 : long now ) {
1155 :
1156 150 : fd_crds_key_t key[1];
1157 150 : make_contact_info_key( peer_pubkey, key );
1158 :
1159 150 : fd_crds_entry_t * peer_ci = lookup_map_ele_query( crds->lookup_map, key, NULL, crds->pool );
1160 : /* TODO: error handling? This technically should never hit */
1161 150 : if( FD_UNLIKELY( !peer_ci ) ) return;
1162 150 : uchar old_status = peer_ci->contact_info.is_active;
1163 150 : peer_ci->contact_info.is_active = status;
1164 :
1165 150 : if( FD_UNLIKELY( old_status!=status ) ) {
1166 : /* Trigger sampler update */
1167 0 : crds_samplers_upd_peer_at_idx( crds->samplers,
1168 0 : peer_ci,
1169 0 : peer_ci->contact_info.sampler_idx,
1170 0 : now );
1171 0 : }
1172 150 : }
1173 : void
1174 : fd_crds_peer_active( fd_crds_t * crds,
1175 : uchar const * peer_pubkey,
1176 150 : long now ) {
1177 150 : set_peer_active_status( crds, peer_pubkey, 1 /* active */, now );
1178 150 : }
1179 :
1180 : void
1181 : fd_crds_peer_inactive( fd_crds_t * crds,
1182 : uchar const * peer_pubkey,
1183 0 : long now ) {
1184 0 : set_peer_active_status( crds, peer_pubkey, 0 /* inactive */, now );
1185 0 : }
1186 :
1187 : fd_contact_info_t const *
1188 : fd_crds_peer_sample( fd_crds_t const * crds,
1189 0 : fd_rng_t * rng ) {
1190 0 : ulong idx = wpeer_sampler_sample( crds->samplers->pr_sampler,
1191 0 : rng,
1192 0 : crds->samplers->ele_cnt );
1193 0 : if( FD_UNLIKELY( idx==SAMPLE_IDX_SENTINEL ) ) return NULL;
1194 0 : return fd_crds_entry_contact_info( crds->samplers->ele[idx] );
1195 0 : }
1196 :
1197 : fd_contact_info_t const *
1198 : fd_crds_bucket_sample_and_remove( fd_crds_t * crds,
1199 : fd_rng_t * rng,
1200 3 : ulong bucket ) {
1201 3 : ulong idx = wpeer_sampler_sample( &crds->samplers->bucket_samplers[bucket],
1202 3 : rng,
1203 3 : crds->samplers->ele_cnt );
1204 3 : if( FD_UNLIKELY( idx==SAMPLE_IDX_SENTINEL ) ) return NULL;
1205 : /* Disable peer to prevent future sampling until added back with
1206 : fd_crds_bucket_add */
1207 3 : wpeer_sampler_disable( &crds->samplers->bucket_samplers[bucket],
1208 3 : idx,
1209 3 : crds->samplers->ele_cnt );
1210 :
1211 3 : return fd_crds_entry_contact_info( crds->samplers->ele[idx] );
1212 3 : }
1213 :
1214 : void
1215 : fd_crds_bucket_add( fd_crds_t * crds,
1216 : ulong bucket,
1217 0 : uchar const * pubkey ) {
1218 0 : fd_crds_key_t key[1];
1219 0 : make_contact_info_key( pubkey, key );
1220 0 : fd_crds_entry_t * peer_ci = lookup_map_ele_query( crds->lookup_map, key, NULL, crds->pool );
1221 0 : if( FD_UNLIKELY( !peer_ci ) ) {
1222 0 : FD_LOG_DEBUG(( "Sample peer not found in CRDS. Likely dropped." ));
1223 0 : return;
1224 0 : }
1225 0 : wpeer_sampler_t * bucket_sampler = &crds->samplers->bucket_samplers[bucket];
1226 0 : wpeer_sampler_enable( bucket_sampler,
1227 0 : peer_ci->contact_info.sampler_idx,
1228 0 : crds->samplers->ele_cnt );
1229 :
1230 0 : ulong score = wpeer_sampler_bucket_score( peer_ci, bucket );
1231 0 : wpeer_sampler_upd( bucket_sampler,
1232 0 : score,
1233 0 : peer_ci->contact_info.sampler_idx,
1234 0 : crds->samplers->ele_cnt );
1235 0 : }
1236 :
1237 : struct fd_crds_mask_iter_private {
1238 : ulong idx;
1239 : ulong end_hash;
1240 : };
1241 :
1242 : fd_crds_mask_iter_t *
1243 : fd_crds_mask_iter_init( fd_crds_t const * crds,
1244 : ulong mask,
1245 : uint mask_bits,
1246 0 : uchar iter_mem[ static 16UL ] ) {
1247 0 : fd_crds_mask_iter_t * it = (fd_crds_mask_iter_t *)iter_mem;
1248 0 : ulong start_hash = 0 ;
1249 0 : if( FD_LIKELY( mask_bits > 0) ) start_hash = (mask&(~0UL<<(64UL-mask_bits)));
1250 :
1251 0 : it->end_hash = mask;
1252 :
1253 0 : it->idx = hash_treap_idx_ge( crds->hash_treap, start_hash, crds->pool );
1254 0 : return it;
1255 0 : }
1256 :
1257 : fd_crds_mask_iter_t *
1258 0 : fd_crds_mask_iter_next( fd_crds_mask_iter_t * it, fd_crds_t const * crds ) {
1259 0 : fd_crds_entry_t const * val = hash_treap_ele_fast_const( it->idx, crds->pool );
1260 0 : it->idx = val->hash.next;
1261 0 : return it;
1262 0 : }
1263 :
1264 : int
1265 0 : fd_crds_mask_iter_done( fd_crds_mask_iter_t * it, fd_crds_t const * crds ) {
1266 0 : fd_crds_entry_t const * val = hash_treap_ele_fast_const( it->idx, crds->pool );
1267 0 : return hash_treap_idx_is_null( it->idx ) ||
1268 0 : (it->end_hash < val->hash.hash);
1269 0 : }
1270 :
1271 : fd_crds_entry_t const *
1272 0 : fd_crds_mask_iter_entry( fd_crds_mask_iter_t * it, fd_crds_t const * crds ){
1273 0 : return hash_treap_ele_fast_const( it->idx, crds->pool );
1274 0 : }
1275 :
1276 : fd_crds_mask_iter_t *
1277 : fd_crds_purged_mask_iter_init( fd_crds_t const * crds,
1278 : ulong mask,
1279 : uint mask_bits,
1280 0 : uchar iter_mem[ static 16UL ] ){
1281 0 : fd_crds_mask_iter_t * it = (fd_crds_mask_iter_t *)iter_mem;
1282 0 : ulong start_hash = 0;
1283 0 : if( FD_LIKELY( mask_bits > 0 ) ) start_hash = (mask&(~0UL<<(64UL-mask_bits)));
1284 0 : it->end_hash = mask;
1285 :
1286 0 : it->idx = purged_treap_idx_ge( crds->purged.treap, start_hash, crds->purged.pool );
1287 0 : return it;
1288 0 : }
1289 :
1290 : fd_crds_mask_iter_t *
1291 : fd_crds_purged_mask_iter_next( fd_crds_mask_iter_t * it,
1292 0 : fd_crds_t const * crds ){
1293 0 : fd_crds_purged_t const * val = purged_treap_ele_fast_const( it->idx, crds->purged.pool );
1294 0 : it->idx = val->treap.next;
1295 0 : return it;
1296 0 : }
1297 :
1298 : int
1299 : fd_crds_purged_mask_iter_done( fd_crds_mask_iter_t * it,
1300 0 : fd_crds_t const * crds ){
1301 0 : fd_crds_purged_t const * val = purged_treap_ele_fast_const( it->idx, crds->purged.pool );
1302 0 : return purged_treap_idx_is_null( it->idx ) ||
1303 0 : (it->end_hash < fd_ulong_load_8( val->hash ));
1304 0 : }
1305 :
1306 : /* fd_crds_purged_mask_iter_hash returns the hash of the current
1307 : entry in the purged mask iterator. */
1308 : uchar const *
1309 : fd_crds_purged_mask_iter_hash( fd_crds_mask_iter_t * it,
1310 0 : fd_crds_t const * crds ){
1311 0 : fd_crds_purged_t const * val = purged_treap_ele_fast_const( it->idx, crds->purged.pool );
1312 0 : return val->hash;
1313 0 : }
|
