Line data Source code
1 : #include "fd_repair_strat.h"
2 :
3 : FD_FN_CONST ulong
4 117 : fd_repair_strat_align( void ) {
5 117 : return 128UL;
6 117 : }
7 :
8 : FD_FN_CONST ulong
9 18 : fd_repair_strat_footprint( void ) {
10 18 : ulong chain_cnt = fd_repair_strat_peer_map_chain_cnt_est( FD_REPAIR_STRAT_PEER_MAX );
11 18 : ulong l = FD_LAYOUT_INIT;
12 18 : l = FD_LAYOUT_APPEND( l, fd_repair_strat_align(), sizeof(fd_repair_strat_t) );
13 18 : l = FD_LAYOUT_APPEND( l, alignof(fd_repair_strat_peer_t), sizeof(fd_repair_strat_peer_t)*FD_REPAIR_STRAT_PEER_MAX );
14 18 : l = FD_LAYOUT_APPEND( l, alignof(fd_repair_strat_peer_t), sizeof(fd_repair_strat_peer_t)*FD_REPAIR_STRAT_PEER_MAX );
15 18 : l = FD_LAYOUT_APPEND( l, fd_repair_strat_peer_map_align(), fd_repair_strat_peer_map_footprint( chain_cnt ) );
16 18 : l = FD_LAYOUT_APPEND( l, fd_repair_strat_peer_map_align(), fd_repair_strat_peer_map_footprint( chain_cnt ) );
17 18 : l = FD_LAYOUT_APPEND( l, alignof(uint), sizeof(uint)*FD_REPAIR_STRAT_PEER_MAX );
18 126 : for( ulong b=0UL; b<FD_REPAIR_STRAT_BUCKET_CNT; b++ ) l = FD_LAYOUT_APPEND( l, fd_repair_strat_set_align(), fd_repair_strat_set_footprint() );
19 18 : return FD_LAYOUT_FINI( l, fd_repair_strat_align() );
20 18 : }
21 :
22 : void *
23 : fd_repair_strat_new( void * shmem,
24 15 : ulong seed ) {
25 15 : ulong chain_cnt = fd_repair_strat_peer_map_chain_cnt_est( FD_REPAIR_STRAT_PEER_MAX );
26 :
27 15 : FD_SCRATCH_ALLOC_INIT( l, shmem );
28 15 : fd_repair_strat_t * strat = FD_SCRATCH_ALLOC_APPEND( l, fd_repair_strat_align(), sizeof(fd_repair_strat_t) );
29 15 : void * cur_peers = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_repair_strat_peer_t), sizeof(fd_repair_strat_peer_t)*FD_REPAIR_STRAT_PEER_MAX );
30 15 : void * old_peers = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_repair_strat_peer_t), sizeof(fd_repair_strat_peer_t)*FD_REPAIR_STRAT_PEER_MAX );
31 15 : void * cur_map = FD_SCRATCH_ALLOC_APPEND( l, fd_repair_strat_peer_map_align(), fd_repair_strat_peer_map_footprint( chain_cnt ) );
32 15 : void * old_map = FD_SCRATCH_ALLOC_APPEND( l, fd_repair_strat_peer_map_align(), fd_repair_strat_peer_map_footprint( chain_cnt ) );
33 15 : uint * free = FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), sizeof(uint)*FD_REPAIR_STRAT_PEER_MAX );
34 105 : for( ulong b=0UL; b<FD_REPAIR_STRAT_BUCKET_CNT; b++ ) {
35 90 : strat->buckets[ b ] = fd_repair_strat_set_join( fd_repair_strat_set_new( FD_SCRATCH_ALLOC_APPEND( l, fd_repair_strat_set_align(), fd_repair_strat_set_footprint() ) ) );
36 90 : strat->cursor [ b ] = fd_ulong_hash( seed^b ) % FD_REPAIR_STRAT_PEER_MAX; /* random start, so nodes with the same stake order do not herd */
37 90 : }
38 15 : FD_TEST( FD_SCRATCH_ALLOC_FINI( l, fd_repair_strat_align() )==(ulong)shmem + fd_repair_strat_footprint() );
39 :
40 15 : strat->cur.peers = (fd_repair_strat_peer_t *)cur_peers;
41 15 : strat->old.peers = (fd_repair_strat_peer_t *)old_peers;
42 15 : strat->cur.map = fd_repair_strat_peer_map_join( fd_repair_strat_peer_map_new( cur_map, chain_cnt, seed ) );
43 15 : strat->old.map = fd_repair_strat_peer_map_join( fd_repair_strat_peer_map_new( old_map, chain_cnt, seed ) );
44 15 : strat->staked_cnt = 0UL;
45 15 : strat->free = free;
46 15 : strat->free_cnt = FD_REPAIR_STRAT_PEER_MAX;
47 491535 : for( ulong i=0UL; i<FD_REPAIR_STRAT_PEER_MAX; i++ ) free[ i ] = (uint)( FD_REPAIR_STRAT_PEER_MAX-1UL-i ); /* lowest slot on top */
48 15 : strat->pick_cnt = 0UL;
49 15 : strat->seed = seed;
50 15 : strat->epoch_cnt = 0UL;
51 15 : strat->turbine_srtt = 0L;
52 15 : strat->turbine_rttvar = 0L;
53 15 : return shmem;
54 15 : }
55 :
56 : fd_repair_strat_t *
57 15 : fd_repair_strat_join( void * shstrat ) {
58 15 : return (fd_repair_strat_t *)shstrat;
59 15 : }
60 :
61 : static ulong
62 49083 : bucket( fd_repair_strat_peer_t const * peer ) {
63 49083 : long rtt = peer->srtt_ns+(peer->rttvar_ns>>2); /* a quarter weight on jitter: a jittery peer ranks slower, as its hedge does, but a first sample's rttvar of rtt/2 does not demote a new peer */
64 49083 : if( FD_UNLIKELY( !peer->srtt_ns ) ) return FD_REPAIR_STRAT_BUCKET_UNMEASURED;
65 18864 : if( FD_LIKELY ( rtt< 25L*1000000L ) ) return FD_REPAIR_STRAT_BUCKET_25MS;
66 597 : if( FD_LIKELY ( rtt< 50L*1000000L ) ) return FD_REPAIR_STRAT_BUCKET_50MS;
67 567 : if( FD_LIKELY ( rtt<100L*1000000L ) ) return FD_REPAIR_STRAT_BUCKET_100MS;
68 528 : if( FD_LIKELY ( rtt<200L*1000000L ) ) return FD_REPAIR_STRAT_BUCKET_200MS;
69 120 : return FD_REPAIR_STRAT_BUCKET_SLOW;
70 528 : }
71 :
72 : /* place puts slot idx in the bucket of its rtt if it can take another
73 : request, and in no bucket otherwise. */
74 :
75 : static void
76 : place( fd_repair_strat_t * self,
77 49083 : ulong idx ) {
78 49083 : fd_repair_strat_peer_t * peer = &self->cur.peers[ idx ];
79 49083 : fd_repair_strat_set_remove( self->buckets[ peer->bucket ], idx );
80 49083 : peer->bucket = bucket( peer );
81 49083 : if( FD_LIKELY( peer->ip4 && peer->out_cnt<FD_REPAIR_STRAT_INFLIGHT_MAX ) ) fd_repair_strat_set_insert( self->buckets[ peer->bucket ], idx );
82 49083 : }
83 :
84 : void
85 : fd_repair_strat_epoch_advanced( fd_repair_strat_t * self,
86 : fd_stake_weight_t const * ids,
87 24 : ulong id_cnt ) {
88 24 : fd_repair_strat_order_t old = self->cur;
89 24 : self->cur = self->old;
90 24 : self->old = old;
91 :
92 24 : fd_repair_strat_peer_map_reset( self->cur.map );
93 24 : self->epoch_cnt++;
94 168 : for( ulong b=0UL; b<FD_REPAIR_STRAT_BUCKET_CNT; b++ ) {
95 144 : fd_repair_strat_set_null( self->buckets[ b ] );
96 144 : self->cursor[ b ] = fd_ulong_hash( self->seed^(self->epoch_cnt<<8)^b ) % FD_REPAIR_STRAT_PEER_MAX;
97 144 : }
98 :
99 : /* Staked identities take [0,staked_cnt) by stake rank, keeping what
100 : was known about them. */
101 :
102 24 : ulong cnt = fd_ulong_min( id_cnt, FD_REPAIR_STRAT_PEER_MAX );
103 87 : for( ulong rank=0UL; rank<cnt; rank++ ) {
104 63 : fd_repair_strat_peer_t * peer = &self->cur.peers[ rank ];
105 63 : fd_repair_strat_peer_t const * prev = fd_repair_strat_peer_map_ele_query_const( self->old.map, &ids[ rank ].key, NULL, self->old.peers );
106 63 : if( FD_LIKELY( prev ) ) *peer = *prev;
107 9 : else {
108 9 : fd_memset( peer, 0, sizeof(fd_repair_strat_peer_t) );
109 9 : peer->id_key = ids[ rank ].key;
110 9 : peer->bucket = FD_REPAIR_STRAT_BUCKET_UNMEASURED;
111 9 : }
112 63 : fd_repair_strat_peer_map_ele_insert( self->cur.map, peer, self->cur.peers );
113 63 : place( self, rank );
114 63 : }
115 24 : self->staked_cnt = cnt;
116 :
117 : /* Unstaked peers with an address take the slots after. */
118 :
119 24 : self->free_cnt = 0UL;
120 786393 : for( ulong i=FD_REPAIR_STRAT_PEER_MAX; i>cnt; i-- ) self->free[ self->free_cnt++ ] = (uint)( i-1UL );
121 24 : for( fd_repair_strat_peer_map_iter_t iter = fd_repair_strat_peer_map_iter_init( self->old.map, self->old.peers );
122 87 : !fd_repair_strat_peer_map_iter_done( iter, self->old.map, self->old.peers );
123 63 : iter = fd_repair_strat_peer_map_iter_next( iter, self->old.map, self->old.peers ) ) {
124 63 : fd_repair_strat_peer_t const * prev = fd_repair_strat_peer_map_iter_ele_const( iter, self->old.map, self->old.peers );
125 63 : if( FD_UNLIKELY( !prev->ip4 || !self->free_cnt ) ) continue;
126 60 : if( FD_LIKELY( fd_repair_strat_peer_map_ele_query_const( self->cur.map, &prev->id_key, NULL, self->cur.peers ) ) ) continue; /* staked now */
127 9 : ulong idx = self->free[ --self->free_cnt ];
128 9 : self->cur.peers[ idx ] = *prev;
129 9 : fd_repair_strat_peer_map_ele_insert( self->cur.map, &self->cur.peers[ idx ], self->cur.peers );
130 9 : place( self, idx );
131 9 : }
132 24 : }
133 :
134 : void
135 : fd_repair_strat_contact_info_updated( fd_repair_strat_t * self,
136 : fd_pubkey_t const * id_key,
137 : uint ip4,
138 48 : ushort port ) {
139 48 : fd_repair_strat_peer_t * peer = fd_repair_strat_peer_map_ele_query( self->cur.map, id_key, NULL, self->cur.peers );
140 48 : if( FD_UNLIKELY( !peer && !self->free_cnt ) ) return;
141 48 : if( FD_UNLIKELY( !peer ) ) {
142 42 : peer = &self->cur.peers[ self->free[ --self->free_cnt ] ];
143 42 : fd_memset( peer, 0, sizeof(fd_repair_strat_peer_t) );
144 42 : peer->id_key = *id_key;
145 42 : peer->bucket = FD_REPAIR_STRAT_BUCKET_UNMEASURED;
146 42 : fd_repair_strat_peer_map_ele_insert( self->cur.map, peer, self->cur.peers );
147 42 : }
148 48 : if( FD_LIKELY( peer->ip4==ip4 && peer->port==port ) ) return;
149 :
150 48 : peer->ip4 = ip4;
151 48 : peer->port = port;
152 48 : peer->srtt_ns = 0L;
153 48 : peer->rttvar_ns = 0L;
154 48 : peer->ping_ts = 0L;
155 48 : place( self, (ulong)( peer-self->cur.peers ) );
156 48 : }
157 :
158 : void
159 : fd_repair_strat_contact_info_removed( fd_repair_strat_t * self,
160 12 : fd_pubkey_t const * id_key ) {
161 12 : fd_repair_strat_peer_t * peer = fd_repair_strat_peer_map_ele_query( self->cur.map, id_key, NULL, self->cur.peers );
162 12 : if( FD_UNLIKELY( !peer ) ) return;
163 :
164 12 : ulong idx = (ulong)( peer-self->cur.peers );
165 12 : peer->ip4 = 0U;
166 12 : place( self, idx );
167 12 : if( FD_LIKELY( idx<self->staked_cnt ) ) return; /* a staked peer keeps its slot */
168 :
169 9 : fd_repair_strat_peer_map_ele_remove( self->cur.map, id_key, NULL, self->cur.peers );
170 9 : self->free[ self->free_cnt++ ] = (uint)idx;
171 9 : }
172 :
173 : fd_repair_strat_peer_t *
174 : fd_repair_strat_pick( fd_repair_strat_t * self,
175 : long now,
176 : int staked,
177 210 : ulong avoid ) {
178 210 : ulong end = fd_ulong_if( staked, self->staked_cnt, FD_REPAIR_STRAT_PEER_MAX );
179 210 : int explore = !( ++self->pick_cnt % FD_REPAIR_STRAT_EXPLORE );
180 210 : int fallback = 0; /* avoid was available */
181 858 : for( ulong i=0UL; i<FD_REPAIR_STRAT_BUCKET_CNT; i++ ) {
182 843 : ulong b = fd_ulong_if( explore, (i+FD_REPAIR_STRAT_BUCKET_UNMEASURED)%FD_REPAIR_STRAT_BUCKET_CNT, i ); /* unmeasured first when exploring */
183 843 : fd_repair_strat_set_t const * set = self->buckets[ b ];
184 843 : ulong start = self->cursor[ b ];
185 :
186 : /* The next two peers after the cursor, walking to end and then
187 : from the start back to the cursor. */
188 :
189 843 : ulong cand[ 2 ];
190 843 : ulong cand_cnt = 0UL;
191 843 : ulong first = ULONG_MAX; /* first available peer, avoid included */
192 843 : int wrap = 0;
193 843 : ulong idx = fd_repair_strat_set_const_iter_next( set, start );
194 2031 : while( cand_cnt<2UL ) {
195 1917 : if( FD_UNLIKELY( fd_repair_strat_set_const_iter_done( idx ) || idx>=end ) ) {
196 1542 : if( wrap ) break;
197 822 : wrap = 1;
198 822 : idx = fd_repair_strat_set_const_iter_init( set );
199 822 : continue;
200 1542 : }
201 375 : if( FD_UNLIKELY( wrap && idx>start ) ) break;
202 366 : if( FD_LIKELY( self->cur.peers[ idx ].ban_ts<=now ) ) {
203 327 : first = fd_ulong_if( first==ULONG_MAX, idx, first );
204 327 : if( FD_LIKELY( idx!=avoid ) ) cand[ cand_cnt++ ] = idx;
205 18 : else fallback = 1;
206 327 : }
207 366 : idx = fd_repair_strat_set_const_iter_next( set, idx );
208 366 : }
209 843 : if( FD_UNLIKELY( !cand_cnt ) ) continue;
210 :
211 195 : fd_repair_strat_peer_t const * c0 = &self->cur.peers[ cand[ 0 ] ];
212 195 : fd_repair_strat_peer_t const * c1 = &self->cur.peers[ cand[ cand_cnt-1UL ] ];
213 195 : ulong won = fd_ulong_if( (ulong)(c1->srtt_ns+1L)*(c1->out_cnt+1UL)<(ulong)(c0->srtt_ns+1L)*(c0->out_cnt+1UL), cand[ cand_cnt-1UL ], cand[ 0 ] );
214 195 : self->cursor[ b ] = first; /* one peer on, avoided or not, so pairs shift rather than repeat */
215 195 : self->cur.peers[ won ].out_cnt++;
216 195 : place( self, won );
217 195 : return &self->cur.peers[ won ];
218 843 : }
219 15 : if( FD_UNLIKELY( fallback ) ) {
220 3 : self->cur.peers[ avoid ].out_cnt++;
221 3 : place( self, avoid );
222 3 : return &self->cur.peers[ avoid ];
223 3 : }
224 12 : return NULL;
225 15 : }
226 :
227 : fd_repair_strat_peer_t *
228 : fd_repair_strat_pick_staked( fd_repair_strat_t * self,
229 : long now,
230 : ulong avoid,
231 24003 : ulong r ) {
232 24003 : ulong n = fd_ulong_min( self->staked_cnt, FD_REPAIR_STRAT_ESCAPE_TOP );
233 24003 : ulong m = 0UL;
234 120015 : for( ulong idx=0UL; idx<n; idx++ ) {
235 96012 : fd_repair_strat_peer_t const * peer = &self->cur.peers[ idx ];
236 96012 : m += (ulong)( idx!=avoid && peer->ip4 && peer->out_cnt<FD_REPAIR_STRAT_INFLIGHT_MAX && peer->ban_ts<=now );
237 96012 : }
238 24003 : if( FD_UNLIKELY( !m ) ) return NULL;
239 :
240 24000 : ulong k = fd_ulong_min( (r&UINT_MAX)%m, (r>>32)%m ); /* among the available peers, in stake order */
241 44802 : for( ulong idx=0UL; idx<n; idx++ ) {
242 44802 : fd_repair_strat_peer_t * peer = &self->cur.peers[ idx ];
243 44802 : if( idx==avoid || !peer->ip4 || peer->out_cnt>=FD_REPAIR_STRAT_INFLIGHT_MAX || peer->ban_ts>now ) continue;
244 29868 : if( k-- ) continue;
245 24000 : peer->out_cnt++;
246 24000 : place( self, idx );
247 24000 : return peer;
248 29868 : }
249 0 : return NULL;
250 24000 : }
251 :
252 : void
253 : fd_repair_strat_request_done( fd_repair_strat_t * self,
254 : fd_pubkey_t const * id_key,
255 : long rtt,
256 450 : int byid ) {
257 450 : fd_repair_strat_peer_t * peer = fd_repair_strat_peer_map_ele_query( self->cur.map, id_key, NULL, self->cur.peers );
258 450 : if( FD_UNLIKELY( !peer ) ) return; /* forgotten while the request was in flight */
259 :
260 450 : peer->out_cnt -= !!peer->out_cnt;
261 450 : peer->expired_cnt = 0U;
262 450 : peer->byid_expired_cnt = fd_uint_if( byid, 0U, peer->byid_expired_cnt ); /* a positional reply does not vouch for by-id */
263 450 : peer->rttvar_ns = fd_long_if( !!peer->srtt_ns, (3L*peer->rttvar_ns+(long)fd_long_abs( peer->srtt_ns-rtt ))/4L, rtt/2L ); /* RFC 6298 */
264 450 : peer->srtt_ns = fd_long_if( !!peer->srtt_ns, (7L*peer->srtt_ns+rtt)/8L, rtt );
265 450 : place( self, (ulong)( peer-self->cur.peers ) );
266 450 : }
267 :
268 : void
269 : fd_repair_strat_request_failed( fd_repair_strat_t * self,
270 : fd_pubkey_t const * id_key,
271 24087 : long ban_ts ) {
272 24087 : fd_repair_strat_peer_t * peer = fd_repair_strat_peer_map_ele_query( self->cur.map, id_key, NULL, self->cur.peers );
273 24087 : if( FD_UNLIKELY( !peer ) ) return; /* forgotten while the request was in flight */
274 :
275 24087 : peer->out_cnt -= !!peer->out_cnt;
276 24087 : peer->ban_ts = fd_long_max( peer->ban_ts, ban_ts ); /* an expiry does not shorten a verify failure's ban */
277 24087 : place( self, (ulong)( peer-self->cur.peers ) );
278 24087 : }
279 :
280 : int
281 : fd_repair_strat_request_expired( fd_repair_strat_t * self,
282 : fd_pubkey_t const * id_key,
283 : long ban_ts,
284 216 : int byid ) {
285 216 : fd_repair_strat_peer_t * peer = fd_repair_strat_peer_map_ele_query( self->cur.map, id_key, NULL, self->cur.peers );
286 216 : if( FD_UNLIKELY( !peer ) ) return 0; /* forgotten while the request was in flight */
287 :
288 216 : peer->out_cnt -= !!peer->out_cnt;
289 216 : peer->byid_expired_cnt += (uint)!!byid;
290 216 : int byid_ban = peer->byid_expired_cnt>=FD_REPAIR_STRAT_BYID_MAX;
291 216 : int ban = fd_int_if( byid_ban, 1, -(++peer->expired_cnt>=FD_REPAIR_STRAT_EXPIRED_MAX) );
292 216 : if( FD_UNLIKELY( ban ) ) {
293 9 : peer->ban_ts = fd_long_max( peer->ban_ts, ban_ts );
294 9 : peer->expired_cnt = 0U;
295 9 : peer->byid_expired_cnt = 0U;
296 9 : }
297 216 : place( self, (ulong)( peer-self->cur.peers ) );
298 216 : return ban;
299 216 : }
300 :
301 : void
302 : fd_repair_strat_turbine_done( fd_repair_strat_t * self,
303 : fd_pubkey_t const * id_key,
304 411 : long t ) {
305 411 : self->turbine_rttvar = fd_long_if( !!self->turbine_srtt, (3L*self->turbine_rttvar+(long)fd_long_abs( self->turbine_srtt-t ))/4L, t/2L );
306 411 : self->turbine_srtt = fd_long_if( !!self->turbine_srtt, (7L*self->turbine_srtt+t)/8L, t );
307 :
308 411 : fd_repair_strat_peer_t * peer = id_key ? fd_repair_strat_peer_map_ele_query( self->cur.map, id_key, NULL, self->cur.peers ) : NULL;
309 411 : if( FD_UNLIKELY( !peer ) ) return;
310 411 : peer->turbine_rttvar = fd_long_if( !!peer->turbine_srtt, (3L*peer->turbine_rttvar+(long)fd_long_abs( peer->turbine_srtt-t ))/4L, t/2L );
311 411 : peer->turbine_srtt = fd_long_if( !!peer->turbine_srtt, (7L*peer->turbine_srtt+t)/8L, t );
312 411 : peer->turbine_cnt++;
313 411 : }
314 :
315 : long
316 : fd_repair_strat_eager_ns( fd_repair_strat_t * self,
317 33 : fd_pubkey_t const * id_key ) {
318 33 : fd_repair_strat_peer_t const * peer = id_key ? fd_repair_strat_peer_map_ele_query_const( self->cur.map, id_key, NULL, self->cur.peers ) : NULL;
319 33 : int own = peer && peer->turbine_cnt>=FD_REPAIR_STRAT_TURBINE_MIN;
320 33 : long srtt = own ? peer->turbine_srtt : self->turbine_srtt;
321 33 : long rttvar = own ? peer->turbine_rttvar : self->turbine_rttvar;
322 33 : if( FD_UNLIKELY( !srtt ) ) return FD_REPAIR_STRAT_EAGER_MAX_NS;
323 27 : return fd_long_min( fd_long_max( srtt+4L*rttvar, FD_REPAIR_STRAT_EAGER_MIN_NS ), FD_REPAIR_STRAT_EAGER_MAX_NS );
324 33 : }
325 :
326 : fd_repair_strat_peer_t *
327 : fd_repair_strat_query( fd_repair_strat_t * self,
328 234 : fd_pubkey_t const * id_key ) {
329 234 : return fd_repair_strat_peer_map_ele_query( self->cur.map, id_key, NULL, self->cur.peers );
330 234 : }
331 :
332 : ulong
333 : fd_repair_strat_slot_remap( fd_repair_strat_t const * self,
334 18 : ulong idx ) {
335 18 : if( FD_UNLIKELY( idx>=FD_REPAIR_STRAT_PEER_MAX ) ) return ULONG_MAX;
336 15 : fd_repair_strat_peer_t const * peer = fd_repair_strat_peer_map_ele_query_const( self->cur.map, &self->old.peers[ idx ].id_key, NULL, self->cur.peers );
337 15 : return peer ? (ulong)( peer-self->cur.peers ) : ULONG_MAX;
338 18 : }
|
