Line data Source code
1 : #define _GNU_SOURCE
2 :
3 : #include "../../disco/topo/fd_topo.h"
4 : #include "../../disco/fd_clock_tile.h"
5 : #include <linux/futex.h>
6 : #include "generated/fd_rotor_tile_seccomp.h"
7 : #include "../../disco/shred/fd_shred_tile.h"
8 : #include "../../choreo/rotor/fd_rotor.h"
9 : #include "../../choreo/votor/ag_cert.h"
10 : #include "../genesis/fd_genesi_tile.h"
11 : #include "../restore/utils/fd_ssmsg.h"
12 : #include "../votor/fd_votor_tile.h"
13 : #include "../replay/fd_replay_tile.h"
14 : #include "../../disco/keyguard/fd_keyload.h"
15 : #include "../../disco/keyguard/fd_keyswitch.h"
16 : #include "../../flamenco/gossip/fd_gossip_message.h"
17 : #include "../../flamenco/leaders/fd_leaders_base.h"
18 : #include "../../flamenco/leaders/fd_multi_epoch_leaders.h"
19 : #include "../../choreo/rotor/fd_rotor_serde.h"
20 : #include "../../choreo/rotor/fd_rotor_strat.h"
21 : #include "../../ballet/sha256/fd_sha256.h"
22 : #include "../../disco/keyguard/fd_keyguard.h"
23 : #include "../../disco/events/generated/fd_event_gen.h"
24 : #include "../../disco/metrics/fd_metrics.h"
25 : #include "../../disco/net/fd_net_tile.h"
26 : #include "../../disco/shred/fd_rnonce_ss.h"
27 : #include "../../util/net/fd_net_headers.h"
28 : #include "../../util/pod/fd_pod_format.h"
29 : #include "fd_rotor_tile.h"
30 :
31 : /* timeout: tracks a repair request we've made that expires at the given
32 : time. If the repair request is not responded by timeout, retry. */
33 :
34 : struct timeout {
35 : long timeout; /* ns */
36 : ulong slot;
37 : ulong blk; /* blk_pool idx, checked against slot when due */
38 : uint fec_idx; /* FEC index, WINDOW_INDEX only */
39 : uchar tag; /* FD_ROTOR_SERDE_TAG_*, WINDOW_INDEX for any FEC set */
40 : uchar attempt;
41 : ushort last_peer; /* 1+strat slot of the peer last asked, 0 if none */
42 : uint rcvd; /* the FEC set's rcvd at the last attempt */
43 : uint stuck; /* attempts in a row that brought no new shred */
44 : };
45 : typedef struct timeout timeout_t;
46 : FD_STATIC_ASSERT( FD_ROTOR_STRAT_PEER_MAX<USHORT_MAX, last_peer );
47 :
48 : #define PRQ_NAME timeout_prq
49 0 : #define PRQ_T timeout_t
50 : #include "../../util/tmpl/fd_prq.c"
51 :
52 : /* pending: an outstanding request we're waiting for to either receive a
53 : response or get timed out. one-to-many relationship with timeout
54 : (there can be many pendings per timeout). */
55 :
56 0 : #define PENDING_TTL ((long)FD_RNONCE_SS_DELTA_MAX<<24) /* a shred reply's nonce stops verifying after this */
57 0 : #define SIGN_NS (6500L) /* fd_ed25519_sign of a request, test_ed25519 --bench */
58 0 : #define FEC_SET_P90 (38UL) /* p90 fec_set_cnt */
59 :
60 : struct pending {
61 : ulong key;
62 : ulong next; /* pool next */
63 : struct {
64 : ulong prev;
65 : ulong next;
66 : } map;
67 : struct {
68 : ulong prev;
69 : ulong next;
70 : } dlist;
71 : long ts; /* sent timestamp, nanos */
72 : ulong slot;
73 : ulong blk; /* blk_pool idx */
74 : fd_mr32_t dmr;
75 : fd_pubkey_t peer; /* id_key of the peer picked */
76 : uint fec_idx; /* FEC index */
77 : uint fec_set_cnt; /* the blk's when sent */
78 : uint tag; /* FD_ROTOR_SERDE_TAG_* of the request */
79 : uint mask; /* bit i set while shred i of the FEC set is unanswered, bit 0 for a HIGHEST or ORPHAN; the pending outlives its last answer */
80 : uint answered; /* shreds matched so far */
81 : };
82 : typedef struct pending pending_t;
83 :
84 : #define POOL_NAME pending_pool
85 0 : #define POOL_T pending_t
86 : #include "../../util/tmpl/fd_pool.c"
87 :
88 : #define MAP_NAME pending_map
89 0 : #define MAP_ELE_T pending_t
90 0 : #define MAP_PREV map.prev
91 0 : #define MAP_NEXT map.next
92 : #define MAP_MULTI 1
93 : #define MAP_OPTIMIZE_RANDOM_ACCESS_REMOVAL 1
94 : #include "../../util/tmpl/fd_map_chain.c"
95 :
96 : #define DLIST_NAME pending_dlist
97 : #define DLIST_ELE_T pending_t
98 0 : #define DLIST_PREV dlist.prev
99 0 : #define DLIST_NEXT dlist.next
100 : #include "../../util/tmpl/fd_dlist.c"
101 :
102 : /* Messages waiting for their signature, idx in the sig's high 32 bits. */
103 :
104 : #define SIGN_MAX (16UL)
105 0 : #define SIGN_RESERVE (8UL) /* sign slots only pongs may take */
106 0 : #define PING_MAX (4UL) /* pings per after_credit */
107 :
108 : struct sign {
109 : ulong next;
110 : uchar buf[ FD_ROTOR_SER_MAX ];
111 : ulong sz;
112 : ulong sig_off; /* where the signature goes in buf */
113 : uint saddr; /* 0 lets the net tile choose */
114 : uint daddr;
115 : ushort dport; /* net order */
116 : };
117 : typedef struct sign sign_t;
118 :
119 : #define POOL_NAME sign_pool
120 0 : #define POOL_T sign_t
121 : #include "../../util/tmpl/fd_pool.c"
122 :
123 : /* Peers with a new address get one signed request to elicit a ping. */
124 :
125 : #define DEQUE_NAME ping_deque
126 0 : #define DEQUE_T fd_pubkey_t
127 : #include "../../util/tmpl/fd_deque_dynamic.c"
128 :
129 0 : #define PING_TTL (500L*1000000L) /* a ping is answered this long after we sent its peer a request, Agave's 2*DELTA */
130 0 : #define BAN_TTL (10L*1000000000L) /* a peer whose response fails to verify is not picked for this long */
131 0 : #define STUCK_MAX (3U) /* attempts with no new shred before a retry leaves the bucket order */
132 0 : #define NO_PEER (10L*1000000L) /* a req with no peer to pick waits this long */
133 :
134 : static fd_ed25519_sig_t const sig_null = {0};
135 :
136 0 : #define IN_KIND_GENESIS (0)
137 0 : #define IN_KIND_SHRED (1)
138 0 : #define IN_KIND_SNAP (2)
139 0 : #define IN_KIND_VOTOR (3)
140 0 : #define IN_KIND_NET (4)
141 0 : #define IN_KIND_GOSSIP (5)
142 0 : #define IN_KIND_REPLAY (6)
143 0 : #define IN_KIND_SIGN (7)
144 0 : #define IN_KIND_EPOCH (8)
145 :
146 : struct fd_rotor_tile {
147 :
148 : /* Signing */
149 :
150 : fd_pubkey_t identity_key;
151 : int staked;
152 : fd_keyswitch_t * keyswitch;
153 : int halt_signing;
154 : sign_t * sign_pool;
155 : ulong sign_cnt;
156 : struct {
157 : ulong idx;
158 : fd_wksp_t * mem;
159 : ulong chunk0;
160 : ulong wmark;
161 : ulong chunk;
162 : ulong free;
163 : } sign[SIGN_MAX];
164 : uchar sign_buf[FD_ED25519_SIG_SZ];
165 :
166 : /* Catchup */
167 :
168 : ulong cert_slot0; /* staked: first cert slot, 0 if none yet */
169 : ulong turbine_slot0; /* unstaked: first turbine slot, 0 if none yet */
170 : ulong catchup_slot; /* highest slot past the window with a verified cert (staked) or signed future shred (unstaked), 0 if none yet */
171 : ulong highest_fec_complete_slot; /* highest slot a FEC set completed for off the network, our leader sets excluded: the cluster tip, 0 if none yet */
172 :
173 : /* Snapshot */
174 :
175 : ulong snapshot_slot; /* root of the snapshot being loaded, applied at FD_SSMSG_DONE */
176 : fd_mr32_t snapshot_blk_mr;
177 :
178 : /* Epoch metadata */
179 :
180 : fd_multi_epoch_leaders_t * mleaders;
181 :
182 : /* Rotor data structures */
183 :
184 : ulong seed;
185 : fd_rotor_t * rotor;
186 : fd_rotor_strat_t * strat;
187 :
188 : /* Requests */
189 :
190 : fd_clock_tile_t clock[1];
191 : fd_rng_t rng[1];
192 : fd_rnonce_ss_t rnonce_ss[1];
193 : timeout_t * eager_prq;
194 : timeout_t * notar_prq;
195 : timeout_t * final_prq;
196 : pending_t * pending_pool;
197 : pending_map_t * pending_map;
198 : pending_dlist_t * pending_dlist;
199 :
200 : /* Networking */
201 :
202 : int allow_private_address;
203 : fd_pubkey_t * ping_deque;
204 : fd_ip4_udp_hdrs_t net_hdr[1];
205 : ushort net_id;
206 : uchar net_buf[ FD_NET_MTU ];
207 :
208 : /* Links */
209 :
210 : int in_kind[32];
211 : union {
212 : struct {
213 : fd_wksp_t * mem;
214 : ulong chunk0;
215 : ulong wmark;
216 : ulong mtu;
217 : ulong sign_idx; /* IN_KIND_SIGN, its sign tile in sign */
218 : };
219 : fd_net_rx_bounds_t net_rx; /* IN_KIND_NET */
220 : } in[ 32 ];
221 : ulong chunk; /* reliable ins, after_frag */
222 :
223 : ulong net_out_idx;
224 : fd_wksp_t * net_out_mem;
225 : ulong net_out_chunk0;
226 : ulong net_out_wmark;
227 : ulong net_out_chunk;
228 :
229 : ulong replay_out_idx;
230 : fd_wksp_t * replay_out_mem;
231 : ulong replay_out_chunk0;
232 : ulong replay_out_wmark;
233 : ulong replay_out_chunk;
234 :
235 : ulong rserve_out_idx; /* ULONG_MAX if rserve is disabled */
236 : fd_wksp_t * rserve_out_mem;
237 : ulong rserve_out_chunk0;
238 : ulong rserve_out_wmark;
239 : ulong rserve_out_chunk;
240 :
241 : /* Events */
242 :
243 : fd_event_block_received_t * event;
244 :
245 : /* Metrics */
246 :
247 : struct {
248 : ulong pkt_tx;
249 : ulong request_tx[ FD_ROTOR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID+1U ]; /* by FD_ROTOR_SERDE_TAG_* */
250 : ulong ping_tx;
251 : ulong slot_highest_repaired;
252 : ulong slot_current;
253 : ulong fec_delivered;
254 : ulong shred_old;
255 : ulong shred_rx[ FD_METRICS_ENUM_SHRED_RESPONSE_MATCH_RESULT_CNT ];
256 : ulong meta_rx;
257 : ulong meta_malformed;
258 : ulong meta_unsolicited;
259 : ulong parent_fec_count_ok;
260 : ulong parent_fec_count_failed;
261 : ulong fec_root_ok;
262 : ulong fec_root_failed;
263 : ulong replay_root_advanced;
264 : ulong replay_missing_fec;
265 : ulong ping_malformed;
266 : ulong ping_unknown_peer;
267 : ulong ping_signature_failed;
268 : ulong req_cancelled;
269 : ulong req_no_peer;
270 : ulong req_hedged;
271 : ulong req_expired;
272 : ulong req_stuck_escaped;
273 : ulong req_late_attempt;
274 : ulong peer_byid_banned;
275 : ulong byid_expired;
276 : fd_histf_t response_latency[ 1 ];
277 : fd_histf_t retry_delay[ 1 ];
278 : fd_histf_t escape_hedge[ 1 ];
279 : } metrics[ 1 ];
280 : };
281 : typedef struct fd_rotor_tile fd_rotor_tile_t;
282 :
283 : /* push_timeout inserts req into its blk's role's timeout_prq. A full
284 : timeout_prq drops it and rewinds the blk, back into its blk_treap,
285 : so the work is discovered again. */
286 :
287 : static void
288 : push_timeout( fd_rotor_tile_t * ctx,
289 : fd_rotor_blk_t * blk,
290 0 : timeout_t const * req ) {
291 0 : fd_rotor_slot_meta_t const * meta = fd_rotor_slot_meta( ctx->rotor, blk->slot );
292 0 : ulong idx = fd_rotor_blk_pool_idx( ctx->rotor->blk_pool, blk );
293 0 : timeout_t * prq = meta->final==idx ? ctx->final_prq : meta->notar==idx ? ctx->notar_prq : ctx->eager_prq;
294 0 : if( FD_LIKELY( timeout_prq_cnt( prq )<timeout_prq_max( prq ) ) ) {
295 0 : timeout_prq_insert( prq, req );
296 0 : return;
297 0 : }
298 0 : if( FD_LIKELY( req->tag==FD_ROTOR_SERDE_TAG_WINDOW_INDEX ) ) blk->wait_fec_cnt = fd_uint_min( blk->wait_fec_cnt, req->fec_idx );
299 0 : blk->meta_req = ( blk->meta_req && req->tag!=FD_ROTOR_SERDE_TAG_PARENT_AND_FEC_SET_COUNT );
300 0 : blk->highest_req = ( blk->highest_req && req->tag!=FD_ROTOR_SERDE_TAG_HIGHEST_WINDOW_INDEX );
301 0 : blk->orphan_req = ( blk->orphan_req && req->tag!=FD_ROTOR_SERDE_TAG_ORPHAN );
302 0 : if( FD_UNLIKELY( blk->in_blk_treap ) ) return;
303 0 : fd_rotor_treap_ele_insert( meta->final==idx ? ctx->rotor->final_treap : meta->notar==idx ? ctx->rotor->notar_treap : ctx->rotor->eager_treap, blk, ctx->rotor->blk_pool );
304 0 : blk->in_blk_treap = 1;
305 0 : }
306 :
307 : /* poll_timeout polls for one or more requests that have timed out. */
308 :
309 : static fd_rotor_blk_t *
310 : poll_timeout( fd_rotor_tile_t * ctx,
311 : long now,
312 0 : timeout_t * out ) {
313 0 : timeout_t * prqs[3] = { ctx->final_prq, ctx->notar_prq, ctx->eager_prq };
314 0 : for( ulong c=0UL; c<3; c++ ) {
315 0 : timeout_t * prq = prqs[ c ];
316 0 : while( timeout_prq_cnt( prq ) && prq[ 0 ].timeout<=now ) {
317 0 : timeout_t req = prq[ 0 ];
318 0 : timeout_prq_remove_min( prq );
319 :
320 0 : if( FD_UNLIKELY( req.slot <= ctx->rotor->root ) ) continue;
321 0 : fd_rotor_blk_t * blk = fd_rotor_blk_map_ele_query( ctx->rotor->blk_map, &req.slot, NULL, ctx->rotor->blk_pool );
322 0 : while( blk && fd_rotor_blk_pool_idx( ctx->rotor->blk_pool, blk )!=req.blk ) blk = (fd_rotor_blk_t *)fd_rotor_blk_map_ele_next_const( blk, NULL, ctx->rotor->blk_pool ); /* the req outlived its blk */
323 0 : if( FD_UNLIKELY( !blk ) ) { ctx->metrics->req_cancelled++; continue; }
324 :
325 0 : fd_rotor_slot_meta_t const * meta = fd_rotor_slot_meta( ctx->rotor, blk->slot );
326 0 : if( FD_UNLIKELY( meta->final!=req.blk && meta->notar!=req.blk && meta->eager==req.blk && !blk->eager ) ) { ctx->metrics->req_cancelled++; continue; } /* the slot's eager blk, never repaired */
327 0 : fd_rotor_fec_t const * fec = fd_rotor_fec_pool_ele_const( ctx->rotor->fec_pool, fd_ulong_if( req.tag==FD_ROTOR_SERDE_TAG_WINDOW_INDEX, blk->fecs[ req.fec_idx ], fd_rotor_fec_pool_idx_null( ctx->rotor->fec_pool ) ) );
328 0 : if( FD_UNLIKELY( req.tag==FD_ROTOR_SERDE_TAG_PARENT_AND_FEC_SET_COUNT && blk->parent_slot!=ULONG_MAX ) ) { ctx->metrics->req_cancelled++; continue; }
329 0 : if( FD_UNLIKELY( req.tag==FD_ROTOR_SERDE_TAG_WINDOW_INDEX && fec && fec->complete ) ) { ctx->metrics->req_cancelled++; continue; }
330 0 : if( FD_UNLIKELY( req.tag==FD_ROTOR_SERDE_TAG_WINDOW_INDEX && !blk->eager && req.fec_idx>=blk->cmpl_fec_cnt ) ) { ctx->metrics->req_cancelled++; continue; } /* of a freed blk whose pool idx was reused */
331 0 : if( FD_UNLIKELY( req.tag==FD_ROTOR_SERDE_TAG_HIGHEST_WINDOW_INDEX && blk->cmpl_fec_cnt ) ) { ctx->metrics->req_cancelled++; continue; }
332 0 : fd_rotor_blk_t const * parent = fd_rotor_blk_pool_ele_const( ctx->rotor->blk_pool, blk->parent );
333 0 : if( FD_UNLIKELY( req.tag==FD_ROTOR_SERDE_TAG_ORPHAN && ( blk->parent_slot<=ctx->rotor->root || ( parent && parent->parent_slot!=ULONG_MAX ) ) ) ) { blk->orphan_req = 0; ctx->metrics->req_cancelled++; continue; } /* its ancestry showed up, died with the root, or a lower blk asks */
334 0 : long t_eager = fd_rotor_strat_eager_ns( ctx->strat, fd_multi_epoch_leaders_get_leader_for_slot( ctx->mleaders, blk->slot ) );
335 0 : if( FD_UNLIKELY( req.tag==FD_ROTOR_SERDE_TAG_HIGHEST_WINDOW_INDEX && blk->rcvd_fec_ts+t_eager>now ) ) { /* turbine is not quiet yet */
336 0 : req.timeout = blk->rcvd_fec_ts+t_eager;
337 0 : push_timeout( ctx, blk, &req );
338 0 : continue;
339 0 : }
340 0 : *out = req;
341 0 : return blk;
342 0 : }
343 0 : }
344 0 : return NULL;
345 0 : }
346 :
347 : /* discover new requests translating rotor treaps => repair requests. */
348 :
349 : static void
350 : discover( fd_rotor_tile_t * ctx,
351 : fd_rotor_blk_t * blk,
352 0 : long now ) {
353 0 : fd_rotor_slot_meta_t const * meta = fd_rotor_slot_meta( ctx->rotor, blk->slot );
354 0 : ulong idx = fd_rotor_blk_pool_idx( ctx->rotor->blk_pool, blk );
355 0 : if( FD_UNLIKELY( blk->slot<=ctx->rotor->root ) ) return;
356 0 : if( FD_UNLIKELY( meta->final!=idx && meta->notar!=idx && meta->eager==idx && !blk->eager ) ) return; /* the slot's eager blk, never repaired */
357 :
358 0 : int eager = blk->eager;
359 0 : long t_eager = fd_rotor_strat_eager_ns( ctx->strat, fd_multi_epoch_leaders_get_leader_for_slot( ctx->mleaders, blk->slot ) );
360 0 : timeout_t req = { .slot = blk->slot, .blk = fd_rotor_blk_pool_idx( ctx->rotor->blk_pool, blk ) };
361 :
362 0 : if( FD_UNLIKELY( !eager && blk->parent_slot==ULONG_MAX && blk->meta_req ) ) return;
363 0 : if( FD_UNLIKELY( !eager && blk->parent_slot==ULONG_MAX ) ) {
364 0 : blk->meta_req = 1;
365 0 : req.tag = FD_ROTOR_SERDE_TAG_PARENT_AND_FEC_SET_COUNT;
366 0 : req.timeout = now;
367 0 : push_timeout( ctx, blk, &req );
368 0 : return;
369 0 : }
370 :
371 0 : fd_rotor_blk_t const * parent = fd_rotor_blk_pool_ele_const( ctx->rotor->blk_pool, blk->parent );
372 0 : int orphan_root = !parent || parent->parent_slot==ULONG_MAX; /* the root of its disconnected subtree, the blks under it wait on it */
373 0 : int orphaned = orphan_root && !blk->orphan_req && blk->parent_slot!=ULONG_MAX && blk->parent_slot>ctx->rotor->root && memcmp( &blk->parent_blk_mr, &hash_null, sizeof(fd_mr32_t) ) && !( eager && meta->invalidated );
374 0 : if( FD_UNLIKELY( orphaned ) ) {
375 0 : blk->orphan_req = 1;
376 0 : req.tag = FD_ROTOR_SERDE_TAG_ORPHAN; /* its ancestry is unknown, learn up to 11 of it in one reply */
377 0 : req.timeout = fd_long_if( eager, blk->rcvd_fec_ts+t_eager, now ); /* an eager blk's parent may still be on turbine */
378 0 : push_timeout( ctx, blk, &req );
379 0 : }
380 :
381 0 : if( FD_UNLIKELY( eager && !blk->cmpl_fec_cnt && !blk->highest_req ) ) {
382 0 : blk->highest_req = 1;
383 0 : req.tag = FD_ROTOR_SERDE_TAG_HIGHEST_WINDOW_INDEX;
384 0 : req.timeout = blk->rcvd_fec_ts+t_eager;
385 0 : push_timeout( ctx, blk, &req );
386 0 : }
387 :
388 0 : timeout_t * prq = meta->final==idx ? ctx->final_prq : meta->notar==idx ? ctx->notar_prq : ctx->eager_prq;
389 0 : uint cmpl_fec_cnt = fd_uint_if( !!blk->cmpl_fec_cnt, blk->cmpl_fec_cnt, fd_uint_if( eager, blk->rcvd_fec_cnt, 0U ) );
390 0 : uint fec_idx = blk->wait_fec_cnt;
391 0 : for( ; fec_idx<cmpl_fec_cnt && timeout_prq_cnt( prq )<timeout_prq_max( prq ); fec_idx++ ) {
392 0 : fd_rotor_fec_t const * fec = fd_rotor_fec_pool_ele_const( ctx->rotor->fec_pool, blk->fecs[ fec_idx ] );
393 0 : if( FD_LIKELY( fec && fec->complete ) ) continue;
394 0 : req.tag = FD_ROTOR_SERDE_TAG_WINDOW_INDEX;
395 0 : req.fec_idx = fec_idx;
396 0 : req.timeout = fd_long_if( eager, ( fec ? fec->first_ts : blk->rcvd_fec_ts )+t_eager, now ); /* a partial set waits from its first shred, a missing one from the last set turbine showed */
397 0 : push_timeout( ctx, blk, &req );
398 0 : }
399 0 : blk->wait_fec_cnt = fec_idx;
400 0 : if( FD_LIKELY( timeout_prq_cnt( prq )<timeout_prq_max( prq ) ) ) return;
401 0 : if( FD_UNLIKELY( blk->in_blk_treap ) ) return;
402 0 : fd_rotor_treap_ele_insert( meta->final==idx ? ctx->rotor->final_treap : meta->notar==idx ? ctx->rotor->notar_treap : ctx->rotor->eager_treap, blk, ctx->rotor->blk_pool ); /* timeout_prq filled up, resume later */
403 0 : blk->in_blk_treap = 1;
404 0 : }
405 :
406 : FD_STATIC_ASSERT( FD_FEC_BLK_MAX<=FD_EVENT_BLOCK_RECEIVED_FEC_SETS_MAX, block_received_fec_sets );
407 :
408 : static void
409 : report_block_received( void * ctx_,
410 0 : fd_rotor_blk_t const * blk ) {
411 0 : if( FD_LIKELY( !fd_event_tl ) ) return;
412 0 : fd_rotor_tile_t * ctx = (fd_rotor_tile_t *)ctx_;
413 0 : uint fec_cnt = fd_uint_if( !!blk->cmpl_fec_cnt, blk->cmpl_fec_cnt, blk->rcvd_fec_cnt ); /* an incomplete blk reports what it received */
414 0 : fd_rotor_fec_t const * last = fd_rotor_fec_pool_ele_const( ctx->rotor->fec_pool, fec_cnt ? blk->fecs[ fec_cnt-1U ] : fd_rotor_fec_pool_idx_null( ctx->rotor->fec_pool ) );
415 0 : int is_leader = last && last->is_leader;
416 0 : int known_id = fd_rotor_slot_meta( ctx->rotor, blk->slot )->eager!=fd_rotor_blk_pool_idx( ctx->rotor->blk_pool, blk );
417 0 : fd_event_block_received_t * ev = ctx->event;
418 0 : fd_memset( ev, 0, FD_EVENT_BLOCK_RECEIVED_PREFIX_SZ );
419 0 : ev->slot = blk->slot;
420 0 : ev->parent_slot = blk->parent_slot;
421 0 : ev->cancelled = !!blk->telemetry.cancelled_reason;
422 0 : ev->cancelled_reason = fd_int_if( !!blk->telemetry.cancelled_reason, blk->telemetry.cancelled_reason, FD_EVENT_BLOCK_RECEIVED_CANCELLED_REASON_NOT_CANCELLED );
423 0 : ev->cancelled_time = fd_ulong_if( ev->cancelled, (ulong)blk->telemetry.cancelled_ts, 0UL );
424 0 : ev->notarized = known_id && !is_leader;
425 0 : ev->is_leader = is_leader;
426 0 : ev->caught_up = fd_ulong_max( ctx->metrics->slot_current, ctx->catchup_slot )<=ctx->metrics->slot_highest_repaired+4UL;
427 0 : ev->fec_set_count = blk->cmpl_fec_cnt;
428 0 : ev->slot_complete = !!blk->cmpl_fec_cnt;
429 0 : ev->equivocation_detected_shred = blk->telemetry.cancelled_reason==FD_EVENT_BLOCK_RECEIVED_CANCELLED_REASON_MERKLE_ROOT_MISMATCH;
430 0 : ev->last_completed_fec_set_index = blk->telemetry.last_cmpl_fec_idx;
431 0 : ev->last_shred_received_time = (ulong)blk->telemetry.last_shred_ts;
432 0 : ev->turbine_shred_received = blk->telemetry.turbine_shred_cnt;
433 0 : ev->repair_shred_received = blk->telemetry.repair_shred_cnt;
434 0 : ev->recovered_shred_count = blk->telemetry.recovered_shred_cnt;
435 0 : if( FD_LIKELY( !is_leader ) ) { /* our own blocks are never repaired */
436 0 : ev->repair_request_window_count = blk->telemetry.req_window_cnt;
437 0 : ev->repair_request_highest_window_count = blk->telemetry.req_highest_cnt;
438 0 : ev->repair_request_orphan_count = blk->telemetry.req_orphan_cnt;
439 0 : ev->repair_request_shred_for_block_id_count = blk->telemetry.req_shred_bid_cnt;
440 0 : ev->repair_request_parent_fec_count = blk->telemetry.req_parent_cnt;
441 0 : ev->repair_request_fec_root_count = blk->telemetry.req_fec_root_cnt;
442 0 : ev->first_repair_request_time = (ulong)blk->telemetry.first_req_ts;
443 0 : ev->repair_shred_responses_received = blk->telemetry.shred_res_cnt;
444 0 : ev->parent_fec_count_responses_received = blk->telemetry.parent_res_cnt;
445 0 : ev->fec_root_responses_received = blk->telemetry.fec_root_res_cnt;
446 0 : ev->last_repair_received_time = (ulong)blk->telemetry.last_shred_res_ts;
447 0 : ev->first_meta_received_time = (ulong)blk->telemetry.first_meta_res_ts;
448 0 : }
449 0 : memcpy( ev->block_id, blk->dmr.uc, sizeof(fd_mr32_t) );
450 0 : memcpy( ev->parent_block_id, blk->parent_blk_mr.uc, sizeof(fd_mr32_t) );
451 0 : ev->fec_sets_cnt = fd_ulong_min( fec_cnt, FD_EVENT_BLOCK_RECEIVED_FEC_SETS_MAX );
452 0 : for( ulong k=0UL; k<ev->fec_sets_cnt; k++ ) {
453 0 : fd_rotor_fec_t const * fec = fd_rotor_fec_pool_ele_const( ctx->rotor->fec_pool, blk->fecs[ k ] );
454 0 : fd_event_block_received_fec_sets_t * f = &ev->fec_sets[ k ];
455 0 : fd_memset( f, 0, sizeof(fd_event_block_received_fec_sets_t) );
456 0 : if( FD_UNLIKELY( !fec ) ) continue; /* an incomplete blk can miss FEC sets */
457 0 : memcpy( f->merkle_root, fec->mr32.uc, sizeof(fd_mr32_t) );
458 0 : f->index = (uint)k*FD_FEC_SHRED_CNT;
459 0 : f->data_shreds_received_mask = fec->data_rcvd;
460 0 : f->parity_shreds_received_mask = fec->parity_rcvd;
461 0 : f->repair_shreds_received_mask = fec->repair_rcvd;
462 0 : f->first_shred_received_time = (ulong)fec->first_ts;
463 0 : f->completed_time = (ulong)fec->cmpl_ts;
464 0 : f->final_shred_source = fd_int_if( fec->is_leader, FD_EVENT_BLOCK_RECEIVED_FEC_SETS_FINAL_SHRED_SOURCE_LEADER, fec->final_src );
465 0 : ev->first_shred_received_time = fd_ulong_if( fec->first_ts && ( !ev->first_shred_received_time || (ulong)fec->first_ts<ev->first_shred_received_time ), (ulong)fec->first_ts, ev->first_shred_received_time );
466 0 : }
467 0 : fd_event_report_block_received( ev );
468 0 : }
469 :
470 :
471 : static inline void
472 : handle_shred( fd_rotor_tile_t * ctx,
473 : ulong sig,
474 : uchar const * chunk,
475 0 : long rx_ts ) {
476 0 : if( FD_UNLIKELY( fd_shred_sig_src( sig )==SHRED_SIG_FEC_EVICTED ) ) {
477 0 : fd_fec_evicted_t const * evicted = (fd_fec_evicted_t const *)fd_type_pun_const( chunk );
478 0 : if( FD_UNLIKELY( evicted->slot<=ctx->rotor->root || evicted->slot>=ctx->rotor->root+ctx->rotor->slot_max ) ) return; /* outside the window */
479 0 : if( FD_UNLIKELY( evicted->fec_set_idx>=ctx->rotor->fec_blk_max*FD_FEC_SHRED_CNT ) ) return;
480 0 : fd_mr20_t key; memcpy( key.uc, evicted->merkle_root.uc, sizeof(fd_mr20_t) );
481 0 : fd_rotor_fec_evicted( ctx->rotor, evicted->slot, evicted->fec_set_idx, &key );
482 0 : return;
483 0 : }
484 :
485 0 : ulong slot = ((fd_shred_base_t const *)fd_type_pun_const( chunk ))->shred.slot; /* complete and shred messages share the header position */
486 0 : ctx->metrics->shred_old += (ulong)( slot<=ctx->rotor->root && fd_shred_sig_src( sig )<=SHRED_SIG_SRC_BAD_REPAIR );
487 0 : ctx->metrics->slot_current = fd_ulong_if( slot>ctx->rotor->root && fd_shred_sig_src( sig )<=SHRED_SIG_SRC_BAD_REPAIR, fd_ulong_max( ctx->metrics->slot_current, slot ), ctx->metrics->slot_current );
488 0 : int turbine_first = slot>ctx->rotor->root && slot<ctx->rotor->root+ctx->rotor->slot_max && fd_shred_sig_src( sig )==SHRED_SIG_SRC_TURBINE && !ctx->turbine_slot0;
489 0 : ctx->turbine_slot0 = fd_ulong_if( turbine_first, slot, ctx->turbine_slot0 );
490 0 : if( FD_UNLIKELY( slot<=ctx->rotor->root || slot>=ctx->rotor->root+ctx->rotor->slot_max ) ) return; /* outside the window */
491 :
492 0 : if( FD_UNLIKELY( turbine_first && !ctx->staked ) ) fd_rotor_slot_catchup( ctx->rotor, slot ); /* unstaked has no certs, turbine is untrusted so staked never does this */
493 :
494 0 : switch( fd_shred_sig_src( sig ) ) {
495 0 : case SHRED_SIG_FEC_COMPLETE:
496 0 : case SHRED_SIG_FEC_COMPLETE_LEADER:
497 0 : case SHRED_SIG_FEC_COMPLETE_AGAIN: {
498 0 : fd_fec_complete_t const * complete = (fd_fec_complete_t const *)fd_type_pun_const( chunk );
499 0 : fd_shred_t const * shred = &complete->last_shred_hdr;
500 :
501 0 : ctx->highest_fec_complete_slot = fd_ulong_max( ctx->highest_fec_complete_slot, shred->slot );
502 0 : if( FD_UNLIKELY( shred->idx>=ctx->rotor->fec_blk_max*FD_FEC_SHRED_CNT ) ) return;
503 0 : if( FD_UNLIKELY( !shred->data.parent_off || shred->data.parent_off>shred->slot-ctx->rotor->root ) ) return; /* parent is the slot itself, or below the root */
504 :
505 0 : fd_rotor_fec_t const * fec = fd_rotor_fec_complete( ctx->rotor, shred, &complete->merkle_root, sig==SHRED_SIG_FEC_COMPLETE_LEADER, (uint)complete->turbine_shred_cnt, (uint)complete->repair_shred_cnt, (uint)complete->reconstructed_shred_cnt, complete->data_shred_rcvd, complete->parity_shred_rcvd, complete->repair_data_rcvd, rx_ts );
506 0 : if( FD_UNLIKELY( sig!=SHRED_SIG_FEC_COMPLETE || !fec || !fec->first_ts ) ) break; /* a first completion from the network */
507 :
508 0 : fd_rotor_strat_turbine_done( ctx->strat, fd_multi_epoch_leaders_get_leader_for_slot( ctx->mleaders, shred->slot ), rx_ts-fec->first_ts );
509 0 : break;
510 0 : }
511 0 : case SHRED_SIG_SRC_TURBINE:
512 0 : case SHRED_SIG_SRC_LEADER:
513 0 : case SHRED_SIG_SRC_RECONSTRUCTED:
514 0 : case SHRED_SIG_SRC_REPAIR:
515 0 : case SHRED_SIG_SRC_BAD_REPAIR: {
516 0 : fd_shred_base_t const * msg = (fd_shred_base_t const *)fd_type_pun_const( chunk );
517 0 : fd_shred_t const * shred = &msg->shred;
518 0 : long now = fd_clock_tile_now( ctx->clock );
519 :
520 0 : if( FD_UNLIKELY( fd_shred_sig_res( sig )==SHRED_SIG_RESULT_EQVOC ) ) {
521 0 : fd_rotor_slot_invalidated( ctx->rotor, shred->slot, FD_EVENT_BLOCK_RECEIVED_CANCELLED_REASON_MERKLE_ROOT_MISMATCH, now );
522 0 : return;
523 0 : }
524 :
525 0 : if( FD_UNLIKELY( !fd_shred_is_data( fd_shred_type( shred->variant ) ) ) ) return;
526 0 : if( FD_UNLIKELY( shred->idx>=ctx->rotor->fec_blk_max*FD_FEC_SHRED_CNT ) ) return;
527 :
528 : /* A repair shred answers the request whose nonce it carries: the
529 : first one ends that request's pick, the last one ends the
530 : request. A HighestWindowIndex answer that does not complete the
531 : slot arrives as BAD_REPAIR. */
532 :
533 0 : int normal = fd_rnonce_ss_normal_repair( msg->rnonce );
534 0 : int reply = fd_shred_sig_src( sig )==SHRED_SIG_SRC_REPAIR || fd_shred_sig_src( sig )==SHRED_SIG_SRC_BAD_REPAIR;
535 0 : ulong key = (ulong)(uint)slot<<32 | fd_uint_if( normal, shred->idx/FD_FEC_SHRED_CNT, (uint)ctx->rotor->fec_blk_max );
536 0 : uint bit = fd_uint_if( normal, 1U<<( shred->idx%FD_FEC_SHRED_CNT ), 1U );
537 0 : pending_t * pending = reply ? pending_map_ele_query( ctx->pending_map, &key, NULL, ctx->pending_pool ) : NULL;
538 0 : for( ; pending; pending = (pending_t *)pending_map_ele_next_const( pending, NULL, ctx->pending_pool ) ) {
539 0 : if( FD_UNLIKELY( !( pending->mask & bit ) ) ) continue; /* already answered, or not asked of this request */
540 0 : if( FD_LIKELY( fd_rnonce_ss_compute( ctx->rnonce_ss, normal, slot, shred->idx, pending->ts )!=msg->rnonce ) ) continue;
541 0 : if( FD_LIKELY( !pending->answered ) ) { /* rtt to the shred tile's network rx, not to this tile's turn at the frag */
542 0 : long rtt = fd_long_max( rx_ts-pending->ts, 1L );
543 0 : fd_rotor_strat_request_done( ctx->strat, &pending->peer, rtt, pending->tag==FD_ROTOR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID );
544 0 : fd_histf_sample( ctx->metrics->response_latency, (ulong)rtt );
545 0 : }
546 0 : pending->mask &= ~bit;
547 0 : pending->answered++;
548 0 : ctx->metrics->shred_rx[ fd_ulong_if( pending->tag==FD_ROTOR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID, FD_METRICS_ENUM_SHRED_RESPONSE_MATCH_RESULT_V_FOR_BLOCK_ID_IDX, FD_METRICS_ENUM_SHRED_RESPONSE_MATCH_RESULT_V_SLOT_INDEX_IDX ) ]++;
549 0 : fd_rotor_blk_t * blk = fd_rotor_blk_map_ele_query( ctx->rotor->blk_map, &pending->slot, NULL, ctx->rotor->blk_pool );
550 0 : while( blk && memcmp( &blk->dmr, &pending->dmr, sizeof(fd_mr32_t) ) ) blk = (fd_rotor_blk_t *)fd_rotor_blk_map_ele_next_const( blk, NULL, ctx->rotor->blk_pool ); /* not by pending->blk, it may be freed or reused, and a notar blk lives on as eager after dedup */
551 0 : if( FD_LIKELY( blk && ( pending->tag==FD_ROTOR_SERDE_TAG_WINDOW_INDEX || pending->tag==FD_ROTOR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID ) ) ) {
552 0 : blk->telemetry.shred_res_cnt++;
553 0 : blk->telemetry.last_shred_res_ts = rx_ts;
554 0 : }
555 0 : if( FD_LIKELY( !pending->mask ) ) { /* every shred answered */
556 0 : pending_map_ele_remove_fast( ctx->pending_map, pending, ctx->pending_pool );
557 0 : pending_dlist_ele_remove ( ctx->pending_dlist, pending, ctx->pending_pool );
558 0 : pending_pool_ele_release ( ctx->pending_pool, pending );
559 0 : }
560 0 : break;
561 0 : }
562 0 : ctx->metrics->shred_rx[ FD_METRICS_ENUM_SHRED_RESPONSE_MATCH_RESULT_V_UNMATCHED_IDX ] += (ulong)( reply && !pending );
563 :
564 0 : if( FD_UNLIKELY( !shred->data.parent_off || shred->data.parent_off>shred->slot-ctx->rotor->root ) ) return; /* parent is the slot itself, or below the root */
565 :
566 0 : int src = fd_shred_sig_src( sig )==SHRED_SIG_SRC_TURBINE ? FD_EVENT_BLOCK_RECEIVED_FEC_SETS_FINAL_SHRED_SOURCE_TURBINE :
567 0 : fd_shred_sig_src( sig )==SHRED_SIG_SRC_LEADER ? FD_EVENT_BLOCK_RECEIVED_FEC_SETS_FINAL_SHRED_SOURCE_LEADER :
568 0 : reply ? FD_EVENT_BLOCK_RECEIVED_FEC_SETS_FINAL_SHRED_SOURCE_REPAIR :
569 0 : 0; /* reconstructed, not received */
570 0 : fd_rotor_shred_insert( ctx->rotor, shred, &msg->merkle_root, src, rx_ts );
571 0 : break;
572 0 : }
573 0 : default: FD_LOG_ERR(( "unhandled shred sig src %u", fd_shred_sig_src( sig ) ));
574 0 : }
575 0 : }
576 :
577 : static void
578 : handle_snapshot( fd_rotor_tile_t * ctx,
579 : ulong sig,
580 0 : uchar const * chunk ) {
581 0 : switch( fd_ssmsg_sig_message( sig ) ) {
582 0 : case FD_SSMSG_MANIFEST_FULL:
583 0 : case FD_SSMSG_MANIFEST_INCREMENTAL: {
584 0 : fd_snapshot_manifest_t const * manifest = (fd_snapshot_manifest_t const *)fd_type_pun_const( chunk );
585 0 : ctx->snapshot_slot = manifest->slot;
586 0 : memcpy( ctx->snapshot_blk_mr.uc, manifest->block_id, sizeof(fd_mr32_t) );
587 0 : break;
588 0 : }
589 0 : case FD_SSMSG_DONE:
590 0 : fd_rotor_init( ctx->rotor, ctx->snapshot_slot, &ctx->snapshot_blk_mr, report_block_received, ctx );
591 0 : break;
592 0 : default:
593 0 : FD_LOG_ERR(( "unexpected snapshot message %lu", fd_ssmsg_sig_message( sig ) ));
594 0 : }
595 0 : }
596 :
597 : static inline void
598 : handle_votor( fd_rotor_tile_t * ctx,
599 : ulong sig,
600 0 : uchar const * chunk ) {
601 0 : fd_votor_msg_t const * msg = (fd_votor_msg_t const *)fd_type_pun_const( chunk );
602 0 : long now = fd_clock_tile_now( ctx->clock );
603 :
604 0 : switch( sig ) {
605 0 : case FD_VOTOR_SIG_CERTED: {
606 0 : fd_votor_certed_t const * certed = &msg->certed;
607 0 : if( FD_UNLIKELY( certed->slot<=ctx->rotor->root || certed->slot>=ctx->rotor->root+ctx->rotor->slot_max ) ) return; /* outside the window */
608 0 : if( FD_UNLIKELY( ctx->staked && !ctx->cert_slot0 ) ) {
609 0 : ctx->cert_slot0 = certed->slot;
610 0 : fd_rotor_slot_catchup( ctx->rotor, certed->slot );
611 0 : }
612 0 : switch( certed->kind ) {
613 0 : case AG_CERT_KIND_FINAL:
614 0 : case AG_CERT_KIND_FAST_FINAL: fd_rotor_blk_finalized( ctx->rotor, certed->slot, &certed->block_id, now ); break;
615 0 : case AG_CERT_KIND_NOTAR: fd_rotor_blk_notarized( ctx->rotor, certed->slot, &certed->block_id ); break;
616 0 : case AG_CERT_KIND_NOTAR_FALLBACK: break;
617 0 : case AG_CERT_KIND_SKIP: break; /* a skip cert alone is not final, see fd_rotor_slot_skipped */
618 0 : default: FD_LOG_ERR(( "unhandled cert kind %u", certed->kind ));
619 0 : }
620 0 : break;
621 0 : }
622 0 : case FD_VOTOR_SIG_REPAIR: {
623 0 : ulong slot = msg->repair.slot;
624 0 : if( FD_UNLIKELY( slot<=ctx->rotor->root ) ) return;
625 0 : if( FD_UNLIKELY( slot>=ctx->rotor->root+ctx->rotor->slot_max ) ) {
626 0 : if( FD_UNLIKELY( !ctx->staked || slot<=ctx->catchup_slot ) ) return;
627 0 : ctx->catchup_slot = slot;
628 0 : fd_rotor_slot_catchup( ctx->rotor, ctx->rotor->root+ctx->rotor->slot_max );
629 0 : return;
630 0 : }
631 0 : fd_rotor_blk_notarized( ctx->rotor, slot, &msg->repair.block_id );
632 0 : break;
633 0 : }
634 0 : case FD_VOTOR_SIG_QUORUM: {
635 0 : ulong slot = msg->quorum.slot;
636 0 : if( FD_UNLIKELY( slot<=ctx->rotor->root || slot>=ctx->rotor->root+ctx->rotor->slot_max ) ) return; /* outside the window */
637 0 : switch( msg->quorum.kind ) {
638 0 : case FD_VOTOR_QUORUM_KIND_SAFE_TO_NOTAR: break; /* replay already has the block */
639 0 : case FD_VOTOR_QUORUM_KIND_IMPLICITLY_FINALIZED: fd_rotor_blk_finalized( ctx->rotor, slot, &msg->quorum.block_id, now ); break;
640 0 : case FD_VOTOR_QUORUM_KIND_IMPLICITLY_SKIPPED: fd_rotor_slot_skipped ( ctx->rotor, slot ); break;
641 0 : default: FD_LOG_ERR(( "unhandled quorum kind %u", msg->quorum.kind ));
642 0 : }
643 0 : break;
644 0 : }
645 0 : default: FD_LOG_ERR(( "unhandled votor sig %lu", sig ));
646 0 : }
647 0 : }
648 :
649 : static inline void
650 : handle_replay( fd_rotor_tile_t * ctx,
651 : ulong sig,
652 0 : uchar const * chunk ) {
653 0 : switch( sig ) {
654 0 : case REPLAY_SIG_ROOT_ADVANCED: {
655 0 : fd_replay_root_advanced_t const * root = (fd_replay_root_advanced_t const *)fd_type_pun_const( chunk );
656 0 : ctx->metrics->replay_root_advanced++;
657 0 : if( FD_UNLIKELY( root->slot<=ctx->rotor->root ) ) return;
658 0 : fd_rotor_root_advanced( ctx->rotor, root->slot, &root->block_id );
659 0 : if( FD_UNLIKELY( ctx->catchup_slot>ctx->rotor->root ) ) fd_rotor_slot_catchup( ctx->rotor, fd_ulong_min( ctx->catchup_slot+1UL, ctx->rotor->root+ctx->rotor->slot_max ) );
660 0 : break;
661 0 : }
662 0 : case REPLAY_SIG_SLOT_DEAD: {
663 0 : fd_replay_slot_dead_t const * dead = (fd_replay_slot_dead_t const *)fd_type_pun_const( chunk );
664 0 : if( FD_UNLIKELY( dead->slot<=ctx->rotor->root ) ) return;
665 0 : fd_rotor_blk_dead( ctx->rotor, dead->slot, &dead->block_id );
666 0 : break;
667 0 : }
668 0 : case REPLAY_SIG_MISSING_FEC:
669 0 : ctx->metrics->replay_missing_fec++;
670 0 : ctx->rotor->reconsume = 1;
671 0 : break;
672 0 : default: FD_LOG_ERR(( "unhandled replay sig %lu", sig ));
673 0 : }
674 0 : }
675 :
676 : static void
677 : handle_gossip( fd_rotor_tile_t * ctx,
678 : ulong sig,
679 0 : fd_gossip_update_message_t const * msg ) {
680 :
681 0 : fd_pubkey_t id_key;
682 0 : memcpy( id_key.uc, msg->origin, sizeof(fd_pubkey_t) );
683 :
684 0 : switch( sig ) {
685 0 : case FD_GOSSIP_UPDATE_TAG_CONTACT_INFO: {
686 0 : fd_gossip_socket_t const * socket = &msg->contact_info->value->sockets[FD_GOSSIP_CONTACT_INFO_SOCKET_SERVE_REPAIR];
687 0 : uint ip4 = fd_uint_if( !socket->is_ipv6, socket->ip4, 0U );
688 0 : ushort port = fd_ushort_bswap( socket->port );
689 0 : if( FD_UNLIKELY( fd_pubkey_eq( &id_key, &ctx->identity_key ) || !ip4 || !port || fd_ip4_addr_is_mcast( ip4 ) || ( !ctx->allow_private_address && !fd_ip4_addr_is_public( ip4 ) ) ) ) {
690 0 : fd_rotor_strat_contact_info_removed( ctx->strat, &id_key );
691 0 : return;
692 0 : }
693 :
694 0 : fd_rotor_strat_contact_info_updated( ctx->strat, &id_key, ip4, port );
695 0 : fd_rotor_strat_peer_t const * peer = fd_rotor_strat_query( ctx->strat, &id_key );
696 0 : if( FD_UNLIKELY( peer && !peer->ping_ts && !ping_deque_full( ctx->ping_deque ) ) ) ping_deque_push_tail( ctx->ping_deque, id_key ); /* new address */
697 0 : break;
698 0 : }
699 0 : case FD_GOSSIP_UPDATE_TAG_CONTACT_INFO_REMOVE:
700 0 : fd_rotor_strat_contact_info_removed( ctx->strat, &id_key );
701 0 : break;
702 0 : default:
703 0 : FD_LOG_ERR(( "unexpected gossip sig %lu", sig ));
704 0 : }
705 0 : }
706 :
707 : /* Only answer a ping if it comes from the address of a peer we sent a
708 : request to in the last PING_TTL. */
709 :
710 : static void
711 : handle_ping( fd_rotor_tile_t * ctx,
712 : fd_stem_context_t * stem,
713 : uchar const * data,
714 : ulong data_sz,
715 : fd_ip4_hdr_t const * ip4,
716 : fd_udp_hdr_t const * udp,
717 0 : long now ) {
718 0 : fd_pubkey_t from;
719 0 : fd_hash_t token;
720 0 : fd_ed25519_sig_t sig;
721 0 : if( FD_UNLIKELY( !fd_rotor_ping_de( data, data_sz, &from, &token, sig ) && !fd_rotor_block_id_ping_de( data, data_sz, &from, &token, sig ) ) ) { ctx->metrics->ping_malformed++; return; }
722 :
723 0 : fd_rotor_strat_peer_t * peer = fd_rotor_strat_query( ctx->strat, &from );
724 0 : if( FD_UNLIKELY( !peer ) ) { ctx->metrics->ping_unknown_peer++; return; }
725 0 : if( FD_UNLIKELY( peer->ip4!=ip4->saddr || fd_ushort_bswap( peer->port )!=udp->net_sport ) ) return;
726 0 : if( FD_UNLIKELY( now-peer->ping_ts>=PING_TTL ) ) return;
727 0 : if( FD_UNLIKELY( ctx->halt_signing || !sign_pool_free( ctx->sign_pool ) ) ) return;
728 :
729 0 : peer->ping_ts = now-PING_TTL; /* one verify per request we sent */
730 0 : fd_sha512_t sha[1];
731 0 : if( FD_UNLIKELY( fd_ed25519_verify( token.uc, sizeof(fd_hash_t), sig, from.uc, sha )!=FD_ED25519_SUCCESS ) ) { ctx->metrics->ping_signature_failed++; return; }
732 :
733 0 : ulong s = 0UL;
734 0 : for( ulong i=1UL; i<ctx->sign_cnt; i++ ) s = fd_ulong_if( ctx->sign[ i ].free>ctx->sign[ s ].free, i, s );
735 0 : uchar * preimage = fd_chunk_to_laddr( ctx->sign[ s ].mem, ctx->sign[ s ].chunk );
736 0 : memcpy( preimage, "SOLANA_PING_PONG", 16UL );
737 0 : memcpy( preimage+16UL, token.uc, sizeof(fd_hash_t) );
738 0 : fd_rotor_pong_t pong;
739 0 : fd_sha256_hash( preimage, 16UL+sizeof(fd_hash_t), pong.hash.uc );
740 :
741 0 : sign_t * pending = sign_pool_ele_acquire( ctx->sign_pool );
742 0 : *pending = (sign_t){ .sig_off = FD_ROTOR_PONG_SER_SZ-FD_ED25519_SIG_SZ, .saddr = ip4->daddr, .daddr = ip4->saddr, .dport = udp->net_sport };
743 0 : pending->sz = fd_rotor_pong_ser( &pong, sig_null, &ctx->identity_key, pending->buf );
744 0 : fd_stem_publish( stem, ctx->sign[ s ].idx, (sign_pool_idx( ctx->sign_pool, pending )<<32)|(ulong)FD_KEYGUARD_SIGN_TYPE_SHA256_ED25519, ctx->sign[ s ].chunk, 16UL+sizeof(fd_hash_t), 0UL, 0UL, 0UL );
745 0 : ctx->sign[ s ].chunk = fd_dcache_compact_next( ctx->sign[ s ].chunk, 16UL+sizeof(fd_hash_t), ctx->sign[ s ].chunk0, ctx->sign[ s ].wmark );
746 0 : ctx->sign[ s ].free--;
747 0 : ctx->metrics->request_tx[ FD_ROTOR_SERDE_TAG_PONG ]++;
748 0 : }
749 :
750 : static void
751 : handle_parent_and_fec_set_count_res( fd_rotor_tile_t * ctx,
752 : uchar const * data,
753 : ulong data_sz,
754 : fd_ip4_hdr_t const * ip4,
755 : fd_udp_hdr_t const * udp,
756 0 : long now ) {
757 0 : uint fec_set_cnt;
758 0 : ulong parent_slot;
759 0 : fd_mr32_t parent_blk_mr;
760 0 : uchar proof[ FD_ROTOR_PROOF_MAX ];
761 0 : ulong proof_sz;
762 0 : uint nonce;
763 0 : ctx->metrics->meta_rx++;
764 0 : if( FD_UNLIKELY( fd_rotor_res_parent_fec_set_count_de( data, data_sz, &fec_set_cnt, &parent_slot, &parent_blk_mr, proof, &proof_sz, &nonce )!=data_sz ) ) { ctx->metrics->meta_malformed++; return; }
765 :
766 0 : pending_t * pending = pending_map_ele_query( ctx->pending_map, &(ulong){ nonce }, NULL, ctx->pending_pool );
767 0 : if( FD_UNLIKELY( !pending || pending->tag!=FD_ROTOR_SERDE_TAG_PARENT_AND_FEC_SET_COUNT ) ) { ctx->metrics->meta_unsolicited++; return; }
768 :
769 0 : fd_rotor_strat_peer_t const * peer = fd_rotor_strat_query( ctx->strat, &pending->peer );
770 0 : int valid = !fd_rotor_res_parent_fec_set_count_verify( fec_set_cnt, parent_slot, &parent_blk_mr, proof, proof_sz, &pending->dmr );
771 0 : int from = peer && peer->ip4==ip4->saddr && fd_ushort_bswap( peer->port )==udp->net_sport;
772 0 : ctx->metrics->parent_fec_count_failed += (ulong)!valid;
773 0 : if( FD_UNLIKELY( !valid && !from ) ) return;
774 :
775 0 : pending_t sent = *pending;
776 0 : pending_map_ele_remove_fast( ctx->pending_map, pending, ctx->pending_pool );
777 0 : pending_dlist_ele_remove ( ctx->pending_dlist, pending, ctx->pending_pool );
778 0 : pending_pool_ele_release ( ctx->pending_pool, pending );
779 0 : if( FD_UNLIKELY( !valid ) ) {
780 0 : fd_rotor_strat_request_failed( ctx->strat, &sent.peer, now+BAN_TTL );
781 0 : return;
782 0 : }
783 0 : fd_rotor_strat_request_done( ctx->strat, &sent.peer, now-sent.ts, 0 );
784 0 : fd_histf_sample( ctx->metrics->response_latency, (ulong)( now-sent.ts ) );
785 0 : ctx->metrics->parent_fec_count_ok++;
786 :
787 0 : if( FD_UNLIKELY( sent.slot<=ctx->rotor->root ) ) return; /* rooted while in flight */
788 0 : if( FD_UNLIKELY( parent_slot<ctx->rotor->root ) ) return; /* on a fork below the root */
789 0 : if( FD_UNLIKELY( parent_slot>=sent.slot ) ) return; /* the leader signed a bad parent */
790 :
791 0 : fd_rotor_blk_parented( ctx->rotor, sent.slot, &sent.dmr, parent_slot, &parent_blk_mr, fec_set_cnt, now );
792 0 : }
793 :
794 : static void
795 : handle_fec_set_root_res( fd_rotor_tile_t * ctx,
796 : uchar const * data,
797 : ulong data_sz,
798 : fd_ip4_hdr_t const * ip4,
799 : fd_udp_hdr_t const * udp,
800 0 : long now ) {
801 0 : fd_mr20_t fec_mr;
802 0 : uchar proof[ FD_ROTOR_PROOF_MAX ];
803 0 : ulong proof_sz;
804 0 : uint nonce;
805 0 : ctx->metrics->meta_rx++;
806 0 : if( FD_UNLIKELY( fd_rotor_res_fec_set_root_de( data, data_sz, fec_mr.uc, proof, &proof_sz, &nonce )!=data_sz ) ) { ctx->metrics->meta_malformed++; return; }
807 :
808 0 : pending_t * pending = pending_map_ele_query( ctx->pending_map, &(ulong){ nonce }, NULL, ctx->pending_pool );
809 0 : if( FD_UNLIKELY( !pending || pending->tag!=FD_ROTOR_SERDE_TAG_FEC_SET_ROOT ) ) { ctx->metrics->meta_unsolicited++; return; }
810 :
811 0 : fd_rotor_strat_peer_t const * peer = fd_rotor_strat_query( ctx->strat, &pending->peer );
812 0 : int valid = !fd_rotor_res_fec_set_root_verify( fec_mr.uc, proof, proof_sz, &pending->dmr, pending->fec_idx*FD_FEC_SHRED_CNT, pending->fec_set_cnt );
813 0 : int from = peer && peer->ip4==ip4->saddr && fd_ushort_bswap( peer->port )==udp->net_sport;
814 0 : ctx->metrics->fec_root_failed += (ulong)!valid;
815 0 : if( FD_UNLIKELY( !valid && !from ) ) return;
816 :
817 0 : pending_t sent = *pending;
818 0 : pending_map_ele_remove_fast( ctx->pending_map, pending, ctx->pending_pool );
819 0 : pending_dlist_ele_remove ( ctx->pending_dlist, pending, ctx->pending_pool );
820 0 : pending_pool_ele_release ( ctx->pending_pool, pending );
821 0 : if( FD_UNLIKELY( !valid ) ) {
822 0 : fd_rotor_strat_request_failed( ctx->strat, &sent.peer, now+BAN_TTL );
823 0 : return;
824 0 : }
825 0 : fd_rotor_strat_request_done( ctx->strat, &sent.peer, now-sent.ts, 0 );
826 0 : fd_histf_sample( ctx->metrics->response_latency, (ulong)( now-sent.ts ) );
827 0 : ctx->metrics->fec_root_ok++;
828 :
829 0 : if( FD_UNLIKELY( sent.slot<=ctx->rotor->root ) ) return; /* rooted while in flight */
830 0 : fd_rotor_fec_notarized( ctx->rotor, sent.slot, &sent.dmr, sent.fec_idx*FD_FEC_SHRED_CNT, &fec_mr, now );
831 0 : }
832 :
833 : static void
834 : handle_net( fd_rotor_tile_t * ctx,
835 : fd_stem_context_t * stem,
836 0 : ulong sz ) {
837 0 : uchar * data;
838 0 : ulong data_sz;
839 0 : fd_ip4_hdr_t * ip4;
840 0 : fd_udp_hdr_t * udp;
841 0 : if( FD_UNLIKELY( !fd_ip4_udp_hdr_strip( ctx->net_buf, sz, &data, &data_sz, NULL, &ip4, &udp ) ) ) return;
842 0 : if( FD_UNLIKELY( data_sz<sizeof(uint) ) ) return;
843 :
844 0 : long now = fd_clock_tile_now( ctx->clock );
845 0 : if( FD_UNLIKELY( data_sz==FD_ROTOR_PING_DE_SZ ) ) { /* no response has a ping's size */
846 0 : handle_ping( ctx, stem, data, data_sz, ip4, udp, now );
847 0 : return;
848 0 : }
849 0 : switch( FD_LOAD( uint, data ) ) {
850 0 : case FD_ROTOR_SERDE_TAG_PARENT_FEC_SET_COUNT_RES: handle_parent_and_fec_set_count_res( ctx, data, data_sz, ip4, udp, now ); break;
851 0 : case FD_ROTOR_SERDE_TAG_FEC_SET_ROOT_RES: handle_fec_set_root_res ( ctx, data, data_sz, ip4, udp, now ); break;
852 0 : default: break;
853 0 : }
854 0 : }
855 :
856 : /* handle_sign sends the message that was waiting for this signature. */
857 :
858 : static void
859 : handle_sign( fd_rotor_tile_t * ctx,
860 : fd_stem_context_t * stem,
861 : ulong in_idx,
862 0 : ulong sig ) {
863 0 : ctx->sign[ ctx->in[ in_idx ].sign_idx ].free++;
864 0 : sign_t * pending = sign_pool_ele( ctx->sign_pool, sig>>32 );
865 0 : memcpy( pending->buf+pending->sig_off, ctx->sign_buf, FD_ED25519_SIG_SZ );
866 :
867 0 : uchar * packet = fd_chunk_to_laddr( ctx->net_out_mem, ctx->net_out_chunk );
868 0 : fd_ip4_udp_hdrs_t * hdr = (fd_ip4_udp_hdrs_t *)packet;
869 0 : *hdr = *ctx->net_hdr;
870 0 : hdr->ip4->saddr = pending->saddr;
871 0 : hdr->ip4->daddr = pending->daddr;
872 0 : hdr->ip4->net_id = fd_ushort_bswap( ctx->net_id++ );
873 0 : hdr->ip4->net_tot_len = fd_ushort_bswap( (ushort)( pending->sz+sizeof(fd_ip4_hdr_t)+sizeof(fd_udp_hdr_t) ) );
874 0 : hdr->ip4->check = fd_ip4_hdr_check_fast( hdr->ip4 );
875 0 : hdr->udp->net_dport = pending->dport;
876 0 : hdr->udp->net_len = fd_ushort_bswap( (ushort)( pending->sz+sizeof(fd_udp_hdr_t) ) );
877 0 : memcpy( packet+sizeof(fd_ip4_udp_hdrs_t), pending->buf, pending->sz );
878 :
879 0 : ulong sz = pending->sz+sizeof(fd_ip4_udp_hdrs_t);
880 0 : ulong ts = fd_frag_meta_ts_comp( fd_tickcount() );
881 0 : fd_stem_publish( stem, ctx->net_out_idx, fd_disco_netmux_sig( pending->daddr, pending->dport, pending->daddr, DST_PROTO_OUTGOING, sizeof(fd_ip4_udp_hdrs_t) ), ctx->net_out_chunk, sz, 0UL, ts, ts );
882 0 : ctx->net_out_chunk = fd_dcache_compact_next( ctx->net_out_chunk, sz, ctx->net_out_chunk0, ctx->net_out_wmark );
883 0 : sign_pool_ele_release( ctx->sign_pool, pending );
884 0 : ctx->metrics->pkt_tx++;
885 0 : }
886 :
887 : static inline void
888 0 : metrics_write( fd_rotor_tile_t * ctx ) {
889 0 : ulong queued = fd_rotor_treap_ele_cnt( ctx->rotor->eager_treap ) + fd_rotor_treap_ele_cnt( ctx->rotor->notar_treap ) + fd_rotor_treap_ele_cnt( ctx->rotor->final_treap );
890 0 : ulong waiting = timeout_prq_cnt( ctx->eager_prq ) + timeout_prq_cnt ( ctx->notar_prq ) + timeout_prq_cnt( ctx->final_prq );
891 :
892 0 : FD_MCNT_SET( ROTOR, PKT_TX, ctx->metrics->pkt_tx );
893 0 : FD_MCNT_SET( ROTOR, REQUEST_TX_WINDOW_INDEX, ctx->metrics->request_tx[ FD_ROTOR_SERDE_TAG_WINDOW_INDEX ] );
894 0 : FD_MCNT_SET( ROTOR, REQUEST_TX_HIGHEST_WINDOW_INDEX, ctx->metrics->request_tx[ FD_ROTOR_SERDE_TAG_HIGHEST_WINDOW_INDEX ] );
895 0 : FD_MCNT_SET( ROTOR, REQUEST_TX_ORPHAN, ctx->metrics->request_tx[ FD_ROTOR_SERDE_TAG_ORPHAN ] );
896 0 : FD_MCNT_SET( ROTOR, REQUEST_TX_PARENT_FEC_SET_COUNT, ctx->metrics->request_tx[ FD_ROTOR_SERDE_TAG_PARENT_AND_FEC_SET_COUNT ] );
897 0 : FD_MCNT_SET( ROTOR, REQUEST_TX_FEC_SET_ROOT, ctx->metrics->request_tx[ FD_ROTOR_SERDE_TAG_FEC_SET_ROOT ] );
898 0 : FD_MCNT_SET( ROTOR, REQUEST_TX_WINDOW_INDEX_FOR_BLOCK_ID, ctx->metrics->request_tx[ FD_ROTOR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID ] );
899 0 : FD_MCNT_SET( ROTOR, REQUEST_TX_PONG, ctx->metrics->request_tx[ FD_ROTOR_SERDE_TAG_PONG ] );
900 0 : FD_MCNT_SET( ROTOR, PING_PROBE_TX, ctx->metrics->ping_tx );
901 :
902 0 : FD_MGAUGE_SET( ROTOR, HIGHEST_DELIVERED_SLOT, ctx->metrics->slot_highest_repaired );
903 0 : FD_MGAUGE_SET( ROTOR, HIGHEST_RECEIVED_SLOT, ctx->metrics->slot_current );
904 0 : FD_MGAUGE_SET( ROTOR, FIRST_TURBINE_SLOT, ctx->turbine_slot0 );
905 0 : FD_MGAUGE_SET( ROTOR, CATCHUP_SLOT, ctx->catchup_slot );
906 0 : FD_MGAUGE_SET( ROTOR, BLOCK_PENDING, queued );
907 0 : FD_MGAUGE_SET( ROTOR, REQUEST_IN_FLIGHT, pending_pool_used( ctx->pending_pool ) );
908 0 : FD_MGAUGE_SET( ROTOR, TIMEOUT_PENDING, waiting );
909 0 : FD_MGAUGE_SET( ROTOR, EAGER_DELAY_NANOS, (ulong)fd_rotor_strat_eager_ns( ctx->strat, NULL ) );
910 0 : FD_MGAUGE_SET( ROTOR, SIGN_REQUEST_PENDING, sign_pool_used( ctx->sign_pool ) );
911 :
912 0 : FD_MCNT_SET( ROTOR, FEC_DELIVERED, ctx->metrics->fec_delivered );
913 0 : FD_MCNT_SET( ROTOR, TIMEOUT_CANCELLED, ctx->metrics->req_cancelled );
914 0 : FD_MCNT_SET( ROTOR, TIMEOUT_NO_PEER_DEFERRED, ctx->metrics->req_no_peer );
915 0 : FD_MCNT_SET( ROTOR, REQUEST_HEDGED, ctx->metrics->req_hedged );
916 0 : FD_MCNT_SET( ROTOR, REQUEST_EXPIRED, ctx->metrics->req_expired );
917 0 : FD_MCNT_SET( ROTOR, REQUEST_BY_ID_EXPIRED, ctx->metrics->byid_expired );
918 0 : FD_MCNT_SET( ROTOR, REQUEST_STUCK_ESCAPED, ctx->metrics->req_stuck_escaped );
919 0 : FD_MCNT_SET( ROTOR, REQUEST_LATE_RETRIED, ctx->metrics->req_late_attempt );
920 0 : FD_MCNT_SET( ROTOR, PEER_BY_ID_BANNED, ctx->metrics->peer_byid_banned );
921 :
922 0 : FD_MCNT_SET ( ROTOR, SHRED_OLD_IGNORED, ctx->metrics->shred_old );
923 0 : FD_MCNT_ENUM_COPY( ROTOR, SHRED_RX, ctx->metrics->shred_rx );
924 :
925 0 : FD_MCNT_SET( ROTOR, META_RX, ctx->metrics->meta_rx );
926 0 : FD_MCNT_SET( ROTOR, META_MALFORMED_REJECTED, ctx->metrics->meta_malformed );
927 0 : FD_MCNT_SET( ROTOR, META_UNSOLICITED_REJECTED, ctx->metrics->meta_unsolicited );
928 0 : FD_MCNT_SET( ROTOR, PARENT_FEC_SET_COUNT_ACCEPTED, ctx->metrics->parent_fec_count_ok );
929 0 : FD_MCNT_SET( ROTOR, PARENT_FEC_SET_COUNT_FAILED, ctx->metrics->parent_fec_count_failed );
930 0 : FD_MCNT_SET( ROTOR, FEC_SET_ROOT_ACCEPTED, ctx->metrics->fec_root_ok );
931 0 : FD_MCNT_SET( ROTOR, FEC_SET_ROOT_FAILED, ctx->metrics->fec_root_failed );
932 :
933 0 : FD_MCNT_SET( ROTOR, REPLAY_ROOT_ADVANCED_RX, ctx->metrics->replay_root_advanced );
934 0 : FD_MCNT_SET( ROTOR, REPLAY_MISSING_FEC_RX, ctx->metrics->replay_missing_fec );
935 :
936 0 : FD_MCNT_SET( ROTOR, PING_REJECTED_MALFORMED, ctx->metrics->ping_malformed );
937 0 : FD_MCNT_SET( ROTOR, PING_REJECTED_UNKNOWN_PEER, ctx->metrics->ping_unknown_peer );
938 0 : FD_MCNT_SET( ROTOR, PING_REJECTED_SIGNATURE_FAILED, ctx->metrics->ping_signature_failed );
939 :
940 0 : FD_MHIST_COPY( ROTOR, RESPONSE_LATENCY_NANOS, ctx->metrics->response_latency );
941 0 : FD_MHIST_COPY( ROTOR, RETRY_DELAY_NANOS, ctx->metrics->retry_delay );
942 0 : FD_MHIST_COPY( ROTOR, ESCAPE_PEER_HEDGE_DELAY_NANOS, ctx->metrics->escape_hedge );
943 0 : }
944 :
945 : FD_FN_CONST static inline ulong
946 0 : scratch_align( void ) {
947 0 : return 128UL;
948 0 : }
949 :
950 : FD_FN_PURE static inline ulong
951 0 : scratch_footprint( fd_topo_tile_t const * tile ) {
952 0 : ulong sign_max = tile->rotor.repair_sign_depth*tile->rotor.repair_sign_cnt;
953 0 : ulong timeout_max = tile->rotor.slot_max*(FEC_SET_P90+1UL); /* one blk of the role per slot, each a FEC set req for its p90 FEC sets and a HIGHEST_WINDOW_INDEX or PARENT_AND_FEC_SET_COUNT, a full prq rewinds its blk */
954 0 : ulong pending_max = tile->rotor.repair_sign_cnt*(ulong)( PENDING_TTL/SIGN_NS ); /* what the sign tiles can sign within PENDING_TTL, each pending is at least one */
955 0 : ulong chain_cnt = pending_map_chain_cnt_est( pending_max );
956 0 : ulong l = FD_LAYOUT_INIT;
957 0 : l = FD_LAYOUT_APPEND( l, alignof(fd_rotor_tile_t), sizeof(fd_rotor_tile_t) );
958 0 : l = FD_LAYOUT_APPEND( l, fd_rotor_align(), fd_rotor_footprint( tile->rotor.slot_max, tile->rotor.fec_max, tile->rotor.max_shreds_per_block/FD_FEC_SHRED_CNT ) );
959 0 : l = FD_LAYOUT_APPEND( l, fd_rotor_strat_align(), fd_rotor_strat_footprint() );
960 0 : l = FD_LAYOUT_APPEND( l, fd_multi_epoch_leaders_align(), fd_multi_epoch_leaders_footprint() );
961 0 : l = FD_LAYOUT_APPEND( l, ping_deque_align(), ping_deque_footprint( FD_ROTOR_STRAT_PEER_MAX ) );
962 0 : l = FD_LAYOUT_APPEND( l, timeout_prq_align(), timeout_prq_footprint( timeout_max ) );
963 0 : l = FD_LAYOUT_APPEND( l, timeout_prq_align(), timeout_prq_footprint( timeout_max ) );
964 0 : l = FD_LAYOUT_APPEND( l, timeout_prq_align(), timeout_prq_footprint( timeout_max ) );
965 0 : l = FD_LAYOUT_APPEND( l, pending_pool_align(), pending_pool_footprint( pending_max ) );
966 0 : l = FD_LAYOUT_APPEND( l, pending_map_align(), pending_map_footprint( chain_cnt ) );
967 0 : l = FD_LAYOUT_APPEND( l, pending_dlist_align(), pending_dlist_footprint() );
968 0 : l = FD_LAYOUT_APPEND( l, sign_pool_align(), sign_pool_footprint( sign_max ) );
969 0 : l = FD_LAYOUT_APPEND( l, alignof(fd_event_block_received_t), sizeof(fd_event_block_received_t) );
970 0 : return FD_LAYOUT_FINI( l, scratch_align() );
971 0 : }
972 :
973 : static inline void
974 : after_credit( fd_rotor_tile_t * ctx,
975 : fd_stem_context_t * stem,
976 : int * opt_poll_in,
977 0 : int * charge_busy ) {
978 :
979 : /* Send one reassembled FEC to replay, then repair, which publishes
980 : only on unreliable links. */
981 :
982 0 : if( FD_LIKELY( !fd_rotor_deque_empty( ctx->rotor->reasm_deque ) ) ) {
983 0 : *opt_poll_in = 0;
984 :
985 0 : fd_rotor_deque_t out = fd_rotor_deque_pop_head( ctx->rotor->reasm_deque );
986 0 : fd_rotor_blk_t const * blk = fd_rotor_blk_pool_ele_const( ctx->rotor->blk_pool, out.blk_idx );
987 0 : fd_rotor_fec_t const * fec = fd_rotor_fec_pool_ele_const( ctx->rotor->fec_pool, blk->fecs[ out.fec_idx ] );
988 0 : fd_rotor_replay_fec_t * msg = fd_chunk_to_laddr( ctx->replay_out_mem, ctx->replay_out_chunk );
989 0 : fd_memset( msg, 0, sizeof(fd_rotor_replay_fec_t) );
990 0 : msg->slot = blk->slot;
991 0 : msg->fec_set_idx = out.fec_idx*FD_FEC_SHRED_CNT;
992 0 : msg->mr = fec->mr32;
993 0 : msg->parent_slot = blk->parent_slot;
994 0 : msg->parent_block_id = blk->parent_blk_mr;
995 0 : msg->slot_complete = out.fec_idx+1U==blk->cmpl_fec_cnt;
996 0 : msg->data_complete = fec->data_complete;
997 0 : msg->is_leader = fec->is_leader;
998 0 : msg->known_id = fd_rotor_slot_meta( ctx->rotor, blk->slot )->eager!=out.blk_idx;
999 0 : msg->block_id = blk->dmr;
1000 0 : msg->metrics.highest_fec_complete_slot = fd_ulong_max( ctx->highest_fec_complete_slot, ctx->catchup_slot );
1001 0 : msg->metrics.fec_completed_ts_nanos = fec->cmpl_ts; /* shred tile rx of the shred that completed it, 0 for our own */
1002 0 : msg->metrics.votor_repaired = (uchar)msg->known_id;
1003 0 : fd_stem_publish( stem, ctx->replay_out_idx, ROTOR_SIG_FEC_REPLAY, ctx->replay_out_chunk, sizeof(fd_rotor_replay_fec_t), 0UL, 0UL, fd_frag_meta_ts_comp( fd_tickcount() ) );
1004 0 : ctx->replay_out_chunk = fd_dcache_compact_next( ctx->replay_out_chunk, sizeof(fd_rotor_replay_fec_t), ctx->replay_out_chunk0, ctx->replay_out_wmark );
1005 0 : ctx->metrics->fec_delivered++;
1006 0 : ctx->metrics->slot_highest_repaired = fd_ulong_if( msg->slot_complete, fd_ulong_max( ctx->metrics->slot_highest_repaired, blk->slot ), ctx->metrics->slot_highest_repaired );
1007 :
1008 0 : if( FD_UNLIKELY( msg->slot_complete && ctx->rserve_out_idx!=ULONG_MAX && blk->cmpl_fec_cnt<=FD_FEC_BLK_MAX ) ) { /* rserve serves protocol sized blocks only */
1009 0 : fd_rotor_block_t * block = fd_chunk_to_laddr( ctx->rserve_out_mem, ctx->rserve_out_chunk );
1010 0 : block->slot = blk->slot;
1011 0 : block->block_id = blk->dmr;
1012 0 : block->parent_slot = blk->parent_slot;
1013 0 : block->parent_block_id = blk->parent_blk_mr;
1014 0 : block->fec_set_cnt = blk->cmpl_fec_cnt;
1015 0 : for( uint k=0U; k<blk->cmpl_fec_cnt; k++ ) memcpy( block->merkle_roots[ k ], fd_rotor_fec_pool_ele_const( ctx->rotor->fec_pool, blk->fecs[ k ] )->key.uc, FD_SHRED_MERKLE_NODE_SZ );
1016 0 : ulong sz = FD_ROTOR_BLOCK_SZ( blk->cmpl_fec_cnt );
1017 0 : fd_stem_publish( stem, ctx->rserve_out_idx, ROTOR_SIG_BLOCK, ctx->rserve_out_chunk, sz, 0UL, 0UL, fd_frag_meta_ts_comp( fd_tickcount() ) );
1018 0 : ctx->rserve_out_chunk = fd_dcache_compact_next( ctx->rserve_out_chunk, sz, ctx->rserve_out_chunk0, ctx->rserve_out_wmark );
1019 0 : }
1020 0 : *charge_busy = 1;
1021 0 : }
1022 :
1023 : /* Then expire unanswered requests, discover as many blks off the
1024 : blk_treaps as there are free sign slots (a blk yields at least one
1025 : req, a req at least one sign request), and send pings and due
1026 : reqs while sign slots and pending_pool entries last. A req is one
1027 : peer pick. */
1028 :
1029 0 : long now = fd_clock_tile_now( ctx->clock );
1030 :
1031 0 : while( !pending_dlist_is_empty( ctx->pending_dlist, ctx->pending_pool ) ) {
1032 0 : pending_t * pending = pending_dlist_ele_peek_head( ctx->pending_dlist, ctx->pending_pool );
1033 0 : if( FD_LIKELY( now-pending->ts<PENDING_TTL ) ) break;
1034 :
1035 : /* No rtt sample: silence also means a deduped reply, or a highest
1036 : or orphan request no peer can answer yet. Only a run of shred
1037 : and meta requests left unanswered bans the peer. A request
1038 : answered in part settled its pick on the first shred. */
1039 :
1040 0 : int answerable = pending->slot>ctx->rotor->root && pending->tag!=FD_ROTOR_SERDE_TAG_HIGHEST_WINDOW_INDEX && pending->tag!=FD_ROTOR_SERDE_TAG_ORPHAN;
1041 0 : int byid = pending->tag==FD_ROTOR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID;
1042 0 : ctx->metrics->byid_expired += (ulong)( answerable && byid && !pending->answered );
1043 0 : if( FD_UNLIKELY( pending->answered ) ) {}
1044 0 : else if( FD_LIKELY ( answerable ) ) ctx->metrics->peer_byid_banned += (ulong)( fd_rotor_strat_request_expired( ctx->strat, &pending->peer, now+BAN_TTL, byid )==1 );
1045 0 : else fd_rotor_strat_request_failed ( ctx->strat, &pending->peer, 0 );
1046 0 : ctx->metrics->req_expired+= (ulong)!pending->answered;
1047 0 : pending_dlist_ele_pop_head ( ctx->pending_dlist, ctx->pending_pool );
1048 0 : pending_map_ele_remove_fast( ctx->pending_map, pending, ctx->pending_pool );
1049 0 : pending_pool_ele_release ( ctx->pending_pool, pending );
1050 0 : }
1051 :
1052 0 : for( ulong d=sign_pool_free( ctx->sign_pool ); d; d-- ) {
1053 0 : fd_rotor_t * rotor = ctx->rotor;
1054 0 : fd_rotor_treap_t * treap = fd_rotor_treap_ele_cnt( rotor->final_treap ) ? rotor->final_treap :
1055 0 : fd_rotor_treap_ele_cnt( rotor->notar_treap ) ? rotor->notar_treap :
1056 0 : fd_rotor_treap_ele_cnt( rotor->eager_treap ) ? rotor->eager_treap :
1057 0 : NULL;
1058 0 : if( FD_UNLIKELY( !treap ) ) break;
1059 0 : fd_rotor_blk_t * blk = fd_rotor_treap_fwd_iter_ele( fd_rotor_treap_fwd_iter_init( treap, rotor->blk_pool ), rotor->blk_pool );
1060 0 : fd_rotor_treap_ele_remove( treap, blk, rotor->blk_pool );
1061 0 : blk->in_blk_treap = 0;
1062 0 : discover( ctx, blk, now );
1063 0 : }
1064 0 : if( FD_UNLIKELY( ctx->halt_signing ) ) return;
1065 :
1066 : /* A ping is any signed request: the peer pings before it answers. No
1067 : one has slot 0 under the null block id, so no one answers it. */
1068 :
1069 0 : ulong sent = 0UL;
1070 0 : ulong ts = (ulong)now/1000000UL;
1071 0 : while( sent<PING_MAX
1072 0 : && !ping_deque_empty( ctx->ping_deque )
1073 0 : && sign_pool_free( ctx->sign_pool )>SIGN_RESERVE ) {
1074 0 : fd_pubkey_t id_key = ping_deque_pop_head( ctx->ping_deque );
1075 0 : fd_rotor_strat_peer_t * peer = fd_rotor_strat_query( ctx->strat, &id_key );
1076 0 : sent++;
1077 0 : if( FD_UNLIKELY( !peer || !peer->ip4 || peer->ping_ts ) ) continue; /* gone, or sent to since */
1078 0 : peer->ping_ts = now;
1079 :
1080 0 : sign_t * pending = sign_pool_ele_acquire( ctx->sign_pool );
1081 0 : *pending = (sign_t){ .sig_off = sizeof(uint), .daddr = peer->ip4, .dport = fd_ushort_bswap( peer->port ) };
1082 0 : pending->sz = fd_rotor_req_parent_and_fec_set_count_ser( &(fd_rotor_parent_fec_set_count_t){ .slot = 0UL }, sig_null, &ctx->identity_key, &id_key, ts, fd_rng_uint( ctx->rng ), pending->buf );
1083 :
1084 0 : ulong s = 0UL;
1085 0 : for( ulong i=1UL; i<ctx->sign_cnt; i++ ) s = fd_ulong_if( ctx->sign[ i ].free>ctx->sign[ s ].free, i, s );
1086 0 : ulong sz = fd_rotor_req_sig_ser( pending->buf, pending->sz, fd_chunk_to_laddr( ctx->sign[ s ].mem, ctx->sign[ s ].chunk ) );
1087 0 : fd_stem_publish( stem, ctx->sign[ s ].idx, (sign_pool_idx( ctx->sign_pool, pending )<<32)|(ulong)FD_KEYGUARD_SIGN_TYPE_ED25519, ctx->sign[ s ].chunk, sz, 0UL, 0UL, 0UL );
1088 0 : ctx->sign[ s ].chunk = fd_dcache_compact_next( ctx->sign[ s ].chunk, sz, ctx->sign[ s ].chunk0, ctx->sign[ s ].wmark );
1089 0 : ctx->sign[ s ].free--;
1090 0 : ctx->metrics->ping_tx++;
1091 0 : *charge_busy = 1;
1092 0 : }
1093 :
1094 0 : while( sign_pool_free( ctx->sign_pool )>=SIGN_RESERVE+FD_FEC_SHRED_CNT ) {
1095 0 : timeout_t req;
1096 0 : fd_rotor_blk_t * blk = poll_timeout( ctx, now, &req );
1097 0 : if( FD_UNLIKELY( !blk ) ) break;
1098 :
1099 0 : fd_rotor_fec_t const * fec = fd_rotor_fec_pool_ele_const( ctx->rotor->fec_pool, fd_ulong_if( req.tag==FD_ROTOR_SERDE_TAG_WINDOW_INDEX, blk->fecs[ req.fec_idx ], fd_rotor_fec_pool_idx_null( ctx->rotor->fec_pool ) ) );
1100 0 : int eager = blk->eager;
1101 0 : uint tag = req.tag!=FD_ROTOR_SERDE_TAG_WINDOW_INDEX ? req.tag : /* a FEC set req becomes what its blk still lacks */
1102 0 : eager ? FD_ROTOR_SERDE_TAG_WINDOW_INDEX :
1103 0 : !fec ? FD_ROTOR_SERDE_TAG_FEC_SET_ROOT :
1104 0 : FD_ROTOR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID;
1105 0 : int meta = tag==FD_ROTOR_SERDE_TAG_PARENT_AND_FEC_SET_COUNT || tag==FD_ROTOR_SERDE_TAG_FEC_SET_ROOT;
1106 0 : uint mask = fd_uint_if( tag==FD_ROTOR_SERDE_TAG_WINDOW_INDEX || tag==FD_ROTOR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID, fec ? ~fec->rcvd : UINT_MAX, 1U ); /* bit i is a message for shred i of the FEC set */
1107 0 : if( FD_UNLIKELY( !mask ) ) { /* every shred arrived, the FEC set completes without repair */
1108 0 : req.timeout = now+fd_rotor_strat_eager_ns( ctx->strat, fd_multi_epoch_leaders_get_leader_for_slot( ctx->mleaders, blk->slot ) );
1109 0 : push_timeout( ctx, blk, &req );
1110 0 : continue;
1111 0 : }
1112 :
1113 : /* A retry asks someone else. After STUCK_MAX attempts that brought
1114 : no new shred, the fastest peers are not answering: pick a staked
1115 : peer at random instead. */
1116 0 : uint rcvd = fec ? fec->rcvd : 0U;
1117 0 : uint stuck = fd_uint_if( req.attempt && !( rcvd & ~req.rcvd ), req.stuck+1U, 0U ); /* only new shreds are progress, not an eviction; kept only once sent */
1118 0 : int escape = stuck>=STUCK_MAX && !blk->eager;
1119 0 : fd_rotor_strat_peer_t * peer = escape ? fd_rotor_strat_pick_staked( ctx->strat, now, (ulong)req.last_peer-1UL, fd_rng_ulong( ctx->rng ) ) : NULL;
1120 0 : ctx->metrics->req_stuck_escaped += (ulong)!!peer;
1121 0 : if( FD_UNLIKELY( peer ) ) fd_histf_sample( ctx->metrics->escape_hedge, (ulong)fd_long_if( !!peer->srtt, peer->srtt+4L*peer->rttvar, FD_ROTOR_STRAT_HEDGE_NS ) ); /* uncapped */
1122 0 : if( FD_LIKELY( !peer ) ) peer = fd_rotor_strat_pick( ctx->strat, now, !blk->eager && req.attempt, (ulong)req.last_peer-1UL );
1123 0 : if( FD_UNLIKELY( !peer ) ) {
1124 0 : ctx->metrics->req_no_peer++;
1125 0 : req.timeout = now+NO_PEER;
1126 0 : push_timeout( ctx, blk, &req );
1127 0 : break;
1128 0 : }
1129 0 : peer->ping_ts = now;
1130 0 : req.stuck = stuck;
1131 0 : req.rcvd = rcvd;
1132 :
1133 0 : uint nonce = fd_rng_uint( ctx->rng );
1134 0 : while( meta && pending_map_ele_query_const( ctx->pending_map, &(ulong){ nonce }, NULL, ctx->pending_pool ) ) nonce = fd_rng_uint( ctx->rng );
1135 :
1136 0 : if( FD_UNLIKELY( !pending_pool_free( ctx->pending_pool ) ) ) { /* evict the oldest, a reply to it no longer matches */
1137 0 : pending_t * oldest = pending_dlist_ele_pop_head( ctx->pending_dlist, ctx->pending_pool );
1138 0 : if( FD_LIKELY( !oldest->answered ) ) fd_rotor_strat_request_failed( ctx->strat, &oldest->peer, 0 );
1139 0 : pending_map_ele_remove_fast( ctx->pending_map, oldest, ctx->pending_pool );
1140 0 : pending_pool_ele_release ( ctx->pending_pool, oldest );
1141 0 : }
1142 0 : pending_t * pending = pending_pool_ele_acquire( ctx->pending_pool );
1143 0 : pending->key = fd_ulong_if( meta, (ulong)nonce, (ulong)(uint)blk->slot<<32 | fd_uint_if( req.tag==FD_ROTOR_SERDE_TAG_HIGHEST_WINDOW_INDEX || req.tag==FD_ROTOR_SERDE_TAG_ORPHAN, (uint)ctx->rotor->fec_blk_max, req.fec_idx ) ); /* an ORPHAN's first reply is of the slot itself, matched like a HIGHEST */
1144 0 : pending->ts = now;
1145 0 : pending->slot = blk->slot;
1146 0 : pending->blk = req.blk;
1147 0 : pending->dmr = blk->dmr;
1148 0 : pending->peer = peer->id_key;
1149 0 : pending->fec_idx = req.fec_idx;
1150 0 : pending->fec_set_cnt = blk->cmpl_fec_cnt;
1151 0 : pending->tag = tag;
1152 0 : pending->mask = mask;
1153 0 : pending->answered = 0U;
1154 0 : pending_map_ele_insert ( ctx->pending_map, pending, ctx->pending_pool );
1155 0 : pending_dlist_ele_push_tail( ctx->pending_dlist, pending, ctx->pending_pool );
1156 :
1157 0 : for( uint m=mask; m; m=fd_uint_pop_lsb( m ) ) {
1158 0 : uint idx = req.fec_idx*FD_FEC_SHRED_CNT+(uint)fd_uint_find_lsb( m );
1159 0 : uint rnonce = fd_uint_if( meta, nonce, fd_rnonce_ss_compute( ctx->rnonce_ss, tag!=FD_ROTOR_SERDE_TAG_HIGHEST_WINDOW_INDEX && tag!=FD_ROTOR_SERDE_TAG_ORPHAN, blk->slot, idx, now ) );
1160 0 : sign_t * sign = sign_pool_ele_acquire( ctx->sign_pool );
1161 0 : *sign = (sign_t){ .sig_off = sizeof(uint), .daddr = peer->ip4, .dport = fd_ushort_bswap( peer->port ) };
1162 0 : switch( tag ) {
1163 0 : case FD_ROTOR_SERDE_TAG_PARENT_AND_FEC_SET_COUNT: sign->sz = fd_rotor_req_parent_and_fec_set_count_ser ( &(fd_rotor_parent_fec_set_count_t){ .slot = blk->slot, .block_id = blk->dmr }, sig_null, &ctx->identity_key, &peer->id_key, ts, rnonce, sign->buf ); break;
1164 0 : case FD_ROTOR_SERDE_TAG_FEC_SET_ROOT: sign->sz = fd_rotor_req_fec_set_root_ser ( &(fd_rotor_fec_set_root_t){ .slot = blk->slot, .block_id = blk->dmr, .fec_set_idx = idx }, sig_null, &ctx->identity_key, &peer->id_key, ts, rnonce, sign->buf ); break;
1165 0 : case FD_ROTOR_SERDE_TAG_ORPHAN: sign->sz = fd_rotor_req_orphan_ser ( &(fd_rotor_orphan_t){ .slot = blk->slot }, sig_null, &ctx->identity_key, &peer->id_key, ts, rnonce, sign->buf ); break;
1166 0 : case FD_ROTOR_SERDE_TAG_HIGHEST_WINDOW_INDEX: sign->sz = fd_rotor_req_highest_window_index_ser ( &(fd_rotor_highest_shred_t){ .slot = blk->slot, .shred_idx = blk->rcvd_fec_cnt*FD_FEC_SHRED_CNT }, sig_null, &ctx->identity_key, &peer->id_key, ts, rnonce, sign->buf ); break;
1167 0 : case FD_ROTOR_SERDE_TAG_WINDOW_INDEX: sign->sz = fd_rotor_req_window_index_ser ( &(fd_rotor_shred_t){ .slot = blk->slot, .shred_idx = idx }, sig_null, &ctx->identity_key, &peer->id_key, ts, rnonce, sign->buf ); break;
1168 0 : case FD_ROTOR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID: sign->sz = fd_rotor_req_window_index_for_block_id_ser( &(fd_rotor_shred_for_block_id_t){ .slot = blk->slot, .shred_idx = idx, .block_id = blk->dmr }, sig_null, &ctx->identity_key, &peer->id_key, ts, rnonce, sign->buf ); break;
1169 0 : }
1170 :
1171 0 : ulong s = 0UL;
1172 0 : for( ulong i=1UL; i<ctx->sign_cnt; i++ ) s = fd_ulong_if( ctx->sign[ i ].free>ctx->sign[ s ].free, i, s );
1173 0 : ulong sz = fd_rotor_req_sig_ser( sign->buf, sign->sz, fd_chunk_to_laddr( ctx->sign[ s ].mem, ctx->sign[ s ].chunk ) );
1174 0 : fd_stem_publish( stem, ctx->sign[ s ].idx, (sign_pool_idx( ctx->sign_pool, sign )<<32)|(ulong)FD_KEYGUARD_SIGN_TYPE_ED25519, ctx->sign[ s ].chunk, sz, 0UL, 0UL, 0UL );
1175 0 : ctx->sign[ s ].chunk = fd_dcache_compact_next( ctx->sign[ s ].chunk, sz, ctx->sign[ s ].chunk0, ctx->sign[ s ].wmark );
1176 0 : ctx->sign[ s ].free--;
1177 0 : }
1178 0 : ctx->metrics->request_tx[ tag ] += (ulong)fd_uint_popcnt( mask );
1179 0 : ctx->metrics->req_hedged += (ulong)!!req.attempt;
1180 0 : ctx->metrics->req_late_attempt += (ulong)( req.attempt>=3U );
1181 :
1182 0 : blk->telemetry.first_req_ts = fd_long_if( !blk->telemetry.first_req_ts, now, blk->telemetry.first_req_ts );
1183 0 : uint * req_cnt = NULL;
1184 0 : switch( tag ) {
1185 0 : case FD_ROTOR_SERDE_TAG_WINDOW_INDEX: req_cnt = &blk->telemetry.req_window_cnt; break;
1186 0 : case FD_ROTOR_SERDE_TAG_HIGHEST_WINDOW_INDEX: req_cnt = &blk->telemetry.req_highest_cnt; break;
1187 0 : case FD_ROTOR_SERDE_TAG_ORPHAN: req_cnt = &blk->telemetry.req_orphan_cnt; break;
1188 0 : case FD_ROTOR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID: req_cnt = &blk->telemetry.req_shred_bid_cnt; break;
1189 0 : case FD_ROTOR_SERDE_TAG_PARENT_AND_FEC_SET_COUNT: req_cnt = &blk->telemetry.req_parent_cnt; break;
1190 0 : case FD_ROTOR_SERDE_TAG_FEC_SET_ROOT: req_cnt = &blk->telemetry.req_fec_root_cnt; break;
1191 0 : }
1192 0 : if( FD_LIKELY( req_cnt ) ) *req_cnt += (uint)fd_uint_popcnt( mask );
1193 :
1194 0 : long hedge = fd_long_if( escape, 1L<<24, fd_long_min( fd_long_max( fd_rotor_strat_hedge_ns( peer ), 1L<<24 ), FD_ROTOR_STRAT_HEDGE_NS ) );
1195 0 : req.timeout = now+hedge; /* no backoff: a re-pick gets a new shred nonce, and a random escape pick does not slow the next one */
1196 0 : req.attempt = (uchar)fd_uint_min( req.attempt+1U, UCHAR_MAX );
1197 0 : req.last_peer = (ushort)( 1UL+(ulong)( peer-ctx->strat->cur.peers ) );
1198 0 : fd_histf_sample( ctx->metrics->retry_delay, (ulong)( req.timeout-now ) );
1199 0 : push_timeout( ctx, blk, &req );
1200 0 : *charge_busy = 1;
1201 0 : }
1202 0 : }
1203 :
1204 : static int
1205 : before_frag( fd_rotor_tile_t * ctx,
1206 : ulong in_idx,
1207 : ulong seq FD_PARAM_UNUSED,
1208 0 : ulong sig ) {
1209 0 : switch( ctx->in_kind[ in_idx ] ) {
1210 0 : case IN_KIND_GENESIS:
1211 0 : return 0;
1212 0 : case IN_KIND_SHRED: {
1213 0 : if( FD_UNLIKELY( ctx->rotor->root==ULONG_MAX ) ) return -1; /* hold until the snapshot or genesis gives the root */
1214 0 : if( FD_LIKELY( fd_shred_sig_src( sig )!=SHRED_SIG_FUTURE ) ) return 0;
1215 0 : ulong slot = fd_shred_sig_future_slot( sig );
1216 0 : if( FD_LIKELY( ctx->staked || slot<ctx->rotor->root+ctx->rotor->slot_max || slot<=ctx->catchup_slot ) ) return 1;
1217 0 : ctx->catchup_slot = slot;
1218 0 : fd_rotor_slot_catchup( ctx->rotor, ctx->rotor->root+ctx->rotor->slot_max );
1219 0 : return 1;
1220 0 : }
1221 0 : case IN_KIND_SNAP: {
1222 0 : ulong msg = fd_ssmsg_sig_message( sig );
1223 0 : return msg!=FD_SSMSG_MANIFEST_FULL && msg!=FD_SSMSG_MANIFEST_INCREMENTAL && msg!=FD_SSMSG_DONE;
1224 0 : }
1225 0 : case IN_KIND_VOTOR:
1226 0 : if( FD_UNLIKELY( ctx->rotor->root==ULONG_MAX ) ) return -1; /* hold until the snapshot or genesis gives the root */
1227 0 : return sig!=FD_VOTOR_SIG_CERTED && sig!=FD_VOTOR_SIG_REPAIR && sig!=FD_VOTOR_SIG_QUORUM;
1228 0 : case IN_KIND_REPLAY:
1229 0 : return sig!=REPLAY_SIG_ROOT_ADVANCED && sig!=REPLAY_SIG_SLOT_DEAD && sig!=REPLAY_SIG_MISSING_FEC;
1230 0 : case IN_KIND_EPOCH:
1231 0 : return 0;
1232 0 : case IN_KIND_GOSSIP:
1233 0 : return sig!=FD_GOSSIP_UPDATE_TAG_CONTACT_INFO && sig!=FD_GOSSIP_UPDATE_TAG_CONTACT_INFO_REMOVE;
1234 0 : case IN_KIND_NET:
1235 0 : return fd_disco_netmux_sig_proto( sig )!=DST_PROTO_REPAIR;
1236 0 : case IN_KIND_SIGN:
1237 0 : return 0;
1238 0 : default:
1239 0 : FD_LOG_ERR(( "unexpected in_kind %d", ctx->in_kind[ in_idx ] ));
1240 0 : }
1241 0 : }
1242 :
1243 : static void
1244 : during_frag( fd_rotor_tile_t * ctx,
1245 : ulong in_idx,
1246 : ulong seq FD_PARAM_UNUSED,
1247 : ulong sig FD_PARAM_UNUSED,
1248 : ulong chunk,
1249 : ulong sz,
1250 0 : ulong ctl ) {
1251 0 : switch( ctx->in_kind[ in_idx ] ) {
1252 0 : case IN_KIND_NET:
1253 0 : memcpy( ctx->net_buf, fd_net_rx_translate_frag( &ctx->in[ in_idx ].net_rx, chunk, ctl, sz ), sz );
1254 0 : break;
1255 0 : case IN_KIND_SIGN:
1256 0 : if( FD_UNLIKELY( chunk<ctx->in[ in_idx ].chunk0 || chunk>ctx->in[ in_idx ].wmark || sz>ctx->in[ in_idx ].mtu ) ) {
1257 0 : FD_LOG_ERR(( "chunk %lu sz %lu from in_kind %d out of bounds, chunk0 %lu wmark %lu",
1258 0 : chunk, sz, ctx->in_kind[ in_idx ], ctx->in[ in_idx ].chunk0, ctx->in[ in_idx ].wmark ));
1259 0 : }
1260 0 : memcpy( ctx->sign_buf, fd_chunk_to_laddr_const( ctx->in[ in_idx ].mem, chunk ), FD_ED25519_SIG_SZ );
1261 0 : break;
1262 0 : default: /* reliable, read in place by after_frag */
1263 0 : if( FD_UNLIKELY( sz!=0UL && ( chunk<ctx->in[ in_idx ].chunk0 || chunk>ctx->in[ in_idx ].wmark || sz>ctx->in[ in_idx ].mtu ) ) ) { /* a snapshot DONE has no payload */
1264 0 : FD_LOG_ERR(( "chunk %lu sz %lu from in_kind %d out of bounds, chunk0 %lu wmark %lu",
1265 0 : chunk, sz, ctx->in_kind[ in_idx ], ctx->in[ in_idx ].chunk0, ctx->in[ in_idx ].wmark ));
1266 0 : }
1267 0 : ctx->chunk = chunk;
1268 0 : break;
1269 0 : }
1270 0 : }
1271 :
1272 : static void
1273 : after_frag( fd_rotor_tile_t * ctx,
1274 : ulong in_idx,
1275 : ulong seq FD_PARAM_UNUSED,
1276 : ulong sig,
1277 : ulong sz,
1278 : ulong tsorig,
1279 : ulong tspub FD_PARAM_UNUSED,
1280 0 : fd_stem_context_t * stem ) {
1281 0 : uchar const * src = fd_chunk_to_laddr_const( ctx->in[ in_idx ].mem, ctx->chunk );
1282 :
1283 0 : switch( ctx->in_kind[ in_idx ] ) {
1284 0 : case IN_KIND_NET:
1285 0 : handle_net( ctx, stem, sz );
1286 0 : break;
1287 0 : case IN_KIND_SIGN:
1288 0 : handle_sign( ctx, stem, in_idx, sig );
1289 0 : break;
1290 0 : case IN_KIND_GENESIS: {
1291 0 : fd_genesis_meta_t const * meta = (fd_genesis_meta_t const *)fd_type_pun_const( src );
1292 0 : if( FD_UNLIKELY( !meta->bootstrap ) ) break; /* booting from a snapshot */
1293 0 : fd_rotor_init( ctx->rotor, 0UL, &hash_null, report_block_received, ctx );
1294 0 : break;
1295 0 : }
1296 0 : case IN_KIND_SHRED: { /* tsorig is the shred tile's network rx of the shred, the completing one for a FEC set */
1297 0 : long now_tick = fd_tickcount();
1298 0 : long rx_tick = fd_frag_meta_ts_decomp( tsorig, now_tick );
1299 0 : if( FD_UNLIKELY( rx_tick>now_tick ) ) rx_tick -= 1L<<32;
1300 0 : handle_shred( ctx, sig, src, fd_clock_tile_tickcount_to_wallclock( ctx->clock, rx_tick ) );
1301 0 : break;
1302 0 : }
1303 0 : case IN_KIND_SNAP:
1304 0 : handle_snapshot( ctx, sig, src );
1305 0 : break;
1306 0 : case IN_KIND_VOTOR:
1307 0 : handle_votor( ctx, sig, src );
1308 0 : break;
1309 0 : case IN_KIND_REPLAY:
1310 0 : handle_replay( ctx, sig, src );
1311 0 : break;
1312 0 : case IN_KIND_EPOCH: {
1313 0 : fd_epoch_info_msg_t const * msg = (fd_epoch_info_msg_t const *)fd_type_pun_const( src );
1314 0 : fd_rotor_strat_epoch_advanced( ctx->strat, fd_epoch_info_msg_id_weights( msg ), msg->staked_id_cnt );
1315 0 : timeout_t * prqs[3] = { ctx->final_prq, ctx->notar_prq, ctx->eager_prq };
1316 0 : for( ulong c=0UL; c<3UL; c++ ) { /* the re-sort moved peers, keep retries avoiding the same one */
1317 0 : for( ulong i=0UL; i<timeout_prq_cnt( prqs[ c ] ); i++ ) {
1318 0 : timeout_t * req = &prqs[ c ][ i ];
1319 0 : req->last_peer = (ushort)( 1UL+fd_rotor_strat_slot_remap( ctx->strat, (ulong)req->last_peer-1UL ) );
1320 0 : }
1321 0 : }
1322 0 : fd_rotor_strat_peer_t const * self = fd_rotor_strat_query( ctx->strat, &ctx->identity_key );
1323 0 : ctx->staked = self && (ulong)( self-ctx->strat->cur.peers )<ctx->strat->staked_cnt;
1324 0 : fd_multi_epoch_leaders_epoch_msg_init( ctx->mleaders, msg );
1325 0 : fd_multi_epoch_leaders_epoch_msg_fini( ctx->mleaders );
1326 0 : break;
1327 0 : }
1328 0 : case IN_KIND_GOSSIP:
1329 0 : handle_gossip( ctx, sig, (fd_gossip_update_message_t const *)fd_type_pun_const( src ) );
1330 0 : break;
1331 0 : default:
1332 0 : FD_LOG_ERR(( "unexpected in_kind %d", ctx->in_kind[ in_idx ] ));
1333 0 : }
1334 0 : }
1335 :
1336 : /* next_deadline is when the tile next has work without a frag: the
1337 : earliest timeout due, or the oldest pending expiring. */
1338 :
1339 : static long
1340 0 : next_deadline( fd_rotor_tile_t * ctx ) {
1341 0 : long due = LONG_MAX;
1342 0 : due = fd_long_if( !!timeout_prq_cnt( ctx->eager_prq ), fd_long_min( due, ctx->eager_prq[ 0 ].timeout ), due );
1343 0 : due = fd_long_if( !!timeout_prq_cnt( ctx->notar_prq ), fd_long_min( due, ctx->notar_prq[ 0 ].timeout ), due );
1344 0 : due = fd_long_if( !!timeout_prq_cnt( ctx->final_prq ), fd_long_min( due, ctx->final_prq[ 0 ].timeout ), due );
1345 0 : if( FD_LIKELY( !pending_dlist_is_empty( ctx->pending_dlist, ctx->pending_pool ) ) ) due = fd_long_min( due, pending_dlist_ele_peek_head( ctx->pending_dlist, ctx->pending_pool )->ts+PENDING_TTL );
1346 0 : if( FD_UNLIKELY( due==LONG_MAX ) ) return LONG_MAX;
1347 0 : return fd_clock_tile_wallclock_to_tickcount( ctx->clock, due );
1348 0 : }
1349 :
1350 : static inline void
1351 0 : during_housekeeping( fd_rotor_tile_t * ctx ) {
1352 0 : if( FD_UNLIKELY( fd_clock_tile_recal_due( ctx->clock ) ) ) fd_clock_tile_recal( ctx->clock );
1353 :
1354 0 : if( FD_UNLIKELY( fd_keyswitch_state_query( ctx->keyswitch )==FD_KEYSWITCH_STATE_UNHALT_PENDING ) ) {
1355 0 : FD_CHECK_CRIT( ctx->halt_signing, "state machine corruption" );
1356 0 : ctx->halt_signing = 0;
1357 :
1358 : /* Peers key their ping-pong cache by our pubkey, so every peer is
1359 : cold for the new identity. Re-warm them all. */
1360 0 : ping_deque_remove_all( ctx->ping_deque );
1361 0 : for( ulong i=0UL; i<FD_ROTOR_STRAT_PEER_MAX && !ping_deque_full( ctx->ping_deque ); i++ ) {
1362 0 : fd_rotor_strat_peer_t * peer = &ctx->strat->cur.peers[ i ];
1363 0 : if( FD_LIKELY( !peer->ip4 ) ) continue; /* free slot, or no address */
1364 0 : peer->ping_ts = 0L;
1365 0 : ping_deque_push_tail( ctx->ping_deque, peer->id_key );
1366 0 : }
1367 :
1368 0 : fd_keyswitch_state( ctx->keyswitch, FD_KEYSWITCH_STATE_COMPLETED );
1369 0 : }
1370 :
1371 0 : if( FD_UNLIKELY( fd_keyswitch_state_query( ctx->keyswitch )==FD_KEYSWITCH_STATE_SWITCH_PENDING ) ) {
1372 0 : ctx->halt_signing = 1;
1373 0 : memcpy( ctx->identity_key.uc, ctx->keyswitch->bytes, 32UL );
1374 0 : fd_rotor_strat_peer_t const * self = fd_rotor_strat_query( ctx->strat, &ctx->identity_key );
1375 0 : ctx->staked = self && (ulong)( self-ctx->strat->cur.peers )<ctx->strat->staked_cnt;
1376 0 : if( FD_LIKELY( sign_pool_free( ctx->sign_pool )==sign_pool_max( ctx->sign_pool ) ) ) fd_keyswitch_state( ctx->keyswitch, FD_KEYSWITCH_STATE_COMPLETED ); /* every sign request returned */
1377 0 : }
1378 0 : }
1379 :
1380 : static void
1381 : privileged_init( fd_topo_t const * topo,
1382 0 : fd_topo_tile_t const * tile ) {
1383 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
1384 :
1385 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
1386 0 : fd_rotor_tile_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_rotor_tile_t), sizeof(fd_rotor_tile_t) );
1387 0 : fd_memset( ctx, 0, sizeof(fd_rotor_tile_t) );
1388 :
1389 0 : FD_TEST( fd_rng_secure( &ctx->seed, sizeof(ulong) ) );
1390 :
1391 0 : uchar const * identity_key = fd_keyload_load( tile->rotor.identity_key_path, /* pubkey only: */ 1 );
1392 0 : fd_memcpy( ctx->identity_key.uc, identity_key, sizeof(fd_pubkey_t) );
1393 :
1394 0 : ulong rnonce_ss_id = fd_pod_queryf_ulong( topo->props, ULONG_MAX, "rnonce_ss" );
1395 0 : FD_TEST( rnonce_ss_id!=ULONG_MAX );
1396 0 : memcpy( ctx->rnonce_ss, fd_topo_obj_laddr( topo, rnonce_ss_id ), sizeof(fd_rnonce_ss_t) );
1397 0 : }
1398 :
1399 : static void
1400 : unprivileged_init( fd_topo_t const * topo,
1401 0 : fd_topo_tile_t const * tile ) {
1402 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
1403 0 : ulong sign_max = tile->rotor.repair_sign_depth*tile->rotor.repair_sign_cnt;
1404 0 : ulong timeout_max = tile->rotor.slot_max*(FEC_SET_P90+1UL);
1405 0 : ulong pending_max = tile->rotor.repair_sign_cnt*(ulong)( PENDING_TTL/SIGN_NS );
1406 0 : ulong chain_cnt = pending_map_chain_cnt_est( pending_max );
1407 :
1408 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
1409 0 : fd_rotor_tile_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_rotor_tile_t), sizeof(fd_rotor_tile_t) );
1410 0 : ctx->rotor = FD_SCRATCH_ALLOC_APPEND( l, fd_rotor_align(), fd_rotor_footprint( tile->rotor.slot_max, tile->rotor.fec_max, tile->rotor.max_shreds_per_block/FD_FEC_SHRED_CNT ) );
1411 0 : void * strat = FD_SCRATCH_ALLOC_APPEND( l, fd_rotor_strat_align(), fd_rotor_strat_footprint() );
1412 0 : void * mleaders = FD_SCRATCH_ALLOC_APPEND( l, fd_multi_epoch_leaders_align(), fd_multi_epoch_leaders_footprint() );
1413 0 : void * ping_deque = FD_SCRATCH_ALLOC_APPEND( l, ping_deque_align(), ping_deque_footprint( FD_ROTOR_STRAT_PEER_MAX ) );
1414 0 : void * eager_prq = FD_SCRATCH_ALLOC_APPEND( l, timeout_prq_align(), timeout_prq_footprint( timeout_max ) );
1415 0 : void * notar_prq = FD_SCRATCH_ALLOC_APPEND( l, timeout_prq_align(), timeout_prq_footprint( timeout_max ) );
1416 0 : void * final_prq = FD_SCRATCH_ALLOC_APPEND( l, timeout_prq_align(), timeout_prq_footprint( timeout_max ) );
1417 0 : void * pending_pool = FD_SCRATCH_ALLOC_APPEND( l, pending_pool_align(), pending_pool_footprint( pending_max ) );
1418 0 : void * pending_map = FD_SCRATCH_ALLOC_APPEND( l, pending_map_align(), pending_map_footprint( chain_cnt ) );
1419 0 : void * pending_dlist = FD_SCRATCH_ALLOC_APPEND( l, pending_dlist_align(), pending_dlist_footprint() );
1420 0 : void * sign_pool = FD_SCRATCH_ALLOC_APPEND( l, sign_pool_align(), sign_pool_footprint( sign_max ) );
1421 0 : ctx->event = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_event_block_received_t), sizeof(fd_event_block_received_t) );
1422 0 : ulong scratch_top = FD_SCRATCH_ALLOC_FINI( l, scratch_align() );
1423 0 : if( FD_UNLIKELY( scratch_top > (ulong)scratch + scratch_footprint( tile ) ) )
1424 0 : FD_LOG_ERR(( "scratch overflow %lu %lu %lu", scratch_top - (ulong)scratch - scratch_footprint( tile ), scratch_top, (ulong)scratch + scratch_footprint( tile ) ));
1425 :
1426 0 : ctx->rotor = fd_rotor_join( fd_rotor_new( ctx->rotor, tile->rotor.slot_max, tile->rotor.fec_max, tile->rotor.max_shreds_per_block/FD_FEC_SHRED_CNT, ctx->seed ) );
1427 0 : FD_TEST( ctx->rotor );
1428 0 : ctx->mleaders = fd_multi_epoch_leaders_join( fd_multi_epoch_leaders_new( mleaders ) );
1429 0 : FD_TEST( ctx->mleaders );
1430 :
1431 0 : ctx->keyswitch = fd_keyswitch_join( fd_topo_obj_laddr( topo, tile->id_keyswitch_obj_id ) );
1432 0 : FD_TEST( ctx->keyswitch );
1433 :
1434 0 : ctx->strat = fd_rotor_strat_join( fd_rotor_strat_new( strat, ctx->seed ) );
1435 0 : ctx->ping_deque = ping_deque_join( ping_deque_new( ping_deque, FD_ROTOR_STRAT_PEER_MAX ) );
1436 0 : ctx->eager_prq = timeout_prq_join( timeout_prq_new( eager_prq, timeout_max ) );
1437 0 : ctx->notar_prq = timeout_prq_join( timeout_prq_new( notar_prq, timeout_max ) );
1438 0 : ctx->final_prq = timeout_prq_join( timeout_prq_new( final_prq, timeout_max ) );
1439 0 : ctx->pending_pool = pending_pool_join( pending_pool_new( pending_pool, pending_max ) );
1440 0 : ctx->pending_map = pending_map_join( pending_map_new( pending_map, chain_cnt, ctx->seed ) );
1441 0 : ctx->pending_dlist = pending_dlist_join( pending_dlist_new( pending_dlist ) );
1442 0 : ctx->sign_pool = sign_pool_join( sign_pool_new( sign_pool, sign_max ) );
1443 0 : FD_TEST( ctx->strat && ctx->ping_deque && ctx->eager_prq && ctx->notar_prq && ctx->final_prq && ctx->pending_pool && ctx->pending_map && ctx->pending_dlist && ctx->sign_pool );
1444 0 : FD_TEST( fd_rng_join( fd_rng_new( ctx->rng, (uint)ctx->seed, ctx->seed>>32 ) ) );
1445 0 : fd_clock_tile_init( ctx->clock );
1446 0 : FD_TEST( fd_histf_join( fd_histf_new( ctx->metrics->response_latency, FD_MHIST_MIN( ROTOR, RESPONSE_LATENCY_NANOS ), FD_MHIST_MAX( ROTOR, RESPONSE_LATENCY_NANOS ) ) ) );
1447 0 : FD_TEST( fd_histf_join( fd_histf_new( ctx->metrics->retry_delay, FD_MHIST_MIN( ROTOR, RETRY_DELAY_NANOS ), FD_MHIST_MAX( ROTOR, RETRY_DELAY_NANOS ) ) ) );
1448 0 : FD_TEST( fd_histf_join( fd_histf_new( ctx->metrics->escape_hedge, FD_MHIST_MIN( ROTOR, ESCAPE_PEER_HEDGE_DELAY_NANOS ), FD_MHIST_MAX( ROTOR, ESCAPE_PEER_HEDGE_DELAY_NANOS ) ) ) );
1449 :
1450 0 : FD_TEST( tile->in_cnt<=sizeof(ctx->in_kind)/sizeof(ctx->in_kind[0]) );
1451 0 : for( ulong i=0UL; i<tile->in_cnt; i++ ) {
1452 0 : fd_topo_link_t const * link = &topo->links[ tile->in_link_id[ i ] ];
1453 :
1454 0 : if ( FD_LIKELY( !strcmp( link->name, "genesi_out" ) ) ) ctx->in_kind[ i ] = IN_KIND_GENESIS;
1455 0 : else if( FD_LIKELY( !strcmp( link->name, "shred_out" ) ) ) ctx->in_kind[ i ] = IN_KIND_SHRED;
1456 0 : else if( FD_LIKELY( !strcmp( link->name, "snapin_manif" ) ) ) ctx->in_kind[ i ] = IN_KIND_SNAP;
1457 0 : else if( FD_LIKELY( !strcmp( link->name, "votor_out" ) ) ) ctx->in_kind[ i ] = IN_KIND_VOTOR;
1458 0 : else if( FD_LIKELY( !strcmp( link->name, "net_repair" ) ) ) ctx->in_kind[ i ] = IN_KIND_NET;
1459 0 : else if( FD_LIKELY( !strcmp( link->name, "gossip_ciaddr" ) ) ) ctx->in_kind[ i ] = IN_KIND_GOSSIP;
1460 0 : else if( FD_LIKELY( !strcmp( link->name, "replay_slot" ) ) ) ctx->in_kind[ i ] = IN_KIND_REPLAY;
1461 0 : else if( FD_LIKELY( !strcmp( link->name, "replay_rotor" ) ) ) ctx->in_kind[ i ] = IN_KIND_REPLAY;
1462 0 : else if( FD_LIKELY( !strcmp( link->name, "sign_repair" ) ) ) ctx->in_kind[ i ] = IN_KIND_SIGN;
1463 0 : else if( FD_LIKELY( !strcmp( link->name, "replay_epoch" ) ) ) ctx->in_kind[ i ] = IN_KIND_EPOCH;
1464 0 : else FD_LOG_ERR(( "rotor tile has unexpected input link %lu %s", i, link->name ));
1465 0 : if( FD_UNLIKELY( ctx->in_kind[ i ]==IN_KIND_NET ) ) { /* net frags are bounded by fd_net_rx, not a dcache */
1466 0 : fd_net_rx_bounds_init( &ctx->in[ i ].net_rx, link->dcache );
1467 0 : continue;
1468 0 : }
1469 :
1470 0 : ctx->in[ i ].mem = topo->workspaces[ topo->objs[ link->dcache_obj_id ].wksp_id ].wksp;
1471 0 : ctx->in[ i ].chunk0 = fd_dcache_compact_chunk0( ctx->in[ i ].mem, link->dcache );
1472 0 : ctx->in[ i ].wmark = fd_dcache_compact_wmark ( ctx->in[ i ].mem, link->dcache, link->mtu );
1473 0 : ctx->in[ i ].mtu = link->mtu;
1474 0 : }
1475 :
1476 0 : ctx->replay_out_idx = fd_topo_find_tile_out_link( topo, tile, "repair_out", 0UL );
1477 0 : FD_TEST( ctx->replay_out_idx!=ULONG_MAX );
1478 0 : fd_topo_link_t const * replay_out = &topo->links[ tile->out_link_id[ ctx->replay_out_idx ] ];
1479 0 : ctx->replay_out_mem = topo->workspaces[ topo->objs[ replay_out->dcache_obj_id ].wksp_id ].wksp;
1480 0 : ctx->replay_out_chunk0 = fd_dcache_compact_chunk0( ctx->replay_out_mem, replay_out->dcache );
1481 0 : ctx->replay_out_wmark = fd_dcache_compact_wmark ( ctx->replay_out_mem, replay_out->dcache, replay_out->mtu );
1482 0 : ctx->replay_out_chunk = ctx->replay_out_chunk0;
1483 :
1484 : /* sign tile s takes repair_sign s and answers on sign_repair s */
1485 :
1486 0 : ctx->sign_cnt = tile->rotor.repair_sign_cnt;
1487 0 : FD_TEST( ctx->sign_cnt<=SIGN_MAX );
1488 0 : for( ulong s=0UL; s<ctx->sign_cnt; s++ ) {
1489 0 : ulong out_idx = fd_topo_find_tile_out_link( topo, tile, "repair_sign", s );
1490 0 : ulong in_idx = fd_topo_find_tile_in_link ( topo, tile, "sign_repair", s );
1491 0 : FD_TEST( out_idx!=ULONG_MAX && in_idx!=ULONG_MAX );
1492 0 : fd_topo_link_t const * sign_out = &topo->links[ tile->out_link_id[ out_idx ] ];
1493 0 : FD_TEST( sign_out->mtu>=FD_ROTOR_SIG_SER_MAX );
1494 0 : ctx->sign[ s ].idx = out_idx;
1495 0 : ctx->sign[ s ].mem = topo->workspaces[ topo->objs[ sign_out->dcache_obj_id ].wksp_id ].wksp;
1496 0 : ctx->sign[ s ].chunk0 = fd_dcache_compact_chunk0( ctx->sign[ s ].mem, sign_out->dcache );
1497 0 : ctx->sign[ s ].wmark = fd_dcache_compact_wmark ( ctx->sign[ s ].mem, sign_out->dcache, sign_out->mtu );
1498 0 : ctx->sign[ s ].chunk = ctx->sign[ s ].chunk0;
1499 0 : ctx->sign[ s ].free = tile->rotor.repair_sign_depth;
1500 0 : ctx->in[ in_idx ].sign_idx = s;
1501 0 : }
1502 :
1503 0 : ctx->net_out_idx = fd_topo_find_tile_out_link( topo, tile, "repair_net", 0UL );
1504 0 : FD_TEST( ctx->net_out_idx!=ULONG_MAX );
1505 0 : fd_topo_link_t const * net_out = &topo->links[ tile->out_link_id[ ctx->net_out_idx ] ];
1506 0 : ctx->net_out_mem = topo->workspaces[ topo->objs[ net_out->dcache_obj_id ].wksp_id ].wksp;
1507 0 : ctx->net_out_chunk0 = fd_dcache_compact_chunk0( ctx->net_out_mem, net_out->dcache );
1508 0 : ctx->net_out_wmark = fd_dcache_compact_wmark ( ctx->net_out_mem, net_out->dcache, net_out->mtu );
1509 0 : ctx->net_out_chunk = ctx->net_out_chunk0;
1510 0 : fd_ip4_udp_hdr_init( ctx->net_hdr, 0UL, 0U, tile->rotor.repair_client_listen_port );
1511 0 : ctx->allow_private_address = tile->rotor.allow_private_address;
1512 :
1513 0 : ctx->rserve_out_idx = fd_topo_find_tile_out_link( topo, tile, "rotor_rserve", 0UL );
1514 0 : if( FD_LIKELY( ctx->rserve_out_idx!=ULONG_MAX ) ) {
1515 0 : fd_topo_link_t const * rserve_out = &topo->links[ tile->out_link_id[ ctx->rserve_out_idx ] ];
1516 0 : ctx->rserve_out_mem = topo->workspaces[ topo->objs[ rserve_out->dcache_obj_id ].wksp_id ].wksp;
1517 0 : ctx->rserve_out_chunk0 = fd_dcache_compact_chunk0( ctx->rserve_out_mem, rserve_out->dcache );
1518 0 : ctx->rserve_out_wmark = fd_dcache_compact_wmark ( ctx->rserve_out_mem, rserve_out->dcache, rserve_out->mtu );
1519 0 : ctx->rserve_out_chunk = ctx->rserve_out_chunk0;
1520 0 : }
1521 0 : }
1522 :
1523 : static ulong
1524 : populate_allowed_seccomp( fd_topo_t const * topo FD_PARAM_UNUSED,
1525 : fd_topo_tile_t const * tile FD_PARAM_UNUSED,
1526 : ulong out_cnt,
1527 0 : struct sock_filter * out ) {
1528 0 : populate_sock_filter_policy_fd_rotor_tile( out_cnt, out, (uint)fd_log_private_logfile_fd() );
1529 0 : return sock_filter_policy_fd_rotor_tile_instr_cnt;
1530 0 : }
1531 :
1532 : static ulong
1533 : populate_allowed_fds( fd_topo_t const * topo FD_PARAM_UNUSED,
1534 : fd_topo_tile_t const * tile FD_PARAM_UNUSED,
1535 : ulong out_fds_cnt,
1536 0 : int * out_fds ) {
1537 0 : if( FD_UNLIKELY( out_fds_cnt<2UL ) ) FD_LOG_ERR(( "out_fds_cnt %lu", out_fds_cnt ));
1538 :
1539 0 : ulong out_cnt = 0UL;
1540 0 : out_fds[ out_cnt++ ] = 2; /* stderr */
1541 0 : if( FD_LIKELY( -1!=fd_log_private_logfile_fd() ) )
1542 0 : out_fds[ out_cnt++ ] = fd_log_private_logfile_fd(); /* logfile */
1543 0 : return out_cnt;
1544 0 : }
1545 :
1546 0 : #define STEM_BURST (1UL)
1547 0 : #define STEM_LAZY (128L*3000L)
1548 :
1549 0 : #define STEM_CALLBACK_CONTEXT_TYPE fd_rotor_tile_t
1550 0 : #define STEM_CALLBACK_CONTEXT_ALIGN alignof(fd_rotor_tile_t)
1551 :
1552 0 : #define STEM_CALLBACK_METRICS_WRITE metrics_write
1553 0 : #define STEM_CALLBACK_DURING_HOUSEKEEPING during_housekeeping
1554 0 : #define STEM_CALLBACK_NEXT_DEADLINE next_deadline
1555 0 : #define STEM_CALLBACK_AFTER_CREDIT after_credit
1556 0 : #define STEM_CALLBACK_BEFORE_FRAG before_frag
1557 0 : #define STEM_CALLBACK_DURING_FRAG during_frag
1558 0 : #define STEM_CALLBACK_AFTER_FRAG after_frag
1559 :
1560 : #include "../../disco/stem/fd_stem.c"
1561 :
1562 : static ulong
1563 0 : max_event_sz( fd_topo_tile_t const * tile FD_PARAM_UNUSED ) {
1564 0 : return sizeof(fd_event_block_received_t);
1565 0 : }
1566 :
1567 : fd_topo_run_tile_t fd_tile_rotor = {
1568 : .name = "rotor",
1569 : .max_event_sz = max_event_sz,
1570 : .populate_allowed_seccomp = populate_allowed_seccomp,
1571 : .populate_allowed_fds = populate_allowed_fds,
1572 : .scratch_align = scratch_align,
1573 : .scratch_footprint = scratch_footprint,
1574 : .unprivileged_init = unprivileged_init,
1575 : .privileged_init = privileged_init,
1576 : .run = stem_run,
1577 : };
|
