Line data Source code
1 : #include "fd_sspeer_selector.h"
2 : #include "../../../util/bits/fd_sat.h"
3 : #include "../../../util/log/fd_log.h"
4 :
5 : struct fd_sspeer_private {
6 : fd_sspeer_key_t key;
7 : fd_ip4_port_t addr;
8 : ulong full_slot;
9 : ulong incr_slot;
10 : uchar full_hash[ FD_HASH_FOOTPRINT ];
11 : uchar incr_hash[ FD_HASH_FOOTPRINT ];
12 : ulong latency;
13 : ulong score;
14 : int valid;
15 :
16 : struct {
17 : ulong next;
18 : } pool;
19 :
20 : struct {
21 : ulong next;
22 : ulong prev;
23 : } map_by_key;
24 :
25 : struct {
26 : ulong next;
27 : ulong prev;
28 : } map_by_addr;
29 :
30 : struct {
31 : ulong parent;
32 : ulong left;
33 : ulong right;
34 : ulong prio;
35 : } score_treap;
36 : };
37 :
38 : typedef struct fd_sspeer_private fd_sspeer_private_t;
39 :
40 : #define POOL_NAME peer_pool
41 372 : #define POOL_T fd_sspeer_private_t
42 : #define POOL_IDX_T ulong
43 14356341 : #define POOL_NEXT pool.next
44 : #include "../../../util/tmpl/fd_pool.c"
45 :
46 : #define MAP_NAME peer_map_by_key
47 1611 : #define MAP_KEY key
48 813 : #define MAP_ELE_T fd_sspeer_private_t
49 : #define MAP_KEY_T fd_sspeer_key_t
50 1092 : #define MAP_PREV map_by_key.prev
51 2490 : #define MAP_NEXT map_by_key.next
52 681 : #define MAP_KEY_EQ(k0,k1) (fd_sspeer_key_eq(k0,k1))
53 2766 : #define MAP_KEY_HASH(key,seed) (fd_sspeer_key_hash(key,seed))
54 : #define MAP_OPTIMIZE_RANDOM_ACCESS_REMOVAL 1
55 : #include "../../../util/tmpl/fd_map_chain.c"
56 :
57 : #define MAP_NAME peer_map_by_addr
58 1503 : #define MAP_KEY addr
59 786 : #define MAP_ELE_T fd_sspeer_private_t
60 : #define MAP_KEY_T fd_ip4_port_t
61 1122 : #define MAP_PREV map_by_addr.prev
62 2184 : #define MAP_NEXT map_by_addr.next
63 81 : #define MAP_KEY_EQ(k0,k1) ((k0)->l==(k1)->l)
64 1629 : #define MAP_KEY_HASH(key,seed) (seed^(key)->l)
65 : #define MAP_OPTIMIZE_RANDOM_ACCESS_REMOVAL 1
66 : #define MAP_MULTI 1
67 : #include "../../../util/tmpl/fd_map_chain.c"
68 :
69 1371 : #define COMPARE_WORSE(x,y) ( (x)->score<(y)->score )
70 :
71 : #define TREAP_T fd_sspeer_private_t
72 : #define TREAP_NAME score_treap
73 : #define TREAP_QUERY_T void * /* We don't use query ... */
74 : #define TREAP_CMP(a,b) (__extension__({ (void)(a); (void)(b); -1; })) /* which means we don't need to give a real
75 : implementation to cmp either */
76 5073 : #define TREAP_IDX_T ulong
77 1371 : #define TREAP_LT COMPARE_WORSE
78 3990 : #define TREAP_PARENT score_treap.parent
79 3789 : #define TREAP_LEFT score_treap.left
80 2538 : #define TREAP_RIGHT score_treap.right
81 14356662 : #define TREAP_PRIO score_treap.prio
82 : #include "../../../util/tmpl/fd_treap.c"
83 :
84 906 : #define DEFAULT_SLOTS_BEHIND (1000UL*1000UL) /* 1,000,000 slots behind */
85 : /* Intentionally less than DEFAULT_SLOTS_BEHIND so that peers with an
86 : unknown slot are penalized more than the worst known peer. */
87 873 : #define MAX_SLOTS_BEHIND_CAP (864000UL) /* ~2 epochs worth of slots */
88 : /* Assumed latency (in nanos) for peers that have not been pinged yet.
89 : Pings are sent immediately on peer discovery, so this default is
90 : short-lived. 100ms is a neutral middle-ground: high enough that
91 : any peer with a measured latency is preferred, low enough that slot
92 : distance still meaningfully differentiates unpinged peers. */
93 912 : #define DEFAULT_PEER_LATENCY (100UL*1000UL*1000UL) /* 100ms */
94 906 : #define DEFAULT_SLOTS_BEHIND_PENALTY (200UL)
95 :
96 : #define FD_SSPEER_SELECTOR_DEBUG 0
97 :
98 : struct fd_sspeer_selector_private {
99 : fd_sspeer_private_t * pool;
100 : peer_map_by_key_t * map_by_key;
101 : peer_map_by_addr_t * map_by_addr;
102 : score_treap_t * score_treap;
103 : score_treap_t * shadow_score_treap;
104 : ulong * peer_idx_list;
105 : fd_sscluster_slot_t cluster_slot;
106 : ulong max_peers;
107 : int cluster_slot_dirty;
108 :
109 : ulong magic; /* ==FD_SSPEER_SELECTOR_MAGIC */
110 : };
111 :
112 : FD_FN_CONST ulong
113 768 : fd_sspeer_selector_align( void ) {
114 768 : return fd_ulong_max( alignof( fd_sspeer_selector_t), fd_ulong_max( peer_pool_align(),
115 768 : fd_ulong_max( peer_map_by_key_align(), fd_ulong_max( peer_map_by_addr_align(),
116 768 : fd_ulong_max( score_treap_align(), alignof(ulong) ) ) ) ) );
117 768 : }
118 :
119 : FD_FN_CONST ulong
120 54 : fd_sspeer_selector_footprint( ulong max_peers ) {
121 54 : ulong l;
122 54 : l = FD_LAYOUT_INIT;
123 54 : l = FD_LAYOUT_APPEND( l, alignof(fd_sspeer_selector_t), sizeof(fd_sspeer_selector_t) );
124 54 : l = FD_LAYOUT_APPEND( l, peer_pool_align(), peer_pool_footprint( 2UL*max_peers ) );
125 54 : l = FD_LAYOUT_APPEND( l, peer_map_by_key_align(), peer_map_by_key_footprint( peer_map_by_key_chain_cnt_est( 2UL*max_peers ) ) );
126 54 : l = FD_LAYOUT_APPEND( l, peer_map_by_addr_align(), peer_map_by_addr_footprint( peer_map_by_addr_chain_cnt_est( 2UL*max_peers ) ) );
127 54 : l = FD_LAYOUT_APPEND( l, score_treap_align(), score_treap_footprint( max_peers ) );
128 54 : l = FD_LAYOUT_APPEND( l, score_treap_align(), score_treap_footprint( max_peers ) );
129 54 : l = FD_LAYOUT_APPEND( l, alignof(ulong), max_peers * sizeof(ulong) ); /* pool indices for shadow treap swap */
130 54 : return FD_LAYOUT_FINI( l, fd_sspeer_selector_align() );
131 54 : }
132 :
133 : void *
134 : fd_sspeer_selector_new( void * shmem,
135 : ulong max_peers,
136 129 : ulong seed ) {
137 129 : if( FD_UNLIKELY( !shmem ) ) {
138 0 : FD_LOG_WARNING(( "NULL shmem" ));
139 0 : return NULL;
140 0 : }
141 :
142 129 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shmem, fd_sspeer_selector_align() ) ) ) {
143 0 : FD_LOG_WARNING(( "unaligned shmem" ));
144 0 : return NULL;
145 0 : }
146 :
147 129 : if( FD_UNLIKELY( max_peers < 1UL ) ) {
148 0 : FD_LOG_WARNING(( "max_peers must be at least 1" ));
149 0 : return NULL;
150 0 : }
151 :
152 129 : FD_SCRATCH_ALLOC_INIT( l, shmem );
153 129 : fd_sspeer_selector_t * selector = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_sspeer_selector_t), sizeof(fd_sspeer_selector_t) );
154 129 : void * _pool = FD_SCRATCH_ALLOC_APPEND( l, peer_pool_align(), peer_pool_footprint( 2UL*max_peers ) );
155 129 : void * _map = FD_SCRATCH_ALLOC_APPEND( l, peer_map_by_key_align(), peer_map_by_key_footprint( peer_map_by_key_chain_cnt_est( 2UL*max_peers ) ) );
156 129 : void * _multimap_by_addr = FD_SCRATCH_ALLOC_APPEND( l, peer_map_by_addr_align(), peer_map_by_addr_footprint( peer_map_by_addr_chain_cnt_est( 2UL*max_peers ) ) );
157 129 : void * _score_treap = FD_SCRATCH_ALLOC_APPEND( l, score_treap_align(), score_treap_footprint( max_peers ) );
158 129 : void * _shadow_score_treap = FD_SCRATCH_ALLOC_APPEND( l, score_treap_align(), score_treap_footprint( max_peers ) );
159 129 : void * _peer_idx_list = FD_SCRATCH_ALLOC_APPEND( l, alignof(ulong), max_peers * sizeof(ulong) ); /* pool indices for shadow treap swap */
160 :
161 129 : selector->pool = peer_pool_join( peer_pool_new( _pool, 2UL*max_peers ) );
162 : /* Seed treap priorities so the treap is balanced. */
163 129 : score_treap_seed( selector->pool, 2UL*max_peers, seed );
164 129 : selector->map_by_key = peer_map_by_key_join( peer_map_by_key_new( _map, peer_map_by_key_chain_cnt_est( 2UL*max_peers ), seed ) );
165 129 : selector->map_by_addr = peer_map_by_addr_join( peer_map_by_addr_new( _multimap_by_addr, peer_map_by_addr_chain_cnt_est( 2UL*max_peers ), seed ) );
166 129 : selector->score_treap = score_treap_join( score_treap_new( _score_treap, max_peers ) );
167 129 : selector->shadow_score_treap = score_treap_join( score_treap_new( _shadow_score_treap, max_peers ) );
168 129 : selector->peer_idx_list = (ulong *)_peer_idx_list;
169 129 : selector->max_peers = max_peers;
170 :
171 129 : selector->cluster_slot.full = 0UL;
172 129 : selector->cluster_slot.incremental = FD_SSPEER_SLOT_UNKNOWN;
173 129 : selector->cluster_slot_dirty = 0;
174 :
175 129 : FD_COMPILER_MFENCE();
176 129 : FD_VOLATILE( selector->magic ) = FD_SSPEER_SELECTOR_MAGIC;
177 129 : FD_COMPILER_MFENCE();
178 :
179 129 : return (void *)selector;
180 129 : }
181 :
182 : fd_sspeer_selector_t *
183 129 : fd_sspeer_selector_join( void * shselector ) {
184 129 : if( FD_UNLIKELY( !shselector ) ) {
185 0 : FD_LOG_WARNING(( "NULL shselector" ));
186 0 : return NULL;
187 0 : }
188 :
189 129 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shselector, fd_sspeer_selector_align() ) ) ) {
190 0 : FD_LOG_WARNING(( "misaligned shselector" ));
191 0 : return NULL;
192 0 : }
193 :
194 129 : fd_sspeer_selector_t * selector = (fd_sspeer_selector_t *)shselector;
195 :
196 129 : if( FD_UNLIKELY( selector->magic!=FD_SSPEER_SELECTOR_MAGIC ) ) {
197 0 : FD_LOG_WARNING(( "bad magic" ));
198 0 : return NULL;
199 0 : }
200 :
201 129 : return selector;
202 129 : }
203 :
204 : void *
205 114 : fd_sspeer_selector_leave( fd_sspeer_selector_t * selector ) {
206 114 : if( FD_UNLIKELY( !selector ) ) {
207 0 : FD_LOG_WARNING(( "NULL selector" ));
208 0 : return NULL;
209 0 : }
210 :
211 114 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)selector, fd_sspeer_selector_align() ) ) ) {
212 0 : FD_LOG_WARNING(( "misaligned selector" ));
213 0 : return NULL;
214 0 : }
215 :
216 114 : if( FD_UNLIKELY( selector->magic!=FD_SSPEER_SELECTOR_MAGIC ) ) {
217 0 : FD_LOG_WARNING(( "bad magic" ));
218 0 : return NULL;
219 0 : }
220 :
221 114 : selector->pool = peer_pool_leave( selector->pool );
222 114 : selector->map_by_key = peer_map_by_key_leave( selector->map_by_key );
223 114 : selector->map_by_addr = peer_map_by_addr_leave( selector->map_by_addr );
224 114 : selector->score_treap = score_treap_leave( selector->score_treap );
225 114 : selector->shadow_score_treap = score_treap_leave( selector->shadow_score_treap );
226 :
227 114 : return (void *)selector;
228 114 : }
229 :
230 : void *
231 114 : fd_sspeer_selector_delete( void * shselector ) {
232 114 : if( FD_UNLIKELY( !shselector ) ) {
233 0 : FD_LOG_WARNING(( "NULL shselector" ));
234 0 : return NULL;
235 0 : }
236 :
237 114 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shselector, fd_sspeer_selector_align() ) ) ) {
238 0 : FD_LOG_WARNING(( "misaligned shselector" ));
239 0 : return NULL;
240 0 : }
241 :
242 114 : fd_sspeer_selector_t * selector = (fd_sspeer_selector_t *)shselector;
243 :
244 114 : if( FD_UNLIKELY( selector->magic!=FD_SSPEER_SELECTOR_MAGIC ) ) {
245 0 : FD_LOG_WARNING(( "bad magic" ));
246 0 : return NULL;
247 0 : }
248 :
249 114 : selector->pool = peer_pool_delete( selector->pool );
250 114 : selector->map_by_key = peer_map_by_key_delete( selector->map_by_key );
251 114 : selector->map_by_addr = peer_map_by_addr_delete( selector->map_by_addr );
252 114 : selector->score_treap = score_treap_delete( selector->score_treap );
253 114 : selector->shadow_score_treap = score_treap_delete( selector->shadow_score_treap );
254 :
255 114 : FD_COMPILER_MFENCE();
256 114 : FD_VOLATILE( selector->magic ) = 0UL;
257 114 : FD_COMPILER_MFENCE();
258 :
259 114 : return (void *)selector;
260 114 : }
261 :
262 : /* Calculates a score for a peer given its latency and its resolved
263 : full and incremental slots */
264 : static ulong
265 : fd_sspeer_selector_score( fd_sspeer_selector_t const * selector,
266 : ulong peer_latency,
267 : ulong full_slot,
268 906 : ulong incr_slot ) {
269 906 : peer_latency = peer_latency!=FD_SSPEER_LATENCY_UNKNOWN ? peer_latency : DEFAULT_PEER_LATENCY;
270 :
271 906 : ulong slots_behind = DEFAULT_SLOTS_BEHIND;
272 :
273 906 : if( FD_LIKELY( full_slot!=FD_SSPEER_SLOT_UNKNOWN ) ) {
274 873 : if( FD_LIKELY( incr_slot!=FD_SSPEER_SLOT_UNKNOWN &&
275 873 : selector->cluster_slot.incremental!=FD_SSPEER_SLOT_UNKNOWN ) ) {
276 321 : slots_behind = selector->cluster_slot.incremental>incr_slot ? selector->cluster_slot.incremental - incr_slot : 0UL;
277 552 : } else {
278 : /* Either the peer has no incremental or the cluster has no
279 : incremental reference yet. Fall back to comparing full_slot
280 : against the cluster full slot. */
281 552 : slots_behind = selector->cluster_slot.full>full_slot ? selector->cluster_slot.full - full_slot : 0UL;
282 552 : }
283 873 : slots_behind = fd_ulong_min( slots_behind, MAX_SLOTS_BEHIND_CAP );
284 873 : }
285 :
286 : /* Using saturating arithmetic to avoid overflow and cap at
287 : FD_SSPEER_SCORE_MAX. */
288 906 : ulong penalty = fd_ulong_sat_mul( DEFAULT_SLOTS_BEHIND_PENALTY,
289 906 : fd_ulong_sat_mul( slots_behind, slots_behind ) );
290 906 : ulong score = fd_ulong_sat_add( peer_latency, penalty );
291 906 : return fd_ulong_min( score, FD_SSPEER_SCORE_MAX );
292 906 : }
293 :
294 : /* Validates slot arguments for both new and existing peers. Returns
295 : 0 on success, -1 on failure due to incr_slot<full_slot, and -2 on
296 : failure due to full_slot==UNKNOWN with incr_slot!=UNKNOWN. The
297 : caller passes in the effective (already-resolved) full_slot and
298 : incr_slot values. No log on failure (the caller is responsible
299 : for logging whenever needed).
300 :
301 : Two invariants are enforced:
302 : 1. When both slots are known, incr_slot must be >= full_slot.
303 : 2. An incremental slot requires a known full slot. */
304 : static int
305 : fd_sspeer_validate_slot_args( ulong full_slot,
306 660 : ulong incr_slot ) {
307 660 : if( FD_UNLIKELY( incr_slot!=FD_SSPEER_SLOT_UNKNOWN &&
308 660 : full_slot!=FD_SSPEER_SLOT_UNKNOWN &&
309 660 : incr_slot<full_slot ) ) {
310 6 : return -1;
311 6 : }
312 :
313 654 : if( FD_UNLIKELY( full_slot==FD_SSPEER_SLOT_UNKNOWN &&
314 654 : incr_slot!=FD_SSPEER_SLOT_UNKNOWN ) ) {
315 9 : return -2;
316 9 : }
317 :
318 645 : return 0;
319 654 : }
320 :
321 : /* Updates a peer's score with new values for latency and/or resolved
322 : full/incremental slots. Returns FD_SSPEER_UPDATE_SUCCESS on
323 : success, or the specific fd_sspeer_validate_slot_args error code
324 : on failure without modifying the peer or any data structure.
325 :
326 : Slot-based incremental clearing: when the caller provides
327 : incr_slot==UNKNOWN and full_slot!=UNKNOWN, the peer's existing
328 : incremental data is cleared if it is stale (peer->incr_slot <
329 : full_slot). Otherwise, the existing incremental data is preserved. */
330 : static int
331 : fd_sspeer_selector_update( fd_sspeer_selector_t * selector,
332 : fd_sspeer_private_t * peer,
333 : ulong latency,
334 : ulong full_slot,
335 : ulong incr_slot,
336 : uchar const full_hash[ FD_HASH_FOOTPRINT ],
337 96 : uchar const incr_hash[ FD_HASH_FOOTPRINT ] ) {
338 96 : ulong peer_latency = latency!=FD_SSPEER_LATENCY_UNKNOWN ? latency : peer->latency;
339 96 : ulong peer_full_slot = full_slot!=FD_SSPEER_SLOT_UNKNOWN ? full_slot : peer->full_slot;
340 :
341 96 : ulong peer_incr_slot;
342 96 : int clear_incr = 0;
343 96 : if( incr_slot!=FD_SSPEER_SLOT_UNKNOWN ) {
344 36 : peer_incr_slot = incr_slot;
345 60 : } else if( full_slot!=FD_SSPEER_SLOT_UNKNOWN &&
346 60 : peer->incr_slot!=FD_SSPEER_SLOT_UNKNOWN &&
347 60 : peer->incr_slot<full_slot ) {
348 : /* The caller is providing a new full_slot that has advanced past
349 : the peer's existing incremental, the incremental is stale. */
350 12 : peer_incr_slot = FD_SSPEER_SLOT_UNKNOWN;
351 12 : clear_incr = 1;
352 48 : } else {
353 48 : peer_incr_slot = peer->incr_slot;
354 48 : }
355 :
356 96 : int validate_err = fd_sspeer_validate_slot_args( peer_full_slot, peer_incr_slot );
357 96 : if( FD_UNLIKELY( validate_err ) ) return validate_err;
358 :
359 90 : score_treap_ele_remove( selector->score_treap, peer, selector->pool );
360 :
361 90 : peer->score = fd_sspeer_selector_score( selector, peer_latency, peer_full_slot, peer_incr_slot );
362 :
363 90 : peer->latency = peer_latency;
364 90 : peer->full_slot = peer_full_slot;
365 90 : peer->incr_slot = peer_incr_slot;
366 90 : if( FD_LIKELY( full_hash ) ) {
367 36 : fd_memcpy( peer->full_hash, full_hash, FD_HASH_FOOTPRINT );
368 36 : }
369 90 : if( FD_UNLIKELY( clear_incr ) ) {
370 12 : fd_memset( peer->incr_hash, 0, FD_HASH_FOOTPRINT );
371 78 : } else if( FD_LIKELY( incr_hash ) ) {
372 15 : fd_memcpy( peer->incr_hash, incr_hash, FD_HASH_FOOTPRINT );
373 15 : }
374 :
375 90 : score_treap_ele_insert( selector->score_treap, peer, selector->pool );
376 90 : return FD_SSPEER_UPDATE_SUCCESS;
377 96 : }
378 :
379 : ulong
380 : fd_sspeer_selector_update_on_ping( fd_sspeer_selector_t * selector,
381 : fd_ip4_port_t addr,
382 21 : ulong latency ) {
383 21 : ulong ele_idx = peer_map_by_addr_idx_query_const( selector->map_by_addr, &addr, ULONG_MAX, selector->pool );
384 21 : ulong cnt = 0UL;
385 45 : for(;;) {
386 45 : if( FD_UNLIKELY( ele_idx==ULONG_MAX ) ) break;
387 24 : fd_sspeer_private_t * peer = selector->pool + ele_idx;
388 : /* Update cannot fail here: slots are FD_SSPEER_SLOT_UNKNOWN and
389 : hashes are NULL, so fd_sspeer_validate_slot_args always
390 : returns FD_SSPEER_UPDATE_SUCCESS (no clear_incr trigger,
391 : no incr<full violation). */
392 24 : int update_status = fd_sspeer_selector_update( selector, peer, latency,
393 24 : FD_SSPEER_SLOT_UNKNOWN, FD_SSPEER_SLOT_UNKNOWN,
394 24 : NULL, NULL );
395 24 : if( FD_UNLIKELY( update_status!=FD_SSPEER_UPDATE_SUCCESS ) ) {
396 : /* A warning is a tradeoff between crashing with FD_LOG_CRIT and
397 : potentially missing the log altogether with FD_LOG_DEBUG. */
398 0 : if( peer->key.is_url ) {
399 0 : FD_LOG_WARNING(( "unexpected selector update returned %d for peer %s " FD_IP4_ADDR_FMT ":%hu",
400 0 : update_status, peer->key.url.hostname,
401 0 : FD_IP4_ADDR_FMT_ARGS( peer->key.url.resolved_addr.addr ), fd_ushort_bswap( peer->key.url.resolved_addr.port ) ));
402 0 : } else {
403 0 : FD_BASE58_ENCODE_32_BYTES( peer->key.pubkey->uc, peer_pubkey_b58 );
404 0 : FD_LOG_WARNING(( "unexpected selector update returned %d for peer %s " FD_IP4_ADDR_FMT ":%hu",
405 0 : update_status, peer_pubkey_b58,
406 0 : FD_IP4_ADDR_FMT_ARGS( peer->addr.addr ), fd_ushort_bswap( peer->addr.port ) ));
407 0 : }
408 0 : }
409 24 : ele_idx = peer_map_by_addr_idx_next_const( ele_idx, ULONG_MAX, selector->pool );
410 24 : cnt++;
411 24 : }
412 21 : return cnt;
413 21 : }
414 :
415 : ulong
416 : fd_sspeer_selector_add( fd_sspeer_selector_t * selector,
417 : fd_sspeer_key_t const * key,
418 : fd_ip4_port_t addr,
419 : ulong latency,
420 : ulong full_slot,
421 : ulong incr_slot,
422 : uchar const full_hash[ FD_HASH_FOOTPRINT ],
423 675 : uchar const incr_hash[ FD_HASH_FOOTPRINT ] ) {
424 675 : if( FD_UNLIKELY( key==NULL ) ) return FD_SSPEER_SCORE_INVALID;
425 : /* A peer without a valid address cannot be added to the selector.
426 : For an existing peer changing from a valid address to 0, it is
427 : the caller's responsibility to remove them. */
428 666 : if( FD_UNLIKELY( !addr.l ) ) return FD_SSPEER_SCORE_INVALID;
429 :
430 : /* Reject implausible slot values. FD_SSPEER_SLOT_UNKNOWN is allowed. */
431 654 : if( FD_UNLIKELY( full_slot!=FD_SSPEER_SLOT_UNKNOWN && full_slot>=FD_SSPEER_PLAUSIBLE_MAX_SLOT ) ) return FD_SSPEER_SCORE_INVALID;
432 648 : if( FD_UNLIKELY( incr_slot!=FD_SSPEER_SLOT_UNKNOWN && incr_slot>=FD_SSPEER_PLAUSIBLE_MAX_SLOT ) ) return FD_SSPEER_SCORE_INVALID;
433 :
434 645 : fd_sspeer_private_t * peer = peer_map_by_key_ele_query( selector->map_by_key, key, NULL, selector->pool );
435 645 : if( FD_LIKELY( peer ) ) {
436 72 : ulong old_full = peer->full_slot;
437 72 : ulong old_incr = peer->incr_slot;
438 72 : int update_status = fd_sspeer_selector_update( selector, peer, latency, full_slot, incr_slot, full_hash, incr_hash );
439 72 : if( FD_UNLIKELY( update_status!=FD_SSPEER_UPDATE_SUCCESS ) ) return FD_SSPEER_SCORE_INVALID;
440 66 : if( FD_UNLIKELY( peer->full_slot!=old_full || peer->incr_slot!=old_incr ) ) selector->cluster_slot_dirty = 1;
441 : /* Update the addr map after the selector update so that the peer
442 : is not mutated when the update fails. */
443 66 : if( FD_UNLIKELY( peer->addr.l!=addr.l ) ) {
444 6 : peer_map_by_addr_ele_remove_fast( selector->map_by_addr, peer, selector->pool );
445 6 : peer->addr = addr;
446 6 : peer_map_by_addr_ele_insert( selector->map_by_addr, peer, selector->pool );
447 6 : }
448 573 : } else {
449 573 : if( FD_UNLIKELY( !peer_pool_free( selector->pool ) ) ) {
450 0 : FD_LOG_WARNING(( "peer selector pool exhausted" ));
451 0 : return FD_SSPEER_SCORE_INVALID;
452 0 : }
453 573 : if( FD_UNLIKELY( score_treap_ele_cnt(selector->score_treap)>=selector->max_peers ) ) {
454 9 : FD_LOG_WARNING(( "peer selector at max capacity" ));
455 9 : return FD_SSPEER_SCORE_INVALID;
456 9 : }
457 :
458 564 : if( FD_UNLIKELY( fd_sspeer_validate_slot_args( full_slot, incr_slot ) ) ) {
459 9 : return FD_SSPEER_SCORE_INVALID;
460 9 : }
461 :
462 555 : peer = peer_pool_ele_acquire( selector->pool );
463 555 : peer->key = *key;
464 555 : peer->addr = addr;
465 555 : peer->latency = latency;
466 555 : peer->score = fd_sspeer_selector_score( selector, latency, full_slot, incr_slot );
467 555 : peer->full_slot = full_slot;
468 555 : peer->incr_slot = incr_slot;
469 555 : if( FD_LIKELY( full_hash ) ) fd_memcpy( peer->full_hash, full_hash, FD_HASH_FOOTPRINT );
470 543 : else fd_memset( peer->full_hash, 0, FD_HASH_FOOTPRINT );
471 : /* full_hash and incr_hash are treated independently here. */
472 555 : if( FD_LIKELY( incr_hash ) ) fd_memcpy( peer->incr_hash, incr_hash, FD_HASH_FOOTPRINT );
473 543 : else fd_memset( peer->incr_hash, 0, FD_HASH_FOOTPRINT );
474 555 : peer_map_by_key_ele_insert( selector->map_by_key, peer, selector->pool );
475 555 : peer_map_by_addr_ele_insert( selector->map_by_addr, peer, selector->pool );
476 555 : score_treap_ele_insert( selector->score_treap, peer, selector->pool );
477 555 : if( FD_LIKELY( full_slot!=FD_SSPEER_SLOT_UNKNOWN || incr_slot!=FD_SSPEER_SLOT_UNKNOWN ) ) selector->cluster_slot_dirty = 1;
478 555 : }
479 621 : peer->valid = peer->full_slot!=FD_SSPEER_SLOT_UNKNOWN;
480 621 : return peer->score;
481 645 : }
482 :
483 : void
484 : fd_sspeer_selector_remove( fd_sspeer_selector_t * selector,
485 516 : fd_sspeer_key_t const * key ) {
486 516 : if( FD_UNLIKELY( key==NULL ) ) return;
487 510 : fd_sspeer_private_t * peer = peer_map_by_key_ele_query( selector->map_by_key, key, NULL, selector->pool );
488 510 : if( FD_UNLIKELY( peer==NULL ) ) return;
489 495 : score_treap_ele_remove( selector->score_treap, peer, selector->pool );
490 495 : peer_map_by_key_ele_remove_fast( selector->map_by_key, peer, selector->pool );
491 495 : peer_map_by_addr_ele_remove_fast( selector->map_by_addr, peer, selector->pool );
492 495 : peer_pool_ele_release( selector->pool, peer );
493 495 : selector->cluster_slot_dirty = 1;
494 495 : }
495 :
496 : void
497 : fd_sspeer_selector_remove_by_addr( fd_sspeer_selector_t * selector,
498 48 : fd_ip4_port_t addr ) {
499 105 : for(;;) {
500 105 : fd_sspeer_private_t * peer = peer_map_by_addr_ele_remove( selector->map_by_addr, &addr, NULL, selector->pool );
501 105 : if( FD_UNLIKELY( peer==NULL ) ) break;
502 57 : score_treap_ele_remove( selector->score_treap, peer, selector->pool );
503 57 : peer_map_by_key_ele_remove_fast( selector->map_by_key, peer, selector->pool );
504 57 : peer_pool_ele_release( selector->pool, peer );
505 57 : selector->cluster_slot_dirty = 1;
506 57 : }
507 48 : }
508 :
509 : fd_sspeer_t
510 : fd_sspeer_selector_best( fd_sspeer_selector_t * selector,
511 : int incremental,
512 303 : ulong base_slot ) {
513 303 : if( FD_UNLIKELY( incremental && base_slot==FD_SSPEER_SLOT_UNKNOWN ) ) {
514 3 : FD_LOG_WARNING(( "incremental selection requires a valid base_slot" ));
515 3 : return (fd_sspeer_t){
516 3 : .key = { .is_url = 0 },
517 3 : .addr = { .l=0UL },
518 3 : .full_slot = FD_SSPEER_SLOT_UNKNOWN,
519 3 : .incr_slot = FD_SSPEER_SLOT_UNKNOWN,
520 3 : .score = FD_SSPEER_SCORE_INVALID,
521 3 : .full_hash = {0},
522 3 : .incr_hash = {0},
523 3 : };
524 3 : }
525 :
526 300 : for( score_treap_fwd_iter_t iter = score_treap_fwd_iter_init( selector->score_treap, selector->pool );
527 351 : !score_treap_fwd_iter_done( iter );
528 306 : iter = score_treap_fwd_iter_next( iter, selector->pool ) ) {
529 306 : fd_sspeer_private_t const * peer = score_treap_fwd_iter_ele_const( iter, selector->pool );
530 : /* For full selection (!incremental), any valid peer is eligible.
531 : For incremental selection, the peer must serve the same base full
532 : snapshot and must actually offer an incremental snapshot. */
533 306 : if( FD_LIKELY( peer->valid &&
534 306 : (!incremental ||
535 306 : (peer->full_slot==base_slot && peer->incr_slot!=FD_SSPEER_SLOT_UNKNOWN) ) ) ) {
536 255 : fd_sspeer_t best = {
537 255 : .key = peer->key,
538 255 : .addr = peer->addr,
539 255 : .full_slot = peer->full_slot,
540 255 : .incr_slot = peer->incr_slot,
541 255 : .score = peer->score,
542 255 : };
543 255 : fd_memcpy( best.full_hash, peer->full_hash, FD_HASH_FOOTPRINT );
544 255 : fd_memcpy( best.incr_hash, peer->incr_hash, FD_HASH_FOOTPRINT );
545 255 : return best;
546 255 : }
547 306 : }
548 :
549 45 : return (fd_sspeer_t){
550 45 : .key = { .is_url = 0 },
551 45 : .addr = { .l=0UL },
552 45 : .full_slot = FD_SSPEER_SLOT_UNKNOWN,
553 45 : .incr_slot = FD_SSPEER_SLOT_UNKNOWN,
554 45 : .score = FD_SSPEER_SCORE_INVALID,
555 45 : .full_hash = {0},
556 45 : .incr_hash = {0},
557 45 : };
558 300 : }
559 :
560 : void
561 177 : fd_sspeer_selector_process_cluster_slot( fd_sspeer_selector_t * selector ) {
562 177 : if( FD_LIKELY( !selector->cluster_slot_dirty ) ) return;
563 165 : selector->cluster_slot_dirty = 0;
564 :
565 : /* Compute max full_slot and incr_slot across all tracked peers. */
566 165 : ulong max_full = 0UL;
567 165 : ulong max_incr = 0UL;
568 165 : ulong full_cnt = 0UL;
569 165 : ulong incr_cnt = 0UL;
570 165 : for( score_treap_fwd_iter_t iter = score_treap_fwd_iter_init( selector->score_treap, selector->pool );
571 531 : !score_treap_fwd_iter_done( iter );
572 366 : iter = score_treap_fwd_iter_next( iter, selector->pool ) ) {
573 366 : fd_sspeer_private_t const * peer = score_treap_fwd_iter_ele_const( iter, selector->pool );
574 366 : if( FD_LIKELY( peer->full_slot!=FD_SSPEER_SLOT_UNKNOWN ) ) {
575 360 : max_full = fd_ulong_max( max_full, peer->full_slot );
576 360 : full_cnt++;
577 360 : }
578 366 : if( FD_LIKELY( peer->incr_slot!=FD_SSPEER_SLOT_UNKNOWN ) ) {
579 264 : max_incr = fd_ulong_max( max_incr, peer->incr_slot );
580 264 : incr_cnt++;
581 264 : }
582 366 : }
583 :
584 : /* If no known incr_slot among tracked peers, treat as unknown. */
585 165 : if( FD_UNLIKELY( incr_cnt==0UL ) ) {
586 42 : max_incr = FD_SSPEER_SLOT_UNKNOWN;
587 42 : }
588 :
589 : /* If no known full_slot among tracked peers, reset so that a
590 : subsequent add() is not scored against a stale cluster slot. */
591 165 : if( FD_UNLIKELY( full_cnt==0UL ) ) {
592 27 : selector->cluster_slot.full = 0UL;
593 27 : selector->cluster_slot.incremental = FD_SSPEER_SLOT_UNKNOWN;
594 27 : return;
595 27 : }
596 :
597 : /* The full and incremental maxima are computed from different peer
598 : subsets, so max_incr < max_full is possible. Clamp to maintain
599 : the invariant that incremental >= full. */
600 138 : if( FD_UNLIKELY( max_incr!=FD_SSPEER_SLOT_UNKNOWN &&
601 138 : max_incr<max_full ) ) {
602 3 : max_incr = max_full;
603 3 : }
604 :
605 : /* If neither max changed, no rescore needed. */
606 138 : if( FD_LIKELY( max_full==selector->cluster_slot.full &&
607 138 : max_incr==selector->cluster_slot.incremental ) ) return;
608 :
609 114 : selector->cluster_slot.full = max_full;
610 114 : selector->cluster_slot.incremental = max_incr;
611 :
612 : /* Rescore all peers using the shadow treap swap mechanism. */
613 114 : ulong idx = 0UL;
614 114 : for( score_treap_fwd_iter_t iter = score_treap_fwd_iter_init( selector->score_treap, selector->pool );
615 375 : !score_treap_fwd_iter_done( iter );
616 261 : iter = score_treap_fwd_iter_next( iter, selector->pool ) ) {
617 261 : fd_sspeer_private_t * peer = score_treap_fwd_iter_ele( iter, selector->pool );
618 261 : fd_sspeer_private_t * shadow_peer = peer_pool_ele_acquire( selector->pool );
619 261 : shadow_peer->latency = peer->latency;
620 261 : shadow_peer->full_slot = peer->full_slot;
621 261 : shadow_peer->incr_slot = peer->incr_slot;
622 261 : shadow_peer->addr = peer->addr;
623 261 : shadow_peer->key = peer->key;
624 261 : shadow_peer->score = fd_sspeer_selector_score( selector, shadow_peer->latency, shadow_peer->full_slot, shadow_peer->incr_slot );
625 261 : shadow_peer->valid = peer->valid;
626 261 : fd_memcpy( shadow_peer->full_hash, peer->full_hash, FD_HASH_FOOTPRINT );
627 261 : fd_memcpy( shadow_peer->incr_hash, peer->incr_hash, FD_HASH_FOOTPRINT );
628 261 : score_treap_ele_insert( selector->shadow_score_treap, shadow_peer, selector->pool );
629 261 : selector->peer_idx_list[ idx++ ] = peer_pool_idx( selector->pool, peer );
630 261 : peer_map_by_key_ele_remove_fast( selector->map_by_key, peer, selector->pool );
631 261 : peer_map_by_addr_ele_remove_fast( selector->map_by_addr, peer, selector->pool );
632 261 : peer_map_by_key_ele_insert( selector->map_by_key, shadow_peer, selector->pool );
633 261 : peer_map_by_addr_ele_insert( selector->map_by_addr, shadow_peer, selector->pool );
634 261 : }
635 :
636 : /* clear score treap */
637 375 : for( ulong i=0UL; i<idx; i++ ) {
638 261 : fd_sspeer_private_t * peer = peer_pool_ele( selector->pool, selector->peer_idx_list[ i ] );
639 261 : score_treap_ele_remove( selector->score_treap, peer, selector->pool );
640 261 : peer_pool_ele_release( selector->pool, peer );
641 261 : }
642 :
643 114 : score_treap_t * tmp = selector->score_treap;
644 114 : selector->score_treap = selector->shadow_score_treap;
645 114 : selector->shadow_score_treap = tmp;
646 :
647 : #if FD_SSPEER_SELECTOR_DEBUG
648 : FD_TEST( score_treap_verify( selector->score_treap, selector->pool )==0 );
649 : #endif
650 114 : }
651 :
652 : fd_sscluster_slot_t
653 114 : fd_sspeer_selector_cluster_slot( fd_sspeer_selector_t * selector ) {
654 114 : return selector->cluster_slot;
655 114 : }
656 :
657 : ulong
658 231 : fd_sspeer_selector_peer_map_by_key_ele_cnt( fd_sspeer_selector_t * selector ) {
659 231 : ulong cnt = 0UL;
660 231 : for( peer_map_by_key_iter_t iter = peer_map_by_key_iter_init( selector->map_by_key, selector->pool );
661 828 : !peer_map_by_key_iter_done( iter, selector->map_by_key, selector->pool );
662 597 : iter = peer_map_by_key_iter_next( iter, selector->map_by_key, selector->pool ) ) {
663 597 : cnt++;
664 597 : }
665 231 : return cnt;
666 231 : }
667 :
668 : ulong
669 168 : fd_sspeer_selector_peer_map_by_addr_ele_cnt( fd_sspeer_selector_t * selector ) {
670 168 : ulong cnt = 0UL;
671 168 : for( peer_map_by_addr_iter_t iter = peer_map_by_addr_iter_init( selector->map_by_addr, selector->pool );
672 558 : !peer_map_by_addr_iter_done( iter, selector->map_by_addr, selector->pool );
673 390 : iter = peer_map_by_addr_iter_next( iter, selector->map_by_addr, selector->pool ) ) {
674 390 : cnt++;
675 390 : }
676 168 : return cnt;
677 168 : }
|
