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 0 : #define POOL_T fd_sspeer_private_t
42 : #define POOL_IDX_T ulong
43 0 : #define POOL_NEXT pool.next
44 : #include "../../../util/tmpl/fd_pool.c"
45 :
46 : #define MAP_NAME peer_map_by_key
47 0 : #define MAP_KEY key
48 0 : #define MAP_ELE_T fd_sspeer_private_t
49 : #define MAP_KEY_T fd_sspeer_key_t
50 0 : #define MAP_PREV map_by_key.prev
51 0 : #define MAP_NEXT map_by_key.next
52 0 : #define MAP_KEY_EQ(k0,k1) (fd_sspeer_key_eq(k0,k1))
53 0 : #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 0 : #define MAP_KEY addr
59 0 : #define MAP_ELE_T fd_sspeer_private_t
60 : #define MAP_KEY_T fd_ip4_port_t
61 0 : #define MAP_PREV map_by_addr.prev
62 0 : #define MAP_NEXT map_by_addr.next
63 0 : #define MAP_KEY_EQ(k0,k1) ((k0)->l==(k1)->l)
64 0 : #define MAP_KEY_HASH(key,seed) (fd_ulong_hash( (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 0 : #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 0 : #define TREAP_IDX_T ulong
77 0 : #define TREAP_LT COMPARE_WORSE
78 0 : #define TREAP_PARENT score_treap.parent
79 0 : #define TREAP_LEFT score_treap.left
80 0 : #define TREAP_RIGHT score_treap.right
81 0 : #define TREAP_PRIO score_treap.prio
82 : #include "../../../util/tmpl/fd_treap.c"
83 :
84 0 : #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 0 : #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 0 : #define DEFAULT_PEER_LATENCY (100UL*1000UL*1000UL) /* 100ms */
94 0 : #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 0 : fd_sspeer_selector_align( void ) {
114 0 : return fd_ulong_max( alignof( fd_sspeer_selector_t), fd_ulong_max( peer_pool_align(),
115 0 : fd_ulong_max( peer_map_by_key_align(), fd_ulong_max( peer_map_by_addr_align(),
116 0 : fd_ulong_max( score_treap_align(), alignof(ulong) ) ) ) ) );
117 0 : }
118 :
119 : FD_FN_CONST ulong
120 0 : fd_sspeer_selector_footprint( ulong max_peers ) {
121 0 : ulong l;
122 0 : l = FD_LAYOUT_INIT;
123 0 : l = FD_LAYOUT_APPEND( l, alignof(fd_sspeer_selector_t), sizeof(fd_sspeer_selector_t) );
124 0 : l = FD_LAYOUT_APPEND( l, peer_pool_align(), peer_pool_footprint( 2UL*max_peers ) );
125 0 : 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 0 : 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 0 : l = FD_LAYOUT_APPEND( l, score_treap_align(), score_treap_footprint( max_peers ) );
128 0 : l = FD_LAYOUT_APPEND( l, score_treap_align(), score_treap_footprint( max_peers ) );
129 0 : l = FD_LAYOUT_APPEND( l, alignof(ulong), max_peers * sizeof(ulong) ); /* pool indices for shadow treap swap */
130 0 : return FD_LAYOUT_FINI( l, fd_sspeer_selector_align() );
131 0 : }
132 :
133 : void *
134 : fd_sspeer_selector_new( void * shmem,
135 : ulong max_peers,
136 0 : ulong seed ) {
137 0 : if( FD_UNLIKELY( !shmem ) ) {
138 0 : FD_LOG_WARNING(( "NULL shmem" ));
139 0 : return NULL;
140 0 : }
141 :
142 0 : 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 0 : 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 0 : FD_SCRATCH_ALLOC_INIT( l, shmem );
153 0 : fd_sspeer_selector_t * selector = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_sspeer_selector_t), sizeof(fd_sspeer_selector_t) );
154 0 : void * _pool = FD_SCRATCH_ALLOC_APPEND( l, peer_pool_align(), peer_pool_footprint( 2UL*max_peers ) );
155 0 : 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 0 : 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 0 : void * _score_treap = FD_SCRATCH_ALLOC_APPEND( l, score_treap_align(), score_treap_footprint( max_peers ) );
158 0 : void * _shadow_score_treap = FD_SCRATCH_ALLOC_APPEND( l, score_treap_align(), score_treap_footprint( max_peers ) );
159 0 : void * _peer_idx_list = FD_SCRATCH_ALLOC_APPEND( l, alignof(ulong), max_peers * sizeof(ulong) ); /* pool indices for shadow treap swap */
160 :
161 0 : selector->pool = peer_pool_join( peer_pool_new( _pool, 2UL*max_peers ) );
162 : /* Seed treap priorities so the treap is balanced. */
163 0 : score_treap_seed( selector->pool, 2UL*max_peers, seed );
164 0 : 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 0 : 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 0 : selector->score_treap = score_treap_join( score_treap_new( _score_treap, max_peers ) );
167 0 : selector->shadow_score_treap = score_treap_join( score_treap_new( _shadow_score_treap, max_peers ) );
168 0 : selector->peer_idx_list = (ulong *)_peer_idx_list;
169 0 : selector->max_peers = max_peers;
170 :
171 0 : selector->cluster_slot.full = 0UL;
172 0 : selector->cluster_slot.incremental = FD_SSPEER_SLOT_UNKNOWN;
173 0 : selector->cluster_slot_dirty = 0;
174 :
175 0 : FD_COMPILER_MFENCE();
176 0 : FD_VOLATILE( selector->magic ) = FD_SSPEER_SELECTOR_MAGIC;
177 0 : FD_COMPILER_MFENCE();
178 :
179 0 : return (void *)selector;
180 0 : }
181 :
182 : fd_sspeer_selector_t *
183 0 : fd_sspeer_selector_join( void * shselector ) {
184 0 : if( FD_UNLIKELY( !shselector ) ) {
185 0 : FD_LOG_WARNING(( "NULL shselector" ));
186 0 : return NULL;
187 0 : }
188 :
189 0 : 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 0 : fd_sspeer_selector_t * selector = (fd_sspeer_selector_t *)shselector;
195 :
196 0 : if( FD_UNLIKELY( selector->magic!=FD_SSPEER_SELECTOR_MAGIC ) ) {
197 0 : FD_LOG_WARNING(( "bad magic" ));
198 0 : return NULL;
199 0 : }
200 :
201 0 : return selector;
202 0 : }
203 :
204 : void *
205 0 : fd_sspeer_selector_leave( fd_sspeer_selector_t * selector ) {
206 0 : if( FD_UNLIKELY( !selector ) ) {
207 0 : FD_LOG_WARNING(( "NULL selector" ));
208 0 : return NULL;
209 0 : }
210 :
211 0 : 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 0 : if( FD_UNLIKELY( selector->magic!=FD_SSPEER_SELECTOR_MAGIC ) ) {
217 0 : FD_LOG_WARNING(( "bad magic" ));
218 0 : return NULL;
219 0 : }
220 :
221 0 : selector->pool = peer_pool_leave( selector->pool );
222 0 : selector->map_by_key = peer_map_by_key_leave( selector->map_by_key );
223 0 : selector->map_by_addr = peer_map_by_addr_leave( selector->map_by_addr );
224 0 : selector->score_treap = score_treap_leave( selector->score_treap );
225 0 : selector->shadow_score_treap = score_treap_leave( selector->shadow_score_treap );
226 :
227 0 : return (void *)selector;
228 0 : }
229 :
230 : void *
231 0 : fd_sspeer_selector_delete( void * shselector ) {
232 0 : if( FD_UNLIKELY( !shselector ) ) {
233 0 : FD_LOG_WARNING(( "NULL shselector" ));
234 0 : return NULL;
235 0 : }
236 :
237 0 : 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 0 : fd_sspeer_selector_t * selector = (fd_sspeer_selector_t *)shselector;
243 :
244 0 : if( FD_UNLIKELY( selector->magic!=FD_SSPEER_SELECTOR_MAGIC ) ) {
245 0 : FD_LOG_WARNING(( "bad magic" ));
246 0 : return NULL;
247 0 : }
248 :
249 0 : selector->pool = peer_pool_delete( selector->pool );
250 0 : selector->map_by_key = peer_map_by_key_delete( selector->map_by_key );
251 0 : selector->map_by_addr = peer_map_by_addr_delete( selector->map_by_addr );
252 0 : selector->score_treap = score_treap_delete( selector->score_treap );
253 0 : selector->shadow_score_treap = score_treap_delete( selector->shadow_score_treap );
254 :
255 0 : FD_COMPILER_MFENCE();
256 0 : FD_VOLATILE( selector->magic ) = 0UL;
257 0 : FD_COMPILER_MFENCE();
258 :
259 0 : return (void *)selector;
260 0 : }
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 0 : ulong incr_slot ) {
269 0 : peer_latency = peer_latency!=FD_SSPEER_LATENCY_UNKNOWN ? peer_latency : DEFAULT_PEER_LATENCY;
270 :
271 0 : ulong slots_behind = DEFAULT_SLOTS_BEHIND;
272 :
273 0 : if( FD_LIKELY( full_slot!=FD_SSPEER_SLOT_UNKNOWN ) ) {
274 0 : if( FD_LIKELY( incr_slot!=FD_SSPEER_SLOT_UNKNOWN &&
275 0 : selector->cluster_slot.incremental!=FD_SSPEER_SLOT_UNKNOWN ) ) {
276 0 : slots_behind = selector->cluster_slot.incremental>incr_slot ? selector->cluster_slot.incremental - incr_slot : 0UL;
277 0 : } 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 0 : slots_behind = selector->cluster_slot.full>full_slot ? selector->cluster_slot.full - full_slot : 0UL;
282 0 : }
283 0 : slots_behind = fd_ulong_min( slots_behind, MAX_SLOTS_BEHIND_CAP );
284 0 : }
285 :
286 : /* Using saturating arithmetic to avoid overflow and cap at
287 : FD_SSPEER_SCORE_MAX. */
288 0 : ulong penalty = fd_ulong_sat_mul( DEFAULT_SLOTS_BEHIND_PENALTY,
289 0 : fd_ulong_sat_mul( slots_behind, slots_behind ) );
290 0 : ulong score = fd_ulong_sat_add( peer_latency, penalty );
291 0 : return fd_ulong_min( score, FD_SSPEER_SCORE_MAX );
292 0 : }
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 0 : ulong incr_slot ) {
307 0 : if( FD_UNLIKELY( incr_slot!=FD_SSPEER_SLOT_UNKNOWN &&
308 0 : full_slot!=FD_SSPEER_SLOT_UNKNOWN &&
309 0 : incr_slot<full_slot ) ) {
310 0 : return -1;
311 0 : }
312 :
313 0 : if( FD_UNLIKELY( full_slot==FD_SSPEER_SLOT_UNKNOWN &&
314 0 : incr_slot!=FD_SSPEER_SLOT_UNKNOWN ) ) {
315 0 : return -2;
316 0 : }
317 :
318 0 : return 0;
319 0 : }
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 0 : uchar const incr_hash[ FD_HASH_FOOTPRINT ] ) {
338 0 : ulong peer_latency = latency!=FD_SSPEER_LATENCY_UNKNOWN ? latency : peer->latency;
339 0 : ulong peer_full_slot = full_slot!=FD_SSPEER_SLOT_UNKNOWN ? full_slot : peer->full_slot;
340 :
341 0 : ulong peer_incr_slot;
342 0 : int clear_incr = 0;
343 0 : if( incr_slot!=FD_SSPEER_SLOT_UNKNOWN ) {
344 0 : peer_incr_slot = incr_slot;
345 0 : } else if( full_slot!=FD_SSPEER_SLOT_UNKNOWN &&
346 0 : peer->incr_slot!=FD_SSPEER_SLOT_UNKNOWN &&
347 0 : 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 0 : peer_incr_slot = FD_SSPEER_SLOT_UNKNOWN;
351 0 : clear_incr = 1;
352 0 : } else {
353 0 : peer_incr_slot = peer->incr_slot;
354 0 : }
355 :
356 0 : int validate_err = fd_sspeer_validate_slot_args( peer_full_slot, peer_incr_slot );
357 0 : if( FD_UNLIKELY( validate_err ) ) return validate_err;
358 :
359 0 : score_treap_ele_remove( selector->score_treap, peer, selector->pool );
360 :
361 0 : peer->score = fd_sspeer_selector_score( selector, peer_latency, peer_full_slot, peer_incr_slot );
362 :
363 0 : peer->latency = peer_latency;
364 0 : peer->full_slot = peer_full_slot;
365 0 : peer->incr_slot = peer_incr_slot;
366 0 : if( FD_LIKELY( full_hash ) ) {
367 0 : fd_memcpy( peer->full_hash, full_hash, FD_HASH_FOOTPRINT );
368 0 : }
369 0 : if( FD_UNLIKELY( clear_incr ) ) {
370 0 : fd_memset( peer->incr_hash, 0, FD_HASH_FOOTPRINT );
371 0 : } else if( FD_LIKELY( incr_hash ) ) {
372 0 : fd_memcpy( peer->incr_hash, incr_hash, FD_HASH_FOOTPRINT );
373 0 : }
374 :
375 0 : score_treap_ele_insert( selector->score_treap, peer, selector->pool );
376 0 : return FD_SSPEER_UPDATE_SUCCESS;
377 0 : }
378 :
379 : ulong
380 : fd_sspeer_selector_update_on_ping( fd_sspeer_selector_t * selector,
381 : fd_ip4_port_t addr,
382 0 : ulong latency ) {
383 0 : ulong ele_idx = peer_map_by_addr_idx_query_const( selector->map_by_addr, &addr, ULONG_MAX, selector->pool );
384 0 : ulong cnt = 0UL;
385 0 : for(;;) {
386 0 : if( FD_UNLIKELY( ele_idx==ULONG_MAX ) ) break;
387 0 : 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 0 : int update_status = fd_sspeer_selector_update( selector, peer, latency,
393 0 : FD_SSPEER_SLOT_UNKNOWN, FD_SSPEER_SLOT_UNKNOWN,
394 0 : NULL, NULL );
395 0 : 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 0 : ele_idx = peer_map_by_addr_idx_next_const( ele_idx, ULONG_MAX, selector->pool );
410 0 : cnt++;
411 0 : }
412 0 : return cnt;
413 0 : }
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 0 : uchar const incr_hash[ FD_HASH_FOOTPRINT ] ) {
424 0 : 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 0 : if( FD_UNLIKELY( !addr.l ) ) return FD_SSPEER_SCORE_INVALID;
429 :
430 : /* Reject implausible slot values. FD_SSPEER_SLOT_UNKNOWN is allowed. */
431 0 : if( FD_UNLIKELY( full_slot!=FD_SSPEER_SLOT_UNKNOWN && full_slot>=FD_SSPEER_PLAUSIBLE_MAX_SLOT ) ) return FD_SSPEER_SCORE_INVALID;
432 0 : if( FD_UNLIKELY( incr_slot!=FD_SSPEER_SLOT_UNKNOWN && incr_slot>=FD_SSPEER_PLAUSIBLE_MAX_SLOT ) ) return FD_SSPEER_SCORE_INVALID;
433 :
434 0 : fd_sspeer_private_t * peer = peer_map_by_key_ele_query( selector->map_by_key, key, NULL, selector->pool );
435 0 : if( FD_LIKELY( peer ) ) {
436 0 : ulong old_full = peer->full_slot;
437 0 : ulong old_incr = peer->incr_slot;
438 0 : int update_status = fd_sspeer_selector_update( selector, peer, latency, full_slot, incr_slot, full_hash, incr_hash );
439 0 : if( FD_UNLIKELY( update_status!=FD_SSPEER_UPDATE_SUCCESS ) ) return FD_SSPEER_SCORE_INVALID;
440 0 : 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 0 : if( FD_UNLIKELY( peer->addr.l!=addr.l ) ) {
444 0 : peer_map_by_addr_ele_remove_fast( selector->map_by_addr, peer, selector->pool );
445 0 : peer->addr = addr;
446 0 : peer_map_by_addr_ele_insert( selector->map_by_addr, peer, selector->pool );
447 0 : }
448 0 : } else {
449 0 : 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 0 : if( FD_UNLIKELY( score_treap_ele_cnt(selector->score_treap)>=selector->max_peers ) ) {
454 0 : FD_LOG_WARNING(( "peer selector at max capacity" ));
455 0 : return FD_SSPEER_SCORE_INVALID;
456 0 : }
457 :
458 0 : if( FD_UNLIKELY( fd_sspeer_validate_slot_args( full_slot, incr_slot ) ) ) {
459 0 : return FD_SSPEER_SCORE_INVALID;
460 0 : }
461 :
462 0 : peer = peer_pool_ele_acquire( selector->pool );
463 0 : peer->key = *key;
464 0 : peer->addr = addr;
465 0 : peer->latency = latency;
466 0 : peer->score = fd_sspeer_selector_score( selector, latency, full_slot, incr_slot );
467 0 : peer->full_slot = full_slot;
468 0 : peer->incr_slot = incr_slot;
469 0 : if( FD_LIKELY( full_hash ) ) fd_memcpy( peer->full_hash, full_hash, FD_HASH_FOOTPRINT );
470 0 : else fd_memset( peer->full_hash, 0, FD_HASH_FOOTPRINT );
471 : /* full_hash and incr_hash are treated independently here. */
472 0 : if( FD_LIKELY( incr_hash ) ) fd_memcpy( peer->incr_hash, incr_hash, FD_HASH_FOOTPRINT );
473 0 : else fd_memset( peer->incr_hash, 0, FD_HASH_FOOTPRINT );
474 0 : peer_map_by_key_ele_insert( selector->map_by_key, peer, selector->pool );
475 0 : peer_map_by_addr_ele_insert( selector->map_by_addr, peer, selector->pool );
476 0 : score_treap_ele_insert( selector->score_treap, peer, selector->pool );
477 0 : if( FD_LIKELY( full_slot!=FD_SSPEER_SLOT_UNKNOWN || incr_slot!=FD_SSPEER_SLOT_UNKNOWN ) ) selector->cluster_slot_dirty = 1;
478 0 : }
479 0 : peer->valid = peer->full_slot!=FD_SSPEER_SLOT_UNKNOWN;
480 0 : return peer->score;
481 0 : }
482 :
483 : void
484 : fd_sspeer_selector_remove( fd_sspeer_selector_t * selector,
485 0 : fd_sspeer_key_t const * key ) {
486 0 : if( FD_UNLIKELY( key==NULL ) ) return;
487 0 : fd_sspeer_private_t * peer = peer_map_by_key_ele_query( selector->map_by_key, key, NULL, selector->pool );
488 0 : if( FD_UNLIKELY( peer==NULL ) ) return;
489 0 : score_treap_ele_remove( selector->score_treap, peer, selector->pool );
490 0 : peer_map_by_key_ele_remove_fast( selector->map_by_key, peer, selector->pool );
491 0 : peer_map_by_addr_ele_remove_fast( selector->map_by_addr, peer, selector->pool );
492 0 : peer_pool_ele_release( selector->pool, peer );
493 0 : selector->cluster_slot_dirty = 1;
494 0 : }
495 :
496 : void
497 : fd_sspeer_selector_remove_by_addr( fd_sspeer_selector_t * selector,
498 0 : fd_ip4_port_t addr ) {
499 0 : for(;;) {
500 0 : fd_sspeer_private_t * peer = peer_map_by_addr_ele_remove( selector->map_by_addr, &addr, NULL, selector->pool );
501 0 : if( FD_UNLIKELY( peer==NULL ) ) break;
502 0 : score_treap_ele_remove( selector->score_treap, peer, selector->pool );
503 0 : peer_map_by_key_ele_remove_fast( selector->map_by_key, peer, selector->pool );
504 0 : peer_pool_ele_release( selector->pool, peer );
505 0 : selector->cluster_slot_dirty = 1;
506 0 : }
507 0 : }
508 :
509 : fd_sspeer_t
510 : fd_sspeer_selector_best( fd_sspeer_selector_t * selector,
511 : int incremental,
512 : ulong base_slot,
513 0 : ulong min_reachable_slot ) {
514 0 : if( FD_UNLIKELY( incremental && base_slot==FD_SSPEER_SLOT_UNKNOWN ) ) {
515 0 : FD_LOG_WARNING(( "incremental selection requires a valid base_slot" ));
516 0 : return (fd_sspeer_t){
517 0 : .key = { .is_url = 0 },
518 0 : .addr = { .l=0UL },
519 0 : .full_slot = FD_SSPEER_SLOT_UNKNOWN,
520 0 : .incr_slot = FD_SSPEER_SLOT_UNKNOWN,
521 0 : .score = FD_SSPEER_SCORE_INVALID,
522 0 : .full_hash = {0},
523 0 : .incr_hash = {0},
524 0 : };
525 0 : }
526 :
527 0 : for( score_treap_fwd_iter_t iter = score_treap_fwd_iter_init( selector->score_treap, selector->pool );
528 0 : !score_treap_fwd_iter_done( iter );
529 0 : iter = score_treap_fwd_iter_next( iter, selector->pool ) ) {
530 0 : fd_sspeer_private_t const * peer = score_treap_fwd_iter_ele_const( iter, selector->pool );
531 : /* For full selection (!incremental), any valid peer is eligible.
532 : For incremental selection, the peer must serve the same base full
533 : snapshot and must actually offer an incremental snapshot. */
534 0 : ulong reachable_slot = fd_ulong_if( peer->incr_slot!=FD_SSPEER_SLOT_UNKNOWN, peer->incr_slot, peer->full_slot );
535 0 : if( FD_LIKELY( peer->valid &&
536 0 : reachable_slot>=min_reachable_slot &&
537 0 : (!incremental ||
538 0 : (peer->full_slot==base_slot && peer->incr_slot!=FD_SSPEER_SLOT_UNKNOWN) ) ) ) {
539 0 : fd_sspeer_t best = {
540 0 : .key = peer->key,
541 0 : .addr = peer->addr,
542 0 : .full_slot = peer->full_slot,
543 0 : .incr_slot = peer->incr_slot,
544 0 : .score = peer->score,
545 0 : };
546 0 : fd_memcpy( best.full_hash, peer->full_hash, FD_HASH_FOOTPRINT );
547 0 : fd_memcpy( best.incr_hash, peer->incr_hash, FD_HASH_FOOTPRINT );
548 0 : return best;
549 0 : }
550 0 : }
551 :
552 0 : return (fd_sspeer_t){
553 0 : .key = { .is_url = 0 },
554 0 : .addr = { .l=0UL },
555 0 : .full_slot = FD_SSPEER_SLOT_UNKNOWN,
556 0 : .incr_slot = FD_SSPEER_SLOT_UNKNOWN,
557 0 : .score = FD_SSPEER_SCORE_INVALID,
558 0 : .full_hash = {0},
559 0 : .incr_hash = {0},
560 0 : };
561 0 : }
562 :
563 : void
564 0 : fd_sspeer_selector_process_cluster_slot( fd_sspeer_selector_t * selector ) {
565 0 : if( FD_LIKELY( !selector->cluster_slot_dirty ) ) return;
566 0 : selector->cluster_slot_dirty = 0;
567 :
568 : /* Compute max full_slot and incr_slot across all tracked peers. */
569 0 : ulong max_full = 0UL;
570 0 : ulong max_incr = 0UL;
571 0 : ulong full_cnt = 0UL;
572 0 : ulong incr_cnt = 0UL;
573 0 : for( score_treap_fwd_iter_t iter = score_treap_fwd_iter_init( selector->score_treap, selector->pool );
574 0 : !score_treap_fwd_iter_done( iter );
575 0 : iter = score_treap_fwd_iter_next( iter, selector->pool ) ) {
576 0 : fd_sspeer_private_t const * peer = score_treap_fwd_iter_ele_const( iter, selector->pool );
577 0 : if( FD_LIKELY( peer->full_slot!=FD_SSPEER_SLOT_UNKNOWN ) ) {
578 0 : max_full = fd_ulong_max( max_full, peer->full_slot );
579 0 : full_cnt++;
580 0 : }
581 0 : if( FD_LIKELY( peer->incr_slot!=FD_SSPEER_SLOT_UNKNOWN ) ) {
582 0 : max_incr = fd_ulong_max( max_incr, peer->incr_slot );
583 0 : incr_cnt++;
584 0 : }
585 0 : }
586 :
587 : /* If no known incr_slot among tracked peers, treat as unknown. */
588 0 : if( FD_UNLIKELY( incr_cnt==0UL ) ) {
589 0 : max_incr = FD_SSPEER_SLOT_UNKNOWN;
590 0 : }
591 :
592 : /* If no known full_slot among tracked peers, reset so that a
593 : subsequent add() is not scored against a stale cluster slot. */
594 0 : if( FD_UNLIKELY( full_cnt==0UL ) ) {
595 0 : selector->cluster_slot.full = 0UL;
596 0 : selector->cluster_slot.incremental = FD_SSPEER_SLOT_UNKNOWN;
597 0 : return;
598 0 : }
599 :
600 : /* The full and incremental maxima are computed from different peer
601 : subsets, so max_incr < max_full is possible. Clamp to maintain
602 : the invariant that incremental >= full. */
603 0 : if( FD_UNLIKELY( max_incr!=FD_SSPEER_SLOT_UNKNOWN &&
604 0 : max_incr<max_full ) ) {
605 0 : max_incr = max_full;
606 0 : }
607 :
608 : /* If neither max changed, no rescore needed. */
609 0 : if( FD_LIKELY( max_full==selector->cluster_slot.full &&
610 0 : max_incr==selector->cluster_slot.incremental ) ) return;
611 :
612 0 : selector->cluster_slot.full = max_full;
613 0 : selector->cluster_slot.incremental = max_incr;
614 :
615 : /* Rescore all peers using the shadow treap swap mechanism. */
616 0 : ulong idx = 0UL;
617 0 : for( score_treap_fwd_iter_t iter = score_treap_fwd_iter_init( selector->score_treap, selector->pool );
618 0 : !score_treap_fwd_iter_done( iter );
619 0 : iter = score_treap_fwd_iter_next( iter, selector->pool ) ) {
620 0 : fd_sspeer_private_t * peer = score_treap_fwd_iter_ele( iter, selector->pool );
621 0 : fd_sspeer_private_t * shadow_peer = peer_pool_ele_acquire( selector->pool );
622 0 : shadow_peer->latency = peer->latency;
623 0 : shadow_peer->full_slot = peer->full_slot;
624 0 : shadow_peer->incr_slot = peer->incr_slot;
625 0 : shadow_peer->addr = peer->addr;
626 0 : shadow_peer->key = peer->key;
627 0 : shadow_peer->score = fd_sspeer_selector_score( selector, shadow_peer->latency, shadow_peer->full_slot, shadow_peer->incr_slot );
628 0 : shadow_peer->valid = peer->valid;
629 0 : fd_memcpy( shadow_peer->full_hash, peer->full_hash, FD_HASH_FOOTPRINT );
630 0 : fd_memcpy( shadow_peer->incr_hash, peer->incr_hash, FD_HASH_FOOTPRINT );
631 0 : score_treap_ele_insert( selector->shadow_score_treap, shadow_peer, selector->pool );
632 0 : selector->peer_idx_list[ idx++ ] = peer_pool_idx( selector->pool, peer );
633 0 : peer_map_by_key_ele_remove_fast( selector->map_by_key, peer, selector->pool );
634 0 : peer_map_by_addr_ele_remove_fast( selector->map_by_addr, peer, selector->pool );
635 0 : peer_map_by_key_ele_insert( selector->map_by_key, shadow_peer, selector->pool );
636 0 : peer_map_by_addr_ele_insert( selector->map_by_addr, shadow_peer, selector->pool );
637 0 : }
638 :
639 : /* clear score treap */
640 0 : for( ulong i=0UL; i<idx; i++ ) {
641 0 : fd_sspeer_private_t * peer = peer_pool_ele( selector->pool, selector->peer_idx_list[ i ] );
642 0 : score_treap_ele_remove( selector->score_treap, peer, selector->pool );
643 0 : peer_pool_ele_release( selector->pool, peer );
644 0 : }
645 :
646 0 : score_treap_t * tmp = selector->score_treap;
647 0 : selector->score_treap = selector->shadow_score_treap;
648 0 : selector->shadow_score_treap = tmp;
649 :
650 : #if FD_SSPEER_SELECTOR_DEBUG
651 : FD_TEST( score_treap_verify( selector->score_treap, selector->pool )==0 );
652 : #endif
653 0 : }
654 :
655 : fd_sscluster_slot_t
656 0 : fd_sspeer_selector_cluster_slot( fd_sspeer_selector_t * selector ) {
657 0 : return selector->cluster_slot;
658 0 : }
659 :
660 : ulong
661 0 : fd_sspeer_selector_peer_map_by_key_ele_cnt( fd_sspeer_selector_t * selector ) {
662 0 : ulong cnt = 0UL;
663 0 : for( peer_map_by_key_iter_t iter = peer_map_by_key_iter_init( selector->map_by_key, selector->pool );
664 0 : !peer_map_by_key_iter_done( iter, selector->map_by_key, selector->pool );
665 0 : iter = peer_map_by_key_iter_next( iter, selector->map_by_key, selector->pool ) ) {
666 0 : cnt++;
667 0 : }
668 0 : return cnt;
669 0 : }
670 :
671 : ulong
672 0 : fd_sspeer_selector_peer_map_by_addr_ele_cnt( fd_sspeer_selector_t * selector ) {
673 0 : ulong cnt = 0UL;
674 0 : for( peer_map_by_addr_iter_t iter = peer_map_by_addr_iter_init( selector->map_by_addr, selector->pool );
675 0 : !peer_map_by_addr_iter_done( iter, selector->map_by_addr, selector->pool );
676 0 : iter = peer_map_by_addr_iter_next( iter, selector->map_by_addr, selector->pool ) ) {
677 0 : cnt++;
678 0 : }
679 0 : return cnt;
680 0 : }
|
