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 per_peer = 2UL*sizeof(fd_sspeer_private_t) + 2UL*2UL*sizeof(ulong) + sizeof(ulong);
122 0 : if( FD_UNLIKELY( !max_peers || max_peers>(ULONG_MAX/2UL)/per_peer ) ) return 0UL;
123 0 : ulong l;
124 0 : l = FD_LAYOUT_INIT;
125 0 : l = FD_LAYOUT_APPEND( l, alignof(fd_sspeer_selector_t), sizeof(fd_sspeer_selector_t) );
126 0 : l = FD_LAYOUT_APPEND( l, peer_pool_align(), peer_pool_footprint( 2UL*max_peers ) );
127 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 ) ) );
128 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 ) ) );
129 0 : l = FD_LAYOUT_APPEND( l, score_treap_align(), score_treap_footprint( 2UL*max_peers ) );
130 0 : l = FD_LAYOUT_APPEND( l, score_treap_align(), score_treap_footprint( 2UL*max_peers ) );
131 0 : l = FD_LAYOUT_APPEND( l, alignof(ulong), max_peers * sizeof(ulong) ); /* pool indices for shadow treap swap */
132 0 : return FD_LAYOUT_FINI( l, fd_sspeer_selector_align() );
133 0 : }
134 :
135 : void *
136 : fd_sspeer_selector_new( void * shmem,
137 : ulong max_peers,
138 0 : ulong seed ) {
139 0 : if( FD_UNLIKELY( !shmem ) ) {
140 0 : FD_LOG_WARNING(( "NULL shmem" ));
141 0 : return NULL;
142 0 : }
143 :
144 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shmem, fd_sspeer_selector_align() ) ) ) {
145 0 : FD_LOG_WARNING(( "unaligned shmem" ));
146 0 : return NULL;
147 0 : }
148 :
149 0 : if( FD_UNLIKELY( !fd_sspeer_selector_footprint( max_peers ) ) ) {
150 0 : FD_LOG_WARNING(( "max_peers out of range (%lu)", max_peers ));
151 0 : return NULL;
152 0 : }
153 :
154 0 : FD_SCRATCH_ALLOC_INIT( l, shmem );
155 0 : fd_sspeer_selector_t * selector = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_sspeer_selector_t), sizeof(fd_sspeer_selector_t) );
156 0 : void * _pool = FD_SCRATCH_ALLOC_APPEND( l, peer_pool_align(), peer_pool_footprint( 2UL*max_peers ) );
157 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 ) ) );
158 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 ) ) );
159 0 : void * _score_treap = FD_SCRATCH_ALLOC_APPEND( l, score_treap_align(), score_treap_footprint( 2UL*max_peers ) );
160 0 : void * _shadow_score_treap = FD_SCRATCH_ALLOC_APPEND( l, score_treap_align(), score_treap_footprint( 2UL*max_peers ) );
161 0 : void * _peer_idx_list = FD_SCRATCH_ALLOC_APPEND( l, alignof(ulong), max_peers * sizeof(ulong) ); /* pool indices for shadow treap swap */
162 :
163 0 : selector->pool = peer_pool_join( peer_pool_new( _pool, 2UL*max_peers ) );
164 : /* Seed treap priorities so the treap is balanced. */
165 0 : score_treap_seed( selector->pool, 2UL*max_peers, seed );
166 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 ) );
167 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 ) );
168 0 : selector->score_treap = score_treap_join( score_treap_new( _score_treap, 2UL*max_peers ) );
169 0 : selector->shadow_score_treap = score_treap_join( score_treap_new( _shadow_score_treap, 2UL*max_peers ) );
170 0 : selector->peer_idx_list = (ulong *)_peer_idx_list;
171 0 : selector->max_peers = max_peers;
172 :
173 0 : selector->cluster_slot.full = 0UL;
174 0 : selector->cluster_slot.incremental = FD_SSPEER_SLOT_UNKNOWN;
175 0 : selector->cluster_slot_dirty = 0;
176 :
177 0 : FD_COMPILER_MFENCE();
178 0 : FD_VOLATILE( selector->magic ) = FD_SSPEER_SELECTOR_MAGIC;
179 0 : FD_COMPILER_MFENCE();
180 :
181 0 : return (void *)selector;
182 0 : }
183 :
184 : fd_sspeer_selector_t *
185 0 : fd_sspeer_selector_join( void * shselector ) {
186 0 : if( FD_UNLIKELY( !shselector ) ) {
187 0 : FD_LOG_WARNING(( "NULL shselector" ));
188 0 : return NULL;
189 0 : }
190 :
191 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shselector, fd_sspeer_selector_align() ) ) ) {
192 0 : FD_LOG_WARNING(( "misaligned shselector" ));
193 0 : return NULL;
194 0 : }
195 :
196 0 : fd_sspeer_selector_t * selector = (fd_sspeer_selector_t *)shselector;
197 :
198 0 : if( FD_UNLIKELY( selector->magic!=FD_SSPEER_SELECTOR_MAGIC ) ) {
199 0 : FD_LOG_WARNING(( "bad magic" ));
200 0 : return NULL;
201 0 : }
202 :
203 0 : return selector;
204 0 : }
205 :
206 : void *
207 0 : fd_sspeer_selector_leave( fd_sspeer_selector_t * selector ) {
208 0 : if( FD_UNLIKELY( !selector ) ) {
209 0 : FD_LOG_WARNING(( "NULL selector" ));
210 0 : return NULL;
211 0 : }
212 :
213 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)selector, fd_sspeer_selector_align() ) ) ) {
214 0 : FD_LOG_WARNING(( "misaligned selector" ));
215 0 : return NULL;
216 0 : }
217 :
218 0 : if( FD_UNLIKELY( selector->magic!=FD_SSPEER_SELECTOR_MAGIC ) ) {
219 0 : FD_LOG_WARNING(( "bad magic" ));
220 0 : return NULL;
221 0 : }
222 :
223 0 : selector->pool = peer_pool_leave( selector->pool );
224 0 : selector->map_by_key = peer_map_by_key_leave( selector->map_by_key );
225 0 : selector->map_by_addr = peer_map_by_addr_leave( selector->map_by_addr );
226 0 : selector->score_treap = score_treap_leave( selector->score_treap );
227 0 : selector->shadow_score_treap = score_treap_leave( selector->shadow_score_treap );
228 :
229 0 : return (void *)selector;
230 0 : }
231 :
232 : void *
233 0 : fd_sspeer_selector_delete( void * shselector ) {
234 0 : if( FD_UNLIKELY( !shselector ) ) {
235 0 : FD_LOG_WARNING(( "NULL shselector" ));
236 0 : return NULL;
237 0 : }
238 :
239 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shselector, fd_sspeer_selector_align() ) ) ) {
240 0 : FD_LOG_WARNING(( "misaligned shselector" ));
241 0 : return NULL;
242 0 : }
243 :
244 0 : fd_sspeer_selector_t * selector = (fd_sspeer_selector_t *)shselector;
245 :
246 0 : if( FD_UNLIKELY( selector->magic!=FD_SSPEER_SELECTOR_MAGIC ) ) {
247 0 : FD_LOG_WARNING(( "bad magic" ));
248 0 : return NULL;
249 0 : }
250 :
251 0 : selector->pool = peer_pool_delete( selector->pool );
252 0 : selector->map_by_key = peer_map_by_key_delete( selector->map_by_key );
253 0 : selector->map_by_addr = peer_map_by_addr_delete( selector->map_by_addr );
254 0 : selector->score_treap = score_treap_delete( selector->score_treap );
255 0 : selector->shadow_score_treap = score_treap_delete( selector->shadow_score_treap );
256 :
257 0 : FD_COMPILER_MFENCE();
258 0 : FD_VOLATILE( selector->magic ) = 0UL;
259 0 : FD_COMPILER_MFENCE();
260 :
261 0 : return (void *)selector;
262 0 : }
263 :
264 : /* Calculates a score for a peer given its latency and its resolved
265 : full and incremental slots */
266 : static ulong
267 : fd_sspeer_selector_score( fd_sspeer_selector_t const * selector,
268 : ulong peer_latency,
269 : ulong full_slot,
270 0 : ulong incr_slot ) {
271 0 : peer_latency = peer_latency!=FD_SSPEER_LATENCY_UNKNOWN ? peer_latency : DEFAULT_PEER_LATENCY;
272 :
273 0 : ulong slots_behind = DEFAULT_SLOTS_BEHIND;
274 :
275 0 : if( FD_LIKELY( full_slot!=FD_SSPEER_SLOT_UNKNOWN ) ) {
276 0 : if( FD_LIKELY( incr_slot!=FD_SSPEER_SLOT_UNKNOWN &&
277 0 : selector->cluster_slot.incremental!=FD_SSPEER_SLOT_UNKNOWN ) ) {
278 0 : slots_behind = selector->cluster_slot.incremental>incr_slot ? selector->cluster_slot.incremental - incr_slot : 0UL;
279 0 : } else {
280 : /* Either the peer has no incremental or the cluster has no
281 : incremental reference yet. Fall back to comparing full_slot
282 : against the cluster full slot. */
283 0 : slots_behind = selector->cluster_slot.full>full_slot ? selector->cluster_slot.full - full_slot : 0UL;
284 0 : }
285 0 : slots_behind = fd_ulong_min( slots_behind, MAX_SLOTS_BEHIND_CAP );
286 0 : }
287 :
288 : /* Using saturating arithmetic to avoid overflow and cap at
289 : FD_SSPEER_SCORE_MAX. */
290 0 : ulong penalty = fd_ulong_sat_mul( DEFAULT_SLOTS_BEHIND_PENALTY,
291 0 : fd_ulong_sat_mul( slots_behind, slots_behind ) );
292 0 : ulong score = fd_ulong_sat_add( peer_latency, penalty );
293 0 : return fd_ulong_min( score, FD_SSPEER_SCORE_MAX );
294 0 : }
295 :
296 : /* Validates slot arguments for both new and existing peers. Returns
297 : 0 on success, -1 on failure due to incr_slot<full_slot, and -2 on
298 : failure due to full_slot==UNKNOWN with incr_slot!=UNKNOWN. The
299 : caller passes in the effective (already-resolved) full_slot and
300 : incr_slot values. No log on failure (the caller is responsible
301 : for logging whenever needed).
302 :
303 : Two invariants are enforced:
304 : 1. When both slots are known, incr_slot must be >= full_slot.
305 : 2. An incremental slot requires a known full slot. */
306 : static int
307 : fd_sspeer_validate_slot_args( ulong full_slot,
308 0 : ulong incr_slot ) {
309 0 : if( FD_UNLIKELY( incr_slot!=FD_SSPEER_SLOT_UNKNOWN &&
310 0 : full_slot!=FD_SSPEER_SLOT_UNKNOWN &&
311 0 : incr_slot<full_slot ) ) {
312 0 : return -1;
313 0 : }
314 :
315 0 : if( FD_UNLIKELY( full_slot==FD_SSPEER_SLOT_UNKNOWN &&
316 0 : incr_slot!=FD_SSPEER_SLOT_UNKNOWN ) ) {
317 0 : return -2;
318 0 : }
319 :
320 0 : return 0;
321 0 : }
322 :
323 : /* Updates a peer's score with new values for latency and/or resolved
324 : full/incremental slots. Returns FD_SSPEER_UPDATE_SUCCESS on
325 : success, or the specific fd_sspeer_validate_slot_args error code
326 : on failure without modifying the peer or any data structure.
327 :
328 : Slot-based incremental clearing: when the caller provides
329 : incr_slot==UNKNOWN and full_slot!=UNKNOWN, the peer's existing
330 : incremental data is cleared if it is stale (peer->incr_slot <
331 : full_slot). Otherwise, the existing incremental data is preserved. */
332 : static int
333 : fd_sspeer_selector_update( fd_sspeer_selector_t * selector,
334 : fd_sspeer_private_t * peer,
335 : ulong latency,
336 : ulong full_slot,
337 : ulong incr_slot,
338 : uchar const full_hash[ FD_HASH_FOOTPRINT ],
339 0 : uchar const incr_hash[ FD_HASH_FOOTPRINT ] ) {
340 0 : ulong peer_latency = latency!=FD_SSPEER_LATENCY_UNKNOWN ? latency : peer->latency;
341 0 : ulong peer_full_slot = full_slot!=FD_SSPEER_SLOT_UNKNOWN ? full_slot : peer->full_slot;
342 :
343 0 : ulong peer_incr_slot;
344 0 : int clear_incr = 0;
345 0 : if( incr_slot!=FD_SSPEER_SLOT_UNKNOWN ) {
346 0 : peer_incr_slot = incr_slot;
347 0 : } else if( full_slot!=FD_SSPEER_SLOT_UNKNOWN &&
348 0 : peer->incr_slot!=FD_SSPEER_SLOT_UNKNOWN &&
349 0 : peer->incr_slot<full_slot ) {
350 : /* The caller is providing a new full_slot that has advanced past
351 : the peer's existing incremental, the incremental is stale. */
352 0 : peer_incr_slot = FD_SSPEER_SLOT_UNKNOWN;
353 0 : clear_incr = 1;
354 0 : } else {
355 0 : peer_incr_slot = peer->incr_slot;
356 0 : }
357 :
358 0 : int validate_err = fd_sspeer_validate_slot_args( peer_full_slot, peer_incr_slot );
359 0 : if( FD_UNLIKELY( validate_err ) ) return validate_err;
360 :
361 0 : score_treap_ele_remove( selector->score_treap, peer, selector->pool );
362 :
363 0 : peer->score = fd_sspeer_selector_score( selector, peer_latency, peer_full_slot, peer_incr_slot );
364 :
365 0 : peer->latency = peer_latency;
366 0 : peer->full_slot = peer_full_slot;
367 0 : peer->incr_slot = peer_incr_slot;
368 0 : if( FD_LIKELY( full_hash ) ) {
369 0 : fd_memcpy( peer->full_hash, full_hash, FD_HASH_FOOTPRINT );
370 0 : }
371 0 : if( FD_UNLIKELY( clear_incr ) ) {
372 0 : fd_memset( peer->incr_hash, 0, FD_HASH_FOOTPRINT );
373 0 : } else if( FD_LIKELY( incr_hash ) ) {
374 0 : fd_memcpy( peer->incr_hash, incr_hash, FD_HASH_FOOTPRINT );
375 0 : }
376 :
377 0 : score_treap_ele_insert( selector->score_treap, peer, selector->pool );
378 0 : return FD_SSPEER_UPDATE_SUCCESS;
379 0 : }
380 :
381 : ulong
382 : fd_sspeer_selector_update_on_ping( fd_sspeer_selector_t * selector,
383 : fd_ip4_port_t addr,
384 0 : ulong latency ) {
385 0 : ulong ele_idx = peer_map_by_addr_idx_query_const( selector->map_by_addr, &addr, ULONG_MAX, selector->pool );
386 0 : ulong cnt = 0UL;
387 0 : for(;;) {
388 0 : if( FD_UNLIKELY( ele_idx==ULONG_MAX ) ) break;
389 0 : fd_sspeer_private_t * peer = selector->pool + ele_idx;
390 : /* Update cannot fail here: slots are FD_SSPEER_SLOT_UNKNOWN and
391 : hashes are NULL, so fd_sspeer_validate_slot_args always
392 : returns FD_SSPEER_UPDATE_SUCCESS (no clear_incr trigger,
393 : no incr<full violation). */
394 0 : int update_status = fd_sspeer_selector_update( selector, peer, latency,
395 0 : FD_SSPEER_SLOT_UNKNOWN, FD_SSPEER_SLOT_UNKNOWN,
396 0 : NULL, NULL );
397 0 : if( FD_UNLIKELY( update_status!=FD_SSPEER_UPDATE_SUCCESS ) ) {
398 : /* A warning is a tradeoff between crashing with FD_LOG_CRIT and
399 : potentially missing the log altogether with FD_LOG_DEBUG. */
400 0 : if( peer->key.is_url ) {
401 0 : FD_LOG_WARNING(( "unexpected selector update returned %d for peer %s " FD_IP4_ADDR_FMT ":%hu",
402 0 : update_status, peer->key.url.hostname,
403 0 : FD_IP4_ADDR_FMT_ARGS( peer->key.url.resolved_addr.addr ), fd_ushort_bswap( peer->key.url.resolved_addr.port ) ));
404 0 : } else {
405 0 : FD_BASE58_ENCODE_32_BYTES( peer->key.pubkey->uc, peer_pubkey_b58 );
406 0 : FD_LOG_WARNING(( "unexpected selector update returned %d for peer %s " FD_IP4_ADDR_FMT ":%hu",
407 0 : update_status, peer_pubkey_b58,
408 0 : FD_IP4_ADDR_FMT_ARGS( peer->addr.addr ), fd_ushort_bswap( peer->addr.port ) ));
409 0 : }
410 0 : }
411 0 : ele_idx = peer_map_by_addr_idx_next_const( ele_idx, ULONG_MAX, selector->pool );
412 0 : cnt++;
413 0 : }
414 0 : return cnt;
415 0 : }
416 :
417 : ulong
418 : fd_sspeer_selector_add( fd_sspeer_selector_t * selector,
419 : fd_sspeer_key_t const * key,
420 : fd_ip4_port_t addr,
421 : ulong latency,
422 : ulong full_slot,
423 : ulong incr_slot,
424 : uchar const full_hash[ FD_HASH_FOOTPRINT ],
425 0 : uchar const incr_hash[ FD_HASH_FOOTPRINT ] ) {
426 0 : if( FD_UNLIKELY( key==NULL ) ) return FD_SSPEER_SCORE_INVALID;
427 : /* A peer without a valid address cannot be added to the selector.
428 : For an existing peer changing from a valid address to 0, it is
429 : the caller's responsibility to remove them. */
430 0 : if( FD_UNLIKELY( !addr.l ) ) return FD_SSPEER_SCORE_INVALID;
431 :
432 : /* Reject implausible slot values. FD_SSPEER_SLOT_UNKNOWN is allowed. */
433 0 : if( FD_UNLIKELY( full_slot!=FD_SSPEER_SLOT_UNKNOWN && full_slot>=FD_SSPEER_PLAUSIBLE_MAX_SLOT ) ) return FD_SSPEER_SCORE_INVALID;
434 0 : if( FD_UNLIKELY( incr_slot!=FD_SSPEER_SLOT_UNKNOWN && incr_slot>=FD_SSPEER_PLAUSIBLE_MAX_SLOT ) ) return FD_SSPEER_SCORE_INVALID;
435 :
436 0 : fd_sspeer_private_t * peer = peer_map_by_key_ele_query( selector->map_by_key, key, NULL, selector->pool );
437 0 : if( FD_LIKELY( peer ) ) {
438 0 : ulong old_full = peer->full_slot;
439 0 : ulong old_incr = peer->incr_slot;
440 0 : int update_status = fd_sspeer_selector_update( selector, peer, latency, full_slot, incr_slot, full_hash, incr_hash );
441 0 : if( FD_UNLIKELY( update_status!=FD_SSPEER_UPDATE_SUCCESS ) ) return FD_SSPEER_SCORE_INVALID;
442 0 : if( FD_UNLIKELY( peer->full_slot!=old_full || peer->incr_slot!=old_incr ) ) selector->cluster_slot_dirty = 1;
443 : /* Update the addr map after the selector update so that the peer
444 : is not mutated when the update fails. */
445 0 : if( FD_UNLIKELY( peer->addr.l!=addr.l ) ) {
446 0 : peer_map_by_addr_ele_remove_fast( selector->map_by_addr, peer, selector->pool );
447 0 : peer->addr = addr;
448 0 : peer_map_by_addr_ele_insert( selector->map_by_addr, peer, selector->pool );
449 0 : }
450 0 : } else {
451 0 : if( FD_UNLIKELY( !peer_pool_free( selector->pool ) ) ) {
452 0 : FD_LOG_WARNING(( "peer selector pool exhausted" ));
453 0 : return FD_SSPEER_SCORE_INVALID;
454 0 : }
455 0 : if( FD_UNLIKELY( score_treap_ele_cnt(selector->score_treap)>=selector->max_peers ) ) {
456 0 : FD_LOG_WARNING(( "peer selector at max capacity" ));
457 0 : return FD_SSPEER_SCORE_INVALID;
458 0 : }
459 :
460 0 : if( FD_UNLIKELY( fd_sspeer_validate_slot_args( full_slot, incr_slot ) ) ) {
461 0 : return FD_SSPEER_SCORE_INVALID;
462 0 : }
463 :
464 0 : peer = peer_pool_ele_acquire( selector->pool );
465 0 : peer->key = *key;
466 0 : peer->addr = addr;
467 0 : peer->latency = latency;
468 0 : peer->score = fd_sspeer_selector_score( selector, latency, full_slot, incr_slot );
469 0 : peer->full_slot = full_slot;
470 0 : peer->incr_slot = incr_slot;
471 0 : if( FD_LIKELY( full_hash ) ) fd_memcpy( peer->full_hash, full_hash, FD_HASH_FOOTPRINT );
472 0 : else fd_memset( peer->full_hash, 0, FD_HASH_FOOTPRINT );
473 : /* full_hash and incr_hash are treated independently here. */
474 0 : if( FD_LIKELY( incr_hash ) ) fd_memcpy( peer->incr_hash, incr_hash, FD_HASH_FOOTPRINT );
475 0 : else fd_memset( peer->incr_hash, 0, FD_HASH_FOOTPRINT );
476 0 : peer_map_by_key_ele_insert( selector->map_by_key, peer, selector->pool );
477 0 : peer_map_by_addr_ele_insert( selector->map_by_addr, peer, selector->pool );
478 0 : score_treap_ele_insert( selector->score_treap, peer, selector->pool );
479 0 : if( FD_LIKELY( full_slot!=FD_SSPEER_SLOT_UNKNOWN || incr_slot!=FD_SSPEER_SLOT_UNKNOWN ) ) selector->cluster_slot_dirty = 1;
480 0 : }
481 0 : peer->valid = peer->full_slot!=FD_SSPEER_SLOT_UNKNOWN;
482 0 : return peer->score;
483 0 : }
484 :
485 : void
486 : fd_sspeer_selector_remove( fd_sspeer_selector_t * selector,
487 0 : fd_sspeer_key_t const * key ) {
488 0 : if( FD_UNLIKELY( key==NULL ) ) return;
489 0 : fd_sspeer_private_t * peer = peer_map_by_key_ele_query( selector->map_by_key, key, NULL, selector->pool );
490 0 : if( FD_UNLIKELY( peer==NULL ) ) return;
491 0 : score_treap_ele_remove( selector->score_treap, peer, selector->pool );
492 0 : peer_map_by_key_ele_remove_fast( selector->map_by_key, peer, selector->pool );
493 0 : peer_map_by_addr_ele_remove_fast( selector->map_by_addr, peer, selector->pool );
494 0 : peer_pool_ele_release( selector->pool, peer );
495 0 : selector->cluster_slot_dirty = 1;
496 0 : }
497 :
498 : void
499 : fd_sspeer_selector_remove_by_addr( fd_sspeer_selector_t * selector,
500 0 : fd_ip4_port_t addr ) {
501 0 : for(;;) {
502 0 : fd_sspeer_private_t * peer = peer_map_by_addr_ele_remove( selector->map_by_addr, &addr, NULL, selector->pool );
503 0 : if( FD_UNLIKELY( peer==NULL ) ) break;
504 0 : score_treap_ele_remove( selector->score_treap, peer, selector->pool );
505 0 : peer_map_by_key_ele_remove_fast( selector->map_by_key, peer, selector->pool );
506 0 : peer_pool_ele_release( selector->pool, peer );
507 0 : selector->cluster_slot_dirty = 1;
508 0 : }
509 0 : }
510 :
511 : fd_sspeer_t
512 : fd_sspeer_selector_best( fd_sspeer_selector_t * selector,
513 : int incremental,
514 : ulong base_slot,
515 0 : ulong min_reachable_slot ) {
516 0 : if( FD_UNLIKELY( incremental && base_slot==FD_SSPEER_SLOT_UNKNOWN ) ) {
517 0 : FD_LOG_WARNING(( "incremental selection requires a valid base_slot" ));
518 0 : return (fd_sspeer_t){
519 0 : .key = { .is_url = 0 },
520 0 : .addr = { .l=0UL },
521 0 : .full_slot = FD_SSPEER_SLOT_UNKNOWN,
522 0 : .incr_slot = FD_SSPEER_SLOT_UNKNOWN,
523 0 : .score = FD_SSPEER_SCORE_INVALID,
524 0 : .full_hash = {0},
525 0 : .incr_hash = {0},
526 0 : };
527 0 : }
528 :
529 0 : for( score_treap_fwd_iter_t iter = score_treap_fwd_iter_init( selector->score_treap, selector->pool );
530 0 : !score_treap_fwd_iter_done( iter );
531 0 : iter = score_treap_fwd_iter_next( iter, selector->pool ) ) {
532 0 : fd_sspeer_private_t const * peer = score_treap_fwd_iter_ele_const( iter, selector->pool );
533 : /* For full selection (!incremental), any valid peer is eligible.
534 : For incremental selection, the peer must serve the same base full
535 : snapshot and must actually offer an incremental snapshot. */
536 0 : ulong reachable_slot = fd_ulong_if( peer->incr_slot!=FD_SSPEER_SLOT_UNKNOWN, peer->incr_slot, peer->full_slot );
537 0 : if( FD_LIKELY( peer->valid &&
538 0 : reachable_slot>=min_reachable_slot &&
539 0 : (!incremental ||
540 0 : (peer->full_slot==base_slot && peer->incr_slot!=FD_SSPEER_SLOT_UNKNOWN) ) ) ) {
541 0 : fd_sspeer_t best = {
542 0 : .key = peer->key,
543 0 : .addr = peer->addr,
544 0 : .full_slot = peer->full_slot,
545 0 : .incr_slot = peer->incr_slot,
546 0 : .score = peer->score,
547 0 : };
548 0 : fd_memcpy( best.full_hash, peer->full_hash, FD_HASH_FOOTPRINT );
549 0 : fd_memcpy( best.incr_hash, peer->incr_hash, FD_HASH_FOOTPRINT );
550 0 : return best;
551 0 : }
552 0 : }
553 :
554 0 : return (fd_sspeer_t){
555 0 : .key = { .is_url = 0 },
556 0 : .addr = { .l=0UL },
557 0 : .full_slot = FD_SSPEER_SLOT_UNKNOWN,
558 0 : .incr_slot = FD_SSPEER_SLOT_UNKNOWN,
559 0 : .score = FD_SSPEER_SCORE_INVALID,
560 0 : .full_hash = {0},
561 0 : .incr_hash = {0},
562 0 : };
563 0 : }
564 :
565 : void
566 0 : fd_sspeer_selector_process_cluster_slot( fd_sspeer_selector_t * selector ) {
567 0 : if( FD_LIKELY( !selector->cluster_slot_dirty ) ) return;
568 0 : selector->cluster_slot_dirty = 0;
569 :
570 : /* Compute max full_slot and incr_slot across all tracked peers. */
571 0 : ulong max_full = 0UL;
572 0 : ulong max_incr = 0UL;
573 0 : ulong full_cnt = 0UL;
574 0 : ulong incr_cnt = 0UL;
575 0 : for( score_treap_fwd_iter_t iter = score_treap_fwd_iter_init( selector->score_treap, selector->pool );
576 0 : !score_treap_fwd_iter_done( iter );
577 0 : iter = score_treap_fwd_iter_next( iter, selector->pool ) ) {
578 0 : fd_sspeer_private_t const * peer = score_treap_fwd_iter_ele_const( iter, selector->pool );
579 0 : if( FD_LIKELY( peer->full_slot!=FD_SSPEER_SLOT_UNKNOWN ) ) {
580 0 : max_full = fd_ulong_max( max_full, peer->full_slot );
581 0 : full_cnt++;
582 0 : }
583 0 : if( FD_LIKELY( peer->incr_slot!=FD_SSPEER_SLOT_UNKNOWN ) ) {
584 0 : max_incr = fd_ulong_max( max_incr, peer->incr_slot );
585 0 : incr_cnt++;
586 0 : }
587 0 : }
588 :
589 : /* If no known incr_slot among tracked peers, treat as unknown. */
590 0 : if( FD_UNLIKELY( incr_cnt==0UL ) ) {
591 0 : max_incr = FD_SSPEER_SLOT_UNKNOWN;
592 0 : }
593 :
594 : /* If no known full_slot among tracked peers, reset so that a
595 : subsequent add() is not scored against a stale cluster slot. */
596 0 : if( FD_UNLIKELY( full_cnt==0UL ) ) {
597 0 : selector->cluster_slot.full = 0UL;
598 0 : selector->cluster_slot.incremental = FD_SSPEER_SLOT_UNKNOWN;
599 0 : return;
600 0 : }
601 :
602 : /* The full and incremental maxima are computed from different peer
603 : subsets, so max_incr < max_full is possible. Clamp to maintain
604 : the invariant that incremental >= full. */
605 0 : if( FD_UNLIKELY( max_incr!=FD_SSPEER_SLOT_UNKNOWN &&
606 0 : max_incr<max_full ) ) {
607 0 : max_incr = max_full;
608 0 : }
609 :
610 : /* If neither max changed, no rescore needed. */
611 0 : if( FD_LIKELY( max_full==selector->cluster_slot.full &&
612 0 : max_incr==selector->cluster_slot.incremental ) ) return;
613 :
614 0 : selector->cluster_slot.full = max_full;
615 0 : selector->cluster_slot.incremental = max_incr;
616 :
617 : #if FD_TMPL_USE_HANDHOLDING
618 : if( FD_UNLIKELY( score_treap_ele_cnt( selector->score_treap )>selector->max_peers ) ) FD_LOG_CRIT(( "peer count exceeds max_peers" ));
619 : #endif
620 :
621 : /* Rescore all peers using the shadow treap swap mechanism. */
622 0 : ulong idx = 0UL;
623 0 : for( score_treap_fwd_iter_t iter = score_treap_fwd_iter_init( selector->score_treap, selector->pool );
624 0 : !score_treap_fwd_iter_done( iter );
625 0 : iter = score_treap_fwd_iter_next( iter, selector->pool ) ) {
626 0 : fd_sspeer_private_t * peer = score_treap_fwd_iter_ele( iter, selector->pool );
627 0 : fd_sspeer_private_t * shadow_peer = peer_pool_ele_acquire( selector->pool );
628 0 : shadow_peer->latency = peer->latency;
629 0 : shadow_peer->full_slot = peer->full_slot;
630 0 : shadow_peer->incr_slot = peer->incr_slot;
631 0 : shadow_peer->addr = peer->addr;
632 0 : shadow_peer->key = peer->key;
633 0 : shadow_peer->score = fd_sspeer_selector_score( selector, shadow_peer->latency, shadow_peer->full_slot, shadow_peer->incr_slot );
634 0 : shadow_peer->valid = peer->valid;
635 0 : fd_memcpy( shadow_peer->full_hash, peer->full_hash, FD_HASH_FOOTPRINT );
636 0 : fd_memcpy( shadow_peer->incr_hash, peer->incr_hash, FD_HASH_FOOTPRINT );
637 0 : score_treap_ele_insert( selector->shadow_score_treap, shadow_peer, selector->pool );
638 0 : selector->peer_idx_list[ idx++ ] = peer_pool_idx( selector->pool, peer );
639 0 : peer_map_by_key_ele_remove_fast( selector->map_by_key, peer, selector->pool );
640 0 : peer_map_by_addr_ele_remove_fast( selector->map_by_addr, peer, selector->pool );
641 0 : peer_map_by_key_ele_insert( selector->map_by_key, shadow_peer, selector->pool );
642 0 : peer_map_by_addr_ele_insert( selector->map_by_addr, shadow_peer, selector->pool );
643 0 : }
644 :
645 : /* clear score treap */
646 0 : for( ulong i=0UL; i<idx; i++ ) {
647 0 : fd_sspeer_private_t * peer = peer_pool_ele( selector->pool, selector->peer_idx_list[ i ] );
648 0 : score_treap_ele_remove( selector->score_treap, peer, selector->pool );
649 0 : peer_pool_ele_release( selector->pool, peer );
650 0 : }
651 :
652 0 : score_treap_t * tmp = selector->score_treap;
653 0 : selector->score_treap = selector->shadow_score_treap;
654 0 : selector->shadow_score_treap = tmp;
655 :
656 : #if FD_SSPEER_SELECTOR_DEBUG
657 : FD_TEST( score_treap_verify( selector->score_treap, selector->pool )==0 );
658 : #endif
659 0 : }
660 :
661 : fd_sscluster_slot_t
662 0 : fd_sspeer_selector_cluster_slot( fd_sspeer_selector_t * selector ) {
663 0 : return selector->cluster_slot;
664 0 : }
665 :
666 : ulong
667 0 : fd_sspeer_selector_peer_map_by_key_ele_cnt( fd_sspeer_selector_t * selector ) {
668 0 : ulong cnt = 0UL;
669 0 : for( peer_map_by_key_iter_t iter = peer_map_by_key_iter_init( selector->map_by_key, selector->pool );
670 0 : !peer_map_by_key_iter_done( iter, selector->map_by_key, selector->pool );
671 0 : iter = peer_map_by_key_iter_next( iter, selector->map_by_key, selector->pool ) ) {
672 0 : cnt++;
673 0 : }
674 0 : return cnt;
675 0 : }
676 :
677 : ulong
678 0 : fd_sspeer_selector_peer_map_by_addr_ele_cnt( fd_sspeer_selector_t * selector ) {
679 0 : ulong cnt = 0UL;
680 0 : for( peer_map_by_addr_iter_t iter = peer_map_by_addr_iter_init( selector->map_by_addr, selector->pool );
681 0 : !peer_map_by_addr_iter_done( iter, selector->map_by_addr, selector->pool );
682 0 : iter = peer_map_by_addr_iter_next( iter, selector->map_by_addr, selector->pool ) ) {
683 0 : cnt++;
684 0 : }
685 0 : return cnt;
686 0 : }
|
