Line data Source code
1 : #include "fd_gossip.h"
2 : #include "fd_bloom.h"
3 : #include "fd_gossip_message.h"
4 : #include "fd_gossip_txbuild.h"
5 : #include "fd_active_set.h"
6 : #include "fd_ping_tracker.h"
7 : #include "fd_prune_finder.h"
8 : #include "fd_gossip_wsample.h"
9 : #include "fd_gossip_hset.h"
10 : #include "../../disco/keyguard/fd_keyguard.h"
11 : #include "../../ballet/sha256/fd_sha256.h"
12 : #include "../leaders/fd_leaders_base.h"
13 : #include "../../util/fd_hash32.h"
14 :
15 : FD_STATIC_ASSERT( FD_METRICS_ENUM_GOSSIP_MESSAGE_TYPE_CNT==FD_GOSSIP_MESSAGE_CNT,
16 : "FD_METRICS_ENUM_GOSSIP_MESSAGE_TYPE_CNT must match FD_GOSSIP_MESSAGE_CNT" );
17 :
18 : FD_STATIC_ASSERT( FD_METRICS_ENUM_CRDS_VALUE_TYPE_CNT==FD_GOSSIP_VALUE_CNT,
19 : "FD_METRICS_ENUM_CRDS_VALUE_TYPE_CNT must match FD_GOSSIP_VALUE_CNT" );
20 :
21 0 : #define BLOOM_FALSE_POSITIVE_RATE (0.1)
22 0 : #define BLOOM_NUM_KEYS (8.0)
23 :
24 : struct stake {
25 : fd_pubkey_t pubkey;
26 : ulong stake;
27 :
28 : struct {
29 : ulong prev;
30 : ulong next;
31 : } map;
32 :
33 : struct {
34 : ulong next;
35 : } pool;
36 : };
37 :
38 : typedef struct stake stake_t;
39 :
40 : /* NOTE: Since the staked count is known at the time we populate
41 : the map, we can treat the pool as an array instead. This means we
42 : can bypass the acquire/release model and quickly iterate through the
43 : pool when we repopulate the map on every fd_gossip_stakes_update
44 : iteration. */
45 : #define POOL_NAME stake_pool
46 0 : #define POOL_T stake_t
47 : #define POOL_IDX_T ulong
48 0 : #define POOL_NEXT pool.next
49 : #include "../../util/tmpl/fd_pool.c"
50 :
51 : #define MAP_NAME stake_map
52 0 : #define MAP_KEY pubkey
53 : #define MAP_ELE_T stake_t
54 : #define MAP_KEY_T fd_pubkey_t
55 0 : #define MAP_PREV map.prev
56 0 : #define MAP_NEXT map.next
57 0 : #define MAP_KEY_EQ(k0,k1) fd_pubkey_eq( k0, k1 )
58 0 : #define MAP_KEY_HASH(key,seed) fd_hash32( (key)->uc, (seed) )
59 : #define MAP_OPTIMIZE_RANDOM_ACCESS_REMOVAL 1
60 : #include "../../util/tmpl/fd_map_chain.c"
61 :
62 : struct fd_gossip_private {
63 : uchar identity_pubkey[ 32UL ];
64 : ulong identity_stake;
65 :
66 : fd_gossip_metrics_t metrics[1];
67 :
68 : fd_gossip_wsample_t * wsample;
69 : fd_crds_t * crds;
70 : fd_gossip_purged_t * purged;
71 : fd_active_set_t * active_set;
72 : fd_ping_tracker_t * ping_tracker;
73 : fd_prune_finder_t * prune_finder;
74 :
75 : fd_sha256_t sha256[1];
76 : fd_sha512_t sha512[1];
77 :
78 : ulong entrypoints_cnt;
79 : fd_ip4_port_t entrypoints[ 16UL ];
80 :
81 : fd_rng_t * rng;
82 :
83 : struct {
84 : ulong count;
85 : stake_t * pool;
86 : stake_map_t * map;
87 : } stake;
88 :
89 : struct {
90 : long next_pull_request;
91 : long next_active_set_refresh;
92 : long next_contact_info_refresh;
93 : long next_flush_push_state;
94 : long next_pull_contact_info;
95 : } timers;
96 :
97 : /* Token-bucket rate limiter for outbound pull response data.
98 : Matches Agave's DataBudget: replenished every 100ms with
99 : num_staked*1024 bytes, capped at 5x that amount. Only
100 : pull responses are rate-limited; push messages are not. */
101 : struct {
102 : ulong remaining; /* bytes remaining in budget (signed) */
103 : long last_replenish_nanos; /* last replenish timestamp in nanos */
104 : } outbound_budget;
105 :
106 : /* Per-request iteration budget for the CRDS hash scan in
107 : rx_pull_request. Reset at the start of each request. */
108 : struct {
109 : ulong remaining;
110 : } scan_budget;
111 :
112 : /* Sign requests in flight, oldest first. The signer answers in
113 : order; each holds the finished packet except the signature. */
114 : ulong sign_pend_head;
115 : ulong sign_pend_cnt;
116 : struct {
117 : uchar kind; /* SIGN_PEND_* */
118 : fd_ip4_port_t peer; /* SIGN_PEND_SEND */
119 : ushort sz;
120 : ushort sig_off;
121 : uchar data[ FD_GOSSIP_VALUE_MAX_SZ ];
122 : fd_gossip_value_t value[1]; /* SIGN_PEND_VALUE, SIGN_PEND_CONTACT_INFO */
123 : ulong gen; /* SIGN_PEND_*CONTACT_INFO: my_contact_info.gen when requested */
124 : } sign_pend[ FD_GOSSIP_SIGN_PEND_MAX ];
125 :
126 :
127 : /* Callbacks */
128 : fd_gossip_sign_fn sign_fn;
129 : void * sign_ctx;
130 :
131 : fd_gossip_send_fn send_fn;
132 : void * send_ctx;
133 :
134 : fd_ping_tracker_change_fn ping_tracker_change_fn;
135 : void * ping_tracker_change_fn_ctx;
136 :
137 : struct {
138 : uchar crds_val[ FD_GOSSIP_VALUE_MAX_SZ ];
139 : ulong crds_val_sz;
140 : ulong gen;
141 : int refresh_due;
142 : fd_gossip_value_t ci[1];
143 :
144 : uchar pull_val[ FD_GOSSIP_VALUE_MAX_SZ ];
145 : ulong pull_val_sz;
146 : int pull_pending;
147 : } my_contact_info;
148 :
149 : fd_gossip_out_ctx_t * gossip_net_out;
150 : };
151 :
152 : FD_FN_CONST ulong
153 0 : fd_gossip_align( void ) {
154 0 : return 128uL;
155 0 : }
156 :
157 : FD_FN_CONST ulong
158 : fd_gossip_footprint( ulong max_values,
159 0 : ulong entrypoints_len ) {
160 0 : ulong l;
161 0 : l = FD_LAYOUT_INIT;
162 0 : l = FD_LAYOUT_APPEND( l, alignof(fd_gossip_t), sizeof(fd_gossip_t) );
163 0 : l = FD_LAYOUT_APPEND( l, fd_gossip_purged_align(), fd_gossip_purged_footprint( max_values ) );
164 0 : l = FD_LAYOUT_APPEND( l, fd_gossip_wsample_align(),fd_gossip_wsample_footprint( FD_CONTACT_INFO_TABLE_SIZE ) );
165 0 : l = FD_LAYOUT_APPEND( l, fd_crds_align(), fd_crds_footprint( max_values ) );
166 0 : l = FD_LAYOUT_APPEND( l, fd_active_set_align(), fd_active_set_footprint() );
167 0 : l = FD_LAYOUT_APPEND( l, fd_ping_tracker_align(), fd_ping_tracker_footprint( entrypoints_len ) );
168 0 : l = FD_LAYOUT_APPEND( l, fd_prune_finder_align(), fd_prune_finder_footprint() );
169 0 : l = FD_LAYOUT_APPEND( l, stake_pool_align(), stake_pool_footprint( MAX_SHRED_DESTS ) );
170 0 : l = FD_LAYOUT_APPEND( l, stake_map_align(), stake_map_footprint( stake_map_chain_cnt_est( MAX_SHRED_DESTS ) ) );
171 0 : l = FD_LAYOUT_FINI( l, fd_gossip_align() );
172 0 : return l;
173 0 : }
174 :
175 : static void
176 : ping_tracker_change( void * _ctx,
177 : uchar const * peer_pubkey,
178 : fd_ip4_port_t peer_address,
179 : long now,
180 0 : int change_type ) {
181 0 : fd_gossip_t * ctx = (fd_gossip_t *)_ctx;
182 :
183 0 : if( FD_UNLIKELY( !memcmp( peer_pubkey, ctx->identity_pubkey, 32UL ) ) ) return;
184 :
185 0 : if( FD_LIKELY( change_type==FD_PING_TRACKER_CHANGE_TYPE_ACTIVE ) ) {
186 0 : fd_gossip_purged_drain_no_contact_info( ctx->purged, peer_pubkey );
187 0 : }
188 :
189 0 : ulong ci_idx = fd_crds_ci_idx( ctx->crds, peer_pubkey );
190 :
191 0 : if( FD_UNLIKELY( ci_idx!=ULONG_MAX ) ) {
192 0 : fd_gossip_contact_info_t const * ci = fd_crds_ci( ctx->crds, ci_idx );
193 0 : if( FD_UNLIKELY( ci->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ].is_ipv6 ||
194 0 : ci->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ].ip4!=peer_address.addr ||
195 0 : ci->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ].port!=peer_address.port ) ) ci_idx = ULONG_MAX;
196 0 : }
197 :
198 0 : if( FD_UNLIKELY( ci_idx!=ULONG_MAX ) ) {
199 0 : switch( change_type ) {
200 0 : case FD_PING_TRACKER_CHANGE_TYPE_ACTIVE:
201 0 : fd_gossip_wsample_ping_tracked( ctx->wsample, ci_idx, 1 );
202 0 : break;
203 0 : case FD_PING_TRACKER_CHANGE_TYPE_INACTIVE:
204 0 : case FD_PING_TRACKER_CHANGE_TYPE_INACTIVE_STAKED:
205 0 : fd_gossip_wsample_ping_tracked( ctx->wsample, ci_idx, 0 );
206 0 : fd_active_set_remove_peer( ctx->active_set, ci_idx );
207 0 : break;
208 0 : default: FD_LOG_ERR(( "Unknown change type %d", change_type )); return;
209 0 : }
210 0 : }
211 :
212 0 : ctx->ping_tracker_change_fn( ctx->ping_tracker_change_fn_ctx, peer_pubkey, peer_address, now, change_type );
213 0 : }
214 :
215 0 : #define SIGN_PEND_SEND (0) /* a packet for peer */
216 0 : #define SIGN_PEND_VALUE (1) /* one of our CRDS values: insert and push */
217 0 : #define SIGN_PEND_CONTACT_INFO (2) /* our contact info: store, insert and push */
218 0 : #define SIGN_PEND_PULL_CONTACT_INFO (3) /* our contact info, for pull requests only */
219 :
220 0 : #define PULL_CONTACT_INFO_INTERVAL_NANOS (100L*1000L*1000L)
221 :
222 : static int
223 : sign_request( fd_gossip_t * gossip,
224 : int kind,
225 : fd_ip4_port_t const * peer,
226 : fd_gossip_value_t const * value,
227 : uchar const * data,
228 : ulong sz,
229 : ulong sig_off,
230 : uchar const * signable,
231 : ulong signable_sz,
232 0 : int sign_type ) {
233 0 : if( FD_UNLIKELY( gossip->sign_pend_cnt>=FD_GOSSIP_SIGN_PEND_MAX ) ) {
234 0 : gossip->metrics->sign_pend_full_cnt++;
235 0 : return 0;
236 0 : }
237 :
238 0 : ulong idx = (gossip->sign_pend_head+gossip->sign_pend_cnt)%FD_GOSSIP_SIGN_PEND_MAX;
239 0 : gossip->sign_pend_cnt++;
240 0 : gossip->sign_pend[ idx ].kind = (uchar)kind;
241 0 : gossip->sign_pend[ idx ].peer = peer ? *peer : (fd_ip4_port_t){0};
242 0 : gossip->sign_pend[ idx ].sz = (ushort)sz;
243 0 : gossip->sign_pend[ idx ].sig_off = (ushort)sig_off;
244 0 : fd_memcpy( gossip->sign_pend[ idx ].data, data, sz );
245 0 : if( FD_UNLIKELY( value ) ) *gossip->sign_pend[ idx ].value = *value;
246 :
247 0 : gossip->sign_fn( gossip->sign_ctx, signable, signable_sz, sign_type );
248 :
249 0 : return 1;
250 0 : }
251 :
252 :
253 : void
254 : fd_gossip_sign_response( fd_gossip_t * gossip,
255 : uchar const * signature,
256 : fd_stem_context_t * stem,
257 0 : long now ) {
258 0 : FD_TEST( gossip->sign_pend_cnt );
259 0 : ulong idx = gossip->sign_pend_head;
260 0 : gossip->sign_pend_head = (idx+1UL)%FD_GOSSIP_SIGN_PEND_MAX;
261 0 : gossip->sign_pend_cnt--;
262 :
263 0 : uchar * data = gossip->sign_pend[ idx ].data;
264 0 : ulong sz = gossip->sign_pend[ idx ].sz;
265 0 : fd_memcpy( data+gossip->sign_pend[ idx ].sig_off, signature, 64UL );
266 :
267 0 : switch( gossip->sign_pend[ idx ].kind ) {
268 0 : case SIGN_PEND_SEND: {
269 0 : gossip->send_fn( gossip->send_ctx, stem, data, sz, &gossip->sign_pend[ idx ].peer, (ulong)now );
270 0 : uint tag = FD_LOAD( uint, data );
271 0 : gossip->metrics->message_tx[ tag ]++;
272 0 : gossip->metrics->message_tx_bytes[ tag ] += sz+42UL; /* 42 = sizeof(fd_ip4_udp_hdrs_t) */
273 0 : break;
274 0 : }
275 0 : case SIGN_PEND_VALUE: {
276 0 : fd_gossip_value_t * value = gossip->sign_pend[ idx ].value;
277 0 : fd_memcpy( value->signature, signature, 64UL );
278 0 : int origin_active = 0; /* Value doesn't matter, since is_me=1 it's never used. */
279 0 : if( FD_UNLIKELY( fd_crds_insert( gossip->crds, value, data, sz, gossip->identity_stake, origin_active, 1, 0, now, stem, NULL ) ) ) break;
280 0 : fd_active_set_push( gossip->active_set, data, sz, gossip->identity_pubkey, gossip->identity_stake, stem, now, 1 );
281 0 : break;
282 0 : }
283 0 : case SIGN_PEND_CONTACT_INFO: {
284 : /* The contact info changed again since this was requested */
285 0 : if( FD_UNLIKELY( gossip->sign_pend[ idx ].gen!=gossip->my_contact_info.gen ) ) break;
286 :
287 0 : fd_gossip_value_t * value = gossip->sign_pend[ idx ].value;
288 0 : fd_memcpy( value->signature, signature, 64UL );
289 0 : fd_memcpy( gossip->my_contact_info.crds_val, data, sz );
290 0 : gossip->my_contact_info.crds_val_sz = sz;
291 0 : fd_memcpy( gossip->my_contact_info.pull_val, data, sz );
292 0 : gossip->my_contact_info.pull_val_sz = sz;
293 0 : int origin_active = 0; /* Value doesn't matter, since is_me=1 it's never used. */
294 0 : fd_crds_insert( gossip->crds, value, data, sz, gossip->identity_stake, origin_active, 1, 0, now, stem, NULL );
295 0 : fd_active_set_push( gossip->active_set, data, sz, gossip->identity_pubkey, gossip->identity_stake, stem, now, 1 );
296 0 : break;
297 0 : }
298 0 : case SIGN_PEND_PULL_CONTACT_INFO: {
299 0 : gossip->my_contact_info.pull_pending = 0;
300 0 : if( FD_UNLIKELY( gossip->sign_pend[ idx ].gen!=gossip->my_contact_info.gen ) ) break;
301 0 : fd_memcpy( gossip->my_contact_info.pull_val, data, sz );
302 0 : gossip->my_contact_info.pull_val_sz = sz;
303 0 : break;
304 0 : }
305 0 : default: FD_LOG_CRIT(( "unknown sign pend kind %u", gossip->sign_pend[ idx ].kind ));
306 0 : }
307 0 : }
308 :
309 : ulong
310 0 : fd_gossip_sign_pend_cnt( fd_gossip_t const * gossip ) {
311 0 : return gossip->sign_pend_cnt;
312 0 : }
313 :
314 : static inline void
315 0 : refresh_contact_info( fd_gossip_t * gossip ) {
316 0 : gossip->my_contact_info.gen++;
317 0 : gossip->my_contact_info.refresh_due = 1;
318 0 : }
319 :
320 : static void
321 : request_contact_info( fd_gossip_t * gossip,
322 0 : long now ) {
323 0 : fd_memcpy( gossip->my_contact_info.ci->origin, gossip->identity_pubkey, 32UL );
324 0 : gossip->my_contact_info.ci->wallclock = (ulong)FD_NANOSEC_TO_MILLI( now );
325 0 : uchar serialized[ FD_GOSSIP_VALUE_MAX_SZ ];
326 0 : long sz = fd_gossip_value_serialize( gossip->my_contact_info.ci, serialized, FD_GOSSIP_VALUE_MAX_SZ );
327 0 : FD_TEST( sz!=-1L );
328 :
329 0 : ulong idx = (gossip->sign_pend_head+gossip->sign_pend_cnt)%FD_GOSSIP_SIGN_PEND_MAX;
330 0 : if( FD_UNLIKELY( !sign_request( gossip, SIGN_PEND_CONTACT_INFO, NULL, gossip->my_contact_info.ci, serialized, (ulong)sz, 0UL, serialized+64UL, (ulong)sz-64UL, FD_KEYGUARD_SIGN_TYPE_ED25519 ) ) ) {
331 0 : gossip->my_contact_info.refresh_due = 1; /* queue full: retry next advance */
332 0 : return;
333 0 : }
334 0 : gossip->sign_pend[ idx ].gen = gossip->my_contact_info.gen;
335 0 : }
336 :
337 : static void
338 : request_pull_contact_info( fd_gossip_t * gossip,
339 0 : long now ) {
340 0 : fd_gossip_value_t ci[1] = { *gossip->my_contact_info.ci };
341 0 : fd_memcpy( ci->origin, gossip->identity_pubkey, 32UL );
342 0 : ci->wallclock = (ulong)FD_NANOSEC_TO_MILLI( now );
343 0 : uchar serialized[ FD_GOSSIP_VALUE_MAX_SZ ];
344 0 : long sz = fd_gossip_value_serialize( ci, serialized, FD_GOSSIP_VALUE_MAX_SZ );
345 0 : FD_TEST( sz!=-1L );
346 :
347 0 : ulong idx = (gossip->sign_pend_head+gossip->sign_pend_cnt)%FD_GOSSIP_SIGN_PEND_MAX;
348 0 : if( FD_UNLIKELY( !sign_request( gossip, SIGN_PEND_PULL_CONTACT_INFO, NULL, NULL, serialized, (ulong)sz, 0UL, serialized+64UL, (ulong)sz-64UL, FD_KEYGUARD_SIGN_TYPE_ED25519 ) ) ) return;
349 0 : gossip->sign_pend[ idx ].gen = gossip->my_contact_info.gen;
350 0 : gossip->my_contact_info.pull_pending = 1;
351 0 : }
352 :
353 : void *
354 : fd_gossip_new( void * shmem,
355 : fd_rng_t * rng,
356 : uchar const ping_seed[ static 32 ],
357 : ulong max_values,
358 : ulong entrypoints_len,
359 : fd_ip4_port_t const * entrypoints,
360 : uchar const * identity_pubkey,
361 : fd_gossip_contact_info_t const * my_contact_info,
362 : long now,
363 : fd_gossip_send_fn send_fn,
364 : void * send_ctx,
365 : fd_gossip_sign_fn sign_fn,
366 : void * sign_ctx,
367 : fd_ping_tracker_change_fn ping_tracker_change_fn,
368 : void * ping_tracker_change_fn_ctx,
369 : fd_gossip_activity_update_fn activity_update_fn,
370 : void * activity_update_fn_ctx,
371 : fd_gossip_out_ctx_t * gossip_update_out,
372 0 : fd_gossip_out_ctx_t * gossip_net_out ) {
373 0 : if( FD_UNLIKELY( !shmem ) ) {
374 0 : FD_LOG_WARNING(( "NULL shmem" ));
375 0 : return NULL;
376 0 : }
377 :
378 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shmem, fd_gossip_align() ) ) ) {
379 0 : FD_LOG_WARNING(( "misaligned shmem" ));
380 0 : return NULL;
381 0 : }
382 :
383 0 : if( FD_UNLIKELY( entrypoints_len>16UL ) ) {
384 0 : FD_LOG_WARNING(( "entrypoints_cnt must be in [0, 16]" ));
385 0 : return NULL;
386 0 : }
387 :
388 0 : if( FD_UNLIKELY( !fd_ulong_is_pow2( max_values ) ) ) {
389 0 : FD_LOG_WARNING(( "max_values must be a power of 2" ));
390 0 : return NULL;
391 0 : }
392 :
393 0 : FD_SCRATCH_ALLOC_INIT( l, shmem );
394 0 : fd_gossip_t * gossip = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_gossip_t), sizeof(fd_gossip_t) );
395 0 : void * purged = FD_SCRATCH_ALLOC_APPEND( l, fd_gossip_purged_align(), fd_gossip_purged_footprint( max_values ) );
396 0 : void * wsample = FD_SCRATCH_ALLOC_APPEND( l, fd_gossip_wsample_align(), fd_gossip_wsample_footprint( FD_CONTACT_INFO_TABLE_SIZE ) );
397 0 : void * crds = FD_SCRATCH_ALLOC_APPEND( l, fd_crds_align(), fd_crds_footprint( max_values ) );
398 0 : void * active_set = FD_SCRATCH_ALLOC_APPEND( l, fd_active_set_align(), fd_active_set_footprint() );
399 0 : void * ping_tracker = FD_SCRATCH_ALLOC_APPEND( l, fd_ping_tracker_align(), fd_ping_tracker_footprint( entrypoints_len ) );
400 0 : void * prune_finder = FD_SCRATCH_ALLOC_APPEND( l, fd_prune_finder_align(), fd_prune_finder_footprint() );
401 0 : void * stake_pool = FD_SCRATCH_ALLOC_APPEND( l, stake_pool_align(), stake_pool_footprint( MAX_SHRED_DESTS ) );
402 0 : void * stake_weights = FD_SCRATCH_ALLOC_APPEND( l, stake_map_align(), stake_map_footprint( stake_map_chain_cnt_est( MAX_SHRED_DESTS ) ) );
403 :
404 0 : gossip->gossip_net_out = gossip_net_out;
405 :
406 0 : gossip->entrypoints_cnt = entrypoints_len;
407 0 : fd_memcpy( gossip->entrypoints, entrypoints, entrypoints_len*sizeof(fd_ip4_port_t) );
408 :
409 0 : gossip->purged = fd_gossip_purged_join( fd_gossip_purged_new( purged, rng, max_values ) );
410 0 : FD_TEST( gossip->purged );
411 :
412 0 : gossip->wsample = fd_gossip_wsample_join( fd_gossip_wsample_new( wsample, rng, FD_CONTACT_INFO_TABLE_SIZE ) );
413 0 : FD_TEST( gossip->wsample );
414 :
415 0 : gossip->crds = fd_crds_join( fd_crds_new( crds, entrypoints, entrypoints_len, gossip->wsample, active_set, rng, max_values, gossip->purged, activity_update_fn, activity_update_fn_ctx, gossip_update_out ) );
416 0 : FD_TEST( gossip->crds );
417 :
418 0 : gossip->active_set = fd_active_set_join( fd_active_set_new( active_set, gossip->wsample, gossip->crds, rng, identity_pubkey, 0UL, send_fn, send_ctx ) );
419 0 : FD_TEST( gossip->active_set );
420 :
421 0 : gossip->ping_tracker = fd_ping_tracker_join( fd_ping_tracker_new( ping_tracker, ping_seed, gossip->entrypoints_cnt, gossip->entrypoints, ping_tracker_change, gossip ) );
422 0 : FD_TEST( gossip->ping_tracker );
423 :
424 0 : gossip->prune_finder = fd_prune_finder_join( fd_prune_finder_new( prune_finder, fd_rng_ulong( rng ) ) );
425 0 : FD_TEST( gossip->prune_finder );
426 :
427 0 : gossip->stake.count = 0UL;
428 0 : gossip->stake.pool = stake_pool_join( stake_pool_new( stake_pool, MAX_SHRED_DESTS ) );
429 0 : FD_TEST( gossip->stake.pool );
430 :
431 0 : gossip->stake.map = stake_map_join( stake_map_new( stake_weights, stake_map_chain_cnt_est( MAX_SHRED_DESTS ), fd_rng_ulong( rng ) ) );
432 0 : FD_TEST( gossip->stake.map );
433 :
434 0 : FD_TEST( fd_sha256_join( fd_sha256_new( gossip->sha256 ) ) );
435 0 : FD_TEST( fd_sha512_join( fd_sha512_new( gossip->sha512 ) ) );
436 :
437 0 : gossip->rng = rng;
438 :
439 0 : gossip->timers.next_pull_request = 0L;
440 0 : gossip->timers.next_active_set_refresh = 0L;
441 0 : gossip->timers.next_contact_info_refresh = 0L;
442 0 : gossip->timers.next_flush_push_state = 0L;
443 0 : gossip->timers.next_pull_contact_info = 0L;
444 :
445 0 : gossip->outbound_budget.remaining = 0UL;
446 0 : gossip->outbound_budget.last_replenish_nanos = now;
447 :
448 0 : gossip->sign_pend_head = 0UL;
449 0 : gossip->sign_pend_cnt = 0UL;
450 0 : gossip->my_contact_info.crds_val_sz = 0UL;
451 0 : gossip->my_contact_info.gen = 0UL;
452 0 : gossip->my_contact_info.refresh_due = 0;
453 0 : gossip->my_contact_info.pull_val_sz = 0UL;
454 0 : gossip->my_contact_info.pull_pending = 0;
455 :
456 0 : gossip->send_fn = send_fn;
457 0 : gossip->send_ctx = send_ctx;
458 0 : gossip->sign_fn = sign_fn;
459 0 : gossip->sign_ctx = sign_ctx;
460 0 : gossip->ping_tracker_change_fn = ping_tracker_change_fn;
461 0 : gossip->ping_tracker_change_fn_ctx = ping_tracker_change_fn_ctx;
462 :
463 0 : gossip->my_contact_info.ci->tag = FD_GOSSIP_VALUE_CONTACT_INFO;
464 0 : *gossip->my_contact_info.ci->contact_info = *my_contact_info;
465 0 : fd_memcpy( gossip->identity_pubkey, identity_pubkey, 32UL );
466 0 : gossip->identity_stake = 0UL;
467 0 : refresh_contact_info( gossip );
468 :
469 0 : fd_memset( gossip->metrics, 0, sizeof(fd_gossip_metrics_t) );
470 :
471 0 : return gossip;
472 0 : }
473 :
474 : fd_gossip_t *
475 0 : fd_gossip_join( void * shgossip ) {
476 0 : if( FD_UNLIKELY( !shgossip ) ) {
477 0 : FD_LOG_WARNING(( "NULL shgossip" ));
478 0 : return NULL;
479 0 : }
480 :
481 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shgossip, fd_gossip_align() ) ) ) {
482 0 : FD_LOG_WARNING(( "misaligned shgossip" ));
483 0 : return NULL;
484 0 : }
485 :
486 0 : return (fd_gossip_t *)shgossip;
487 0 : }
488 :
489 : fd_gossip_metrics_t const *
490 0 : fd_gossip_metrics( fd_gossip_t const * gossip ) {
491 0 : return gossip->metrics;
492 0 : }
493 :
494 : fd_crds_metrics_t const *
495 0 : fd_gossip_crds_metrics( fd_gossip_t const * gossip ) {
496 0 : return fd_crds_metrics( gossip->crds );
497 0 : }
498 :
499 : fd_ping_tracker_metrics_t const *
500 0 : fd_gossip_ping_tracker_metrics( fd_gossip_t const * gossip ) {
501 0 : return fd_ping_tracker_metrics( gossip->ping_tracker );
502 0 : }
503 :
504 : fd_gossip_purged_metrics_t const *
505 0 : fd_gossip_purged_metrics2( fd_gossip_t const * gossip ) {
506 0 : return fd_gossip_purged_metrics( gossip->purged );
507 0 : }
508 :
509 : fd_active_set_metrics_t const *
510 0 : fd_gossip_active_set_metrics2( fd_gossip_t const * gossip ) {
511 0 : return fd_active_set_metrics( gossip->active_set );
512 0 : }
513 :
514 : static fd_ip4_port_t
515 0 : random_entrypoint( fd_gossip_t const * gossip ) {
516 0 : ulong idx = fd_rng_ulong_roll( gossip->rng, gossip->entrypoints_cnt );
517 0 : return gossip->entrypoints[ idx ];
518 0 : }
519 :
520 : static ulong
521 : get_stake( fd_gossip_t const * gossip,
522 0 : uchar const * pubkey ) {
523 0 : stake_t const * entry = stake_map_ele_query_const( gossip->stake.map, (fd_pubkey_t const *)pubkey, NULL, gossip->stake.pool );
524 0 : if( FD_UNLIKELY( !entry ) ) return 0UL;
525 0 : return entry->stake;
526 0 : }
527 :
528 : void
529 : fd_gossip_set_identity( fd_gossip_t * gossip,
530 : uchar const * identity_pubkey,
531 : long now,
532 0 : ulong identity_outset ) {
533 0 : int identity_changed = memcmp( gossip->identity_pubkey, identity_pubkey, 32UL );
534 0 : if( FD_UNLIKELY( !identity_changed ) ) return;
535 :
536 0 : ulong new_ci_idx = fd_crds_ci_idx( gossip->crds, identity_pubkey );
537 :
538 : /* The new identity may already exist in CRDS as a normal peer (active
539 : in the wsample and potentially present in the active set). We
540 : must deactivate it before updating identity_pubkey to maintain the
541 : invariant that our own identity is never sampleable. */
542 0 : if( FD_UNLIKELY( new_ci_idx!=ULONG_MAX ) ) fd_active_set_remove_peer( gossip->active_set, new_ci_idx );
543 :
544 : /* Also remove the new identity from the ping tracker (we don't
545 : track ourselves). */
546 0 : fd_ping_tracker_remove( gossip->ping_tracker, identity_pubkey, now );
547 :
548 0 : fd_memcpy( gossip->identity_pubkey, identity_pubkey, 32UL );
549 0 : gossip->identity_stake = get_stake( gossip, identity_pubkey );
550 0 : fd_gossip_wsample_set_identity( gossip->wsample, new_ci_idx );
551 0 : fd_gossip_wsample_self_stake( gossip->wsample, gossip->identity_stake );
552 0 : fd_active_set_set_identity( gossip->active_set, gossip->identity_pubkey, gossip->identity_stake );
553 0 : fd_prune_finder_set_identity( gossip->prune_finder, gossip->identity_pubkey, gossip->identity_stake );
554 :
555 : /* For identity swaps, refresh the contact info outset so this
556 : instance can override older contact info for the same identity. */
557 0 : gossip->my_contact_info.ci->contact_info->outset = identity_outset;
558 0 : refresh_contact_info( gossip );
559 0 : }
560 :
561 : void
562 : fd_gossip_set_shred_version( fd_gossip_t * gossip,
563 0 : ushort shred_version ) {
564 0 : gossip->my_contact_info.ci->contact_info->shred_version = shred_version;
565 0 : refresh_contact_info( gossip );
566 0 : }
567 :
568 : void
569 : fd_gossip_stakes_update( fd_gossip_t * gossip,
570 : fd_stake_weight_t const * stake_weights,
571 0 : ulong stake_weights_cnt ) {
572 0 : for( ulong i=0UL; i<stake_weights_cnt; i++ ) {
573 0 : if( FD_LIKELY( stake_weights[ i ].stake<FD_GOSSIP_STAKED_THRESHOLD ) ) continue;
574 0 : if( FD_UNLIKELY( get_stake( gossip, stake_weights[ i ].key.uc )<FD_GOSSIP_STAKED_THRESHOLD ) )
575 0 : fd_gossip_purged_drain_no_contact_info( gossip->purged, stake_weights[ i ].key.uc );
576 0 : }
577 :
578 0 : stake_map_reset( gossip->stake.map );
579 0 : stake_pool_reset( gossip->stake.pool );
580 :
581 0 : for( ulong i=0UL; i<stake_weights_cnt; i++ ) {
582 0 : stake_t * entry = stake_pool_ele_acquire( gossip->stake.pool );
583 0 : entry->pubkey = stake_weights[i].key;
584 0 : entry->stake = stake_weights[i].stake;
585 0 : stake_map_ele_insert( gossip->stake.map, entry, gossip->stake.pool );
586 0 : }
587 :
588 0 : gossip->identity_stake = get_stake( gossip, gossip->identity_pubkey );
589 0 : fd_gossip_wsample_self_stake( gossip->wsample, gossip->identity_stake );
590 0 : fd_active_set_set_identity( gossip->active_set, gossip->identity_pubkey, gossip->identity_stake );
591 0 : fd_prune_finder_set_identity( gossip->prune_finder, gossip->identity_pubkey, gossip->identity_stake );
592 0 : gossip->stake.count = stake_pool_used( gossip->stake.pool );
593 0 : }
594 :
595 : /* Outbound data budget constants (matching Agave's DataBudget for gossip).
596 : Budget is replenished every BUDGET_REPLENISH_INTERVAL_NS with
597 : num_staked * BUDGET_BYTES_PER_INTERVAL bytes, capped at
598 : BUDGET_MAX_MULTIPLE * num_staked * BUDGET_BYTES_PER_INTERVAL. */
599 :
600 : #define BUDGET_REPLENISH_INTERVAL_NS (100L*1000L*1000L) /* 100 ms */
601 0 : #define BUDGET_BYTES_PER_INTERVAL (1024UL) /* per staked validator */
602 0 : #define BUDGET_MAX_MULTIPLE (5UL) /* max accumulation */
603 0 : #define BUDGET_MIN_STAKED (2UL) /* floor for num_staked */
604 :
605 : /* Lazily replenish the outbound pull-response budget if at least
606 : BUDGET_REPLENISH_INTERVAL_NS have elapsed since last replenish.
607 : Returns current remaining budget in bytes. */
608 :
609 : static inline ulong
610 : outbound_budget_replenish( fd_gossip_t * gossip,
611 0 : long now ) {
612 0 : long elapsed = now-gossip->outbound_budget.last_replenish_nanos;
613 :
614 0 : if( FD_LIKELY( elapsed>=BUDGET_REPLENISH_INTERVAL_NS ) ) {
615 0 : ulong num_staked = fd_ulong_max( gossip->stake.count, BUDGET_MIN_STAKED );
616 0 : ulong increment = num_staked * BUDGET_BYTES_PER_INTERVAL;
617 0 : ulong cap = BUDGET_MAX_MULTIPLE * increment;
618 0 : ulong remaining = gossip->outbound_budget.remaining + increment;
619 0 : gossip->outbound_budget.remaining = fd_ulong_min( remaining, cap );
620 0 : gossip->outbound_budget.last_replenish_nanos = now;
621 0 : }
622 0 : return gossip->outbound_budget.remaining;
623 0 : }
624 :
625 : static inline void
626 : txbuild_flush( fd_gossip_t * gossip,
627 : fd_gossip_txbuild_t * txbuild,
628 : fd_stem_context_t * stem,
629 : fd_ip4_port_t dest_addr,
630 0 : long now ) {
631 0 : if( FD_UNLIKELY( !txbuild->crds_len ) ) return;
632 :
633 : /* Debit the outbound data budget (gossip payload bytes only, not
634 : including IP/UDP headers — matching Agave's DataBudget which
635 : operates on serialized gossip-layer packet sizes). */
636 0 : gossip->outbound_budget.remaining -= fd_ulong_min( txbuild->bytes_len, gossip->outbound_budget.remaining );
637 :
638 0 : gossip->send_fn( gossip->send_ctx, stem, txbuild->bytes, txbuild->bytes_len, &dest_addr, (ulong)now );
639 :
640 0 : gossip->metrics->message_tx[ txbuild->tag ]++;
641 0 : gossip->metrics->message_tx_bytes[ txbuild->tag ] += txbuild->bytes_len+42UL; /* 42 = sizeof(fd_ip4_udp_hdrs_t) */
642 0 : for( ulong i=0UL; i<txbuild->crds_len; i++ ) {
643 0 : gossip->metrics->crds_tx_pull_response[ txbuild->crds[ i ].tag ]++;
644 0 : gossip->metrics->crds_tx_pull_response_bytes[ txbuild->crds[ i ].tag ] += txbuild->crds[ i ].sz;
645 0 : }
646 :
647 0 : fd_gossip_txbuild_init( txbuild, gossip->identity_pubkey, txbuild->tag );
648 0 : }
649 :
650 : /* pull_scan_range iterates CRDS entries in [start_hash, end_hash] and
651 : appends values the caller is missing to pull_resp. Returns 1 if a
652 : budget was exhausted (iteration or outbound data), 0 if the range was
653 : fully scanned. */
654 :
655 : static int
656 : pull_scan_range( fd_gossip_t * gossip,
657 : ulong start_hash,
658 : ulong end_hash,
659 : fd_bloom_t * filter,
660 : ulong adjusted_wallclock_ms,
661 : fd_gossip_txbuild_t * pull_resp,
662 : fd_stem_context_t * stem,
663 : fd_ip4_port_t peer_addr,
664 0 : long now ) {
665 : /* The hset holds the value hashes densely, so the bloom filter is
666 : checked from the scan itself and the (large, cold) CRDS entry is
667 : only touched for the values the caller is missing. Entries come
668 : out in bucket order rather than sorted by prefix; the protocol
669 : does not fix the order of a pull response, and a range covers a
670 : few buckets at most. */
671 0 : fd_gossip_hset_t const * hset = fd_crds_hset( gossip->crds );
672 0 : fd_gossip_hset_iter_t it[1];
673 0 : for( fd_gossip_hset_iter_init( it, hset, start_hash, end_hash );
674 0 : !fd_gossip_hset_iter_done( it );
675 0 : fd_gossip_hset_iter_next( it, hset ) ) {
676 0 : uchar const * hashes = fd_gossip_hset_iter_hashes( it, hset );
677 0 : uint lanes = fd_gossip_hset_iter_lanes( it, hset );
678 0 : if( FD_UNLIKELY( !lanes ) ) continue;
679 : /* The chunk is hashed up front; the budget is still charged and
680 : the lane inspected one at a time, so an exhausted budget returns
681 : at the same lane as before. */
682 0 : uint hits = fd_bloom_contains8( filter, hashes );
683 0 : for( ; lanes; lanes = fd_uint_pop_lsb( lanes ) ) {
684 0 : ulong lane = (ulong)fd_uint_find_lsb( lanes );
685 :
686 0 : if( FD_UNLIKELY( !gossip->scan_budget.remaining ) ) return 1;
687 0 : gossip->scan_budget.remaining--;
688 :
689 0 : if( FD_LIKELY( (hits>>lane)&1U ) ) continue;
690 :
691 0 : fd_crds_entry_t const * candidate = fd_crds_entry_at( gossip->crds, fd_gossip_hset_iter_owner( it, hset, lane ) );
692 :
693 0 : if( FD_UNLIKELY( fd_crds_entry_wallclock( candidate )>adjusted_wallclock_ms ) ) continue;
694 :
695 0 : uchar const * crds_val;
696 0 : ulong crds_size;
697 0 : fd_crds_entry_value( candidate, &crds_val, &crds_size );
698 0 : if( FD_UNLIKELY( !fd_gossip_txbuild_can_fit( pull_resp, crds_size ) ) ) txbuild_flush( gossip, pull_resp, stem, peer_addr, now );
699 0 : fd_gossip_txbuild_append( pull_resp, crds_size, crds_val );
700 :
701 0 : if( FD_UNLIKELY( !gossip->outbound_budget.remaining ) ) return 1;
702 0 : }
703 0 : }
704 0 : return 0;
705 0 : }
706 :
707 : static void
708 : rx_pull_request( fd_gossip_t * gossip,
709 : fd_gossip_pull_request_t const * pr_view,
710 : fd_ip4_port_t peer_addr,
711 : fd_stem_context_t * stem,
712 0 : long now ) {
713 0 : fd_gossip_ping_tracker_track( gossip, pr_view->contact_info->origin, peer_addr, now );
714 :
715 : /* Replenish and check outbound data budget. If the budget is
716 : exhausted, skip generating pull responses entirely. */
717 0 : if( FD_UNLIKELY( !outbound_budget_replenish( gossip, now ) ) ) return;
718 :
719 : /* When responding to a pull request, we skip CRDS entries whose
720 : wallclock is newer than the caller's wallclock + a random jitter.
721 : The jitter is drawn uniformly from [0, TIMEOUT/4) ms, matching
722 : Agave's behavior (CRDS_GOSSIP_PULL_CRDS_TIMEOUT_MS = 15000ms).
723 : This prevents all responders from consistently excluding the same
724 : set of very-recent CRDS values. */
725 0 : #define FD_GOSSIP_PULL_JITTER_BOUND_MS (15000UL/4UL)
726 :
727 : /* Generate a random jitter in [0, 3750) ms, added to the caller's
728 : wallclock. CRDS entries newer than this adjusted threshold are
729 : excluded from the response. The jitter prevents all responders
730 : from consistently excluding the same near-boundary entries,
731 : improving cluster-wide convergence of recent values. */
732 0 : ulong caller_wallclock_ms = pr_view->contact_info->wallclock;
733 0 : ulong jitter_ms = fd_rng_ulong_roll( gossip->rng, FD_GOSSIP_PULL_JITTER_BOUND_MS );
734 0 : ulong adjusted_wallclock_ms = caller_wallclock_ms + jitter_ms;
735 :
736 0 : ulong keys[ sizeof(pr_view->crds_filter->filter->keys)/sizeof(ulong) ];
737 0 : ulong bits[ sizeof(pr_view->crds_filter->filter->bits)/sizeof(ulong) ];
738 0 : fd_memcpy( keys, pr_view->crds_filter->filter->keys, sizeof(pr_view->crds_filter->filter->keys) );
739 0 : fd_memcpy( bits, pr_view->crds_filter->filter->bits, sizeof(pr_view->crds_filter->filter->bits) );
740 :
741 0 : fd_bloom_t filter[1];
742 0 : filter->keys_len = pr_view->crds_filter->filter->keys_len;
743 0 : filter->keys = keys;
744 :
745 0 : filter->bits_len = pr_view->crds_filter->filter->bits_len;
746 0 : filter->bits = bits;
747 :
748 0 : fd_gossip_txbuild_t pull_resp[1];
749 0 : fd_gossip_txbuild_init( pull_resp, gossip->identity_pubkey, FD_GOSSIP_MESSAGE_PULL_RESPONSE );
750 :
751 : /* CPU budget for the CRDS hash scan. An honest sender picks
752 :
753 : mask_bits = ceil(log2(num_items / max_items))
754 : num_items >= MIN_NUM_BLOOM_ITEMS (65536)
755 :
756 : max_items is computed by fd_bloom_max_items as:
757 :
758 : max_items = ceil( max_bits / D(K) )
759 : D(K) = -K / ln(1 - exp(ln(p)/K)), p=0.1
760 :
761 : where K is the number of bloom hash functions. D(K) is minimized
762 : at K=3 where D(3)=4.808. max_bits is bounded by the bloom struct's
763 : bits[] array which holds at most 151 u64 words = 9664 bits (see
764 : fd_gossip_message.h). Therefore,
765 :
766 : max_items <= ceil(9664/4.808) = 2010 < 2048.
767 :
768 : The sender's mask_bits partitions the hash space into 2^mask_bits
769 : partitions, each covering ~max_items of the sender's entries.
770 : What's the worst case iteration count for the receiver given an
771 : honest sender?
772 :
773 : 2^mask_bits >= MIN_NUM_BLOOM_ITEMS / max_items
774 : iterations <= crds_len / 2^mask_bits
775 : iterations <= crds_len * max_items / MIN_NUM_BLOOM_ITEMS
776 : <= crds_len * 2048 / 65536
777 : <= crds_len / 32
778 :
779 : A floor of 1024 handles small tables where the budget
780 : would otherwise round to near-zero. */
781 :
782 0 : gossip->scan_budget.remaining = fd_ulong_max( fd_crds_len( gossip->crds ) / 32UL, 1024UL );
783 :
784 : /* Compute the hash range [start_hash, end_hash] for this mask
785 : partition, then pick a random starting point within it.
786 : We iterate [mid_hash, end_hash] then [start_hash, mid_hash)
787 : so that when the iteration budget truncates the scan, different
788 : entries are skipped each time rather than always penalizing the
789 : tail of the partition. */
790 :
791 0 : ulong start_hash, end_hash;
792 0 : fd_gossip_purged_generate_masks( pr_view->crds_filter->mask, pr_view->crds_filter->mask_bits, &start_hash, &end_hash );
793 0 : ulong range = end_hash - start_hash;
794 0 : ulong mid_hash = start_hash + ( range==ULONG_MAX ? fd_rng_ulong( gossip->rng ) : fd_rng_ulong( gossip->rng ) % (range + 1UL) );
795 :
796 0 : int exhausted = pull_scan_range( gossip, mid_hash, end_hash, filter, adjusted_wallclock_ms, pull_resp, stem, peer_addr, now );
797 0 : if( FD_LIKELY( !exhausted && mid_hash>start_hash ) ) pull_scan_range( gossip, start_hash, mid_hash-1UL, filter, adjusted_wallclock_ms, pull_resp, stem, peer_addr, now );
798 :
799 0 : txbuild_flush( gossip, pull_resp, stem, peer_addr, now );
800 0 : }
801 :
802 : static void
803 : rx_values( fd_gossip_t * gossip,
804 : ulong values_len,
805 : fd_gossip_value_t const * values,
806 : uchar const * payload,
807 : uchar const * failed,
808 : fd_stem_context_t * stem,
809 : int is_push,
810 : long now,
811 0 : long results[ static 17UL ] ) {
812 0 : for( ulong i=0UL; i<values_len; i++ ) {
813 0 : fd_gossip_value_t const * value = &values[ i ];
814 :
815 0 : if( FD_UNLIKELY( failed[ i ] && failed[ i ]!=FD_GOSSIP_FAILED_DUPLICATE ) ) {
816 0 : uchar candidate_hash[ 32UL ];
817 0 : fd_sha256_hash( payload+value->offset, value->length, candidate_hash );
818 0 : if( FD_LIKELY( failed[ i ]==FD_GOSSIP_FAILED_NO_CONTACT_INFO ) ) fd_gossip_purged_insert_no_contact_info( gossip->purged, value->origin, candidate_hash, now );
819 0 : else fd_gossip_purged_insert_failed_insert( gossip->purged, candidate_hash, now );
820 0 : continue;
821 0 : }
822 :
823 0 : ulong origin_stake = get_stake( gossip, value->origin );
824 0 : int is_me = !memcmp( value->origin, gossip->identity_pubkey, 32UL );
825 :
826 0 : int origin_ping_tracker_active = 0;
827 0 : fd_ip4_port_t origin_addr = { 0 };
828 0 : fd_ip4_port_t prev_addr = { 0 };
829 0 : if( FD_UNLIKELY( value->tag==FD_GOSSIP_VALUE_CONTACT_INFO ) ) {
830 0 : origin_addr = (fd_ip4_port_t){
831 0 : .addr = value->contact_info->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ].is_ipv6 ? 0U : value->contact_info->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ].ip4,
832 0 : .port = value->contact_info->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ].port
833 0 : };
834 0 : origin_ping_tracker_active = fd_ping_tracker_active( gossip->ping_tracker, value->origin, origin_addr );
835 0 : ulong prev_ci_idx = fd_crds_ci_idx( gossip->crds, value->origin );
836 0 : if( FD_UNLIKELY( prev_ci_idx!=ULONG_MAX ) ) {
837 0 : fd_gossip_contact_info_t const * prev = fd_crds_ci( gossip->crds, prev_ci_idx );
838 0 : prev_addr.addr = prev->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ].is_ipv6 ? 0U : prev->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ].ip4;
839 0 : prev_addr.port = prev->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ].port;
840 0 : }
841 0 : }
842 :
843 0 : int relay;
844 0 : results[ i ] = fd_crds_insert( gossip->crds, value, payload+value->offset, value->length, origin_stake, origin_ping_tracker_active, is_me, is_push, now, stem, &relay );
845 0 : if( FD_UNLIKELY( results[ i ] ) ) continue;
846 :
847 0 : if( FD_UNLIKELY( value->tag==FD_GOSSIP_VALUE_CONTACT_INFO ) ) {
848 0 : if( FD_LIKELY( !is_me ) ) fd_ping_tracker_track( gossip->ping_tracker, value->origin, origin_stake, origin_addr, now );
849 :
850 : /* Moved to a socket that has not ponged: stop pushing to it. */
851 0 : if( FD_UNLIKELY( prev_addr.addr && prev_addr.l!=origin_addr.l && !origin_ping_tracker_active && origin_stake<FD_GOSSIP_STAKED_THRESHOLD ) ) {
852 0 : fd_active_set_remove_peer( gossip->active_set, fd_crds_ci_idx( gossip->crds, value->origin ) );
853 0 : }
854 :
855 : /* We just learned this peer's contact info. Drain any
856 : no_contact_info hashes associated with this origin from the
857 : purged set so peers re-send those CRDS values. */
858 0 : if( FD_LIKELY( fd_ping_tracker_active( gossip->ping_tracker, value->origin, origin_addr ) ) ) fd_gossip_purged_drain_no_contact_info( gossip->purged, value->origin );
859 0 : }
860 :
861 0 : if( FD_UNLIKELY( !relay ) ) continue;
862 0 : fd_active_set_push( gossip->active_set, payload+value->offset, value->length, value->origin, origin_stake, stem, now, 0 );
863 0 : }
864 0 : }
865 :
866 : static void
867 : rx_pull_response( fd_gossip_t * gossip,
868 : fd_gossip_pull_response_t const * pull_response,
869 : uchar const * payload,
870 : uchar const * failed,
871 : fd_stem_context_t * stem,
872 0 : long now ) {
873 0 : long results[ 17UL ];
874 0 : rx_values( gossip, pull_response->values_len, pull_response->values, payload, failed, stem, 0, now, results );
875 0 : for( ulong i=0UL; i<pull_response->values_len; i++ ) {
876 0 : if( FD_UNLIKELY( failed[ i ] && failed[ i ]!=FD_GOSSIP_FAILED_DUPLICATE ) ) continue;
877 0 : if( FD_LIKELY( !results[ i ] ) ) gossip->metrics->crds_rx_count[ FD_METRICS_ENUM_GOSSIP_CRDS_RESULT_V_INSERTED_PULL_RESPONSE_IDX ]++;
878 0 : else if( results[ i ]<0L ) gossip->metrics->crds_rx_count[ FD_METRICS_ENUM_GOSSIP_CRDS_RESULT_V_STALE_PULL_RESPONSE_IDX ]++;
879 0 : else gossip->metrics->crds_rx_count[ FD_METRICS_ENUM_GOSSIP_CRDS_RESULT_V_DUPLICATE_PULL_RESPONSE_IDX ]++;
880 0 : }
881 0 : }
882 :
883 : /* tx_prune constructs, signs, and sends a prune message telling
884 : `relayer` to stop pushing CRDS values originating from `origin`.
885 :
886 : On-wire layout (bincode):
887 : Protocol tag 4 (FD_GOSSIP_MESSAGE_PRUNE = 3)
888 : sender pubkey 32 (= identity_pubkey, outer PruneMessage field)
889 : PruneData.pubkey 32 (= identity_pubkey)
890 : prunes_len 8
891 : prunes[1] 32
892 : signature 64
893 : destination 32
894 : wallclock 8
895 :
896 : The signable data (input to Ed25519 sign) is the PruneData fields
897 : excluding signature:
898 : prefix[26] + pubkey[32] + prunes_len[8] + prunes[32] + destination[32] + wallclock[8]
899 : This must match fd_keyguard_payload_matches_prune_data (106 + 32 bytes). */
900 :
901 : static void
902 : tx_prune( fd_gossip_t * gossip,
903 : uchar const * relayer,
904 : uchar const * origin,
905 0 : long now ) {
906 0 : ulong ci_idx = fd_crds_ci_idx( gossip->crds, relayer );
907 0 : if( FD_UNLIKELY( ci_idx==ULONG_MAX ) ) return;
908 :
909 0 : fd_gossip_contact_info_t const * ci = fd_crds_ci( gossip->crds, ci_idx );
910 0 : fd_ip4_port_t dest_addr = {
911 0 : .addr = ci->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ].is_ipv6 ? 0U : ci->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ].ip4,
912 0 : .port = ci->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ].port
913 0 : };
914 0 : if( FD_UNLIKELY( !dest_addr.addr || !dest_addr.port ) ) return;
915 :
916 0 : ulong wallclock = (ulong)FD_NANOSEC_TO_MILLI( now );
917 :
918 : /* Build the signable payload:
919 : prefix[26] + pubkey[32] + prunes_len[8] + prunes[32] + destination[32] + wallclock[8] */
920 0 : uchar signable[ 26UL + 32UL + 8UL + 32UL + 32UL + 8UL ];
921 0 : uchar * p = signable;
922 0 : FD_STORE( ulong, p, 18UL ); p += 8UL;
923 0 : fd_memcpy( p, "\xffSOLANA_PRUNE_DATA", 18UL ); p += 18UL;
924 0 : fd_memcpy( p, gossip->identity_pubkey, 32UL ); p += 32UL;
925 0 : FD_STORE( ulong, p, 1UL ); p += 8UL;
926 0 : fd_memcpy( p, origin, 32UL ); p += 32UL;
927 0 : fd_memcpy( p, relayer, 32UL ); p += 32UL;
928 0 : FD_STORE( ulong, p, wallclock ); p += 8UL;
929 :
930 : /* Build the on-wire packet, signature filled in on the response:
931 : tag(4) + sender(32) + pubkey(32) + prunes_len(8) + prunes[32]
932 : + signature(64) + destination(32) + wallclock(8) */
933 0 : uchar pkt[ 4UL + 32UL + 32UL + 8UL + 32UL + 64UL + 32UL + 8UL ];
934 0 : uchar * q = pkt;
935 0 : FD_STORE( uint, q, FD_GOSSIP_MESSAGE_PRUNE ); q += 4UL;
936 0 : fd_memcpy( q, gossip->identity_pubkey, 32UL ); q += 32UL; /* sender */
937 0 : fd_memcpy( q, gossip->identity_pubkey, 32UL ); q += 32UL; /* PruneData.pubkey */
938 0 : FD_STORE( ulong, q, 1UL ); q += 8UL;
939 0 : fd_memcpy( q, origin, 32UL ); q += 32UL;
940 0 : ulong sig_off = (ulong)(q-pkt); q += 64UL;
941 0 : fd_memcpy( q, relayer, 32UL ); q += 32UL;
942 0 : FD_STORE( ulong, q, wallclock ); q += 8UL;
943 :
944 0 : sign_request( gossip, SIGN_PEND_SEND, &dest_addr, NULL, pkt, sizeof(pkt), sig_off, signable, sizeof(signable), FD_KEYGUARD_SIGN_TYPE_ED25519 );
945 0 : }
946 :
947 : static void
948 : tx_prunes( fd_gossip_t * gossip,
949 0 : long now ) {
950 0 : uchar const * relayer;
951 0 : uchar const * origin;
952 0 : while( fd_prune_finder_pop_prune( gossip->prune_finder, &relayer, &origin ) ) {
953 0 : tx_prune( gossip, relayer, origin, now );
954 0 : }
955 0 : }
956 :
957 : static void
958 : rx_push( fd_gossip_t * gossip,
959 : fd_gossip_push_t const * push,
960 : uchar const * payload,
961 : uchar const * failed,
962 : long now,
963 0 : fd_stem_context_t * stem ) {
964 0 : long results[ 17UL ];
965 0 : rx_values( gossip, push->values_len, push->values, payload, failed, stem, 1, now, results );
966 :
967 0 : for( ulong i=0UL; i<push->values_len; i++ ) {
968 0 : if( FD_UNLIKELY( failed[ i ] && failed[ i ]!=FD_GOSSIP_FAILED_DUPLICATE ) ) continue;
969 0 : if( FD_LIKELY( !results[ i ] ) ) gossip->metrics->crds_rx_count[ FD_METRICS_ENUM_GOSSIP_CRDS_RESULT_V_INSERTED_PUSH_IDX ]++;
970 0 : else if( results[ i ]<0L ) gossip->metrics->crds_rx_count[ FD_METRICS_ENUM_GOSSIP_CRDS_RESULT_V_STALE_PUSH_IDX ]++;
971 0 : else gossip->metrics->crds_rx_count[ FD_METRICS_ENUM_GOSSIP_CRDS_RESULT_V_DUPLICATE_PUSH_IDX ]++;
972 :
973 0 : ulong num_dups;
974 0 : if( FD_LIKELY( !results[ i ] ) ) num_dups = 0UL;
975 0 : else if( FD_UNLIKELY( results[ i ]<0L ) ) num_dups = ULONG_MAX; /* stale => never timely */
976 0 : else num_dups = (ulong)results[ i ]-1UL;
977 :
978 0 : ulong origin_stake = get_stake( gossip, push->values[ i ].origin );
979 0 : fd_prune_finder_record( gossip->prune_finder, push->values[ i ].origin, origin_stake, push->from, get_stake( gossip, push->from ), num_dups );
980 0 : }
981 :
982 0 : tx_prunes( gossip, now );
983 0 : }
984 :
985 : static void
986 : rx_prune( fd_gossip_t * gossip,
987 0 : fd_gossip_prune_t const * prune ) {
988 0 : for( ulong i=0UL; i<prune->prunes_len; i++ ) {
989 0 : fd_active_set_prune( gossip->active_set,
990 0 : prune->pubkey,
991 0 : prune->prunes[ i ],
992 0 : get_stake( gossip, prune->prunes[ i ] ) );
993 0 : }
994 0 : }
995 :
996 :
997 : static void
998 : rx_ping( fd_gossip_t * gossip,
999 : fd_gossip_ping_t const * ping,
1000 0 : fd_ip4_port_t peer_address ) {
1001 0 : uchar out_payload[ sizeof(fd_gossip_pong_t)+4UL];
1002 0 : FD_STORE( uint, out_payload, FD_GOSSIP_MESSAGE_PONG );
1003 :
1004 0 : fd_gossip_pong_t * out_pong = (fd_gossip_pong_t *)(out_payload + 4UL);
1005 0 : fd_memcpy( out_pong->from, gossip->identity_pubkey, 32UL );
1006 :
1007 : /* fd_keyguard checks payloads for certain patterns before performing the
1008 : sign. Pattern-matching can't be done on hashed data, so we need
1009 : to supply the pre-hashed image to the sign fn (fd_keyguard will hash when
1010 : supplied with FD_KEYGUARD_SIGN_TYPE_SHA256_ED25519) while also hashing
1011 : the image ourselves onto pong->ping_hash */
1012 :
1013 0 : uchar pre_image[ 48UL ];
1014 0 : fd_memcpy( pre_image, "SOLANA_PING_PONG", 16UL );
1015 0 : fd_memcpy( pre_image+16UL, ping->token, 32UL );
1016 :
1017 0 : fd_sha256_hash( pre_image, 48UL, out_pong->hash );
1018 :
1019 0 : sign_request( gossip, SIGN_PEND_SEND, &peer_address, NULL, out_payload, sizeof(out_payload), 4UL+(ulong)((uchar *)out_pong->signature-(uchar *)out_pong), pre_image, 48UL, FD_KEYGUARD_SIGN_TYPE_SHA256_ED25519 );
1020 0 : }
1021 :
1022 : static void
1023 : rx_pong( fd_gossip_t * gossip,
1024 : fd_gossip_pong_t const * pong,
1025 : fd_ip4_port_t peer_address,
1026 0 : long now ) {
1027 0 : ulong stake = get_stake( gossip, pong->from );
1028 0 : fd_ping_tracker_register( gossip->ping_tracker, pong->from, stake, peer_address, pong->hash, now );
1029 0 : }
1030 :
1031 : void
1032 : fd_gossip_rx( fd_gossip_t * gossip,
1033 : fd_ip4_port_t peer,
1034 : uchar const * data,
1035 : ulong data_sz,
1036 : long now,
1037 0 : fd_stem_context_t * stem ) {
1038 : /* TODO: Implement traffic shaper / bandwidth limiter */
1039 0 : FD_TEST( data_sz>=offsetof(fd_gossip_message_t, push)+offsetof(fd_gossip_push_t, values) ); /* tag and values_len readable */
1040 0 : fd_gossip_message_t const * message = (fd_gossip_message_t const *)data;
1041 0 : ulong msg_sz = fd_gossip_message_used_sz( message );
1042 0 : FD_TEST( data_sz>=msg_sz+FD_GOSSIP_MESSAGE_MAX_CRDS );
1043 0 : uchar const * failed = data+msg_sz;
1044 0 : uchar const * payload = data+msg_sz+FD_GOSSIP_MESSAGE_MAX_CRDS;
1045 :
1046 0 : switch( message->tag ) {
1047 0 : case FD_GOSSIP_MESSAGE_PULL_REQUEST: rx_pull_request( gossip, message->pull_request, peer, stem, now ); break;
1048 0 : case FD_GOSSIP_MESSAGE_PULL_RESPONSE: rx_pull_response( gossip, message->pull_response, payload, failed, stem, now ); break;
1049 0 : case FD_GOSSIP_MESSAGE_PUSH: rx_push( gossip, message->push, payload, failed, now, stem ); break;
1050 0 : case FD_GOSSIP_MESSAGE_PRUNE: rx_prune( gossip, message->prune ); break;
1051 0 : case FD_GOSSIP_MESSAGE_PING: rx_ping( gossip, message->ping, peer ); break;
1052 0 : case FD_GOSSIP_MESSAGE_PONG: rx_pong( gossip, message->pong, peer, now ); break;
1053 0 : default:
1054 0 : FD_LOG_CRIT(( "Unknown gossip message type %u", message->tag ));
1055 0 : break;
1056 0 : }
1057 0 : }
1058 :
1059 : static int
1060 : fd_gossip_push( fd_gossip_t * gossip,
1061 0 : fd_gossip_value_t const * value ) {
1062 0 : uchar serialized[ FD_GOSSIP_VALUE_MAX_SZ ];
1063 0 : long serialized_sz = fd_gossip_value_serialize( value, serialized, sizeof(serialized) );
1064 0 : FD_TEST( serialized_sz!=-1L );
1065 :
1066 : /* Inserted into the CRDS and pushed on the response */
1067 0 : return sign_request( gossip, SIGN_PEND_VALUE, NULL, value, serialized, (ulong)serialized_sz, 0UL, serialized+64UL, (ulong)serialized_sz-64UL, FD_KEYGUARD_SIGN_TYPE_ED25519 ) ? 0 : -1;
1068 0 : }
1069 :
1070 : int
1071 : fd_gossip_push_vote( fd_gossip_t * gossip,
1072 : uchar const * txn,
1073 : ulong txn_sz,
1074 0 : long now ) {
1075 0 : fd_gossip_value_t value = {
1076 0 : .tag = FD_GOSSIP_VALUE_VOTE,
1077 0 : .wallclock = (ulong)FD_NANOSEC_TO_MILLI( now ),
1078 0 : .vote = {{
1079 0 : .index = 0UL, /* TODO */
1080 0 : .transaction_len = txn_sz,
1081 0 : }}
1082 0 : };
1083 0 : fd_memcpy( value.origin, gossip->identity_pubkey, 32UL );
1084 0 : FD_TEST( txn_sz<=sizeof(value.vote->transaction) );
1085 0 : fd_memcpy( value.vote->transaction, txn, txn_sz );
1086 :
1087 0 : return fd_gossip_push( gossip, &value );
1088 0 : }
1089 :
1090 : int
1091 : fd_gossip_push_duplicate_shred( fd_gossip_t * gossip,
1092 : fd_gossip_duplicate_shred_t const * duplicate_shred,
1093 0 : long now ) {
1094 0 : fd_gossip_value_t value = {
1095 0 : .tag = FD_GOSSIP_VALUE_DUPLICATE_SHRED,
1096 0 : .wallclock = (ulong)FD_NANOSEC_TO_MILLI( now ),
1097 0 : };
1098 0 : fd_memcpy( value.origin, gossip->identity_pubkey, 32UL );
1099 0 : *value.duplicate_shred = *duplicate_shred;
1100 :
1101 0 : return fd_gossip_push( gossip, &value );
1102 0 : }
1103 :
1104 : static void
1105 : tx_ping( fd_gossip_t * gossip,
1106 : long now,
1107 0 : int * charge_busy ) {
1108 0 : uchar out_payload[ sizeof(fd_gossip_ping_t) + 4UL ];
1109 0 : FD_STORE( uint, out_payload, FD_GOSSIP_MESSAGE_PING );
1110 :
1111 0 : fd_gossip_ping_t * out_ping = (fd_gossip_ping_t *)( out_payload+4UL );
1112 0 : fd_memcpy( out_ping->from, gossip->identity_pubkey, 32UL );
1113 :
1114 0 : uchar const * peer_pubkey;
1115 0 : uchar const * ping_token;
1116 0 : fd_ip4_port_t const * peer_address;
1117 0 : while( fd_ping_tracker_pop_request( gossip->ping_tracker,
1118 0 : now,
1119 0 : &peer_pubkey,
1120 0 : &peer_address,
1121 0 : &ping_token ) ) {
1122 0 : fd_memcpy( out_ping->token, ping_token, 32UL );
1123 :
1124 0 : sign_request( gossip, SIGN_PEND_SEND, peer_address, NULL, out_payload, sizeof(out_payload), 4UL+(ulong)((uchar *)out_ping->signature-(uchar *)out_ping), out_ping->token, 32UL, FD_KEYGUARD_SIGN_TYPE_ED25519 );
1125 0 : if( charge_busy ) *charge_busy = 1;
1126 0 : }
1127 0 : }
1128 :
1129 : /* Construct and send a pull request to a random peer. The pull
1130 : request contains a bloom filter over our known CRDS hashes so that
1131 : the peer can respond with values we are missing.
1132 :
1133 : NOTE: Divergence from Agave:
1134 : - Agave builds up to 2^mask_bits filters per pull period
1135 : (sampling up to 1024), each covering a distinct partition of
1136 : the hash space. We build and send exactly one filter per
1137 : pull period, covering 1/2^mask_bits of the space.
1138 :
1139 : Maximum bloom filter bits in a PullRequest packet:
1140 :
1141 : PACKET_DATA_SIZE = 1232 (= 1280 - 40 - 8)
1142 :
1143 : Bytes consumed by non-bloom fields:
1144 : discriminant(4) + keys_len(8) + keys(8*num_keys) +
1145 : has_bits(1) + bloom_vec_len(8) + bloom_bits_count(8) +
1146 : bloom_num_bits_set(8) + mask(8) + mask_bits(4)
1147 : + contact_info_crds_val(crds_val_sz)
1148 : = 49 + 8*num_keys + crds_val_sz
1149 :
1150 : The bitvec is serialized as u64 words, so the bitvec storage is
1151 : ceil(num_bits/64)*8 bytes. The remaining packet bytes must
1152 : accommodate this.
1153 :
1154 : Agave determines the max_bytes parameter (input to Bloom::random)
1155 : via an empirical cache (get_max_bloom_filter_bytes). max_bytes*8
1156 : is passed as the max_bits cap to Bloom::random, but actual
1157 : num_bits is only ~83% of max_bits (the E/D ratio for p=0.1).
1158 : We replicate this with a closed-form inversion: the largest
1159 : max_bytes where ceil(num_bits/64)*8 fits in remaining space is
1160 : max_bytes = floor(D * floor(64*W/E) / 8), where W is the max
1161 : number of u64 words, E and D are the bloom filter constants.
1162 :
1163 : num_keys depends on the bloom sizing, which depends on the
1164 : overhead, which depends on num_keys. However there is a closed
1165 : form: compute num_keys from the pessimistic KEYS=8 overhead, then
1166 : recompute the tight overhead with the true num_keys. This always
1167 : converges in one step because the optimal key count is
1168 : D*ln(2) ≈ 3.32 (where D = ln(p)/ln(1/2^ln2)), far from any
1169 : rounding boundary. For p=0.1 and KEYS=8, num_keys is always 3.
1170 :
1171 : NB: The has_bits(1) + bloom_vec_len(8) are only written when
1172 : num_bits>=1. fd_bloom_num_bits clamps to [1, max_bits], so
1173 : num_bits>=1 always holds and this layout is correct. */
1174 :
1175 : static void
1176 : tx_pull_request( fd_gossip_t * gossip,
1177 : fd_stem_context_t * stem,
1178 0 : long now ) {
1179 0 : ulong total_crds_vals = fd_crds_len( gossip->crds ) + fd_gossip_purged_len( gossip->purged );
1180 0 : ulong num_items = fd_ulong_max( 65536UL, total_crds_vals );
1181 0 : ulong crds_val_sz = gossip->my_contact_info.pull_val_sz;
1182 :
1183 : /* Step 1: Compute num_keys from the pessimistic KEYS=8 overhead
1184 : (same initial estimate Agave uses in CrdsFilterSet::new). */
1185 0 : ulong pessimistic_overhead = 49UL + 8UL*(ulong)BLOOM_NUM_KEYS + crds_val_sz;
1186 0 : FD_TEST( pessimistic_overhead<FD_GOSSIP_MTU );
1187 0 : double pessimistic_max_bits = (double)( 8UL*( FD_GOSSIP_MTU - pessimistic_overhead ) );
1188 0 : double pessimistic_items = fd_bloom_max_items( pessimistic_max_bits, BLOOM_NUM_KEYS, BLOOM_FALSE_POSITIVE_RATE );
1189 0 : FD_TEST( pessimistic_items>0.0 );
1190 0 : ulong pessimistic_num_bits = fd_bloom_num_bits( pessimistic_items, BLOOM_FALSE_POSITIVE_RATE, pessimistic_max_bits );
1191 0 : ulong num_keys = fd_bloom_num_keys( (double)pessimistic_num_bits, pessimistic_items );
1192 :
1193 : /* Step 2: Recompute with the tight overhead using the true num_keys.
1194 : Find the largest max_bytes parameter (matching Agave's
1195 : get_max_bloom_filter_bytes cache) such that the resulting bitvec
1196 : fits in the remaining packet space.
1197 :
1198 : Given:
1199 : max_items = ceil(max_bits / D) where D = -K / ln(1-exp(ln(p)/K))
1200 : num_bits = ceil(max_items * E) where E = ln(p) / ln(1/2^ln2)
1201 :
1202 : We need ceil(num_bits/64)*8 <= remaining, i.e. num_bits <= 64*W
1203 : where W = floor(remaining/8). Working backwards:
1204 : max_items <= I where I = floor(64*W / E)
1205 : max_bytes <= D*I / 8
1206 :
1207 : So max_bytes = floor(D * floor(64*W/E) / 8). */
1208 0 : ulong overhead = 49UL + 8UL*num_keys + crds_val_sz;
1209 0 : FD_TEST( overhead<FD_GOSSIP_MTU );
1210 0 : ulong remaining = FD_GOSSIP_MTU - overhead;
1211 0 : ulong max_words = remaining / 8UL; /* max u64 words for bitvec */
1212 :
1213 0 : double E = log( BLOOM_FALSE_POSITIVE_RATE ) / log( 1.0 / pow( 2.0, log( 2.0 ) ) );
1214 0 : double D = -BLOOM_NUM_KEYS / log( 1.0 - exp( log( BLOOM_FALSE_POSITIVE_RATE ) / BLOOM_NUM_KEYS ) );
1215 0 : ulong I = (ulong)floor( 64.0 * (double)max_words / E );
1216 0 : ulong max_bytes = (ulong)floor( D * (double)I / 8.0 );
1217 :
1218 0 : double max_bits = (double)( max_bytes * 8UL );
1219 0 : double max_items = fd_bloom_max_items( max_bits, BLOOM_NUM_KEYS, BLOOM_FALSE_POSITIVE_RATE );
1220 0 : FD_TEST( max_items>0.0 );
1221 0 : ulong num_bits = fd_bloom_num_bits( max_items, BLOOM_FALSE_POSITIVE_RATE, max_bits );
1222 0 : FD_TEST( num_bits>=1UL );
1223 0 : FD_TEST( (num_bits+63UL)/64UL<=max_words ); /* verify bitvec fits */
1224 0 : FD_TEST( fd_bloom_num_keys( (double)num_bits, max_items )==num_keys ); /* verify convergence */
1225 0 : FD_TEST( num_keys<=(ulong)BLOOM_NUM_KEYS );
1226 :
1227 0 : double _mask_bits = ceil( log2( (double)num_items / max_items ) );
1228 0 : uint mask_bits = _mask_bits >= 0.0 ? fd_uint_min( (uint)_mask_bits, 63U ) : 0U;
1229 0 : ulong mask = fd_rng_ulong( gossip->rng ) | (~0UL>>(mask_bits));
1230 :
1231 0 : uchar payload[ FD_GOSSIP_MTU ] = {0};
1232 :
1233 0 : uchar * keys_ptr, * bits_ptr, * bits_set;
1234 0 : long payload_sz = fd_gossip_pull_request_init( payload,
1235 0 : FD_GOSSIP_MTU,
1236 0 : num_keys,
1237 0 : num_bits,
1238 0 : mask,
1239 0 : mask_bits,
1240 0 : gossip->my_contact_info.pull_val,
1241 0 : gossip->my_contact_info.pull_val_sz,
1242 0 : &keys_ptr,
1243 0 : &bits_ptr,
1244 0 : &bits_set );
1245 0 : FD_TEST( -1L!=payload_sz );
1246 :
1247 : /* Bloom filter fields are not naturally aligned in the serialized packet.
1248 : Build the filter in aligned storage and copy it into the packet. */
1249 0 : ulong bloom_keys[ (ulong)BLOOM_NUM_KEYS ];
1250 0 : ulong bloom_bits[ (FD_GOSSIP_MTU+sizeof(ulong)-1UL)/sizeof(ulong) ] = {0};
1251 0 : ulong bloom_word_cnt = (num_bits+63UL)/64UL;
1252 0 : fd_bloom_t filter[1];
1253 0 : fd_bloom_init_inplace( bloom_keys, bloom_bits, num_keys, num_bits, 0, gossip->rng, BLOOM_FALSE_POSITIVE_RATE, filter );
1254 :
1255 0 : ulong start_hash, end_hash;
1256 0 : fd_gossip_purged_generate_masks( mask, mask_bits, &start_hash, &end_hash );
1257 0 : fd_gossip_hset_t const * hsets[ 2 ] = { fd_crds_hset( gossip->crds ), fd_gossip_purged_hset( gossip->purged ) };
1258 : /* Chunks are hashed in pairs (fd_bloom_insert16 interleaves the two
1259 : chains); an odd chunk waits for the next one, the hsets are not
1260 : modified while we scan so its hashes stay put. */
1261 0 : uchar const * pend_hashes = NULL;
1262 0 : uint pend_lanes = 0U;
1263 0 : for( ulong i=0UL; i<2UL; i++ ) {
1264 0 : fd_gossip_hset_iter_t it[1];
1265 0 : for( fd_gossip_hset_iter_init( it, hsets[ i ], start_hash, end_hash );
1266 0 : !fd_gossip_hset_iter_done( it );
1267 0 : fd_gossip_hset_iter_next( it, hsets[ i ] ) ) {
1268 0 : uint lanes = fd_gossip_hset_iter_lanes( it, hsets[ i ] );
1269 0 : if( FD_UNLIKELY( !lanes ) ) continue;
1270 0 : uchar const * hashes = fd_gossip_hset_iter_hashes( it, hsets[ i ] );
1271 0 : if( !pend_lanes ) { pend_hashes = hashes; pend_lanes = lanes; continue; }
1272 0 : fd_bloom_insert16( filter, pend_hashes, pend_lanes, hashes, lanes );
1273 0 : pend_lanes = 0U;
1274 0 : }
1275 0 : }
1276 0 : if( pend_lanes ) fd_bloom_insert8( filter, pend_hashes, pend_lanes );
1277 :
1278 0 : int num_bits_set = 0;
1279 0 : for( ulong i=0UL; i<bloom_word_cnt; i++ ) num_bits_set += fd_ulong_popcnt( bloom_bits[ i ] );
1280 0 : fd_memcpy( keys_ptr, bloom_keys, num_keys*sizeof(ulong) );
1281 0 : fd_memcpy( bits_ptr, bloom_bits, bloom_word_cnt*sizeof(ulong) );
1282 0 : FD_STORE( ulong, bits_set, (ulong)num_bits_set );
1283 :
1284 : /* While no entrypoint's contact info is fresh in the table, send
1285 : every other pull to an entrypoint. */
1286 0 : ulong idx = fd_gossip_wsample_sample_pull_request( gossip->wsample );
1287 0 : if( FD_UNLIKELY( gossip->entrypoints_cnt && !fd_crds_entrypoint_fresh_cnt( gossip->crds ) && fd_rng_uint( gossip->rng )&1U ) ) idx = ULONG_MAX;
1288 0 : fd_ip4_port_t peer_addr;
1289 0 : if( FD_UNLIKELY( idx==ULONG_MAX ) ) {
1290 0 : if( FD_UNLIKELY( !gossip->entrypoints_cnt ) ) {
1291 : /* We are the bootstrapping node, and nobody else is present in
1292 : the cluster. Nowhere to send the pull request. */
1293 0 : return;
1294 0 : }
1295 0 : peer_addr = random_entrypoint( gossip );
1296 0 : } else {
1297 0 : fd_gossip_contact_info_t const * peer = fd_crds_ci( gossip->crds, idx );
1298 0 : peer_addr.addr = peer->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ].is_ipv6 ? 0 : peer->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ].ip4;
1299 0 : peer_addr.port = peer->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ].port;
1300 0 : }
1301 0 : gossip->send_fn( gossip->send_ctx, stem, payload, (ulong)payload_sz, &peer_addr, (ulong)now );
1302 :
1303 0 : gossip->metrics->message_tx[ FD_GOSSIP_MESSAGE_PULL_REQUEST ]++;
1304 0 : gossip->metrics->message_tx_bytes[ FD_GOSSIP_MESSAGE_PULL_REQUEST ] += (ulong)payload_sz + 42UL; /* 42 = sizeof(fd_ip4_udp_hdrs_t) */
1305 0 : }
1306 :
1307 : void
1308 : fd_gossip_advance( fd_gossip_t * gossip,
1309 : long now,
1310 : fd_stem_context_t * stem,
1311 0 : int * charge_busy ) {
1312 0 : outbound_budget_replenish( gossip, now );
1313 :
1314 0 : fd_gossip_purged_expire( gossip->purged, now );
1315 0 : fd_active_set_advance( gossip->active_set, stem, now, charge_busy );
1316 0 : fd_crds_advance( gossip->crds, now, stem, charge_busy );
1317 :
1318 0 : tx_ping( gossip, now, charge_busy );
1319 0 : if( FD_UNLIKELY( now>=gossip->timers.next_pull_contact_info && gossip->my_contact_info.crds_val_sz ) ) {
1320 0 : if( FD_LIKELY( !gossip->my_contact_info.pull_pending ) ) request_pull_contact_info( gossip, now );
1321 0 : gossip->timers.next_pull_contact_info = now+PULL_CONTACT_INFO_INTERVAL_NANOS;
1322 0 : if( charge_busy ) *charge_busy = 1;
1323 0 : }
1324 0 : if( FD_UNLIKELY( now>=gossip->timers.next_pull_request && gossip->my_contact_info.crds_val_sz ) ) {
1325 0 : tx_pull_request( gossip, stem, now );
1326 0 : if( charge_busy ) *charge_busy = 1;
1327 : /* 1.6ms (625/s). Agave sends min(1024, ceil(2^mask_bits/8))
1328 : filters every 500ms. For a typical mainnet table (~65k items,
1329 : mask_bits≈7) that is ~16 filters/500ms = one every 31ms. We
1330 : send a single filter per round, so we fire ~20× more often to
1331 : compensate for sending one filter instead of many per period.
1332 :
1333 : We considered dynamically matching Agave's exact rate by
1334 : computing 500ms/filters_per_round from mask_bits each round,
1335 : but this caused slow table fill on startup (mask_bits starts
1336 : low -> long intervals -> few pulls -> slow CRDS population).
1337 : Adaptive boosting (counter-based, timestamp-based, and
1338 : threshold-based) all added complexity without clear benefit:
1339 : counter decay lost state between send and response arrival,
1340 : timestamp checks never disarmed because trickle inserts kept
1341 : refreshing the window, and threshold heuristics required
1342 : tuning constants that varied by cluster size.
1343 :
1344 : A fixed 1.6ms is simpler and robust: the cost of a redundant
1345 : pull request is negligible (a single 1232-byte packet whose
1346 : reply will be empty if we're already caught up), and it
1347 : guarantees fast table fill on startup without any adaptive
1348 : machinery. */
1349 0 : gossip->timers.next_pull_request = now+1600L*1000L;
1350 0 : }
1351 0 : if( FD_UNLIKELY( now>=gossip->timers.next_contact_info_refresh ) ) {
1352 : /* TODO: Frequency of this? More often if observing? */
1353 0 : gossip->timers.next_contact_info_refresh = now+15L*500L*1000L*1000L; /* TODO: Jitter */
1354 0 : gossip->my_contact_info.refresh_due = 1;
1355 0 : }
1356 0 : if( FD_UNLIKELY( gossip->my_contact_info.refresh_due ) ) {
1357 0 : gossip->my_contact_info.refresh_due = 0;
1358 0 : request_contact_info( gossip, now );
1359 0 : if( charge_busy ) *charge_busy = 1;
1360 0 : }
1361 0 : }
1362 :
1363 : long
1364 0 : fd_gossip_next_deadline( fd_gossip_t const * gossip ) {
1365 0 : if( FD_UNLIKELY( gossip->my_contact_info.refresh_due ) ) return 0L;
1366 0 : long deadline = fd_long_min( gossip->timers.next_contact_info_refresh, fd_active_set_next_deadline( gossip->active_set ) );
1367 0 : deadline = fd_long_min( deadline, fd_ping_tracker_next_deadline( gossip->ping_tracker ) );
1368 0 : if( FD_LIKELY( gossip->my_contact_info.crds_val_sz ) ) deadline = fd_long_min( deadline, gossip->timers.next_pull_request );
1369 0 : return deadline;
1370 0 : }
1371 :
1372 : void
1373 : fd_gossip_ping_tracker_track( fd_gossip_t * gossip,
1374 : uchar const * peer_pubkey,
1375 : fd_ip4_port_t peer_address,
1376 0 : long now ) {
1377 : /* Don't track ourselves. */
1378 0 : if( FD_UNLIKELY( !memcmp( peer_pubkey, gossip->identity_pubkey, 32UL ) ) ) return;
1379 :
1380 0 : ulong origin_stake = get_stake( gossip, peer_pubkey );
1381 0 : fd_ping_tracker_track( gossip->ping_tracker, peer_pubkey, origin_stake, peer_address, now );
1382 0 : }
|
