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