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