Line data Source code
1 : #include "fd_active_set.h"
2 : #include "fd_gossip_txbuild.h"
3 : #include "fd_gossip_wsample.h"
4 : #include "fd_bloom.h"
5 : #include "../../util/net/fd_net_headers.h"
6 :
7 : #define FD_ACTIVE_SET_STAKE_ENTRIES (25UL)
8 : #define FD_ACTIVE_SET_PEERS_PER_ENTRY (12UL)
9 : #define FD_ACTIVE_SET_MAX_PEERS (FD_ACTIVE_SET_STAKE_ENTRIES*FD_ACTIVE_SET_PEERS_PER_ENTRY) /* 300 */
10 0 : #define FD_ACTIVE_SET_PUSH_FANOUT (9UL)
11 :
12 : struct fd_active_set_peer {
13 : long timestamp;
14 : ulong ci_idx;
15 : fd_bloom_t * bloom;
16 : fd_gossip_txbuild_t txbuild[1];
17 :
18 : struct {
19 : ulong prev;
20 : ulong next;
21 : } dlist;
22 : };
23 :
24 : typedef struct fd_active_set_peer fd_active_set_peer_t;
25 :
26 : #define DLIST_NAME push_dlist
27 : #define DLIST_ELE_T fd_active_set_peer_t
28 0 : #define DLIST_PREV dlist.prev
29 0 : #define DLIST_NEXT dlist.next
30 : #include "../../util/tmpl/fd_dlist.c"
31 :
32 : struct fd_active_set_entry {
33 : ulong nodes_idx; /* points to oldest entry in set */
34 : ulong nodes_len;
35 : };
36 :
37 : typedef struct fd_active_set_entry fd_active_set_entry_t;
38 :
39 : struct __attribute__((aligned(FD_ACTIVE_SET_ALIGN))) fd_active_set_private {
40 : fd_active_set_entry_t entries[ FD_ACTIVE_SET_STAKE_ENTRIES ][ 1 ];
41 : fd_active_set_peer_t peers[ FD_ACTIVE_SET_MAX_PEERS ];
42 :
43 : long next_rotate_nanos;
44 : ulong rotate_bucket; /* 0..24, round-robin */
45 :
46 : uchar identity_pubkey[ 32UL ];
47 : ulong identity_stake;
48 :
49 : fd_gossip_wsample_t * wsample;
50 : fd_crds_t * crds;
51 : fd_rng_t * rng;
52 : push_dlist_t * push_dlist;
53 :
54 : fd_gossip_send_fn send_fn;
55 : void * send_fn_ctx;
56 :
57 : fd_active_set_metrics_t metrics[1];
58 :
59 : ulong magic; /* ==FD_ACTIVE_SET_MAGIC */
60 : };
61 :
62 : FD_FN_CONST ulong
63 0 : fd_active_set_align( void ) {
64 0 : return FD_ACTIVE_SET_ALIGN;
65 0 : }
66 :
67 : FD_FN_CONST ulong
68 0 : fd_active_set_footprint( void ) {
69 0 : ulong l;
70 0 : l = FD_LAYOUT_INIT;
71 0 : l = FD_LAYOUT_APPEND( l, FD_ACTIVE_SET_ALIGN, sizeof(fd_active_set_t) );
72 0 : l = FD_LAYOUT_APPEND( l, FD_BLOOM_ALIGN, 25UL*12UL*fd_bloom_footprint( 0.1, 32768UL ) );
73 0 : l = FD_LAYOUT_APPEND( l, push_dlist_align(), push_dlist_footprint() );
74 0 : return FD_LAYOUT_FINI( l, FD_ACTIVE_SET_ALIGN );
75 0 : }
76 :
77 : void *
78 : fd_active_set_new( void * shmem,
79 : fd_gossip_wsample_t * wsample,
80 : fd_crds_t * crds,
81 : fd_rng_t * rng,
82 : uchar const * identity_pubkey,
83 : ulong identity_stake,
84 : fd_gossip_send_fn send_fn,
85 0 : void * send_fn_ctx ) {
86 0 : if( FD_UNLIKELY( !shmem ) ) {
87 0 : FD_LOG_WARNING(( "NULL shmem" ));
88 0 : return NULL;
89 0 : }
90 :
91 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shmem, fd_active_set_align() ) ) ) {
92 0 : FD_LOG_WARNING(( "misaligned shmem" ));
93 0 : return NULL;
94 0 : }
95 :
96 0 : ulong bloom_footprint = fd_bloom_footprint( 0.1, 32768UL );
97 :
98 0 : FD_SCRATCH_ALLOC_INIT( l, shmem );
99 0 : fd_active_set_t * as = FD_SCRATCH_ALLOC_APPEND( l, FD_ACTIVE_SET_ALIGN, sizeof(fd_active_set_t) );
100 0 : uchar * _blooms = FD_SCRATCH_ALLOC_APPEND( l, FD_BLOOM_ALIGN, 25UL*12UL*bloom_footprint );
101 0 : push_dlist_t * _push_dlist = FD_SCRATCH_ALLOC_APPEND( l, push_dlist_align(), push_dlist_footprint() );
102 :
103 0 : as->next_rotate_nanos = 0L;
104 0 : as->rotate_bucket = 0UL;
105 0 : fd_memcpy( as->identity_pubkey, identity_pubkey, 32UL );
106 0 : as->identity_stake = identity_stake;
107 :
108 0 : as->wsample = wsample;
109 0 : as->crds = crds;
110 0 : as->rng = rng;
111 0 : for( ulong i=0UL; i<25UL; i++ ) {
112 0 : fd_active_set_entry_t * entry = as->entries[ i ];
113 0 : entry->nodes_idx = 0UL;
114 0 : entry->nodes_len = 0UL;
115 :
116 0 : for( ulong j=0UL; j<12UL; j++ ) {
117 0 : fd_active_set_peer_t * peer = &as->peers[ i*12UL+j ];
118 0 : peer->bloom = fd_bloom_join( fd_bloom_new( _blooms, rng, 0.1, 32768UL ) );
119 0 : if( FD_UNLIKELY( !peer->bloom ) ) {
120 0 : FD_LOG_WARNING(( "failed to create bloom filter" ));
121 0 : return NULL;
122 0 : }
123 0 : _blooms += bloom_footprint;
124 0 : }
125 0 : }
126 :
127 0 : as->push_dlist = push_dlist_join( push_dlist_new( _push_dlist ) );
128 :
129 0 : as->send_fn = send_fn;
130 0 : as->send_fn_ctx = send_fn_ctx;
131 :
132 0 : memset( as->metrics, 0, sizeof(fd_active_set_metrics_t) );
133 :
134 0 : FD_COMPILER_MFENCE();
135 0 : FD_VOLATILE( as->magic ) = FD_ACTIVE_SET_MAGIC;
136 0 : FD_COMPILER_MFENCE();
137 :
138 0 : return (void *)as;
139 0 : }
140 :
141 : fd_active_set_t *
142 0 : fd_active_set_join( void * shas ) {
143 0 : if( FD_UNLIKELY( !shas ) ) {
144 0 : FD_LOG_WARNING(( "NULL shas" ));
145 0 : return NULL;
146 0 : }
147 :
148 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shas, fd_active_set_align() ) ) ) {
149 0 : FD_LOG_WARNING(( "misaligned shas" ));
150 0 : return NULL;
151 0 : }
152 :
153 0 : fd_active_set_t * as = (fd_active_set_t *)shas;
154 :
155 0 : if( FD_UNLIKELY( as->magic!=FD_ACTIVE_SET_MAGIC ) ) {
156 0 : FD_LOG_WARNING(( "bad magic" ));
157 0 : return NULL;
158 0 : }
159 :
160 0 : return as;
161 0 : }
162 :
163 : fd_active_set_metrics_t const *
164 0 : fd_active_set_metrics( fd_active_set_t const * active_set ) {
165 0 : return active_set->metrics;
166 0 : }
167 :
168 : void
169 : fd_active_set_set_identity( fd_active_set_t * active_set,
170 : uchar const * identity_pubkey,
171 0 : ulong identity_stake ) {
172 0 : fd_memcpy( active_set->identity_pubkey, identity_pubkey, 32UL );
173 0 : active_set->identity_stake = identity_stake;
174 0 : }
175 :
176 : void
177 : fd_active_set_prune( fd_active_set_t * active_set,
178 : uchar const * push_dest,
179 : uchar const * origin,
180 0 : ulong origin_stake ) {
181 0 : if( FD_UNLIKELY( !memcmp( active_set->identity_pubkey, origin, 32UL ) ) ) return;
182 :
183 0 : ulong bucket = fd_active_set_stake_bucket( fd_ulong_min( active_set->identity_stake, origin_stake ) );
184 0 : for( ulong i=0UL; i<active_set->entries[ bucket ]->nodes_len; i++ ) {
185 0 : ulong peer_idx = (active_set->entries[ bucket ]->nodes_idx+i) % 12UL;
186 0 : uchar const * peer_pubkey = fd_crds_ci_pubkey( active_set->crds, active_set->peers[ bucket*12UL+peer_idx ].ci_idx );
187 0 : if( FD_UNLIKELY( !memcmp( peer_pubkey, push_dest, 32UL ) ) ) {
188 0 : fd_bloom_insert( active_set->peers[ bucket*12UL+peer_idx ].bloom, origin, 32UL );
189 0 : return;
190 0 : }
191 0 : }
192 0 : }
193 :
194 : static void
195 : push_flush( fd_active_set_t * active_set,
196 : fd_active_set_peer_t * peer,
197 : fd_stem_context_t * stem,
198 0 : long now ) {
199 0 : if( FD_UNLIKELY( !peer->txbuild->crds_len ) ) return;
200 :
201 0 : fd_gossip_contact_info_t const * ci = fd_crds_ci( active_set->crds, peer->ci_idx );
202 : // TODO: Support ipv6, or prevent ending up in set
203 0 : fd_ip4_port_t dest_addr = {
204 0 : .addr = ci->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ].is_ipv6 ? 0U : ci->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ].ip4,
205 0 : .port = ci->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ].port,
206 0 : };
207 :
208 0 : push_dlist_ele_remove( active_set->push_dlist, peer, active_set->peers );
209 :
210 0 : active_set->send_fn( active_set->send_fn_ctx, stem, peer->txbuild->bytes, peer->txbuild->bytes_len, &dest_addr, (ulong)now );
211 :
212 0 : active_set->metrics->message_tx[ peer->txbuild->tag ]++;
213 0 : active_set->metrics->message_tx_bytes[ peer->txbuild->tag ] += peer->txbuild->bytes_len+42UL; /* 42 = sizeof(fd_ip4_udp_hdrs_t) */
214 0 : for( ulong i=0UL; i<peer->txbuild->crds_len; i++ ) {
215 0 : active_set->metrics->crds_tx_push[ peer->txbuild->crds[ i ].tag ]++;
216 0 : active_set->metrics->crds_tx_push_bytes[ peer->txbuild->crds[ i ].tag ] += peer->txbuild->crds[ i ].sz;
217 0 : }
218 :
219 0 : fd_gossip_txbuild_init( peer->txbuild, active_set->identity_pubkey, FD_GOSSIP_MESSAGE_PUSH );
220 0 : }
221 :
222 : void
223 : fd_active_set_remove_peer( fd_active_set_t * active_set,
224 0 : ulong ci_idx ) {
225 0 : for( ulong b=0UL; b<25UL; b++ ) {
226 0 : fd_active_set_entry_t * entry = active_set->entries[ b ];
227 :
228 0 : for( ulong i=0UL; i<entry->nodes_len; i++ ) {
229 0 : ulong peer_idx = (entry->nodes_idx+i) % 12UL;
230 0 : if( FD_UNLIKELY( active_set->peers[ b*12UL+peer_idx ].ci_idx==ci_idx ) ) {
231 0 : fd_active_set_peer_t * peer = &active_set->peers[ b*12UL+peer_idx ];
232 0 : if( FD_UNLIKELY( peer->txbuild->crds_len ) ) push_dlist_ele_remove( active_set->push_dlist, peer, active_set->peers );
233 :
234 0 : for( ulong j=i; j<entry->nodes_len-1UL; j++ ) {
235 0 : ulong from_idx = b*12UL+(entry->nodes_idx+j+1UL) % 12UL;
236 0 : ulong to_idx = b*12UL+(entry->nodes_idx+j) % 12UL;
237 0 : fd_bloom_t * to_bloom = active_set->peers[ to_idx ].bloom;
238 0 : active_set->peers[ to_idx ] = active_set->peers[ from_idx ];
239 0 : active_set->peers[ from_idx ].bloom = to_bloom;
240 : /* If the moved element is in the push_dlist, fix up the
241 : dlist links so neighbors point to the new location.
242 : idx_replace reads prev/next from old_idx (from_idx, still
243 : intact) and patches neighbors + sentinel to reference
244 : to_abs instead. */
245 0 : if( FD_UNLIKELY( active_set->peers[ to_idx ].txbuild->crds_len ) ) push_dlist_idx_replace( active_set->push_dlist, to_idx, from_idx, active_set->peers );
246 0 : }
247 0 : entry->nodes_len--;
248 0 : if( FD_UNLIKELY( !entry->nodes_len ) ) entry->nodes_idx = 0UL;
249 0 : break;
250 0 : }
251 0 : }
252 0 : }
253 0 : }
254 :
255 : void
256 : fd_active_set_push( fd_active_set_t * active_set,
257 : uchar const * crds_val,
258 : ulong crds_sz,
259 : uchar const * origin_pubkey,
260 : ulong origin_stake,
261 : fd_stem_context_t * stem,
262 : long now,
263 0 : int flush_immediately ) {
264 0 : ulong stake_bucket = fd_active_set_stake_bucket( fd_ulong_min( active_set->identity_stake, origin_stake ) );
265 0 : fd_active_set_entry_t * entry = active_set->entries[ stake_bucket ];
266 :
267 0 : int originates_from_me = !memcmp( active_set->identity_pubkey, origin_pubkey, 32UL );
268 :
269 : /* One pass hashes the origin under every peer's keys. */
270 0 : fd_bloom_t * blooms[ 12 ];
271 0 : for( ulong i=0UL; i<entry->nodes_len; i++ ) blooms[ i ] = active_set->peers[ stake_bucket*12UL+((entry->nodes_idx+i) % 12UL) ].bloom;
272 0 : uint hits = fd_bloom_contains_multi( blooms, entry->nodes_len, origin_pubkey, 32UL );
273 :
274 0 : ulong pushed = 0UL;
275 0 : for( ulong i=0UL; i<entry->nodes_len && pushed<FD_ACTIVE_SET_PUSH_FANOUT; i++ ) {
276 0 : fd_active_set_peer_t * peer = &active_set->peers[ stake_bucket*12UL+((entry->nodes_idx+i) % 12UL) ];
277 :
278 : /* If the value originated from us, we should always push it, even
279 : if there's a bloom filter hit, since bloom filters can have false
280 : positives and we don't want to accidentally not push our own
281 : values. */
282 0 : if( FD_UNLIKELY( ((hits>>i)&1U) && !originates_from_me ) ) continue;
283 :
284 0 : if( FD_UNLIKELY( !fd_gossip_txbuild_can_fit( peer->txbuild, crds_sz ) ) ) push_flush( active_set, peer, stem, now );
285 0 : if( FD_UNLIKELY( !peer->txbuild->crds_len ) ) {
286 0 : peer->timestamp = now;
287 0 : push_dlist_ele_push_tail( active_set->push_dlist, peer, active_set->peers );
288 0 : }
289 0 : fd_gossip_txbuild_append( peer->txbuild, crds_sz, crds_val );
290 0 : if( FD_UNLIKELY( flush_immediately ) ) push_flush( active_set, peer, stem, now );
291 0 : pushed++;
292 0 : }
293 0 : }
294 :
295 : static inline void
296 : rotate_active_set( fd_active_set_t * active_set,
297 : fd_stem_context_t * stem,
298 0 : long now ) {
299 0 : ulong num_bloom_filter_items = fd_ulong_max( fd_crds_peer_count( active_set->crds ), 512UL );
300 :
301 0 : ulong bucket = active_set->rotate_bucket;
302 0 : active_set->rotate_bucket = (active_set->rotate_bucket+1UL) % 25UL;
303 0 : fd_active_set_entry_t * entry = active_set->entries[ bucket ];
304 :
305 : /* Sample a new peer BEFORE evicting the oldest. This prevents the
306 : case where we evict a peer back into the sampler and then
307 : immediately re-sample it, creating a duplicate. A full entry
308 : swaps its oldest peer; a partial one fills up to 12 at once. */
309 :
310 0 : do {
311 0 : ulong added_ci_idx = fd_gossip_wsample_sample_remove_bucket( active_set->wsample, bucket );
312 0 : if( FD_UNLIKELY( added_ci_idx==ULONG_MAX ) ) return;
313 :
314 0 : ulong replace_idx;
315 0 : if( FD_LIKELY( entry->nodes_len==12UL ) ) {
316 0 : replace_idx = entry->nodes_idx;
317 0 : entry->nodes_idx = (entry->nodes_idx+1UL) % 12UL;
318 :
319 : /* Add the replaced peer back to the sampler. */
320 0 : ulong old_ci_idx = active_set->peers[ bucket*12UL+replace_idx ].ci_idx;
321 0 : fd_gossip_wsample_add_bucket( active_set->wsample, bucket, old_ci_idx );
322 0 : push_flush( active_set, &active_set->peers[ bucket*12UL+replace_idx ], stem, now );
323 0 : } else {
324 0 : replace_idx = (entry->nodes_idx+entry->nodes_len) % 12UL;
325 0 : entry->nodes_len++;
326 0 : }
327 :
328 0 : fd_active_set_peer_t * replace = &active_set->peers[ bucket*12UL+replace_idx ];
329 0 : replace->ci_idx = added_ci_idx;
330 0 : uchar const * new_pubkey = fd_crds_ci_pubkey( active_set->crds, added_ci_idx );
331 :
332 0 : fd_bloom_initialize( replace->bloom, num_bloom_filter_items );
333 0 : fd_bloom_insert( replace->bloom, new_pubkey, 32UL );
334 0 : fd_gossip_txbuild_init( replace->txbuild, active_set->identity_pubkey, FD_GOSSIP_MESSAGE_PUSH );
335 0 : } while( entry->nodes_len<12UL );
336 0 : }
337 :
338 :
339 : void
340 : fd_active_set_advance( fd_active_set_t * active_set,
341 : fd_stem_context_t * stem,
342 : long now,
343 0 : int * charge_busy ) {
344 0 : while( !push_dlist_is_empty( active_set->push_dlist, active_set->peers ) ) {
345 0 : fd_active_set_peer_t * head = push_dlist_ele_peek_head( active_set->push_dlist, active_set->peers );
346 0 : if( FD_LIKELY( head->timestamp>=now-1L*1000L*1000L ) ) break;
347 :
348 0 : push_flush( active_set, head, stem, now );
349 0 : if( charge_busy ) *charge_busy = 1;
350 0 : }
351 :
352 0 : if( FD_UNLIKELY( now>=active_set->next_rotate_nanos ) ) {
353 0 : rotate_active_set( active_set, stem, now );
354 0 : active_set->next_rotate_nanos = now+300L*1000L*1000L;
355 0 : if( charge_busy ) *charge_busy = 1;
356 0 : }
357 0 : }
358 :
359 : long
360 0 : fd_active_set_next_deadline( fd_active_set_t const * active_set ) {
361 0 : long deadline = active_set->next_rotate_nanos;
362 0 : if( FD_LIKELY( !push_dlist_is_empty( active_set->push_dlist, active_set->peers ) ) ) {
363 0 : fd_active_set_peer_t const * head = push_dlist_ele_peek_head_const( active_set->push_dlist, active_set->peers );
364 0 : deadline = fd_long_min( deadline, head->timestamp+1L*1000L*1000L+1L );
365 0 : }
366 0 : return deadline;
367 0 : }
|
