Line data Source code
1 : #define _GNU_SOURCE
2 : #include "../tiles.h"
3 :
4 : #if FD_HAS_X86
5 : #include <x86intrin.h>
6 : #endif
7 :
8 : #include <errno.h>
9 : #include <fcntl.h>
10 : #include <unistd.h>
11 :
12 : #include <linux/futex.h>
13 : #include "generated/fd_shred_tile_seccomp.h"
14 : #include "../../util/pod/fd_pod_format.h"
15 : #include "fd_shredder.h"
16 : #include "fd_shred_batch.h"
17 : #include "fd_shred_dest.h"
18 : #include "fd_fec_resolver.h"
19 : #include "fd_stake_ci.h"
20 : #include "fd_rnonce_ss.h"
21 : #include "fd_shred_tile.h"
22 : #include "../store/fd_store.h"
23 : #include "../keyguard/fd_keyload.h"
24 : #include "../keyguard/fd_keyguard.h"
25 : #include "../keyguard/fd_keyswitch.h"
26 : #include "../fd_disco.h"
27 : #include "../net/fd_net_tile.h"
28 : #include "fd_shred_dest_resolver.h"
29 : #include "../../flamenco/leaders/fd_leaders.h"
30 : #include "../../util/net/fd_net_headers.h"
31 : #include "../../flamenco/gossip/fd_gossip_message.h"
32 : #include "../../flamenco/runtime/sysvar/fd_sysvar_epoch_schedule.h"
33 : #include "../../flamenco/runtime/fd_slot_params.h"
34 : #include "../../discof/replay/fd_replay_tile.h"
35 : #include "../repair/fd_repair_serde.h"
36 : #if FD_HAS_AVX
37 : #include "../../util/simd/fd_nt_memcpy.h"
38 : #else
39 : #define fd_memcpy_tn fd_memcpy
40 : #define fd_memcpy_nt_nofence fd_memcpy
41 : #define _mm_sfence() do {} while( 0 )
42 : #endif
43 :
44 : /* The shred tile handles shreds from two data sources: shreds generated
45 : from microblocks from the leader pipeline, and shreds retransmitted
46 : from the network.
47 :
48 : They have rather different semantics, but at the end of the day, they
49 : both result in a bunch of shreds and FEC sets that need to be sent to
50 : the blockstore and on the network, which is why one tile handles
51 : both.
52 :
53 : We segment the memory for the two types of shreds into two halves of
54 : a dcache because they follow somewhat different flow control
55 : patterns. For flow control, the normal guarantee we want to provide
56 : is that the dcache entry is not overwritten unless the mcache entry
57 : has also been overwritten. The normal way to do this when using both
58 : cyclically and with a 1-to-1 mapping is to make the dcache at least
59 : `burst` entries bigger than the mcache.
60 :
61 : The FEC arena uses an exposure window e. Data is copied out before
62 : after_frag returns, so e is a small bound.
63 :
64 : Leader pipeline: Every entry creates up to FD_SHRED_BATCH_FEC_SETS_MAX
65 : FEC sets, and a set stays in the ring from its creation until the
66 : sign tile's response completes it, so up to FD_SHRED_SIGN_PEND_MAX
67 : are created but not yet exposed. The leader section therefore needs
68 : e+FD_SHRED_SIGN_PEND_MAX entries.
69 :
70 : From the network: The FEC resolver doesn't use a cyclic order, but it
71 : does promise that once it returns an FEC set, it will return at least
72 : complete_depth FEC sets before returning it again. This means we
73 : want at most complete_depth-1 FEC sets exposed, so
74 : complete_depth=e+1 FEC sets. The FEC resolver has the
75 : ability to keep individual shreds for partial_depth calls, but
76 : because in this version of the shred tile, we send each shred to all
77 : its destinations as soon as we get it, we don't need that
78 : functionality, so we set partial_depth=1.
79 :
80 : Adding these up and plugging in the current value of
81 : SIGN_PEND_MAX, we get 2*e+10+fec_resolver_depth FEC sets.
82 :
83 : A note on parallelization. From the network, shreds are distributed
84 : to tiles based on a validator-specific seeded hash of (slot, FEC set
85 : index) so all the shreds for a given FEC set (and any equivocating
86 : FEC set) are processed by the same tile. From the leader pipeline,
87 : the original implementation used to parallelize by batch of
88 : microblocks (so within a block, batches were distributed to different
89 : tiles). To support chained merkle shreds, the current implementation
90 : processes all the batches on tile 0 -- this should be a temporary
91 : state while Solana moves to a newer shred format that support better
92 : parallelization. */
93 :
94 0 : #define IN_KIND_EPOCH ( 1UL)
95 0 : #define IN_KIND_POH ( 3UL)
96 0 : #define IN_KIND_NET ( 4UL)
97 0 : #define IN_KIND_SIGN ( 5UL)
98 : #define IN_KIND_REPAIR ( 6UL)
99 0 : #define IN_KIND_IPECHO ( 7UL)
100 0 : #define IN_KIND_GOSSIP ( 8UL)
101 0 : #define IN_KIND_ROOTEDR (11UL) /* Alpenglow */
102 :
103 0 : #define NET_OUT_IDX 1
104 0 : #define SIGN_OUT_IDX 2
105 : #define IN_LINK_MAX 32UL
106 :
107 : FD_STATIC_ASSERT( sizeof(fd_entry_batch_meta_t)==56UL, poh_shred_mtu );
108 :
109 0 : #define FD_SHRED_ADD_SHRED_EXTRA_RETVAL_CNT 3
110 : FD_STATIC_ASSERT( FD_SHRED_ADD_SHRED_EXTRA_RETVAL_CNT+FD_FEC_RESOLVER_ADD_SHRED_RETVAL_CNT==FD_METRICS_ENUM_SHRED_PROCESS_RESULT_CNT, metrics );
111 : FD_STATIC_ASSERT( FD_FEC_RESOLVER_SHRED_OKAY+FD_FEC_RESOLVER_ADD_SHRED_RETVAL_OFF+FD_SHRED_ADD_SHRED_EXTRA_RETVAL_CNT==FD_METRICS_ENUM_SHRED_PROCESS_RESULT_V_ACCEPTED_IDX, metrics );
112 :
113 : /* Number of entries in the block_ids table. Each entry is 32 byte.
114 : This table is used to keep track of block ids that we create
115 : when we're leader, so that we can access them whenever we need
116 : a *parent* block id for a new block. Larger table allows to
117 : retrieve older parent block ids. Currently it's set for worst
118 : case parent offset of USHORT_MAX (max allowed in a shred),
119 : making the total table 2MiB.
120 : See also comment on chained_merkle_root. */
121 0 : #define BLOCK_IDS_TABLE_CNT USHORT_MAX
122 :
123 : /* See note on parallelization above. Currently we process all batches in tile 0. */
124 : #if 1
125 : #define SHOULD_PROCESS_THESE_SHREDS ( ctx->round_robin_id==0 )
126 : #else
127 : #define SHOULD_PROCESS_THESE_SHREDS ( ctx->batch_cnt%ctx->round_robin_cnt==ctx->round_robin_id )
128 : #endif
129 :
130 : typedef union {
131 : struct {
132 : fd_wksp_t * mem;
133 : ulong chunk0;
134 : ulong wmark;
135 : };
136 : fd_net_rx_bounds_t net_rx;
137 : } fd_shred_in_ctx_t;
138 :
139 : /* A leader FEC set whose sign request went out on shred_sign and has
140 : not been answered. */
141 :
142 : typedef struct {
143 : ulong fec_set_idx;
144 : ulong tsorig;
145 : fd_bmtree_node_t merkle_root;
146 : } fd_shred_sign_pend_t;
147 :
148 : typedef struct {
149 : fd_shredder_t * shredder;
150 : fd_fec_resolver_t * resolver;
151 : ulong shred_limit;
152 : fd_pubkey_t identity_key[1]; /* Just the public key */
153 :
154 : ulong round_robin_id;
155 : ulong round_robin_cnt;
156 : /* Number of batches shredded from PoH during the current slot.
157 : This should be the same for all the shred tiles. */
158 : ulong batch_cnt;
159 : /* Slot of the most recent microblock we've seen from PoH,
160 : or 0 if we haven't seen one yet */
161 : ulong slot;
162 :
163 : fd_rnonce_ss_t repair_nonce_ss[1];
164 :
165 : fd_keyswitch_t * keyswitch;
166 :
167 : fd_fec_set_t * fec_sets;
168 :
169 : fd_stake_ci_t * stake_ci;
170 : /* These are used in between during_frag and after_frag */
171 : ulong new_root;
172 : ulong slot_max;
173 : ulong future_slot;
174 :
175 : ulong poh_in_expect_seq;
176 :
177 : ushort net_id;
178 :
179 : int skip_frag;
180 : int parse_failed; /* during_frag's parse failed and this tile owns the frag, counted in after_frag once not overrun */
181 : int fwd_meta; /* during_frag copied a repair ping or metadata response this tile owns, forwarded in after_frag */
182 :
183 : ulong adtl_dests_leader_cnt;
184 : fd_shred_dest_weighted_t adtl_dests_leader [ FD_TOPO_ADTL_DESTS_MAX ];
185 : ulong adtl_dests_retransmit_cnt;
186 : fd_shred_dest_weighted_t adtl_dests_retransmit[ FD_TOPO_ADTL_DESTS_MAX ];
187 :
188 : fd_ip4_udp_hdrs_t data_shred_net_hdr [1];
189 : fd_ip4_udp_hdrs_t parity_shred_net_hdr[1];
190 :
191 : ulong shredder_fec_set_idx; /* In [0, shredder_max_fec_set_idx) */
192 : ulong shredder_max_fec_set_idx; /* exclusive */
193 :
194 : uchar shredder_merkle_root[32];
195 :
196 : ulong send_fec_set_idx[ FD_SHRED_BATCH_FEC_SETS_MAX ];
197 : ulong send_fec_set_cnt;
198 : ulong tsorig; /* timestamp of the last packet in compressed form */
199 :
200 : /* Leader FEC sets awaiting their signature, oldest first. The sign
201 : tile answers in order on sign_shred, so each response frag belongs
202 : to the head, which it finishes. */
203 : fd_shred_sign_pend_t sign_pend[ FD_SHRED_SIGN_PEND_MAX ];
204 :
205 : ulong sign_pend_head;
206 : ulong sign_pend_cnt;
207 : ulong sign_seq;
208 : uchar signature[ 64 ] __attribute__((aligned(32UL)));
209 :
210 : fd_wksp_t * sign_out_mem;
211 : ulong sign_out_chunk0;
212 : ulong sign_out_wmark;
213 : ulong sign_out_chunk;
214 :
215 : /* Includes Ethernet, IP, UDP headers */
216 : ulong shred_buffer_sz;
217 : uchar shred_buffer[ FD_NET_MTU ];
218 :
219 : /* These seeds get generated in privileged_init but used in
220 : unprivileged_init, so we store them here in between. */
221 : ulong resolver_seed;
222 : ulong shred_dest_seed;
223 :
224 : fd_shred_in_ctx_t in[ IN_LINK_MAX ];
225 : int in_kind[ IN_LINK_MAX ];
226 :
227 : fd_wksp_t * net_out_mem;
228 : ulong net_out_chunk0;
229 : ulong net_out_wmark;
230 : ulong net_out_chunk;
231 :
232 : /* This is the output link for shreds that is currently consumed by
233 : the repair and replay tile. */
234 : ulong shred_out_idx;
235 : fd_wksp_t * shred_out_mem;
236 : ulong shred_out_chunk0;
237 : ulong shred_out_wmark;
238 : ulong shred_out_chunk;
239 :
240 : /* Repair pings and metadata responses arrive on the repair client
241 : port with repair shreds, and are forwarded to the repair tile. */
242 : ulong repair_out_idx;
243 : fd_wksp_t * repair_out_mem;
244 : ulong repair_out_chunk0;
245 : ulong repair_out_wmark;
246 : ulong repair_out_chunk;
247 :
248 : /* Highest slot relayed on turbine, published for rpc
249 : (getMaxRetransmitSlot). NULL if rpc is disabled. */
250 : ulong * rtx_fseq;
251 : ulong rtx_slot;
252 :
253 : fd_store_t * store;
254 : fd_store_map_t map_join[1];
255 : int disk_fd;
256 :
257 : fd_gossip_update_message_t gossip_upd_buf[1];
258 :
259 : struct {
260 : fd_histf_t batch_sz[ 1 ];
261 : fd_histf_t batch_microblock_cnt[ 1 ];
262 : fd_histf_t shredding_timing[ 1 ];
263 : fd_histf_t add_shred_timing[ 1 ];
264 : fd_histf_t fec_fallback_write_timing[ 1 ];
265 : ulong shred_processing_result[ FD_FEC_RESOLVER_ADD_SHRED_RETVAL_CNT+FD_SHRED_ADD_SHRED_EXTRA_RETVAL_CNT ];
266 : ulong invalid_block_id_cnt;
267 : ulong shred_rejected_unchained_cnt;
268 : ulong repair_rcv_cnt;
269 : ulong repair_rcv_bytes;
270 : ulong turbine_rcv_cnt;
271 : ulong turbine_rcv_bytes;
272 : ulong bad_nonce;
273 : ulong fec_fallback_write_cnt;
274 : ulong fec_fallback_write_bytes;
275 : } metrics[ 1 ];
276 :
277 : struct {
278 : ulong pos; /* in payload, range [0, FD_SHRED_BATCH_RAW_BUF_SZ-8UL) */
279 : ulong slot; /* set to 0 when pos==0 */
280 : union {
281 : struct {
282 : ulong microblock_cnt;
283 : uchar payload[ FD_SHRED_BATCH_RAW_BUF_SZ - 8UL ];
284 : };
285 : uchar raw[ FD_SHRED_BATCH_RAW_BUF_SZ ];
286 : };
287 : } pending_batch;
288 :
289 : fd_epoch_schedule_t epoch_schedule[1];
290 : fd_shred_features_activation_t features_activation[1];
291 :
292 : /* max_shred_idx is the exclusive upper bound for shred
293 : indices. We need to reject any shred with an
294 : index >= current_max_shred_idx, but we also want to reject anything
295 : that is part of an FEC set whose highest shred index would reach
296 : the bound.
297 :
298 : Because this bound can change with feature gates, for example the
299 : reduce_slot_time feature gates, we store the bound for the
300 : previous, current, and next regimes, along with the slots at which
301 : the current and next bounds take effect. This lets us apply, to
302 : each shred, the limit effective at that shreds slot.
303 :
304 : For a given shred slot, the max_shred_idx for that shred is:
305 : next_max_shred_idx_start_slot <= slot -> next_max_shred_idx
306 : current_max_shred_idx_start_slot <= slot < next_max_shred_idx_start_slot -> current_max_shred_idx
307 : slot < current_max_shred_idx_start_slot -> prev_max_shred_idx
308 : */
309 : ulong prev_max_shred_idx;
310 : ulong current_max_shred_idx;
311 : ulong next_max_shred_idx;
312 : ulong current_max_shred_idx_start_slot;
313 : ulong next_max_shred_idx_start_slot;
314 : ulong bench_max_shred_idx; /* [development.bench.max_shreds_per_block], floors the chain's limits */
315 : /* too large to be left in the stack */
316 : fd_shred_dest_idx_t scratchpad_dests[ FD_SHRED_DEST_MAX_FANOUT*(FD_REEDSOL_DATA_SHREDS_MAX+FD_REEDSOL_PARITY_SHREDS_MAX) ];
317 :
318 : uchar * chained_merkle_root;
319 : fd_bmtree_node_t out_merkle_roots[ FD_SHRED_BATCH_FEC_SETS_MAX ];
320 : uchar block_ids[ BLOCK_IDS_TABLE_CNT ][ FD_SHRED_MERKLE_ROOT_SZ ];
321 :
322 : fd_sha512_t sha512[1];
323 : uchar bmtree_mem[ FD_BMTREE_COMMIT_FOOTPRINT( FD_SHRED_MERKLE_LAYER_CNT ) ] __attribute__((aligned(FD_BMTREE_COMMIT_ALIGN)));
324 : } fd_shred_ctx_t;
325 :
326 : /* shred features are generally considered active at the epoch *following*
327 : the epoch in which the feature gate is activated.
328 :
329 : As an optimization, when the activation slot is received, it is converted
330 : into the first slot of the subsequent epoch. This allows for a more
331 : efficient check (shred_slot >= feature_slot) and avoids the overhead of
332 : repeatedly converting slots into epochs for comparison.
333 :
334 : In Agave, this is done with check_feature_activation():
335 : https://github.com/anza-xyz/agave/blob/v3.1.4/turbine/src/cluster_nodes.rs#L771
336 : https://github.com/anza-xyz/agave/blob/v3.1.4/core/src/shred_fetch_stage.rs#L456 */
337 : static inline ulong
338 0 : fd_shred_get_feature_activation_slot0( ulong feature_slot, fd_shred_ctx_t * ctx ) {
339 : /* if the feature does not have an activation slot yet, return ULONG_MAX */
340 0 : if( FD_UNLIKELY( feature_slot==ULONG_MAX ) ) {
341 0 : return ULONG_MAX;
342 0 : }
343 : /* if we don't have an epoch schedule yet, return ULONG_MAX */
344 0 : if( FD_UNLIKELY( ctx->epoch_schedule->slots_per_epoch==0 ) ) {
345 0 : return ULONG_MAX;
346 0 : }
347 : /* compute the activation epoch, add one, return the first slot. */
348 0 : ulong feature_epoch = 1 + fd_slot_to_epoch( ctx->epoch_schedule, feature_slot, NULL );
349 0 : return fd_epoch_slot0( ctx->epoch_schedule, feature_epoch );
350 0 : }
351 :
352 : FD_FN_CONST static inline ulong
353 0 : scratch_align( void ) {
354 0 : return 128UL;
355 0 : }
356 :
357 : FD_FN_PURE static inline ulong
358 0 : scratch_footprint( fd_topo_tile_t const * tile ) {
359 :
360 0 : ulong fec_resolver_footprint = fd_fec_resolver_footprint( tile->shred.fec_resolver_depth, 1UL, tile->shred.fec_exposure+1UL,
361 0 : 128UL * tile->shred.fec_resolver_depth );
362 0 : ulong l = FD_LAYOUT_INIT;
363 0 : l = FD_LAYOUT_APPEND( l, alignof(fd_shred_ctx_t), sizeof(fd_shred_ctx_t) );
364 0 : l = FD_LAYOUT_APPEND( l, fd_stake_ci_align(), fd_stake_ci_footprint() );
365 0 : l = FD_LAYOUT_APPEND( l, fd_fec_resolver_align(), fec_resolver_footprint );
366 0 : l = FD_LAYOUT_APPEND( l, fd_shredder_align(), fd_shredder_footprint() );
367 0 : return FD_LAYOUT_FINI( l, scratch_align() );
368 0 : }
369 :
370 : static inline void
371 0 : during_housekeeping( fd_shred_ctx_t * ctx ) {
372 0 : if( FD_UNLIKELY( fd_keyswitch_state_query( ctx->keyswitch )==FD_KEYSWITCH_STATE_SWITCH_PENDING ) ) {
373 0 : ulong seq_must_complete = ctx->keyswitch->param;
374 :
375 0 : if( FD_UNLIKELY( fd_seq_lt( ctx->poh_in_expect_seq, seq_must_complete ) || ctx->sign_pend_cnt ) ) {
376 : /* See fd_keyswitch.h, we need to flush any in-flight shreds from
377 : the leader pipeline before switching key. Sets still awaiting
378 : their signature were requested under the old key. */
379 0 : FD_LOG_WARNING(( "Flushing in-flight unpublished shreds, must reach seq %lu, currently at %lu with %lu awaiting signature ...", seq_must_complete, ctx->poh_in_expect_seq, ctx->sign_pend_cnt ));
380 0 : return;
381 0 : }
382 :
383 0 : memcpy( ctx->identity_key->uc, ctx->keyswitch->bytes, 32UL );
384 0 : fd_stake_ci_set_identity( ctx->stake_ci, ctx->identity_key );
385 0 : fd_keyswitch_state( ctx->keyswitch, FD_KEYSWITCH_STATE_COMPLETED );
386 0 : }
387 0 : }
388 :
389 : static inline void
390 0 : metrics_write( fd_shred_ctx_t * ctx ) {
391 0 : FD_MHIST_COPY( SHRED, REPAIR_LEAD_DELAY_NANOS, fd_fec_resolver_completion_lag_hist( ctx->resolver ) );
392 0 : FD_MHIST_COPY( SHRED, BATCH_SIZE_BYTES, ctx->metrics->batch_sz );
393 0 : FD_MHIST_COPY( SHRED, MICROBLOCK_PER_BATCH, ctx->metrics->batch_microblock_cnt );
394 0 : FD_MHIST_COPY( SHRED, BATCH_SHRED_DURATION_NANOS, ctx->metrics->shredding_timing );
395 0 : FD_MHIST_COPY( SHRED, ADD_SHRED_DURATION_NANOS, ctx->metrics->add_shred_timing );
396 0 : FD_MHIST_COPY( SHRED, STORE_FEC_SYNC_SPILL_DURATION_NANOS, ctx->metrics->fec_fallback_write_timing );
397 0 : FD_MCNT_SET ( SHRED, REPAIR_RX, ctx->metrics->repair_rcv_cnt );
398 0 : FD_MCNT_SET ( SHRED, REPAIR_RX_BYTES, ctx->metrics->repair_rcv_bytes );
399 0 : FD_MCNT_SET ( SHRED, TURBINE_RX, ctx->metrics->turbine_rcv_cnt );
400 0 : FD_MCNT_SET ( SHRED, TURBINE_RX_BYTES, ctx->metrics->turbine_rcv_bytes );
401 0 : FD_MCNT_SET ( SHRED, REPAIR_NONCE_FAILED, ctx->metrics->bad_nonce );
402 0 : FD_MCNT_SET ( SHRED, STORE_FEC_SYNC_SPILLED, ctx->metrics->fec_fallback_write_cnt );
403 0 : FD_MCNT_SET ( SHRED, STORE_FEC_SYNC_SPILLED_BYTES, ctx->metrics->fec_fallback_write_bytes );
404 :
405 0 : FD_MCNT_SET ( SHRED, PARENT_BLOCK_ID_ZEROED, ctx->metrics->invalid_block_id_cnt );
406 0 : FD_MCNT_SET ( SHRED, UNCHAINED_REJECTED, ctx->metrics->shred_rejected_unchained_cnt );
407 :
408 0 : FD_MCNT_ENUM_COPY( SHRED, PROCESSED, ctx->metrics->shred_processing_result );
409 0 : }
410 :
411 : static inline int
412 : before_frag( fd_shred_ctx_t * ctx,
413 : ulong in_idx,
414 : ulong seq,
415 0 : ulong sig ) {
416 0 : if( FD_UNLIKELY( ctx->in_kind[ in_idx ]==IN_KIND_IPECHO ) ) {
417 0 : FD_TEST( sig!=0UL && sig<=USHORT_MAX );
418 0 : fd_shredder_set_shred_version ( ctx->shredder, (ushort)sig );
419 0 : fd_fec_resolver_set_shred_version( ctx->resolver, (ushort)sig );
420 0 : return 1;
421 0 : }
422 :
423 0 : if( FD_UNLIKELY( !ctx->shredder->shred_version ) ) return -1;
424 :
425 0 : if( FD_LIKELY( ctx->in_kind[ in_idx ]==IN_KIND_POH ) ) {
426 0 : if( FD_UNLIKELY( (fd_disco_poh_sig_pkt_type( sig )==POH_PKT_TYPE_MICROBLOCK) &
427 0 : (ctx->sign_pend_cnt+FD_SHRED_BATCH_FEC_SETS_MAX>FD_SHRED_SIGN_PEND_MAX) ) ) return -1;
428 0 : ctx->poh_in_expect_seq = seq+1UL;
429 0 : return fd_disco_poh_sig_pkt_type( sig )!=POH_PKT_TYPE_MICROBLOCK;
430 0 : }
431 0 : if( FD_LIKELY( ctx->in_kind[ in_idx ]==IN_KIND_NET ) ) {
432 0 : return (int)(fd_disco_netmux_sig_proto( sig )!=DST_PROTO_SHRED) & (int)(fd_disco_netmux_sig_proto( sig )!=DST_PROTO_REPAIR);
433 0 : }
434 0 : if( FD_UNLIKELY( ctx->in_kind[ in_idx ]==IN_KIND_GOSSIP ) ){
435 0 : return sig!=FD_GOSSIP_UPDATE_TAG_CONTACT_INFO &&
436 0 : sig!=FD_GOSSIP_UPDATE_TAG_CONTACT_INFO_REMOVE;
437 0 : }
438 0 : if( FD_UNLIKELY( ctx->in_kind[ in_idx ]==IN_KIND_ROOTEDR ) ) {
439 0 : return sig!=REPLAY_SIG_ROOT_ADVANCED; /* only care about root_advanced messages */
440 0 : }
441 0 : return 0;
442 0 : }
443 :
444 : static void
445 : during_frag( fd_shred_ctx_t * ctx,
446 : ulong in_idx,
447 : ulong seq,
448 : ulong sig,
449 : ulong chunk,
450 : ulong sz,
451 0 : ulong ctl ) {
452 :
453 0 : ctx->skip_frag = 0;
454 0 : ctx->parse_failed = 0;
455 0 : ctx->fwd_meta = 0;
456 :
457 0 : ctx->tsorig = fd_frag_meta_ts_comp( fd_tickcount() );
458 :
459 0 : if( FD_UNLIKELY( ctx->in_kind[ in_idx ]==IN_KIND_REPAIR ) ) {
460 0 : if( FD_UNLIKELY( chunk<ctx->in[ in_idx ].chunk0 || chunk>ctx->in[ in_idx ].wmark || sz>FD_NET_MTU ) )
461 0 : FD_LOG_ERR(( "chunk %lu %lu corrupt, not in range [%lu,%lu]", chunk, sz,
462 0 : ctx->in[ in_idx ].chunk0, ctx->in[ in_idx ].wmark ));
463 :
464 0 : uchar const * dcache_entry = fd_chunk_to_laddr_const( ctx->in[ in_idx ].mem, chunk );
465 0 : fd_memcpy( ctx->shred_buffer, dcache_entry, sz );
466 0 : return;
467 0 : }
468 :
469 0 : if( FD_UNLIKELY( ctx->in_kind[ in_idx ]==IN_KIND_GOSSIP ) ) {
470 0 : if( FD_UNLIKELY( chunk<ctx->in[ in_idx ].chunk0 || chunk>ctx->in[ in_idx ].wmark || sz>sizeof(fd_gossip_update_message_t) ) )
471 0 : FD_LOG_ERR(( "chunk %lu %lu corrupt, not in range [%lu,%lu]", chunk, sz,
472 0 : ctx->in[ in_idx ].chunk0, ctx->in[ in_idx ].wmark ));
473 0 : uchar const * gossip_upd_msg = fd_chunk_to_laddr_const( ctx->in[ in_idx ].mem, chunk );
474 0 : fd_memcpy( ctx->gossip_upd_buf, gossip_upd_msg, sz );
475 0 : return;
476 0 : }
477 :
478 : /* Alpenglow */
479 0 : if( FD_UNLIKELY( ctx->in_kind[ in_idx ]==IN_KIND_ROOTEDR ) ) {
480 0 : if( FD_UNLIKELY( chunk<ctx->in[ in_idx ].chunk0 || chunk>ctx->in[ in_idx ].wmark || sz<sizeof(fd_replay_root_advanced_t) ) )
481 0 : FD_LOG_ERR(( "chunk %lu %lu corrupt, not in range [%lu,%lu]", chunk, sz,
482 0 : ctx->in[ in_idx ].chunk0, ctx->in[ in_idx ].wmark ));
483 0 : fd_replay_root_advanced_t const * rooted_msg = fd_chunk_to_laddr_const( ctx->in[ in_idx ].mem, chunk );
484 0 : ctx->new_root = rooted_msg->slot;
485 0 : return;
486 0 : }
487 :
488 0 : if( FD_UNLIKELY( ctx->in_kind[ in_idx ]==IN_KIND_EPOCH ) ) {
489 0 : if( FD_UNLIKELY( chunk<ctx->in[ in_idx ].chunk0 || chunk>ctx->in[ in_idx ].wmark ) )
490 0 : FD_LOG_ERR(( "chunk %lu %lu corrupt, not in range [%lu,%lu]", chunk, sz,
491 0 : ctx->in[ in_idx ].chunk0, ctx->in[ in_idx ].wmark ));
492 :
493 0 : uchar const * dcache_entry = fd_chunk_to_laddr_const( ctx->in[ in_idx ].mem, chunk );
494 0 : fd_epoch_info_msg_t const * epoch_msg = fd_type_pun_const( dcache_entry );
495 :
496 0 : FD_TEST( epoch_msg->staked_vote_cnt<=MAX_STAKE_WEIGHTS );
497 0 : FD_TEST( epoch_msg->staked_id_cnt<=MAX_STAKE_WEIGHTS );
498 :
499 0 : fd_stake_ci_epoch_msg_init( ctx->stake_ci, epoch_msg );
500 :
501 0 : *ctx->epoch_schedule = epoch_msg->epoch_schedule;
502 :
503 0 : fd_slot_params_t slot_params = fd_slot_params_lookup( &FD_SLOT_PARAMS_400MS,
504 0 : &epoch_msg->features,
505 0 : &epoch_msg->epoch_schedule,
506 0 : epoch_msg->start_slot );
507 :
508 0 : ctx->current_max_shred_idx = fd_ulong_max( slot_params.max_shred_idx, ctx->bench_max_shred_idx );
509 0 : ctx->current_max_shred_idx_start_slot = fd_slot_params_effective_slot( &slot_params,
510 0 : &epoch_msg->features,
511 0 : &epoch_msg->epoch_schedule );
512 0 : ctx->next_max_shred_idx_start_slot = fd_slot_params_next_effective_slot( &slot_params,
513 0 : &epoch_msg->features,
514 0 : &epoch_msg->epoch_schedule );
515 0 : ctx->prev_max_shred_idx = fd_ulong_max( fd_slot_params_lookup( &FD_SLOT_PARAMS_400MS,
516 0 : &epoch_msg->features,
517 0 : &epoch_msg->epoch_schedule,
518 0 : fd_ulong_sat_sub( ctx->current_max_shred_idx_start_slot, 1UL ) ).max_shred_idx, ctx->bench_max_shred_idx );
519 0 : ctx->next_max_shred_idx = fd_ulong_max( fd_slot_params_lookup( &FD_SLOT_PARAMS_400MS,
520 0 : &epoch_msg->features,
521 0 : &epoch_msg->epoch_schedule,
522 0 : ctx->next_max_shred_idx_start_slot ).max_shred_idx, ctx->bench_max_shred_idx );
523 0 : ctx->features_activation->enforce_fixed_fec_set = fd_shred_get_feature_activation_slot0(
524 0 : epoch_msg->features.enforce_fixed_fec_set, ctx );
525 :
526 0 : return;
527 0 : }
528 :
529 0 : if( FD_UNLIKELY( ctx->in_kind[ in_idx ]==IN_KIND_SIGN ) ) {
530 0 : if( FD_UNLIKELY( chunk<ctx->in[ in_idx ].chunk0 || chunk>ctx->in[ in_idx ].wmark || sz!=64UL ) ) FD_LOG_ERR(( "chunk %lu %lu corrupt, not in range [%lu,%lu]", chunk, sz, ctx->in[ in_idx ].chunk0, ctx->in[ in_idx ].wmark ));
531 :
532 0 : if( FD_UNLIKELY( seq!=ctx->sign_seq ) ) FD_LOG_CRIT(( "sign response seq %lu, expected %lu", seq, ctx->sign_seq ));
533 0 : fd_memcpy( ctx->signature, fd_chunk_to_laddr_const( ctx->in[ in_idx ].mem, chunk ), 64UL );
534 0 : return;
535 0 : }
536 :
537 0 : if( FD_UNLIKELY( ctx->in_kind[ in_idx ]==IN_KIND_POH ) ) {
538 0 : ctx->send_fec_set_cnt = 0UL;
539 :
540 : /* This is a frag from the PoH tile. We'll copy it to our pending
541 : microblock batch and shred it if necessary (last in block or
542 : above watermark). We just go ahead and shred it here, even
543 : though we may get overrun. If we do end up getting overrun, we
544 : just won't send these shreds out and we'll reuse the FEC set for
545 : the next one. From a higher level though, if we do get overrun,
546 : a bunch of shreds will never be transmitted, and we'll end up
547 : producing a block that never lands on chain. */
548 :
549 0 : uchar const * dcache_entry = fd_chunk_to_laddr_const( ctx->in[ in_idx ].mem, chunk );
550 0 : if( FD_UNLIKELY( chunk<ctx->in[ in_idx ].chunk0 || chunk>ctx->in[ in_idx ].wmark || sz>FD_POH_SHRED_MTU ||
551 0 : sz<(sizeof(fd_entry_batch_meta_t)+sizeof(fd_entry_batch_header_t)) ) )
552 0 : FD_LOG_ERR(( "chunk %lu %lu corrupt, not in range [%lu,%lu]", chunk, sz,
553 0 : ctx->in[ in_idx ].chunk0, ctx->in[ in_idx ].wmark ));
554 :
555 0 : fd_entry_batch_meta_t const * entry_meta = (fd_entry_batch_meta_t const *)dcache_entry;
556 0 : uchar const * entry = dcache_entry + sizeof(fd_entry_batch_meta_t);
557 0 : ulong entry_sz = sz - sizeof(fd_entry_batch_meta_t);
558 :
559 : /* It should never be possible for this to fail, but we check it
560 : anyway. */
561 0 : FD_TEST( entry_sz + ctx->pending_batch.pos <= sizeof(ctx->pending_batch.payload) );
562 :
563 0 : ulong target_slot = fd_disco_poh_sig_slot( sig );
564 0 : if( FD_UNLIKELY( (ctx->pending_batch.microblock_cnt>0) & (ctx->pending_batch.slot!=target_slot) ) ) {
565 : /* TODO: The Agave client sends a dummy entry batch with only 1
566 : byte and the block-complete bit set. This helps other
567 : validators know that the block is dead and they should not try
568 : to continue building a fork on it. We probably want a similar
569 : approach eventually. */
570 0 : FD_LOG_WARNING(( "Abandoning %lu microblocks for slot %lu and switching to slot %lu",
571 0 : ctx->pending_batch.microblock_cnt, ctx->pending_batch.slot, target_slot ));
572 0 : ctx->pending_batch.slot = 0UL;
573 0 : ctx->pending_batch.pos = 0UL;
574 0 : ctx->pending_batch.microblock_cnt = 0UL;
575 0 : ctx->batch_cnt = 0UL;
576 :
577 0 : FD_MCNT_INC( SHRED, BATCH_ABANDONED, 1UL );
578 0 : }
579 :
580 0 : ctx->pending_batch.slot = target_slot;
581 : /* We want to send out some shreds immediately when we start a new
582 : slot to help with leader targeting. */
583 0 : int new_slot = 0;
584 0 : if( FD_UNLIKELY( target_slot!=ctx->slot )) {
585 : /* Reset batch count if we are in a new slot */
586 0 : ctx->batch_cnt = 0UL;
587 0 : ctx->slot = target_slot;
588 0 : new_slot = 1;
589 :
590 : /* At the beginning of a new slot, prepare chained_merkle_root.
591 : chained_merkle_root is initialized at the block_id of the parent
592 : block, there's two cases:
593 :
594 : 1. block_id is passed in by the poh tile:
595 : - it's always passed when parent block had a different leader
596 : - it may be passed when we were leader for parent block (there
597 : are race conditions when it's not passed)
598 :
599 : 2. block_id is taken from block_ids table if we were the leader
600 : for the parent block (when we were NOT the leader, because of
601 : equivocation, we can't store block_id in the table)
602 :
603 : chained_merkle_root is stored in block_ids table at target_slot
604 : and it's progressively updated as more microblocks are received.
605 : As a result, when we move to a new slot, the block_ids table at
606 : the old slot will contain the block_id.
607 :
608 : The block_ids table is designed to protect against the race condition
609 : case in 1., therefore the table may not be set in some cases, e.g. if
610 : a validator (re)starts, but in those cases we don't expect the race
611 : condition to apply. */
612 0 : ctx->chained_merkle_root = ctx->block_ids[ target_slot % BLOCK_IDS_TABLE_CNT ];
613 0 : if( FD_UNLIKELY( SHOULD_PROCESS_THESE_SHREDS ) ) {
614 0 : if( FD_LIKELY( entry_meta->parent_block_id_valid ) ) {
615 : /* 1. Initialize chained_merkle_root sent from poh tile */
616 0 : memcpy( ctx->chained_merkle_root, entry_meta->parent_block_id, FD_SHRED_MERKLE_ROOT_SZ );
617 0 : } else {
618 0 : ulong parent_slot = target_slot - entry_meta->parent_offset;
619 0 : fd_epoch_leaders_t const * lsched = fd_stake_ci_get_lsched_for_slot( ctx->stake_ci, parent_slot );
620 0 : fd_pubkey_t const * slot_leader = fd_epoch_leaders_get( lsched, parent_slot );
621 :
622 0 : if( lsched && slot_leader && fd_memeq( slot_leader, ctx->identity_key, sizeof(fd_pubkey_t) ) ) {
623 : /* 2. Initialize chained_merkle_root from block_ids table, if we were the leader */
624 0 : memcpy( ctx->chained_merkle_root, ctx->block_ids[ parent_slot % BLOCK_IDS_TABLE_CNT ], FD_SHRED_MERKLE_ROOT_SZ );
625 0 : } else {
626 : /* This should never happen, log a metric and set chained_merkle_root to 0 */
627 0 : ctx->metrics->invalid_block_id_cnt++;
628 0 : memset( ctx->chained_merkle_root, 0, FD_SHRED_MERKLE_ROOT_SZ );
629 0 : }
630 0 : }
631 0 : }
632 0 : }
633 :
634 0 : if( FD_LIKELY( !SHOULD_PROCESS_THESE_SHREDS ) ) {
635 : /* If we are not processing this batch, filter in after_frag. */
636 0 : ctx->skip_frag = 1;
637 0 : }
638 :
639 0 : ulong pending_batch_wmark = FD_SHRED_BATCH_WMARK_CHAINED;
640 0 : uchar * chained_merkle_root = ctx->chained_merkle_root;
641 :
642 : /* If this microblock completes the block, the batch is then
643 : finalized here. Otherwise, we check whether the new entry
644 : would exceed the pending_batch_wmark. If true, then the
645 : batch is closed now, shredded, and a new batch is started
646 : with the incoming microblock. If false, no shredding takes
647 : place, and the microblock is added to the current batch.
648 : Pack limits entry bytes so this batching cannot exceed
649 : max_shred_idx. */
650 0 : int is_marker = entry_meta->block_complete==-1;
651 0 : int writing_marker = 0;
652 0 : int forced_end_batch = entry_meta->block_complete | new_slot;
653 0 : int batch_would_exceed_wmark = ( ctx->pending_batch.pos + entry_sz ) > pending_batch_wmark;
654 0 : int include_in_current_batch = !is_marker && ( forced_end_batch | ( !batch_would_exceed_wmark ) );
655 0 : int process_current_batch = is_marker ? !!ctx->pending_batch.pos : ( forced_end_batch | batch_would_exceed_wmark );
656 0 : int init_new_batch = !include_in_current_batch;
657 :
658 0 : if( FD_LIKELY( include_in_current_batch ) ) {
659 0 : if( FD_UNLIKELY( SHOULD_PROCESS_THESE_SHREDS ) ) {
660 : /* Ugh, yet another memcpy */
661 0 : fd_memcpy_tn( ctx->pending_batch.payload + ctx->pending_batch.pos, entry, entry_sz );
662 0 : }
663 0 : ctx->pending_batch.pos += entry_sz;
664 0 : ctx->pending_batch.microblock_cnt += 1UL;
665 0 : }
666 :
667 0 : alpenglow_marker:
668 0 : if( FD_LIKELY( process_current_batch )) {
669 0 : ulong batch_sz = sizeof(ulong) + ctx->pending_batch.pos;
670 :
671 0 : if( FD_UNLIKELY( SHOULD_PROCESS_THESE_SHREDS ) ) {
672 : /* If it's our turn, shred this batch. FD_UNLIKELY because shred
673 : tile cnt generally >= 2 */
674 :
675 0 : long shredding_timing = -fd_tickcount();
676 :
677 0 : if( FD_UNLIKELY( !writing_marker ) ) ctx->send_fec_set_cnt = 0UL; /* verbose */
678 :
679 0 : fd_shredder_init_batch( ctx->shredder, ctx->pending_batch.raw, batch_sz, target_slot, entry_meta );
680 :
681 0 : ulong fec_set_cnt = fd_shredder_count_fec_sets( batch_sz );
682 0 : ulong pend_idx = ctx->send_fec_set_cnt;
683 0 : for( ulong j=0UL; j<fec_set_cnt; j++ ) {
684 :
685 0 : fd_fec_set_t * out = ctx->fec_sets + ctx->shredder_fec_set_idx;
686 :
687 0 : FD_TEST( fd_shredder_next_fec_set( ctx->shredder, out, chained_merkle_root ) );
688 0 : FD_TEST( pend_idx<FD_SHRED_BATCH_FEC_SETS_MAX );
689 0 : memcpy( ctx->out_merkle_roots[pend_idx].hash, chained_merkle_root, 32UL );
690 :
691 0 : out->data_shred_rcvd = 0U;
692 0 : out->parity_shred_rcvd = 0U;
693 0 : out->turbine_shred_rcvd = 0UL;
694 0 : out->repair_shred_rcvd = 0UL;
695 :
696 0 : FD_TEST( ctx->send_fec_set_cnt<FD_SHRED_BATCH_FEC_SETS_MAX );
697 0 : ctx->send_fec_set_idx[ ctx->send_fec_set_cnt ] = ctx->shredder_fec_set_idx;
698 0 : ctx->send_fec_set_cnt += 1UL;
699 0 : ctx->shredder_fec_set_idx = (ctx->shredder_fec_set_idx+1UL)%ctx->shredder_max_fec_set_idx;
700 :
701 0 : pend_idx++;
702 0 : }
703 :
704 0 : fd_shredder_fini_batch( ctx->shredder );
705 0 : shredding_timing += fd_tickcount();
706 :
707 : /* Update metrics */
708 0 : fd_histf_sample( ctx->metrics->batch_sz, batch_sz );
709 0 : fd_histf_sample( ctx->metrics->batch_microblock_cnt, ctx->pending_batch.microblock_cnt );
710 0 : fd_histf_sample( ctx->metrics->shredding_timing, (ulong)shredding_timing );
711 0 : } else {
712 0 : if( FD_UNLIKELY( !writing_marker ) ) ctx->send_fec_set_cnt = 0UL; /* verbose */
713 :
714 0 : fd_shredder_skip_batch( ctx->shredder, batch_sz, target_slot );
715 0 : }
716 :
717 0 : ctx->pending_batch.slot = 0UL;
718 0 : ctx->pending_batch.pos = 0UL;
719 0 : ctx->pending_batch.microblock_cnt = 0UL;
720 0 : ctx->batch_cnt++;
721 0 : }
722 :
723 0 : if( FD_UNLIKELY( init_new_batch && !writing_marker ) ) {
724 : /* TODO: this assumes that SHOULD_PROCESS_THESE_SHREDS is
725 : constant across batches. Otherwise, the condition may
726 : need to be removed (or adjusted). */
727 0 : if( FD_UNLIKELY( SHOULD_PROCESS_THESE_SHREDS ) ) {
728 : /* Ugh, yet another memcpy */
729 0 : fd_memcpy_tn( is_marker ? ctx->pending_batch.raw : ctx->pending_batch.payload + 0UL /* verbose */, entry, entry_sz );
730 0 : }
731 0 : ctx->pending_batch.slot = target_slot;
732 0 : ctx->pending_batch.pos = fd_ulong_if( is_marker, entry_sz-sizeof(ulong), entry_sz );
733 0 : ctx->pending_batch.microblock_cnt = fd_ulong_if( is_marker, 0UL, 1UL );
734 0 : }
735 :
736 : /* The entry opened a batch of its own, so close it now. */
737 0 : if( FD_UNLIKELY( is_marker && !writing_marker ) ) {
738 0 : process_current_batch = 1;
739 0 : writing_marker = 1;
740 0 : goto alpenglow_marker;
741 0 : }
742 0 : } else if( FD_UNLIKELY( ctx->in_kind[ in_idx ]==IN_KIND_NET ) ) {
743 : /* The common case, from the net tile. The FEC resolver API does
744 : not present a prepare/commit model. If we get overrun between
745 : when the FEC resolver verifies the signature and when it stores
746 : the local copy, we could end up storing and retransmitting
747 : garbage. Instead we copy it locally, sadly, and only give it to
748 : the FEC resolver when we know it won't be overrun anymore. */
749 0 : uchar const * dcache_entry = fd_net_rx_translate_frag( &ctx->in[ in_idx ].net_rx, chunk, ctl, sz );
750 0 : ulong hdr_sz = fd_disco_netmux_sig_hdr_sz( sig );
751 0 : FD_TEST( hdr_sz <= sz ); /* Should be ensured by the net tile */
752 0 : fd_udp_hdr_t const * udp_hdr = (fd_udp_hdr_t const *)( dcache_entry + hdr_sz - sizeof(fd_udp_hdr_t) );
753 0 : ulong dgram_sz = fd_ulong_min( fd_ulong_sat_sub( fd_ushort_bswap( udp_hdr->net_len ), sizeof(fd_udp_hdr_t) ), sz-hdr_sz );
754 0 : if( FD_UNLIKELY( fd_disco_netmux_sig_proto( sig )==DST_PROTO_REPAIR && dgram_sz<=FD_REPAIR_RES_DE_MAX ) ) {
755 0 : ctx->skip_frag = 1;
756 0 : ctx->fwd_meta = ctx->repair_out_idx!=ULONG_MAX &&
757 0 : fd_disco_netmux_sig_hash( sig )%ctx->round_robin_cnt==ctx->round_robin_id;
758 0 : if( ctx->fwd_meta ) { /* trim trailing bytes outside the datagram */
759 0 : fd_memcpy( ctx->shred_buffer, dcache_entry, hdr_sz+dgram_sz );
760 0 : ctx->shred_buffer_sz = hdr_sz+dgram_sz;
761 0 : }
762 0 : return;
763 0 : }
764 : /* ctx->shred_limit is the smallest per-slot max_shred_idx, so no
765 : later parse is stricter. The result may be garbage if the frag
766 : is overrun meanwhile, which the stem detects before after_frag. */
767 0 : fd_shred_t const * shred = fd_shred_parse( dcache_entry+hdr_sz, sz-hdr_sz, ctx->shred_limit );
768 0 : if( FD_UNLIKELY( !shred ) ) {
769 0 : ulong owner = sz-hdr_sz>=sizeof(ulong) ? fd_ulong_load_8( dcache_entry+hdr_sz )%ctx->round_robin_cnt : 0UL; /* round-robin key, the signature */
770 0 : ctx->parse_failed = owner==ctx->round_robin_id;
771 0 : ctx->skip_frag = 1;
772 0 : return;
773 0 : };
774 :
775 : /* all shreds in the same FEC set will have the same signature
776 : so we can round-robin shreds between the shred tiles based on
777 : just the signature without splitting individual FEC sets. */
778 0 : if( FD_LIKELY( fd_ulong_load_8( shred->signature )%ctx->round_robin_cnt!=ctx->round_robin_id ) ) {
779 0 : ctx->skip_frag = 1;
780 0 : return;
781 0 : }
782 :
783 0 : if( FD_UNLIKELY( fd_disco_netmux_sig_proto( sig )==DST_PROTO_REPAIR ) ) {
784 0 : ctx->metrics->repair_rcv_cnt++;
785 0 : ctx->metrics->repair_rcv_bytes += sz;
786 0 : } else {
787 0 : ctx->metrics->turbine_rcv_cnt++;
788 0 : ctx->metrics->turbine_rcv_bytes += sz;
789 0 : }
790 :
791 : /* Drop unchained merkle shreds */
792 0 : int is_unchained = !fd_shred_is_chained( fd_shred_type( shred->variant ) );
793 0 : if( FD_UNLIKELY( is_unchained ) ) {
794 0 : ctx->metrics->shred_rejected_unchained_cnt++;
795 0 : ctx->skip_frag = 1;
796 0 : return;
797 0 : };
798 :
799 0 : fd_memcpy( ctx->shred_buffer, dcache_entry+hdr_sz, sz-hdr_sz );
800 0 : ctx->shred_buffer_sz = sz-hdr_sz;
801 0 : }
802 0 : }
803 :
804 : static inline void
805 : send_shred( fd_shred_ctx_t * ctx,
806 : fd_stem_context_t * stem,
807 : fd_shred_t const * shred,
808 : fd_shred_dest_weighted_t const * dest,
809 0 : ulong tsorig ) {
810 :
811 0 : if( FD_UNLIKELY( !dest->ip4 ) ) return;
812 :
813 0 : uchar * packet = fd_chunk_to_laddr( ctx->net_out_mem, ctx->net_out_chunk );
814 :
815 0 : int is_data = fd_shred_is_data( fd_shred_type( shred->variant ) );
816 0 : fd_ip4_udp_hdrs_t * hdr = (fd_ip4_udp_hdrs_t *)packet;
817 0 : *hdr = *( is_data ? ctx->data_shred_net_hdr : ctx->parity_shred_net_hdr );
818 :
819 0 : fd_ip4_hdr_t * ip4 = hdr->ip4;
820 0 : ip4->daddr = dest->ip4;
821 0 : ip4->net_id = fd_ushort_bswap( ctx->net_id++ );
822 0 : ip4->check = 0U;
823 0 : ip4->check = fd_ip4_hdr_check_fast( ip4 );
824 :
825 0 : hdr->udp->net_dport = fd_ushort_bswap( dest->port );
826 :
827 0 : ulong shred_sz = fd_ulong_if( is_data, FD_SHRED_MIN_SZ, FD_SHRED_MAX_SZ );
828 0 : #if FD_HAS_AVX
829 : /* We're going to copy this shred potentially a bunch of times without
830 : reading it again, and we'd rather not thrash our cache, so we want
831 : to use non-temporal writes here. We need to make sure we don't
832 : touch the cache line containing the network headers that we just
833 : wrote to though. We know the destination is 64 byte aligned. */
834 0 : fd_memcpy_nt( packet+sizeof(fd_ip4_udp_hdrs_t), shred, shred_sz );
835 : #else
836 : fd_memcpy ( packet+sizeof(fd_ip4_udp_hdrs_t), shred, shred_sz );
837 : #endif
838 :
839 0 : ulong pkt_sz = shred_sz + sizeof(fd_ip4_udp_hdrs_t);
840 0 : ulong tspub = fd_frag_meta_ts_comp( fd_tickcount() );
841 0 : ulong sig = fd_disco_netmux_sig( dest->ip4, dest->port, dest->ip4, DST_PROTO_OUTGOING, sizeof(fd_ip4_udp_hdrs_t) );
842 0 : ulong const chunk = ctx->net_out_chunk;
843 0 : fd_stem_publish( stem, NET_OUT_IDX, sig, chunk, pkt_sz, 0UL, tsorig, tspub );
844 0 : ctx->net_out_chunk = fd_dcache_compact_next( chunk, pkt_sz, ctx->net_out_chunk0, ctx->net_out_wmark );
845 0 : }
846 :
847 : /* Send the set's received (or, if received is 0, its produced and
848 : recovered) shreds to their turbine destinations: the leader's first
849 : hop, or our children. */
850 :
851 : static void
852 : fan_out( fd_shred_ctx_t * ctx,
853 : fd_stem_context_t * stem,
854 : fd_fec_set_t const * set,
855 : int is_leader,
856 : int received,
857 0 : ulong tsorig ) {
858 0 : ulong fanout = 200UL; /* Default Agave's DATA_PLANE_FANOUT = 200UL */
859 0 : fd_shred_t const * shreds[ FD_REEDSOL_DATA_SHREDS_MAX+FD_REEDSOL_PARITY_SHREDS_MAX ];
860 0 : ulong k=0UL;
861 0 : for( ulong i=0UL; i<FD_FEC_SHRED_CNT; i++ )
862 0 : if( !!(set->data_shred_rcvd & (1U<<i))==!!received ) shreds[ k++ ] = set->data_shreds [ i ].s;
863 0 : for( ulong i=0UL; i<FD_FEC_SHRED_CNT; i++ )
864 0 : if( !!(set->parity_shred_rcvd & (1U<<i))==!!received ) shreds[ k++ ] = set->parity_shreds[ i ].s;
865 :
866 0 : if( FD_UNLIKELY( !k ) ) return;
867 0 : fd_shred_dest_t * sdest = fd_stake_ci_get_sdest_for_slot( ctx->stake_ci, shreds[ 0 ]->slot );
868 0 : if( FD_UNLIKELY( !sdest ) ) return;
869 :
870 0 : fd_shred_dest_idx_t * _dests = ctx->scratchpad_dests;
871 :
872 0 : ulong out_stride;
873 0 : ulong max_dest_cnt[1];
874 0 : fd_shred_dest_idx_t * dests;
875 0 : if( FD_LIKELY( !is_leader ) ) {
876 0 : out_stride = k;
877 0 : dests = fd_shred_dest_compute_children( sdest, shreds, k, _dests, k, fanout, fanout, max_dest_cnt );
878 0 : if( FD_UNLIKELY( !dests ) ) return; /* our own slot, or unknown leader: nothing to retransmit */
879 :
880 0 : for( ulong i=0UL; i<k; i++ ) {
881 0 : for( ulong j=0UL; j<ctx->adtl_dests_retransmit_cnt; j++ ) send_shred( ctx, stem, shreds[ i ], ctx->adtl_dests_retransmit+j, tsorig );
882 0 : }
883 0 : } else {
884 0 : for( ulong i=0UL; i<k; i++ ) {
885 0 : for( ulong j=0UL; j<ctx->adtl_dests_leader_cnt; j++ ) send_shred( ctx, stem, shreds[ i ], ctx->adtl_dests_leader+j, tsorig );
886 0 : }
887 0 : out_stride = k;
888 0 : *max_dest_cnt = 2UL; /* Root of turbine tree and next leader */
889 0 : dests = fd_shred_dest_compute_first( sdest, shreds, k, _dests );
890 :
891 : /* Send leader shreds to the next leader. At the epoch boundary,
892 : this will be a no-op so we don't need to get the other sdest. */
893 0 : fd_shred_dest_idx_t next_leader = fd_shred_dest_leader_for_slot( sdest, shreds[ 0 ]->slot + 4UL );
894 0 : for( ulong i=0UL; i<k; i++ ) _dests[ k+i ] = fd_uint_if( next_leader==_dests[ i ], FD_SHRED_DEST_NO_DEST, next_leader );
895 0 : }
896 0 : if( FD_UNLIKELY( !dests ) ) return;
897 :
898 0 : for( ulong i=0UL; i<k; i++ ) {
899 0 : for( ulong j=0UL; j<*max_dest_cnt; j++ ) send_shred( ctx, stem, shreds[ i ], fd_shred_dest_idx_to_dest( sdest, dests[ j*out_stride+i ]), tsorig );
900 0 : }
901 0 : }
902 :
903 : /* Finish a complete FEC set: copy it into the store, tell repair and
904 : replay, and fan it out to the network. */
905 :
906 : static void
907 : complete_fec_set( fd_shred_ctx_t * ctx,
908 : fd_stem_context_t * stem,
909 : ulong fec_set_idx,
910 : fd_bmtree_node_t const * merkle_root,
911 : int is_leader,
912 0 : ulong tsorig ) {
913 0 : fd_fec_set_t * set = ctx->fec_sets + fec_set_idx;
914 :
915 0 : fd_shred_t const * last = set->data_shreds[ FD_FEC_SHRED_CNT - 1 ].s;
916 :
917 : /* Broadcast locally produced shreds before any disk work. */
918 0 : if( FD_UNLIKELY( is_leader ) ) fan_out( ctx, stem, set, is_leader, 0, tsorig );
919 :
920 : /* Compute merkle root and chained merkle root. */
921 :
922 0 : int replay_fwd = 1;
923 0 : int again = 0;
924 0 : if( FD_LIKELY( ctx->store ) ) {
925 0 : fd_hash_t const * mr = fd_type_pun_const( merkle_root );
926 :
927 : /* The payload is filled before the FEC is linked */
928 :
929 0 : if( FD_UNLIKELY( fd_store_query( ctx->map_join, mr ) ) ) {
930 0 : again = 1;
931 0 : } else {
932 0 : fd_store_fec_spill_stats_t spill[1];
933 0 : uint data_idx;
934 0 : uchar * fec_data = fd_store_fec_data_reserve( ctx->store, ctx->disk_fd, &data_idx, spill );
935 0 : if( FD_UNLIKELY( spill->write_cnt ) ) {
936 0 : fd_histf_sample( ctx->metrics->fec_fallback_write_timing, spill->write_ticks );
937 0 : ctx->metrics->fec_fallback_write_cnt += spill->write_cnt;
938 0 : ctx->metrics->fec_fallback_write_bytes += spill->write_bytes;
939 0 : }
940 0 : if( FD_UNLIKELY( !fec_data ) ) FD_LOG_CRIT(( "Store could not allocate a FEC payload" ));
941 0 : uint data_sz = 0U;
942 0 : ushort shred_sz[ FD_FEC_SHRED_CNT ];
943 0 : for( ulong i=0UL; i<FD_FEC_SHRED_CNT; i++ ) {
944 0 : fd_shred_t * data_shred = set->data_shreds[i].s;
945 0 : ulong payload_sz = fd_shred_payload_sz( data_shred );
946 0 : if( FD_UNLIKELY( data_sz + payload_sz > ctx->store->fec_data_max ) ) {
947 0 : FD_LOG_CRIT(( "Shred tile %lu: completed FEC set %lu %u data_sz: %lu exceeds data_max: %lu", ctx->round_robin_id, data_shred->slot, data_shred->fec_set_idx, data_sz + payload_sz, ctx->store->fec_data_max ));
948 0 : }
949 0 : fd_memcpy_nt_nofence( fec_data + data_sz, fd_shred_data_payload( data_shred ), payload_sz );
950 0 : data_sz += (uint)payload_sz;
951 0 : FD_TEST( payload_sz<=USHORT_MAX );
952 0 : shred_sz[ i ] = (ushort)payload_sz;
953 0 : }
954 0 : _mm_sfence();
955 0 : fd_store_fec_t * fec;
956 0 : again = fd_store_insert( ctx->store, ctx->map_join, mr, last->slot, ctx->round_robin_id, data_idx, data_sz, shred_sz, &fec )==FD_MAP_ERR_KEY;
957 0 : }
958 0 : replay_fwd = 1;
959 0 : }
960 :
961 0 : if( FD_LIKELY( ctx->shred_out_idx!=ULONG_MAX && replay_fwd ) ) {
962 :
963 : /* Send all of the data shreds we recovered (weren't received), or
964 : all of them when we are the leader. The chunks are all written
965 : before any is published, so the link burst must cover them or a
966 : chunk could be reused while an unreliable consumer still sees
967 : its old seq. */
968 0 : FD_STATIC_ASSERT( FD_SHRED_STEM_BURST>=32UL, shred_out_burst );
969 0 : ulong missing_chunk[ 32 ];
970 0 : ulong missing_cnt = 0UL;
971 0 : for( int i=0; !again && i<32; i++ ) {
972 0 : if( fd_uint_extract_bit( set->data_shred_rcvd, i )==0 ) {
973 0 : fd_shred_t * const missing = &set->data_shreds[ i ].s[0];
974 :
975 0 : fd_shred_base_t * shred_msg = (fd_shred_base_t *)fd_chunk_to_laddr( ctx->shred_out_mem, ctx->shred_out_chunk );
976 0 : fd_memcpy_nt_nofence( shred_msg->shred_, missing, fd_shred_sz( missing ) );
977 0 : memcpy( &shred_msg->merkle_root, merkle_root->hash, sizeof(fd_hash_t) );
978 :
979 0 : missing_chunk[ missing_cnt++ ] = ctx->shred_out_chunk;
980 0 : ctx->shred_out_chunk = fd_dcache_compact_next( ctx->shred_out_chunk, sizeof(fd_shred_base_t), ctx->shred_out_chunk0, ctx->shred_out_wmark );
981 0 : }
982 0 : }
983 :
984 0 : if( FD_LIKELY( missing_cnt ) ) {
985 0 : _mm_sfence();
986 0 : ulong sig = ((ulong)FD_FEC_RESOLVER_SHRED_COMPLETES << 32UL) | ( is_leader ? SHRED_SIG_SRC_LEADER : SHRED_SIG_SRC_RECONSTRUCTED );
987 0 : ulong tspub = fd_frag_meta_ts_comp( fd_tickcount() );
988 0 : for( ulong i=0UL; i<missing_cnt; i++ ) fd_stem_publish( stem, ctx->shred_out_idx, sig, missing_chunk[ i ], sizeof(fd_shred_base_t), 0UL, tsorig, tspub );
989 0 : }
990 :
991 : /* Replay requires the FEC payload before this notification.
992 : Wire-shred persistence on rserve is asynchronous.
993 :
994 : This message, the shred msg, and the FEC evict msg constitute
995 : the max 3 possible messages to repair/replay per after_frag.
996 : In reality, it is only possible to publish all 3 in the case
997 : where we receive a coding shred first for a FEC set where
998 : (N=1,K=18), which allows for the FEC set to be instantly
999 : completed by the singular coding shred, and that also happens
1000 : to evict a FEC set from the curr_map. When fix-32 arrives, the
1001 : link burst value can be lowered to 2. */
1002 0 : ulong sig = again ? SHRED_SIG_FEC_COMPLETE_AGAIN : ( is_leader ? SHRED_SIG_FEC_COMPLETE_LEADER : SHRED_SIG_FEC_COMPLETE );
1003 :
1004 0 : fd_fec_complete_t * complete_msg = fd_chunk_to_laddr( ctx->shred_out_mem, ctx->shred_out_chunk );
1005 0 : complete_msg->last_shred_hdr = *last;
1006 0 : memcpy( &complete_msg->merkle_root, merkle_root->hash, sizeof(fd_hash_t) );
1007 0 : complete_msg->chained_merkle_root = *(fd_hash_t *)fd_type_pun((uchar *)last + fd_shred_chain_off( last->variant ));
1008 0 : if( FD_LIKELY( sig==SHRED_SIG_FEC_COMPLETE ) ) {
1009 0 : complete_msg->turbine_shred_cnt = (ulong)fd_ulong_popcnt( set->turbine_shred_rcvd );
1010 0 : complete_msg->repair_shred_cnt = (ulong)fd_ulong_popcnt( set->repair_shred_rcvd );
1011 0 : complete_msg->reconstructed_shred_cnt = 2U*(uint)FD_FEC_SHRED_CNT-(uint)(fd_uint_popcnt( set->data_shred_rcvd ) + fd_uint_popcnt( set->parity_shred_rcvd ));
1012 0 : complete_msg->data_shred_rcvd = set->data_shred_rcvd;
1013 0 : complete_msg->parity_shred_rcvd = set->parity_shred_rcvd;
1014 0 : complete_msg->repair_data_rcvd = (uint)set->repair_shred_rcvd; /* data shreds are the low FD_FEC_SHRED_CNT bits */
1015 0 : } else {
1016 0 : complete_msg->turbine_shred_cnt = 0U;
1017 0 : complete_msg->repair_shred_cnt = 0U;
1018 0 : complete_msg->reconstructed_shred_cnt = 0U;
1019 0 : complete_msg->data_shred_rcvd = 0U;
1020 0 : complete_msg->parity_shred_rcvd = 0U;
1021 0 : complete_msg->repair_data_rcvd = 0U;
1022 0 : }
1023 :
1024 0 : fd_stem_publish( stem, ctx->shred_out_idx, sig, ctx->shred_out_chunk, sizeof(fd_fec_complete_t), 0UL, tsorig, fd_frag_meta_ts_comp( fd_tickcount() ) );
1025 0 : ctx->shred_out_chunk = fd_dcache_compact_next( ctx->shred_out_chunk, sizeof(fd_fec_complete_t), ctx->shred_out_chunk0, ctx->shred_out_wmark );
1026 :
1027 0 : }
1028 :
1029 : /* Relay recovered shreds only after replay is told, computing their
1030 : turbine children is costly on staked nodes. A set rebuilt purely
1031 : from repair is not on the live turbine path (catch-up), so its
1032 : recovered shreds are not relayed. */
1033 0 : if( FD_LIKELY( !is_leader && set->turbine_shred_rcvd ) ) fan_out( ctx, stem, set, is_leader, 0, tsorig );
1034 0 : }
1035 :
1036 : static void
1037 : after_frag( fd_shred_ctx_t * ctx,
1038 : ulong in_idx,
1039 : ulong seq,
1040 : ulong sig,
1041 : ulong sz,
1042 : ulong tsorig,
1043 : ulong tspub,
1044 0 : fd_stem_context_t * stem ) {
1045 0 : (void)seq;
1046 0 : (void)sz;
1047 0 : (void)tsorig;
1048 :
1049 0 : if( FD_UNLIKELY( ctx->skip_frag ) ) {
1050 0 : ctx->metrics->shred_processing_result[ FD_METRICS_ENUM_SHRED_PROCESS_RESULT_V_PARSE_FAILED_IDX ] += (ulong)ctx->parse_failed;
1051 0 : if( FD_UNLIKELY( ctx->fwd_meta ) ) {
1052 0 : fd_memcpy( fd_chunk_to_laddr( ctx->repair_out_mem, ctx->repair_out_chunk ), ctx->shred_buffer, ctx->shred_buffer_sz );
1053 0 : fd_stem_publish( stem, ctx->repair_out_idx, sig, ctx->repair_out_chunk, ctx->shred_buffer_sz, 0UL, 0UL, tspub );
1054 0 : ctx->repair_out_chunk = fd_dcache_compact_next( ctx->repair_out_chunk, ctx->shred_buffer_sz, ctx->repair_out_chunk0, ctx->repair_out_wmark );
1055 0 : }
1056 0 : return;
1057 0 : }
1058 :
1059 0 : if( FD_UNLIKELY( ctx->in_kind[ in_idx ]==IN_KIND_EPOCH ) ) {
1060 0 : fd_stake_ci_epoch_msg_fini( ctx->stake_ci );
1061 0 : return;
1062 0 : }
1063 :
1064 0 : if( FD_UNLIKELY( ctx->in_kind[ in_idx ]==IN_KIND_ROOTEDR ) ) {
1065 0 : if( FD_LIKELY( (ctx->new_root > 0UL) & (ctx->new_root<ULONG_MAX) ) ) fd_fec_resolver_advance_slot_old( ctx->resolver, ctx->new_root );
1066 0 : return;
1067 0 : }
1068 :
1069 0 : if( FD_UNLIKELY( ctx->in_kind[ in_idx ]==IN_KIND_GOSSIP ) ) {
1070 0 : if( ctx->gossip_upd_buf->tag==FD_GOSSIP_UPDATE_TAG_CONTACT_INFO ) {
1071 0 : fd_gossip_contact_info_t const * ci = ctx->gossip_upd_buf->contact_info->value;
1072 0 : fd_ip4_port_t tvu_addr;
1073 0 : tvu_addr.addr = ci->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_TVU ].is_ipv6 ? 0U : ci->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_TVU ].ip4;
1074 0 : tvu_addr.port = ci->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_TVU ].port;
1075 0 : if( !tvu_addr.addr || !tvu_addr.port ) {
1076 0 : fd_stake_ci_dest_remove( ctx->stake_ci, fd_type_pun_const( ctx->gossip_upd_buf->origin ) );
1077 0 : } else {
1078 0 : fd_stake_ci_dest_update( ctx->stake_ci, fd_type_pun_const( ctx->gossip_upd_buf->origin ), tvu_addr.addr, fd_ushort_bswap( tvu_addr.port ) );
1079 0 : }
1080 0 : } else if( ctx->gossip_upd_buf->tag==FD_GOSSIP_UPDATE_TAG_CONTACT_INFO_REMOVE ) {
1081 0 : if( FD_UNLIKELY( !memcmp( ctx->identity_key->uc, ctx->gossip_upd_buf->origin, 32UL ) ) ) {
1082 : /* If our own contact info was dropped, we update with dummy IP
1083 : instead of removing since stake_ci expects our contact info
1084 : in the sdests table all the time. fd_stake_ci_new initializes
1085 : both ei->sdests with our contact info so this should always
1086 : update (and not append). */
1087 0 : fd_stake_ci_dest_update( ctx->stake_ci, fd_type_pun_const( ctx->gossip_upd_buf->origin ), 1U, 0U );
1088 0 : } else {
1089 0 : fd_stake_ci_dest_remove( ctx->stake_ci, fd_type_pun_const( ctx->gossip_upd_buf->origin ) );
1090 0 : }
1091 0 : }
1092 0 : return;
1093 0 : }
1094 :
1095 0 : if( FD_UNLIKELY( ctx->in_kind[ in_idx ]==IN_KIND_SIGN ) ) {
1096 0 : FD_TEST( ctx->sign_pend_cnt );
1097 :
1098 0 : ctx->sign_seq++;
1099 0 : fd_shred_sign_pend_t * pend = ctx->sign_pend + ctx->sign_pend_head;
1100 0 : ctx->sign_pend_head = (ctx->sign_pend_head+1UL)%FD_SHRED_SIGN_PEND_MAX;
1101 0 : ctx->sign_pend_cnt--;
1102 :
1103 0 : fd_fec_set_t * set = ctx->fec_sets + pend->fec_set_idx;
1104 0 : for( ulong i=0UL; i<FD_FEC_SHRED_CNT; i++ ) fd_memcpy( set->data_shreds [ i ].s->signature, ctx->signature, FD_ED25519_SIG_SZ );
1105 0 : for( ulong i=0UL; i<FD_FEC_SHRED_CNT; i++ ) fd_memcpy( set->parity_shreds[ i ].s->signature, ctx->signature, FD_ED25519_SIG_SZ );
1106 0 : complete_fec_set( ctx, stem, pend->fec_set_idx, &pend->merkle_root, 1, pend->tsorig );
1107 0 : return;
1108 0 : }
1109 :
1110 0 : if( FD_UNLIKELY( ctx->in_kind[ in_idx ]==IN_KIND_POH ) ) {
1111 0 : FD_TEST( ctx->sign_pend_cnt+ctx->send_fec_set_cnt<=FD_SHRED_SIGN_PEND_MAX );
1112 0 : for( ulong k=0UL; k<ctx->send_fec_set_cnt; k++ ) {
1113 0 : fd_shred_sign_pend_t * pend = ctx->sign_pend + (ctx->sign_pend_head+ctx->sign_pend_cnt++)%FD_SHRED_SIGN_PEND_MAX;
1114 0 : pend->fec_set_idx = ctx->send_fec_set_idx[ k ];
1115 0 : pend->tsorig = ctx->tsorig;
1116 0 : pend->merkle_root = ctx->out_merkle_roots[ k ];
1117 :
1118 0 : memcpy( fd_chunk_to_laddr( ctx->sign_out_mem, ctx->sign_out_chunk ), ctx->out_merkle_roots[ k ].hash, 32UL );
1119 0 : fd_stem_publish( stem, SIGN_OUT_IDX, FD_KEYGUARD_SIGN_TYPE_ED25519, ctx->sign_out_chunk, 32UL, 0UL, ctx->tsorig, fd_frag_meta_ts_comp( fd_tickcount() ) );
1120 0 : ctx->sign_out_chunk = fd_dcache_compact_next( ctx->sign_out_chunk, 32UL, ctx->sign_out_chunk0, ctx->sign_out_wmark );
1121 0 : }
1122 0 : return;
1123 0 : }
1124 0 : if( FD_UNLIKELY( ctx->in_kind[ in_idx ]==IN_KIND_REPAIR ) ) {
1125 0 : return;
1126 0 : }
1127 :
1128 0 : ulong fanout = 200UL; /* Default Agave's DATA_PLANE_FANOUT = 200UL */
1129 :
1130 0 : if( FD_LIKELY( ctx->in_kind[ in_idx ]==IN_KIND_NET ) ) {
1131 0 : uchar * shred_buffer = ctx->shred_buffer;
1132 0 : ulong shred_buffer_sz = ctx->shred_buffer_sz;
1133 :
1134 : /* Parsed in during_frag */
1135 0 : fd_shred_t const * shred = (fd_shred_t const *)shred_buffer;
1136 0 : ulong max_shred_idx = ctx->current_max_shred_idx;
1137 0 : if( FD_UNLIKELY( shred->slot< ctx->current_max_shred_idx_start_slot ) ) max_shred_idx = ctx->prev_max_shred_idx;
1138 0 : if( FD_UNLIKELY( shred->slot>=ctx->next_max_shred_idx_start_slot ) ) max_shred_idx = ctx->next_max_shred_idx;
1139 :
1140 0 : fd_epoch_leaders_t const * lsched = fd_stake_ci_get_lsched_for_slot( ctx->stake_ci, shred->slot );
1141 0 : if( FD_UNLIKELY( !lsched ) ) { ctx->metrics->shred_processing_result[ FD_METRICS_ENUM_SHRED_PROCESS_RESULT_V_LEADER_UNKNOWN_IDX ]++; return; }
1142 :
1143 0 : fd_pubkey_t const * slot_leader = fd_epoch_leaders_get( lsched, shred->slot );
1144 0 : if( FD_UNLIKELY( !slot_leader ) ) { ctx->metrics->shred_processing_result[ FD_METRICS_ENUM_SHRED_PROCESS_RESULT_V_LEADER_UNKNOWN_IDX ]++; return; } /* Count this as bad slot too */
1145 :
1146 : /* Our own block echoed back by a relay: we already have it, so skip
1147 : the verify and recovery. Repair responses for our own slots are
1148 : kept, so a restarted leader can repair its block. */
1149 0 : if( FD_UNLIKELY( fd_disco_netmux_sig_proto( sig )!=DST_PROTO_REPAIR && fd_memeq( slot_leader, ctx->identity_key, sizeof(fd_pubkey_t) ) ) ) {
1150 0 : ctx->metrics->shred_processing_result[ FD_METRICS_ENUM_SHRED_PROCESS_RESULT_V_OWN_LEADER_ECHOED_IDX ]++;
1151 0 : return;
1152 0 : }
1153 :
1154 0 : fd_fec_set_t const * out_fec_set[1];
1155 0 : fd_shred_t const * out_shred[1];
1156 0 : fd_fec_resolver_spilled_t spilled_fec = { 0 };
1157 0 : int from_repair = 0;
1158 :
1159 0 : uint nonce = UINT_MAX;
1160 0 : ulong shred_sz = fd_shred_sz( shred );
1161 0 : if( FD_UNLIKELY( (fd_disco_netmux_sig_proto( sig )==DST_PROTO_REPAIR) & (shred_buffer_sz>=shred_sz+sizeof(uint)) ) ) {
1162 0 : nonce = FD_LOAD(uint, shred_buffer + shred_sz );
1163 0 : long est_now_ns = fd_log_wallclock(); /* TODO: switch to fd_clock for performance */
1164 0 : int slot_complete = fd_shred_is_data( fd_shred_type( shred->variant ) ) && (shred->data.flags & FD_SHRED_DATA_FLAG_SLOT_COMPLETE);
1165 0 : int nonce_okay = fd_rnonce_ss_verify( ctx->repair_nonce_ss, nonce, shred->slot, shred->idx, slot_complete, est_now_ns );
1166 0 : ctx->metrics->bad_nonce += (ulong)(!nonce_okay);
1167 0 : from_repair = nonce_okay;
1168 0 : }
1169 :
1170 0 : uint shred_source = fd_disco_netmux_sig_proto( sig )==DST_PROTO_REPAIR
1171 0 : ? fd_uint_if( from_repair, FD_FEC_RESOLVER_SHRED_SRC_REPAIR, FD_FEC_RESOLVER_SHRED_SRC_BAD_REPAIR )
1172 0 : : FD_FEC_RESOLVER_SHRED_SRC_TURBINE;
1173 :
1174 0 : long add_shred_timing = -fd_tickcount();
1175 0 : int rv = fd_fec_resolver_add_shred( ctx->resolver, shred, shred_buffer_sz, max_shred_idx, shred_source, slot_leader->uc, out_fec_set, out_shred, &ctx->out_merkle_roots[0], &spilled_fec );
1176 0 : add_shred_timing += fd_tickcount();
1177 :
1178 0 : fd_histf_sample( ctx->metrics->add_shred_timing, (ulong)add_shred_timing );
1179 0 : ctx->metrics->shred_processing_result[ rv + FD_FEC_RESOLVER_ADD_SHRED_RETVAL_OFF+FD_SHRED_ADD_SHRED_EXTRA_RETVAL_CNT ]++;
1180 :
1181 0 : if( FD_UNLIKELY( ctx->shred_out_idx!=ULONG_MAX &&
1182 0 : spilled_fec.slot!=0 ) ) {
1183 : /* We've spilled an in-progress FEC set in the fec_resolver. We
1184 : need to let repair know to clear out it's cached info for that
1185 : fec set and re-repair those shreds. */
1186 0 : fd_fec_evicted_t * evicted_msg = (fd_fec_evicted_t *)fd_type_pun( fd_chunk_to_laddr( ctx->shred_out_mem, ctx->shred_out_chunk ) );
1187 0 : evicted_msg->slot = spilled_fec.slot;
1188 0 : evicted_msg->fec_set_idx = spilled_fec.fec_set_idx;
1189 0 : fd_memcpy( evicted_msg->merkle_root.uc, spilled_fec.merkle_root->hash, sizeof(fd_hash_t) );
1190 :
1191 0 : fd_stem_publish( stem, ctx->shred_out_idx, SHRED_SIG_FEC_EVICTED, ctx->shred_out_chunk, sizeof(fd_fec_evicted_t), 0, ctx->tsorig, fd_frag_meta_ts_comp( fd_tickcount() ) );
1192 0 : ctx->shred_out_chunk = fd_dcache_compact_next( ctx->shred_out_chunk, sizeof(fd_fec_evicted_t), ctx->shred_out_chunk0, ctx->shred_out_wmark );
1193 0 : }
1194 :
1195 : /* Send repair future slots past max_live_slots so that unstaked
1196 : nodes can catch up. */
1197 :
1198 0 : if( FD_UNLIKELY( rv==FD_FEC_RESOLVER_SHRED_IGNORED &&
1199 0 : shred_source==FD_FEC_RESOLVER_SHRED_SRC_TURBINE &&
1200 0 : shred->version==ctx->shredder->shred_version &&
1201 0 : shred->slot>ctx->future_slot &&
1202 0 : ctx->new_root<ULONG_MAX &&
1203 0 : shred->slot>=ctx->new_root+ctx->slot_max &&
1204 0 : fd_shred_type( shred->variant )!=FD_SHRED_TYPE_LEGACY_DATA &&
1205 0 : fd_shred_type( shred->variant )!=FD_SHRED_TYPE_LEGACY_CODE ) ) {
1206 0 : fd_bmtree_node_t root[1];
1207 0 : int signed_ = fd_shred_merkle_root( shred, ctx->bmtree_mem, root ) &&
1208 0 : FD_ED25519_SUCCESS==fd_ed25519_verify( root->hash, FD_SHRED_MERKLE_ROOT_SZ, shred->signature, slot_leader->uc, ctx->sha512 );
1209 0 : if( FD_LIKELY( signed_ ) ) {
1210 0 : ctx->future_slot = shred->slot;
1211 0 : fd_stem_publish( stem, ctx->shred_out_idx, fd_shred_sig_future( shred->slot ), 0UL, 0UL, 0UL, ctx->tsorig, fd_frag_meta_ts_comp( fd_tickcount() ) );
1212 0 : }
1213 0 : }
1214 :
1215 0 : if( FD_LIKELY( ctx->shred_out_idx!=ULONG_MAX
1216 0 : && ( ( rv==FD_FEC_RESOLVER_SHRED_OKAY )
1217 0 : | ( rv==FD_FEC_RESOLVER_SHRED_COMPLETES )
1218 0 : | ( rv==FD_FEC_RESOLVER_SHRED_DUPLICATE )
1219 0 : | ( rv==FD_FEC_RESOLVER_SHRED_EQUIVOC ) ) ) ) {
1220 :
1221 : /* Construct the sig from fec_resolver result and shred source. */
1222 :
1223 0 : ulong _sig = fd_disco_netmux_sig_proto( sig )==DST_PROTO_REPAIR
1224 0 : ? ( from_repair /*nonce_okay*/ ? SHRED_SIG_SRC_REPAIR : SHRED_SIG_SRC_BAD_REPAIR )
1225 0 : : ( SHRED_SIG_SRC_TURBINE );
1226 0 : _sig = ((ulong)rv << 32UL) | _sig;
1227 :
1228 : /* Copy the full shred into the frag and publish. */
1229 :
1230 0 : fd_shred_base_t * shred_msg = (fd_shred_base_t *)fd_chunk_to_laddr( ctx->shred_out_mem, ctx->shred_out_chunk );
1231 0 : memcpy( shred_msg->shred_, shred, fd_shred_sz( shred ) );
1232 0 : memcpy( &shred_msg->merkle_root, ctx->out_merkle_roots[0].hash, sizeof(fd_hash_t) );
1233 0 : if( FD_UNLIKELY( fd_disco_netmux_sig_proto( sig )==DST_PROTO_REPAIR ) ) { shred_msg->rnonce = nonce; }
1234 :
1235 0 : ulong tspub = fd_frag_meta_ts_comp( fd_tickcount() );
1236 0 : fd_stem_publish( stem, ctx->shred_out_idx, _sig, ctx->shred_out_chunk, sizeof(fd_shred_base_t), 0UL, ctx->tsorig, tspub );
1237 0 : ctx->shred_out_chunk = fd_dcache_compact_next( ctx->shred_out_chunk, sizeof(fd_shred_base_t), ctx->shred_out_chunk0, ctx->shred_out_wmark );
1238 0 : }
1239 :
1240 0 : if( FD_LIKELY( fd_disco_netmux_sig_proto( sig ) != DST_PROTO_REPAIR &&
1241 0 : ( (rv==FD_FEC_RESOLVER_SHRED_OKAY) | (rv==FD_FEC_RESOLVER_SHRED_COMPLETES) ) ) ) {
1242 : /* Relay this shred */
1243 0 : if( FD_UNLIKELY( ctx->rtx_fseq && shred->slot>ctx->rtx_slot ) ) {
1244 0 : ctx->rtx_slot = shred->slot;
1245 0 : fd_fseq_update( ctx->rtx_fseq, shred->slot );
1246 0 : }
1247 0 : ulong max_dest_cnt[1];
1248 0 : do {
1249 : /* If we've validated the shred and it COMPLETES but we can't
1250 : compute the destination for whatever reason, don't forward
1251 : the shred, but still send it to the blockstore. */
1252 0 : fd_shred_dest_t * sdest = fd_stake_ci_get_sdest_for_slot( ctx->stake_ci, shred->slot );
1253 0 : if( FD_UNLIKELY( !sdest ) ) break;
1254 0 : fd_shred_dest_idx_t * dests = fd_shred_dest_compute_children( sdest, &shred, 1UL, ctx->scratchpad_dests, 1UL, fanout, fanout, max_dest_cnt );
1255 0 : if( FD_UNLIKELY( !dests ) ) break;
1256 :
1257 0 : for( ulong i=0UL; i<ctx->adtl_dests_retransmit_cnt; i++ ) send_shred( ctx, stem, *out_shred, ctx->adtl_dests_retransmit+i, ctx->tsorig );
1258 0 : for( ulong j=0UL; j<*max_dest_cnt; j++ ) send_shred( ctx, stem, *out_shred, fd_shred_dest_idx_to_dest( sdest, dests[ j ] ), ctx->tsorig );
1259 0 : } while( 0 );
1260 0 : }
1261 :
1262 0 : if( FD_LIKELY( rv!=FD_FEC_RESOLVER_SHRED_COMPLETES ) ) return;
1263 :
1264 0 : FD_TEST( ctx->fec_sets <= *out_fec_set );
1265 0 : complete_fec_set( ctx, stem, (ulong)(*out_fec_set - ctx->fec_sets), &ctx->out_merkle_roots[0], 0, ctx->tsorig );
1266 0 : }
1267 0 : }
1268 :
1269 : static void
1270 : privileged_init( fd_topo_t const * topo,
1271 0 : fd_topo_tile_t const * tile ) {
1272 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
1273 0 : FD_TEST( scratch!=NULL );
1274 :
1275 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
1276 0 : fd_shred_ctx_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof( fd_shred_ctx_t ), sizeof( fd_shred_ctx_t ) );
1277 :
1278 0 : if( FD_UNLIKELY( !strcmp( tile->shred.identity_key_path, "" ) ) )
1279 0 : FD_LOG_ERR(( "identity_key_path not set" ));
1280 :
1281 0 : ctx->identity_key[ 0 ] = *(fd_pubkey_t const *)fd_type_pun_const( fd_keyload_load( tile->shred.identity_key_path, /* pubkey only: */ 1 ) );
1282 :
1283 0 : ctx->adtl_dests_retransmit_cnt = tile->shred.adtl_dests_retransmit_cnt;
1284 0 : fd_shred_resolve_additional_destinations( tile->shred.adtl_dests_retransmit,
1285 0 : tile->shred.adtl_dests_retransmit_cnt,
1286 0 : "tiles.shred.additional_shred_destinations_retransmit",
1287 0 : ctx->adtl_dests_retransmit );
1288 0 : ctx->adtl_dests_leader_cnt = tile->shred.adtl_dests_leader_cnt;
1289 0 : fd_shred_resolve_additional_destinations( tile->shred.adtl_dests_leader,
1290 0 : tile->shred.adtl_dests_leader_cnt,
1291 0 : "tiles.shred.additional_shred_destinations_leader",
1292 0 : ctx->adtl_dests_leader );
1293 :
1294 0 : if( FD_UNLIKELY( !fd_rng_secure( &(ctx->resolver_seed), sizeof(ulong) ) ) ) {
1295 0 : FD_LOG_CRIT(( "fd_rng_secure failed" ));
1296 0 : }
1297 0 : if( FD_UNLIKELY( !fd_rng_secure( &(ctx->shred_dest_seed), sizeof(ulong) ) ) ) {
1298 0 : FD_LOG_CRIT(( "fd_rng_secure failed" ));
1299 0 : }
1300 :
1301 0 : ctx->disk_fd = -1;
1302 0 : ulong store_obj_id = fd_pod_queryf_ulong( topo->props, ULONG_MAX, "store" );
1303 0 : if( FD_LIKELY( store_obj_id!=ULONG_MAX ) ) {
1304 0 : fd_store_t * store = fd_store_join( fd_topo_obj_laddr( topo, store_obj_id ) );
1305 0 : FD_TEST( store );
1306 0 : ctx->disk_fd = FD_STORE_FD_RW;
1307 0 : if( FD_UNLIKELY( fcntl( ctx->disk_fd, F_GETFD )<0 ) )
1308 0 : FD_LOG_ERR(( "store file descriptor was not inherited (%i-%s)", errno, fd_io_strerror( errno ) ));
1309 0 : }
1310 :
1311 0 : }
1312 :
1313 : static void
1314 : unprivileged_init( fd_topo_t const * topo,
1315 0 : fd_topo_tile_t const * tile ) {
1316 :
1317 0 : FD_TEST( 0==strcmp( topo->links[tile->out_link_id[ NET_OUT_IDX ]].name, "shred_net" ) );
1318 0 : FD_TEST( 0==strcmp( topo->links[tile->out_link_id[ SIGN_OUT_IDX ]].name, "shred_sign" ) );
1319 :
1320 0 : if( FD_UNLIKELY( !tile->out_cnt ) )
1321 0 : FD_LOG_ERR(( "shred tile has no primary output link" ));
1322 :
1323 0 : ulong fec_exposure = tile->shred.fec_exposure;
1324 0 : if( FD_UNLIKELY( !fec_exposure ) ) FD_LOG_ERR(( "fec_exposure not set" ));
1325 :
1326 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
1327 0 : FD_TEST( scratch!=NULL );
1328 :
1329 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
1330 0 : fd_shred_ctx_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof( fd_shred_ctx_t ), sizeof( fd_shred_ctx_t ) );
1331 :
1332 0 : FD_CHECK_ERR( tile->in_cnt<=IN_LINK_MAX, "too many input links" );
1333 :
1334 0 : ctx->round_robin_cnt = fd_topo_tile_name_cnt( topo, tile->name );
1335 0 : ctx->round_robin_id = tile->kind_id;
1336 0 : ctx->batch_cnt = 0UL;
1337 0 : ctx->slot = ULONG_MAX;
1338 0 : ctx->store = NULL;
1339 :
1340 0 : ulong store_obj_id = fd_pod_queryf_ulong( topo->props, ULONG_MAX, "store" );
1341 0 : if( FD_LIKELY( store_obj_id!=ULONG_MAX ) ) {
1342 0 : ctx->store = fd_store_join( fd_topo_obj_laddr( topo, store_obj_id ) );
1343 0 : FD_TEST( ctx->store->magic==FD_STORE_MAGIC );
1344 0 : FD_TEST( fd_store_map_ljoin( ctx->store, ctx->map_join ) );
1345 0 : }
1346 :
1347 : /* If the default partial_depth is ever changed, correspondingly
1348 : change the size of the fd_fec_intra_pool in fd_fec_repair. */
1349 0 : ulong fec_resolver_footprint = fd_fec_resolver_footprint( tile->shred.fec_resolver_depth, 1UL, fec_exposure+1UL,
1350 0 : 128UL * tile->shred.fec_resolver_depth );
1351 0 : ulong fec_set_cnt = fd_shred_tile_fec_set_cnt( fec_exposure, tile->shred.fec_resolver_depth );
1352 :
1353 0 : void * fec_sets_shmem = NULL;
1354 :
1355 0 : ctx->rtx_fseq = NULL;
1356 0 : ctx->rtx_slot = 0UL;
1357 0 : ulong rtx_obj_id = fd_pod_queryf_ulong( topo->props, ULONG_MAX, "shred_rtx.%lu", tile->kind_id );
1358 0 : if( FD_UNLIKELY( rtx_obj_id!=ULONG_MAX ) ) {
1359 0 : ctx->rtx_fseq = fd_fseq_join( fd_topo_obj_laddr( topo, rtx_obj_id ) );
1360 0 : FD_TEST( ctx->rtx_fseq );
1361 0 : }
1362 :
1363 0 : ctx->repair_out_idx = fd_topo_find_tile_out_link( topo, tile, "shred_repair", ctx->round_robin_id );
1364 0 : if( FD_LIKELY( ctx->repair_out_idx!=ULONG_MAX ) ) {
1365 0 : fd_topo_link_t const * repair_out = &topo->links[ tile->out_link_id[ ctx->repair_out_idx ] ];
1366 0 : FD_TEST( repair_out->mtu>=FD_REPAIR_NET_FRAME_MAX );
1367 0 : ctx->repair_out_mem = topo->workspaces[ topo->objs[ repair_out->dcache_obj_id ].wksp_id ].wksp;
1368 0 : ctx->repair_out_chunk0 = fd_dcache_compact_chunk0( ctx->repair_out_mem, repair_out->dcache );
1369 0 : ctx->repair_out_wmark = fd_dcache_compact_wmark ( ctx->repair_out_mem, repair_out->dcache, repair_out->mtu );
1370 0 : ctx->repair_out_chunk = ctx->repair_out_chunk0;
1371 0 : }
1372 :
1373 0 : ctx->shred_out_idx = fd_topo_find_tile_out_link( topo, tile, "shred_out", ctx->round_robin_id );
1374 0 : if( FD_LIKELY( ctx->shred_out_idx!=ULONG_MAX ) ) {
1375 0 : FD_TEST( fec_exposure==FD_SHRED_FIREDANCER_FEC_EXPOSURE );
1376 0 : fd_topo_link_t const * shred_out = &topo->links[ tile->out_link_id[ ctx->shred_out_idx ] ];
1377 0 : ctx->shred_out_mem = topo->workspaces[ topo->objs[ shred_out->dcache_obj_id ].wksp_id ].wksp;
1378 0 : ctx->shred_out_chunk0 = fd_dcache_compact_chunk0( ctx->shred_out_mem, shred_out->dcache );
1379 0 : ctx->shred_out_wmark = fd_dcache_compact_wmark ( ctx->shred_out_mem, shred_out->dcache, shred_out->mtu );
1380 0 : ctx->shred_out_chunk = ctx->shred_out_chunk0;
1381 0 : FD_TEST( fd_dcache_compact_is_safe( ctx->shred_out_mem, shred_out->dcache, shred_out->mtu, shred_out->depth ) );
1382 0 : ulong fec_sets_required_cnt = fec_set_cnt*ctx->round_robin_cnt;
1383 0 : if( FD_LIKELY( ctx->store ) ) {
1384 0 : FD_TEST( ctx->store->fec_set_cnt>=fec_sets_required_cnt );
1385 0 : fd_fec_set_t * store_fec_sets = fd_store_fec_sets( ctx->store );
1386 0 : FD_TEST( store_fec_sets );
1387 :
1388 0 : fec_sets_shmem = store_fec_sets + ctx->round_robin_id*fec_set_cnt;
1389 0 : } else {
1390 0 : ulong fec_sets_obj_id = fd_pod_query_ulong( topo->props, "fec_sets", ULONG_MAX );
1391 0 : FD_TEST( fec_sets_obj_id!=ULONG_MAX );
1392 :
1393 0 : ulong fec_sets_sz = fd_pod_queryf_ulong( topo->props, 0UL, "obj.%lu.sz", fec_sets_obj_id );
1394 0 : FD_TEST( fec_sets_sz>=fec_sets_required_cnt*sizeof(fd_fec_set_t) );
1395 :
1396 0 : fd_fec_set_t * fec_sets = fd_topo_obj_laddr( topo, fec_sets_obj_id );
1397 0 : FD_TEST( fec_sets );
1398 :
1399 0 : fec_sets_shmem = fec_sets + ctx->round_robin_id*fec_set_cnt;
1400 0 : }
1401 :
1402 0 : ulong rnonce_ss_id = fd_pod_queryf_ulong( topo->props, ULONG_MAX, "rnonce_ss" );
1403 0 : FD_TEST( rnonce_ss_id!=ULONG_MAX );
1404 0 : memcpy( ctx->repair_nonce_ss, fd_topo_obj_laddr( topo, rnonce_ss_id ), sizeof(fd_rnonce_ss_t) );
1405 0 : }
1406 0 : if( FD_UNLIKELY( !fec_sets_shmem ) ) FD_LOG_ERR(( "missing FEC-set backing memory" ));
1407 :
1408 0 : if( FD_UNLIKELY( !tile->shred.fec_resolver_depth ) ) FD_LOG_ERR(( "fec_resolver_depth not set" ));
1409 0 : if( FD_UNLIKELY( !tile->shred.shred_listen_port ) ) FD_LOG_ERR(( "shred_listen_port not set" ));
1410 :
1411 0 : void * _stake_ci = FD_SCRATCH_ALLOC_APPEND( l, fd_stake_ci_align(), fd_stake_ci_footprint() );
1412 0 : void * _resolver = FD_SCRATCH_ALLOC_APPEND( l, fd_fec_resolver_align(), fec_resolver_footprint );
1413 0 : void * _shredder = FD_SCRATCH_ALLOC_APPEND( l, fd_shredder_align(), fd_shredder_footprint() );
1414 :
1415 0 : fd_fec_set_t * fec_sets = (fd_fec_set_t *)fec_sets_shmem;
1416 :
1417 0 : #define NONNULL( x ) (__extension__({ \
1418 0 : __typeof__((x)) __x = (x); \
1419 0 : if( FD_UNLIKELY( !__x ) ) FD_LOG_ERR(( #x " was unexpectedly NULL" )); \
1420 0 : __x; }))
1421 :
1422 0 : ushort expected_shred_version = tile->shred.expected_shred_version;
1423 :
1424 0 : ctx->keyswitch = fd_keyswitch_join( fd_topo_obj_laddr( topo, tile->id_keyswitch_obj_id ) );
1425 0 : FD_TEST( ctx->keyswitch );
1426 :
1427 : /* populate ctx */
1428 0 : fd_topo_link_t const * sign_out = &topo->links[ tile->out_link_id[ SIGN_OUT_IDX ] ];
1429 0 : FD_TEST( sign_out->mtu==32UL );
1430 0 : ctx->sign_out_mem = topo->workspaces[ topo->objs[ sign_out->dcache_obj_id ].wksp_id ].wksp;
1431 0 : ctx->sign_out_chunk0 = fd_dcache_compact_chunk0( ctx->sign_out_mem, sign_out->dcache );
1432 0 : ctx->sign_out_wmark = fd_dcache_compact_wmark ( ctx->sign_out_mem, sign_out->dcache, sign_out->mtu );
1433 0 : ctx->sign_out_chunk = ctx->sign_out_chunk0;
1434 :
1435 0 : ctx->bench_max_shred_idx = tile->shred.bench_max_shreds_per_block;
1436 0 : ulong shred_limit = tile->shred.max_shreds_per_block;
1437 0 : ctx->shred_limit = shred_limit;
1438 0 : fd_fec_set_t * resolver_sets = fec_sets + fec_exposure + FD_SHRED_SIGN_PEND_MAX;
1439 0 : ctx->shredder = NONNULL( fd_shredder_join ( fd_shredder_new ( _shredder ) ) );
1440 0 : ctx->resolver = NONNULL( fd_fec_resolver_join ( fd_fec_resolver_new ( _resolver,
1441 0 : tile->shred.fec_resolver_depth, 1UL,
1442 0 : fec_exposure+1UL,
1443 0 : 128UL * tile->shred.fec_resolver_depth,
1444 0 : tile->shred.slot_max,
1445 0 : resolver_sets,
1446 0 : ctx->resolver_seed ) ) );
1447 0 : FD_TEST( fd_sha512_join( fd_sha512_new( ctx->sha512 ) ) );
1448 0 : ctx->new_root = 0UL;
1449 0 : ctx->slot_max = tile->shred.slot_max;
1450 0 : ctx->future_slot = 0UL;
1451 :
1452 0 : if( FD_LIKELY( !!expected_shred_version ) ) {
1453 0 : fd_shredder_set_shred_version ( ctx->shredder, expected_shred_version );
1454 0 : fd_fec_resolver_set_shred_version( ctx->resolver, expected_shred_version );
1455 0 : }
1456 :
1457 0 : ctx->fec_sets = fec_sets;
1458 :
1459 0 : ctx->stake_ci = fd_stake_ci_join( fd_stake_ci_new( _stake_ci, ctx->identity_key, ctx->shred_dest_seed ) );
1460 :
1461 0 : ctx->net_id = (ushort)0;
1462 :
1463 0 : fd_ip4_udp_hdr_init( ctx->data_shred_net_hdr, FD_SHRED_MIN_SZ, 0, tile->shred.shred_listen_port );
1464 0 : fd_ip4_udp_hdr_init( ctx->parity_shred_net_hdr, FD_SHRED_MAX_SZ, 0, tile->shred.shred_listen_port );
1465 :
1466 0 : for( ulong i=0UL; i<tile->in_cnt; i++ ) {
1467 0 : fd_topo_link_t const * link = &topo->links[ tile->in_link_id[ i ] ];
1468 0 : fd_topo_wksp_t const * link_wksp = &topo->workspaces[ topo->objs[ link->dcache_obj_id ].wksp_id ];
1469 :
1470 0 : if( FD_LIKELY( !strcmp( link->name, "net_shred" ) ) ) {
1471 0 : ctx->in_kind[ i ] = IN_KIND_NET;
1472 0 : fd_net_rx_bounds_init( &ctx->in[ i ].net_rx, link->dcache );
1473 0 : continue; /* only net_rx needs to be set in this case. */
1474 0 : }
1475 0 : else if( FD_LIKELY( !strcmp( link->name, "poh_shred" ) ) ) ctx->in_kind[ i ] = IN_KIND_POH;
1476 0 : else if( FD_LIKELY( !strcmp( link->name, "replay_epoch" ) ) ) ctx->in_kind[ i ] = IN_KIND_EPOCH;
1477 0 : else if( FD_LIKELY( !strcmp( link->name, "sign_shred" ) ) ) ctx->in_kind[ i ] = IN_KIND_SIGN;
1478 0 : else if( FD_LIKELY( !strcmp( link->name, "ipecho_out" ) ) ) ctx->in_kind[ i ] = IN_KIND_IPECHO;
1479 0 : else if( FD_LIKELY( !strcmp( link->name, "replay_slot" ) ) ) ctx->in_kind[ i ] = IN_KIND_ROOTEDR;
1480 0 : else if( FD_LIKELY( !strcmp( link->name, "gossip_ciaddr" ) ) ) ctx->in_kind[ i ] = IN_KIND_GOSSIP;
1481 :
1482 0 : else FD_LOG_ERR(( "shred tile has unexpected input link %lu %s", i, link->name ));
1483 :
1484 0 : if( FD_LIKELY( !!link->mtu ) ) {
1485 0 : ctx->in[ i ].mem = link_wksp->wksp;
1486 0 : ctx->in[ i ].chunk0 = fd_dcache_compact_chunk0( ctx->in[ i ].mem, link->dcache );
1487 0 : ctx->in[ i ].wmark = fd_dcache_compact_wmark ( ctx->in[ i ].mem, link->dcache, link->mtu );
1488 0 : }
1489 0 : }
1490 :
1491 0 : fd_topo_link_t const * net_out = &topo->links[ tile->out_link_id[ NET_OUT_IDX ] ];
1492 :
1493 0 : ctx->net_out_chunk0 = fd_dcache_compact_chunk0( fd_wksp_containing( net_out->dcache ), net_out->dcache );
1494 0 : ctx->net_out_mem = topo->workspaces[ topo->objs[ net_out->dcache_obj_id ].wksp_id ].wksp;
1495 0 : ctx->net_out_wmark = fd_dcache_compact_wmark ( ctx->net_out_mem, net_out->dcache, net_out->mtu );
1496 0 : ctx->net_out_chunk = ctx->net_out_chunk0;
1497 :
1498 0 : if( FD_LIKELY( ctx->shred_out_idx!=ULONG_MAX ) ) {
1499 0 : fd_topo_link_t const * shred_out = &topo->links[ tile->out_link_id[ ctx->shred_out_idx ] ];
1500 0 : ctx->shred_out_mem = topo->workspaces[ topo->objs[ shred_out->dcache_obj_id ].wksp_id ].wksp;
1501 0 : ctx->shred_out_chunk0 = fd_dcache_compact_chunk0( ctx->shred_out_mem, shred_out->dcache );
1502 0 : ctx->shred_out_wmark = fd_dcache_compact_wmark ( ctx->shred_out_mem, shred_out->dcache, shred_out->mtu );
1503 0 : ctx->shred_out_chunk = ctx->shred_out_chunk0;
1504 0 : FD_TEST( fd_dcache_compact_is_safe( ctx->shred_out_mem, shred_out->dcache, shred_out->mtu, shred_out->depth ) );
1505 : /* An unreliable consumer ignores credits, so the dcache must hold
1506 : a full STEM_BURST beyond the mcache depth. */
1507 0 : FD_TEST( shred_out->burst>=FD_SHRED_STEM_BURST );
1508 0 : }
1509 :
1510 0 : ctx->poh_in_expect_seq = 0UL;
1511 :
1512 0 : ctx->sign_pend_head = 0UL;
1513 0 : ctx->sign_pend_cnt = 0UL;
1514 0 : ctx->sign_seq = 0UL;
1515 :
1516 0 : ctx->shredder_fec_set_idx = 0UL;
1517 0 : ctx->shredder_max_fec_set_idx = fec_exposure + FD_SHRED_SIGN_PEND_MAX;
1518 :
1519 0 : ctx->chained_merkle_root = NULL;
1520 0 : memset( ctx->out_merkle_roots, 0, sizeof(ctx->out_merkle_roots) );
1521 :
1522 0 : for( ulong i=0UL; i<FD_SHRED_BATCH_FEC_SETS_MAX; i++ ) { ctx->send_fec_set_idx[ i ] = ULONG_MAX; }
1523 0 : ctx->send_fec_set_cnt = 0UL;
1524 :
1525 0 : ctx->shred_buffer_sz = 0UL;
1526 0 : memset( ctx->shred_buffer, 0xFF, FD_NET_MTU );
1527 :
1528 0 : fd_histf_join( fd_histf_new( ctx->metrics->batch_sz, FD_MHIST_MIN( SHRED, BATCH_SIZE_BYTES ),
1529 0 : FD_MHIST_MAX( SHRED, BATCH_SIZE_BYTES ) ) );
1530 0 : fd_histf_join( fd_histf_new( ctx->metrics->batch_microblock_cnt, FD_MHIST_MIN( SHRED, MICROBLOCK_PER_BATCH ),
1531 0 : FD_MHIST_MAX( SHRED, MICROBLOCK_PER_BATCH ) ) );
1532 0 : fd_histf_join( fd_metrics_histf_new_ns( ctx->metrics->shredding_timing, FD_MHIST_EDGES_NS( SHRED, BATCH_SHRED_DURATION_NANOS ) ) );
1533 0 : fd_histf_join( fd_metrics_histf_new_ns( ctx->metrics->add_shred_timing, FD_MHIST_EDGES_NS( SHRED, ADD_SHRED_DURATION_NANOS ) ) );
1534 0 : fd_histf_join( fd_metrics_histf_new_ns( ctx->metrics->fec_fallback_write_timing, FD_MHIST_EDGES_NS( SHRED, STORE_FEC_SYNC_SPILL_DURATION_NANOS ) ) );
1535 0 : memset( ctx->metrics->shred_processing_result, '\0', sizeof(ctx->metrics->shred_processing_result) );
1536 0 : ctx->metrics->invalid_block_id_cnt = 0UL;
1537 0 : ctx->metrics->shred_rejected_unchained_cnt = 0UL;
1538 0 : ctx->metrics->repair_rcv_cnt = 0UL;
1539 0 : ctx->metrics->repair_rcv_bytes = 0UL;
1540 0 : ctx->metrics->turbine_rcv_cnt = 0UL;
1541 0 : ctx->metrics->turbine_rcv_bytes = 0UL;
1542 0 : ctx->metrics->bad_nonce = 0UL;
1543 0 : ctx->metrics->fec_fallback_write_cnt = 0UL;
1544 0 : ctx->metrics->fec_fallback_write_bytes = 0UL;
1545 :
1546 0 : ctx->pending_batch.microblock_cnt = 0UL;
1547 0 : ctx->pending_batch.pos = 0UL;
1548 0 : ctx->pending_batch.slot = 0UL;
1549 0 : memset( ctx->pending_batch.payload, 0, sizeof(ctx->pending_batch.payload) );
1550 :
1551 0 : memset( ctx->epoch_schedule, 0, sizeof(ctx->epoch_schedule) );
1552 0 : for( ulong i=0UL; i<FD_SHRED_FEATURES_ACTIVATION_SLOT_CNT; i++ ) {
1553 0 : ctx->features_activation->slots[i] = FD_SHRED_FEATURES_ACTIVATION_SLOT_DISABLED;
1554 0 : }
1555 0 : ctx->prev_max_shred_idx = ctx->shred_limit;
1556 0 : ctx->current_max_shred_idx = ctx->shred_limit;
1557 0 : ctx->next_max_shred_idx = ctx->shred_limit;
1558 0 : ctx->current_max_shred_idx_start_slot = 0UL;
1559 0 : ctx->next_max_shred_idx_start_slot = ULONG_MAX;
1560 :
1561 0 : ulong scratch_top = FD_SCRATCH_ALLOC_FINI( l, scratch_align() );
1562 0 : if( FD_UNLIKELY( scratch_top > (ulong)scratch + scratch_footprint( tile ) ) )
1563 0 : FD_LOG_ERR(( "scratch overflow %lu %lu %lu", scratch_top - (ulong)scratch - scratch_footprint( tile ), scratch_top, (ulong)scratch + scratch_footprint( tile ) ));
1564 :
1565 0 : memset( ctx->block_ids, 0, sizeof(ctx->block_ids) );
1566 0 : }
1567 :
1568 : static ulong
1569 : populate_allowed_seccomp( fd_topo_t const * topo,
1570 : fd_topo_tile_t const * tile,
1571 : ulong out_cnt,
1572 0 : struct sock_filter * out ) {
1573 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
1574 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
1575 0 : fd_shred_ctx_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof( fd_shred_ctx_t ), sizeof( fd_shred_ctx_t ) );
1576 :
1577 0 : populate_sock_filter_policy_fd_shred_tile( out_cnt, out, (uint)fd_log_private_logfile_fd(), (uint)ctx->disk_fd );
1578 0 : return sock_filter_policy_fd_shred_tile_instr_cnt;
1579 0 : }
1580 :
1581 : static ulong
1582 : populate_allowed_fds( fd_topo_t const * topo,
1583 : fd_topo_tile_t const * tile,
1584 : ulong out_fds_cnt,
1585 0 : int * out_fds ) {
1586 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
1587 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
1588 0 : fd_shred_ctx_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof( fd_shred_ctx_t ), sizeof( fd_shred_ctx_t ) );
1589 :
1590 0 : if( FD_UNLIKELY( out_fds_cnt<3UL ) ) FD_LOG_ERR(( "out_fds_cnt %lu", out_fds_cnt ));
1591 :
1592 0 : ulong out_cnt = 0UL;
1593 0 : out_fds[ out_cnt++ ] = 2; /* stderr */
1594 0 : if( FD_LIKELY( -1!=fd_log_private_logfile_fd() ) )
1595 0 : out_fds[ out_cnt++ ] = fd_log_private_logfile_fd(); /* logfile */
1596 0 : if( FD_LIKELY( ctx->disk_fd>=0 ) )
1597 0 : out_fds[ out_cnt++ ] = ctx->disk_fd;
1598 0 : return out_cnt;
1599 0 : }
1600 :
1601 : /* One stem iteration finishes at most one FEC set on shred_out: 32
1602 : data shreds and a completion, after at most an eviction and a shred
1603 : on the network path. PoH input only publishes sign requests, at
1604 : most one per FEC set of a batch. */
1605 0 : #define STEM_BURST FD_SHRED_STEM_BURST
1606 :
1607 : /* See explanation in fd_pack */
1608 0 : #define STEM_LAZY (128L*3000L)
1609 :
1610 : /* When leader, poh_shred carries a stream of one microblock per frag
1611 : and the other links are near empty; keep draining it. */
1612 0 : #define STEM_STICKY_POLL_MAX (16UL)
1613 :
1614 0 : #define STEM_CALLBACK_CONTEXT_TYPE fd_shred_ctx_t
1615 0 : #define STEM_CALLBACK_CONTEXT_ALIGN alignof(fd_shred_ctx_t)
1616 :
1617 0 : #define STEM_CALLBACK_DURING_HOUSEKEEPING during_housekeeping
1618 0 : #define STEM_CALLBACK_METRICS_WRITE metrics_write
1619 0 : #define STEM_CALLBACK_BEFORE_FRAG before_frag
1620 0 : #define STEM_CALLBACK_DURING_FRAG during_frag
1621 0 : #define STEM_CALLBACK_AFTER_FRAG after_frag
1622 :
1623 : #include "../stem/fd_stem.c"
1624 :
1625 : fd_topo_run_tile_t fd_tile_shred = {
1626 : .name = "shred",
1627 : .populate_allowed_seccomp = populate_allowed_seccomp,
1628 : .populate_allowed_fds = populate_allowed_fds,
1629 : .scratch_align = scratch_align,
1630 : .scratch_footprint = scratch_footprint,
1631 : .privileged_init = privileged_init,
1632 : .unprivileged_init = unprivileged_init,
1633 : .run = stem_run,
1634 : };
|
