Line data Source code
1 : #include "fd_stem.h"
2 :
3 : /* fd_stem provides services to multiplex multiple streams of input
4 : fragments and present them to a mix of reliable and unreliable
5 : consumers as though they were generated by multiple different
6 : multi-stream producers. The code can be included to generate
7 : a definition of stem_run which can be called as a tile main run
8 : loop.
9 :
10 : The template supports various callback functions which can be
11 : defined like #define STEM_CALLBACK_BEFORE_FRAG before_frag to
12 : tune the behavior of the stem_run loop. The callbacks are:
13 :
14 : SHOULD_SHUTDOWN
15 : It is called at the beginning of each iteration of the stem run loop,
16 : and if it returns non-zero, the stem will exit the run loop and
17 : return from the stem_run function. This is useful for shutting down
18 : the tile.
19 :
20 : DURING_HOUSEKEEPING
21 : Is called during the housekeeping routine, which happens infrequently
22 : on a schedule determined by the stem (based on the lazy parameter,
23 : see fd_tempo.h for more information). It is appropriate to do
24 : slightly expensive things here that wouldn't be OK to do in the main
25 : loop, like updating sequence numbers that are shared with other tiles
26 : (e.g. synchronization information), or sending batched information
27 : somewhere. The ctx is a user-provided context object from when the
28 : stem was initialized.
29 :
30 : METRICS_WRITE
31 : By convention, tiles may wish to accumulate high traffic metrics
32 : locally so they don't cause a lot of cache coherency traffic, and
33 : then periodically publish them to external observers. This callback
34 : is here to support that use case. It occurs infrequently during the
35 : housekeeping loop, and is called inside a compiler fence to ensure
36 : the writes do not get reordered, which may be important for observers
37 : or monitoring tools. The ctx is a user-provided context object from
38 : when the stem tile was initialized.
39 :
40 : BEFORE_CREDIT
41 : Is called every iteration of the stem run loop, whether there is a
42 : new frag ready to receive or not. This callback is also still
43 : invoked even if the stem is backpressured and cannot read any new
44 : fragments while waiting for downstream consumers to catch up. This
45 : callback is useful for things that need to occur even if no new frags
46 : are being handled. For example, servicing network connections could
47 : happen here. The ctx is a user-provided context object from when the
48 : stem tile was initialized. The stem is the stem which is invoking
49 : this callback. The stem should only be used for calling
50 : fd_stem_publish to publish a fragment to downstream consumers.
51 :
52 : The charge_busy argument is 0 by default, and should be set to 1 if
53 : the before_credit function is doing work that should be accounted for
54 : as part of the tiles busy indicator.
55 :
56 : CHECK_CREDIT
57 : Is called every iteration of the stem run loop. This callback
58 : overrides the default backpressure check (can the run loop produce
59 : STEM_BURST frags on any out link without exceeding flow control
60 : credits).
61 :
62 : AFTER_CREDIT
63 : Is called every iteration of the stem run loop, whether there is a
64 : new frag ready to receive or not, except in cases where the stem is
65 : backpressured by a downstream consumer and would not be able to
66 : publish. The callback might be used for publishing new fragments to
67 : downstream consumers in the main loop which are not in response to an
68 : incoming fragment. For example, code that collects incoming
69 : fragments over a period of 1 second and joins them together before
70 : publishing a large block fragment downstream, would publish the block
71 : here. The ctx is a user-provided context object from when the stem
72 : tile was initialized. The stem is the stem which is invoking this
73 : callback. The stem should only be used for calling fd_stem_publish to
74 : publish a fragment to downstream consumers.
75 :
76 : The opt_poll_in argument determines if the stem should proceed with
77 : checking for new fragments to consumer, or should `continue` the main
78 : stem loop to do credit checking again. This could be used if the
79 : after_credit function publishes, and the flow control needs to be
80 : checked again. By default, opt_poll_in is true and the stem will
81 : poll for fragments right away without rerunning the loop or checking
82 : for credits.
83 :
84 : The charge_busy argument is 0 by default, and should be set to 1 if
85 : the after_credit function is doing work that should be accounted for
86 : as part of the tiles busy indicator.
87 :
88 : BEFORE_FRAG
89 : Is called immediately whenever a new fragment has been detected that
90 : was published by an upstream producer. The signature and sequence
91 : number (sig and seq) provided as arguments are read atomically from
92 : shared memory, so must both match each other from the published
93 : fragment (aka. they will not be torn or partially overwritten).
94 : in_idx is an index in [0, num_ins) indicating which producer
95 : published the fragment. No fragment data has been read yet here, nor
96 : has other metadata, for example the size or timestamps of the
97 : fragment. Mainly this callback is useful for deciding whether to
98 : filter the fragment based on its signature. If the return value is
99 : non-zero, the frag will be skipped completely, no fragment data will
100 : be read, and the in will be advanced so that we now wait for the next
101 : fragment. If the return value is -1, then the frag is returned back
102 : to the message queue and will be reprocessed. The ctx is a
103 : user-provided context object from when the stem tile was initialized.
104 :
105 : DURING_FRAG
106 : Is called after the stem has received a new frag from an in, but
107 : before the stem has checked that it was overrun. This callback is
108 : not invoked if the stem is backpressured, as it would not try and
109 : read a frag from an in in the first place (instead, leaving it on the
110 : in mcache to backpressure the upstream producer). in_idx will be the
111 : index of the in that the frag was received from, skipping any unpolled
112 : links. If the producer of the frags is respecting flow control, it is
113 : safe to read frag data in any of the callbacks, but it is suggested to
114 : copy or read frag data within this callback, as if the producer does
115 : not respect flow control, the frag may be torn or corrupt due to an
116 : overrun by the reader. If the frag being read from has been
117 : overwritten while this callback is running, the frag will be ignored
118 : and the stem will not call the after_frag function. Instead it will
119 : recover from the overrun and continue with new frags. This function
120 : cannot fail. The ctx is a user-provided context object from when the
121 : stem tile was initialized. seq, sig, chunk, and sz are the respective
122 : fields from the mcache fragment that was received. If the producer
123 : is not respecting flow control, these may be corrupt or torn and
124 : should not be trusted, except for seq which is read atomically.
125 :
126 : RETURNABLE_FRAG
127 : Is called after the stem has received a new frag from an in, and
128 : assumes that the stem cannot be overrun. This special callback can
129 : instruct the stem not to advance the input sequence number, and
130 : instead return the fragment to the stem to be processed again. This
131 : is useful for processing partial data from fragments without copying
132 : it. This callback is unsafe in general contexts, since it assumes
133 : that the frag will not be overwritten while the callback is running,
134 : and that the frag data is valid throughout the function call. It
135 : should only be used when the stem is guaranteed to not be overrun.
136 : This callback is not invoked if the stem is backpressured, as it
137 : would not try and read a frag from an in in the first place (instead,
138 : leaving it on the in mcache to backpressure the upstream producer).
139 : in_idx will be the index of the in that the frag was received from.
140 : seq, sig, chunk, and sz are the respective fields from the mcache
141 : fragment that was received. tsorig and tspub are the timestamps of
142 : the fragment that was received, and are read atomically from shared
143 : memory, so must both match each other from the published fragment
144 : (aka. they will not be torn or partially overwritten). The ctx is a
145 : user-provided context object from when the stem tile was initialized.
146 : The callback should return 1 if the fragment was not fully processed
147 : and should be returned to the stem for further processing, or 0 if
148 : the fragment was fully processed and the consumer link should be
149 : advanced.
150 :
151 : AFTER_FRAG
152 : Is called immediately after the DURING_FRAG, along with an additional
153 : check that the reader was not overrun while handling the frag. If
154 : the reader was overrun, the frag is abandoned and this function is
155 : not called. This callback is not invoked if the stem is
156 : backpressured, as it would not read a frag in the first place.
157 : in_idx will be the index of the in that the frag was received from,
158 : skipping any unpolled links. You should not read the frag data directly
159 : here, as it might still get overrun, instead it should be copied out of
160 : the frag during the read callback if needed later. This function cannot
161 : fail. The ctx is a user-provided context object from when the stem tile
162 : was initialized. stem should only be used for calling fd_stem_publish
163 : to publish a fragment to downstream consumers. seq is the sequence
164 : number of the fragment that was read from the input mcache. sig, sz,
165 : tsorig, and tspub are the respective fields from the mcache fragment
166 : that was received. If the producer is not respecting flow control,
167 : these may be corrupt or torn and should not be trusted.
168 :
169 : AFTER_POLL_OVERRUN
170 : Is called when an overrun is detected while polling for new frags.
171 : This callback is not called when an overrun is detected in
172 : during_frag. */
173 :
174 : #include "../../util/log/fd_log.h"
175 : #include "../topo/fd_topo.h"
176 : #include "../metrics/fd_metrics.h"
177 : #include "../../tango/fd_tango.h"
178 :
179 : #ifndef STEM_NAME
180 : #define STEM_NAME stem
181 : #endif
182 0 : #define STEM_(n) FD_EXPAND_THEN_CONCAT3(STEM_NAME,_,n)
183 :
184 : #ifndef STEM_BURST
185 : #error "STEM_BURST must be defined"
186 : #endif
187 :
188 : #ifndef STEM_CALLBACK_CONTEXT_TYPE
189 : #error "STEM_CALLBACK_CONTEXT_TYPE must be defined"
190 : #endif
191 :
192 : #ifndef STEM_CALLBACK_CONTEXT_ALIGN
193 : #error "STEM_CALLBACK_CONTEXT_ALIGN must be defined"
194 : #endif
195 :
196 : #ifndef STEM_LAZY
197 0 : #define STEM_LAZY (0L)
198 : #endif
199 :
200 0 : #define STEM_SHUTDOWN_SEQ (ULONG_MAX-1UL)
201 :
202 : static inline void
203 792 : STEM_(in_update)( fd_stem_tile_in_t * in ) {
204 792 : __atomic_store_n( in->fseq, in->seq, __ATOMIC_RELEASE );
205 :
206 792 : volatile ulong * metrics = fd_metrics_link_in( fd_metrics_base_tl, in->idx );
207 :
208 792 : uint * accum = in->accum;
209 792 : ulong a0 = (ulong)accum[0]; ulong a1 = (ulong)accum[1]; ulong a2 = (ulong)accum[2];
210 792 : ulong a3 = (ulong)accum[3]; ulong a4 = (ulong)accum[4]; ulong a5 = (ulong)accum[5];
211 792 : FD_COMPILER_MFENCE();
212 792 : metrics[0] += a0; metrics[1] += a1; metrics[2] += a2;
213 792 : metrics[3] += a3; metrics[4] += a4; metrics[5] += a5;
214 792 : FD_COMPILER_MFENCE();
215 792 : accum[0] = 0U; accum[1] = 0U; accum[2] = 0U;
216 792 : accum[3] = 0U; accum[4] = 0U; accum[5] = 0U;
217 792 : }
218 :
219 : FD_FN_PURE static inline ulong
220 0 : STEM_(scratch_align)( void ) {
221 0 : return FD_STEM_SCRATCH_ALIGN;
222 0 : }
223 :
224 : FD_FN_PURE static inline ulong
225 : STEM_(scratch_footprint)( ulong in_cnt,
226 : ulong out_cnt,
227 0 : ulong cons_cnt ) {
228 0 : ulong l = FD_LAYOUT_INIT;
229 0 : l = FD_LAYOUT_APPEND( l, alignof(fd_stem_tile_in_t), in_cnt*sizeof(fd_stem_tile_in_t) ); /* in */
230 0 : l = FD_LAYOUT_APPEND( l, alignof(ulong), out_cnt*sizeof(ulong) ); /* cr_avail */
231 0 : l = FD_LAYOUT_APPEND( l, alignof(ulong), out_cnt*sizeof(ulong) ); /* out_depth */
232 0 : l = FD_LAYOUT_APPEND( l, alignof(ulong), out_cnt*sizeof(ulong) ); /* out_seq */
233 0 : l = FD_LAYOUT_APPEND( l, alignof(int), out_cnt*sizeof(int) ); /* out_reliable */
234 0 : l = FD_LAYOUT_APPEND( l, alignof(ulong const *), cons_cnt*sizeof(ulong const *) ); /* cons_fseq */
235 0 : l = FD_LAYOUT_APPEND( l, alignof(ulong *), cons_cnt*sizeof(ulong *) ); /* cons_slow */
236 0 : l = FD_LAYOUT_APPEND( l, alignof(ulong), cons_cnt*sizeof(ulong) ); /* cons_out */
237 0 : l = FD_LAYOUT_APPEND( l, alignof(ulong), cons_cnt*sizeof(ulong) ); /* cons_seq */
238 0 : const ulong event_cnt = in_cnt + 1UL + cons_cnt;
239 0 : l = FD_LAYOUT_APPEND( l, alignof(ushort), event_cnt*sizeof(ushort) ); /* event_map */
240 0 : return FD_LAYOUT_FINI( l, STEM_(scratch_align)() );
241 0 : }
242 :
243 : static inline void
244 : STEM_(run1)( ulong in_cnt,
245 : fd_frag_meta_t const ** in_mcache,
246 : ulong ** in_fseq,
247 : ulong out_cnt,
248 : fd_frag_meta_t ** out_mcache,
249 : ulong cons_cnt,
250 : ulong * _cons_out,
251 : ulong ** _cons_fseq,
252 : volatile ulong ** _cons_slow,
253 : ulong burst,
254 : long lazy,
255 : fd_rng_t * rng,
256 : void * scratch,
257 0 : STEM_CALLBACK_CONTEXT_TYPE * ctx ) {
258 : /* in frag stream state */
259 0 : ulong in_seq; /* current position in input poll sequence, in [0,in_cnt) */
260 0 : fd_stem_tile_in_t * in; /* in[in_seq] for in_seq in [0,in_cnt) has information about input fragment stream currently at
261 : position in_seq in the in_idx polling sequence. The ordering of this array is continuously
262 : shuffled to avoid lighthousing effects in the output fragment stream at extreme fan-in and load */
263 :
264 : /* out frag stream state */
265 0 : ulong * out_depth; /* ==fd_mcache_depth( out_mcache[out_idx] ) for out_idx in [0, out_cnt) */
266 0 : ulong * out_seq; /* next mux frag sequence number to publish for out_idx in [0, out_cnt) ]*/
267 0 : int * out_reliable; /* out_reliable[out_idx] is 1 if out_idx has at least one reliable consumer, else 0 */
268 :
269 : /* out flow control state */
270 0 : ulong * cr_avail; /* number of flow control credits available to publish downstream across all outs */
271 0 : ulong min_cr_avail; /* minimum number of flow control credits available to publish downstream */
272 0 : ulong const ** cons_fseq; /* cons_fseq[cons_idx] for cons_idx in [0,cons_cnt) is where to receive fctl credits from consumers */
273 0 : volatile ulong ** cons_slow; /* cons_slow[cons_idx] for cons_idx in [0,cons_cnt) is where to accumulate slow events */
274 0 : ulong * cons_out; /* cons_out[cons_idx] for cons_idx in [0,cons_ct) is which out the consumer consumes from */
275 0 : ulong * cons_seq; /* cons_seq [cons_idx] is the most recent observation of cons_fseq[cons_idx] */
276 :
277 : /* housekeeping state */
278 0 : ulong event_cnt; /* ==in_cnt+cons_cnt+1, total number of housekeeping events */
279 0 : ulong event_seq; /* current position in housekeeping event sequence, in [0,event_cnt) */
280 0 : ushort * event_map; /* current mapping of event_seq to event idx, event_map[ event_seq ] is next event to process */
281 0 : ulong async_min; /* minimum number of ticks between processing a housekeeping event, positive integer power of 2 */
282 :
283 : /* performance metrics */
284 0 : ulong metric_in_backp; /* is the run loop currently backpressured by one or more of the outs, in [0,1] */
285 0 : ulong metric_backp_cnt; /* Accumulates number of transitions of tile to backpressured between housekeeping events */
286 :
287 0 : ulong metric_regime_ticks[ FD_METRICS_ENUM_TILE_REGIME_CNT ]; /* How many ticks the tile has spent in each regime */
288 :
289 0 : if( FD_UNLIKELY( !scratch ) ) FD_LOG_ERR(( "NULL scratch" ));
290 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)scratch, STEM_(scratch_align)() ) ) ) FD_LOG_ERR(( "misaligned scratch" ));
291 :
292 : /* in_backp==1, backp_cnt==0 indicates waiting for initial credits,
293 : cleared during first housekeeping if credits available */
294 0 : metric_in_backp = 1UL;
295 0 : metric_backp_cnt = 0UL;
296 0 : memset( metric_regime_ticks, 0, sizeof( metric_regime_ticks ) );
297 :
298 : /* in frag stream init */
299 :
300 0 : in_seq = 0UL; /* First in to poll */
301 :
302 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
303 0 : in = (fd_stem_tile_in_t *)FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_stem_tile_in_t), in_cnt*sizeof(fd_stem_tile_in_t) );
304 :
305 0 : if( FD_UNLIKELY( !!in_cnt && !in_mcache ) ) FD_LOG_ERR(( "NULL in_mcache" ));
306 0 : if( FD_UNLIKELY( !!in_cnt && !in_fseq ) ) FD_LOG_ERR(( "NULL in_fseq" ));
307 0 : if( FD_UNLIKELY( in_cnt > UINT_MAX ) ) FD_LOG_ERR(( "in_cnt too large" ));
308 0 : for( ulong in_idx=0UL; in_idx<in_cnt; in_idx++ ) {
309 :
310 0 : if( FD_UNLIKELY( !in_mcache[ in_idx ] ) ) FD_LOG_ERR(( "NULL in_mcache[%lu]", in_idx ));
311 0 : if( FD_UNLIKELY( !in_fseq [ in_idx ] ) ) FD_LOG_ERR(( "NULL in_fseq[%lu]", in_idx ));
312 :
313 0 : fd_stem_tile_in_t * this_in = &in[ in_idx ];
314 :
315 0 : this_in->mcache = in_mcache[ in_idx ];
316 0 : this_in->fseq = in_fseq [ in_idx ];
317 :
318 0 : ulong depth = fd_mcache_depth( this_in->mcache );
319 0 : if( FD_UNLIKELY( depth > UINT_MAX ) ) FD_LOG_ERR(( "in_mcache[%lu] too deep", in_idx ));
320 0 : this_in->depth = (uint)depth;
321 0 : this_in->idx = (uint)in_idx;
322 0 : this_in->seq = 0UL;
323 0 : this_in->mline = this_in->mcache + fd_mcache_line_idx( this_in->seq, this_in->depth );
324 :
325 0 : this_in->accum[0] = 0U; this_in->accum[1] = 0U; this_in->accum[2] = 0U;
326 0 : this_in->accum[3] = 0U; this_in->accum[4] = 0U; this_in->accum[5] = 0U;
327 0 : }
328 :
329 : /* out frag stream init */
330 :
331 0 : cr_avail = (ulong *)FD_SCRATCH_ALLOC_APPEND( l, alignof(ulong), out_cnt*sizeof(ulong) );
332 0 : min_cr_avail = fd_ulong_if( cons_cnt>0UL, 0UL, ULONG_MAX );
333 :
334 0 : out_depth = (ulong *)FD_SCRATCH_ALLOC_APPEND( l, alignof(ulong), out_cnt*sizeof(ulong) );
335 0 : out_seq = (ulong *)FD_SCRATCH_ALLOC_APPEND( l, alignof(ulong), out_cnt*sizeof(ulong) );
336 0 : out_reliable = (int *)FD_SCRATCH_ALLOC_APPEND( l, alignof(int), out_cnt*sizeof(int) );
337 :
338 0 : ulong cr_max = fd_ulong_if( !out_cnt || !cons_cnt, 128UL, ULONG_MAX );
339 :
340 0 : for( ulong out_idx=0UL; out_idx<out_cnt; out_idx++ ) {
341 :
342 0 : if( FD_UNLIKELY( !out_mcache[ out_idx ] ) ) FD_LOG_ERR(( "NULL out_mcache[%lu]", out_idx ));
343 :
344 0 : out_depth[ out_idx ] = fd_mcache_depth( out_mcache[ out_idx ] );
345 0 : out_seq[ out_idx ] = 0UL;
346 :
347 0 : cr_avail[ out_idx ] = out_depth[ out_idx ];
348 0 : out_reliable[ out_idx ] = 0;
349 0 : }
350 :
351 0 : cons_fseq = (ulong const **)FD_SCRATCH_ALLOC_APPEND( l, alignof(ulong const *), cons_cnt*sizeof(ulong const *) );
352 0 : cons_slow = (volatile ulong **)FD_SCRATCH_ALLOC_APPEND( l, alignof(ulong *), cons_cnt*sizeof(ulong *) );
353 0 : cons_out = (ulong *) FD_SCRATCH_ALLOC_APPEND( l, alignof(ulong), cons_cnt*sizeof(ulong) );
354 0 : cons_seq = (ulong *) FD_SCRATCH_ALLOC_APPEND( l, alignof(ulong), cons_cnt*sizeof(ulong) );
355 :
356 0 : if( FD_UNLIKELY( !!cons_cnt && !_cons_fseq ) ) FD_LOG_ERR(( "NULL cons_fseq" ));
357 0 : if( FD_UNLIKELY( !!cons_cnt && !_cons_slow ) ) FD_LOG_ERR(( "NULL cons_slow" ));
358 0 : for( ulong cons_idx=0UL; cons_idx<cons_cnt; cons_idx++ ) {
359 0 : if( FD_UNLIKELY( !_cons_fseq[ cons_idx ] ) ) FD_LOG_ERR(( "NULL cons_fseq[%lu]", cons_idx ));
360 0 : if( FD_UNLIKELY( !_cons_slow[ cons_idx ] ) ) FD_LOG_ERR(( "NULL cons_slow[%lu]", cons_idx ));
361 0 : cons_fseq[ cons_idx ] = _cons_fseq[ cons_idx ];
362 0 : cons_out [ cons_idx ] = _cons_out [ cons_idx ];
363 0 : cons_slow[ cons_idx ] = _cons_slow[ cons_idx ];
364 0 : cons_seq [ cons_idx ] = __atomic_load_n( _cons_fseq[ cons_idx ], __ATOMIC_ACQUIRE );
365 :
366 0 : out_reliable[ cons_out[ cons_idx ] ] = 1;
367 0 : cr_max = fd_ulong_min( cr_max, out_depth[ cons_out[ cons_idx ] ] );
368 0 : }
369 :
370 0 : if( FD_UNLIKELY( cons_cnt>0UL && burst>cr_max ) ) FD_LOG_ERR(( "one or more out links have insufficient depth for STEM_BURST %lu. cr_max is %lu", burst, cr_max ));
371 :
372 : /* housekeeping init */
373 :
374 0 : if( lazy<=0L ) lazy = fd_tempo_lazy_default( cr_max );
375 0 : if( FD_UNLIKELY( lazy>(long)1e9 ) ) FD_LOG_ERR(( "excessive stem lazy value: %li", lazy ));
376 0 : FD_LOG_INFO(( "Configuring housekeeping (lazy %li ns)", lazy ));
377 :
378 : /* Initialize the initial event sequence to immediately update
379 : cr_avail on the first run loop iteration and then update all the
380 : ins accordingly. */
381 :
382 0 : event_cnt = in_cnt + 1UL + cons_cnt;
383 0 : event_map = (ushort *)FD_SCRATCH_ALLOC_APPEND( l, alignof(ushort), event_cnt*sizeof(ushort) );
384 0 : event_seq = 0UL; event_map[ event_seq++ ] = (ushort)cons_cnt;
385 0 : for( ulong in_idx=0UL; in_idx< in_cnt; in_idx++ ) event_map[ event_seq++ ] = (ushort)(in_idx+cons_cnt+1UL);
386 0 : for( ulong cons_idx=0UL; cons_idx<cons_cnt; cons_idx++ ) event_map[ event_seq++ ] = (ushort)cons_idx;
387 0 : event_seq = 0UL;
388 :
389 0 : async_min = fd_tempo_async_min( lazy, event_cnt, (float)fd_tempo_tick_per_ns( NULL ) );
390 0 : if( FD_UNLIKELY( !async_min ) ) FD_LOG_ERR(( "bad lazy %lu %lu", (ulong)lazy, event_cnt ));
391 :
392 0 : FD_LOG_INFO(( "Running stem, cr_max = %lu", cr_max ));
393 0 : FD_MGAUGE_SET( TILE, STATUS, 1UL );
394 0 : long then = fd_tickcount();
395 0 : long now = then;
396 0 : for(;;) {
397 :
398 : #ifdef STEM_CALLBACK_SHOULD_SHUTDOWN
399 0 : if( FD_UNLIKELY( STEM_CALLBACK_SHOULD_SHUTDOWN( ctx ) ) ) break;
400 0 : #endif
401 :
402 : /* Do housekeeping at a low rate in the background */
403 :
404 0 : ulong housekeeping_ticks = 0UL;
405 0 : if( FD_UNLIKELY( (now-then)>=0L ) ) {
406 0 : ulong event_idx = (ulong)event_map[ event_seq ];
407 :
408 : /* Do the next async event. event_idx:
409 : <out_cnt - receive credits from out event_idx
410 : ==out_cnt - housekeeping
411 : >out_cnt - send credits to in event_idx - out_cnt - 1.
412 : Branch hints and order are optimized for the case:
413 : out_cnt >~ in_cnt >~ 1. */
414 :
415 0 : if( FD_LIKELY( event_idx<cons_cnt ) ) { /* cons fctl for cons cons_idx */
416 0 : ulong cons_idx = event_idx;
417 :
418 : /* Receive flow control credits from this out. */
419 0 : ulong this_cons_seq = __atomic_load_n( cons_fseq[ cons_idx ], __ATOMIC_ACQUIRE );
420 0 : cons_seq[ cons_idx ] = this_cons_seq;
421 : #ifdef STEM_CALLBACK_RECV_CREDIT
422 0 : STEM_CALLBACK_RECV_CREDIT( ctx, cons_out[ cons_idx ], out_seq[ cons_out[ cons_idx ] ], this_cons_seq );
423 : #endif
424 :
425 0 : } else if( FD_LIKELY( event_idx>cons_cnt ) ) { /* in fctl for in in_idx */
426 0 : ulong in_idx = event_idx - cons_cnt - 1UL;
427 :
428 : /* Send flow control credits and drain flow control diagnostics
429 : for in_idx. */
430 :
431 0 : STEM_(in_update)( &in[ in_idx ] );
432 :
433 0 : } else { /* event_idx==cons_cnt, housekeeping event */
434 :
435 : /* Update metrics counters to external viewers */
436 0 : FD_COMPILER_MFENCE();
437 0 : FD_MGAUGE_SET( TILE, HEARTBEAT_TIMESTAMP_NANOS, (ulong)fd_log_wallclock() );
438 0 : FD_MGAUGE_SET( TILE, IN_BACKPRESSURE, metric_in_backp );
439 0 : FD_MCNT_INC ( TILE, BACKPRESSURE, metric_backp_cnt );
440 0 : FD_MCNT_ENUM_COPY( TILE, REGIME_DURATION_NANOS, metric_regime_ticks );
441 : #ifdef STEM_CALLBACK_METRICS_WRITE
442 0 : STEM_CALLBACK_METRICS_WRITE( ctx );
443 : #endif
444 0 : FD_COMPILER_MFENCE();
445 0 : metric_backp_cnt = 0UL;
446 :
447 : /* Receive flow control credits */
448 0 : if( FD_LIKELY( cons_cnt ) ) {
449 0 : ulong slowest_cons = ULONG_MAX;
450 0 : min_cr_avail = cr_max;
451 0 : for( ulong out_idx=0; out_idx<out_cnt; out_idx++ ) {
452 0 : cr_avail[ out_idx ] = out_depth[ out_idx ];
453 0 : }
454 :
455 0 : for( ulong cons_idx=0UL; cons_idx<cons_cnt; cons_idx++ ) {
456 0 : ulong out_idx = cons_out[ cons_idx ];
457 :
458 : /* Read the fseq boot value (ULONG_MAX) as sequence 0, not
459 : -1, else the producer is one credit short until the
460 : consumer boots. */
461 0 : ulong cseq = fd_ulong_if( cons_seq[ cons_idx ]==ULONG_MAX, 0UL, cons_seq[ cons_idx ] );
462 0 : ulong cons_cr_avail = (ulong)fd_long_max( (long)out_depth[ out_idx ]-fd_long_max( fd_seq_diff( out_seq[ out_idx ], cseq ), 0L ), 0L );
463 :
464 : /* If a reliable consumer exits, they can set the credit
465 : return fseq to STEM_SHUTDOWN_SEQ to indicate they are no
466 : longer actively consuming. */
467 0 : cons_cr_avail = fd_ulong_if( cons_seq[ cons_idx ]==STEM_SHUTDOWN_SEQ, out_depth[ out_idx ], cons_cr_avail );
468 0 : slowest_cons = fd_ulong_if( cons_cr_avail<min_cr_avail, cons_idx, slowest_cons );
469 :
470 0 : cr_avail[ out_idx ] = fd_ulong_min( cr_avail[ out_idx ], cons_cr_avail );
471 0 : min_cr_avail = fd_ulong_min( cons_cr_avail, min_cr_avail );
472 0 : }
473 :
474 : /* See notes above about use of quasi-atomic diagnostic accum */
475 0 : if( FD_LIKELY( slowest_cons!=ULONG_MAX ) ) {
476 0 : FD_COMPILER_MFENCE();
477 0 : (*cons_slow[ slowest_cons ]) += metric_in_backp;
478 0 : FD_COMPILER_MFENCE();
479 0 : }
480 0 : }
481 :
482 : /* Publish producer progress sync word */
483 0 : for( ulong out_idx=0UL; out_idx<out_cnt; out_idx++ ) {
484 0 : fd_mcache_seq_update( fd_mcache_seq_laddr( out_mcache[ out_idx ] ), out_seq[ out_idx ] );
485 0 : }
486 :
487 : #ifdef STEM_CALLBACK_DURING_HOUSEKEEPING
488 0 : STEM_CALLBACK_DURING_HOUSEKEEPING( ctx );
489 : #else
490 : (void)ctx;
491 : #endif
492 0 : }
493 :
494 : /* Select which event to do next (randomized round robin) and
495 : reload the housekeeping timer. */
496 :
497 0 : event_seq++;
498 0 : if( FD_UNLIKELY( event_seq>=event_cnt ) ) {
499 0 : event_seq = 0UL;
500 :
501 : /* Randomize the order of event processing for the next event
502 : event_cnt events to avoid lighthousing effects causing input
503 : credit starvation at extreme fan in/fan out, extreme in load
504 : and high credit return laziness. */
505 :
506 0 : ulong swap_idx = (ulong)fd_rng_uint_roll( rng, (uint)event_cnt );
507 0 : ushort map_tmp = event_map[ swap_idx ];
508 0 : event_map[ swap_idx ] = event_map[ 0 ];
509 0 : event_map[ 0 ] = map_tmp;
510 :
511 : /* We also do the same with the ins to prevent there being a
512 : correlated order frag origins from different inputs
513 : downstream at extreme fan in and extreme in load. */
514 :
515 0 : if( FD_LIKELY( in_cnt>1UL ) ) {
516 0 : swap_idx = (ulong)fd_rng_uint_roll( rng, (uint)in_cnt );
517 0 : fd_stem_tile_in_t in_tmp;
518 0 : in_tmp = in[ swap_idx ];
519 0 : in[ swap_idx ] = in[ 0 ];
520 0 : in[ 0 ] = in_tmp;
521 0 : }
522 0 : }
523 :
524 : /* Reload housekeeping timer */
525 0 : then = now + (long)fd_tempo_async_reload( rng, async_min );
526 0 : long next = fd_tickcount();
527 0 : housekeeping_ticks = (ulong)(next - now);
528 0 : now = next;
529 0 : }
530 :
531 : #if defined(STEM_CALLBACK_BEFORE_CREDIT) || defined(STEM_CALLBACK_AFTER_CREDIT) || defined(STEM_CALLBACK_AFTER_FRAG) || defined(STEM_CALLBACK_RETURNABLE_FRAG)
532 : fd_stem_context_t stem = {
533 : .mcaches = out_mcache,
534 : .depths = out_depth,
535 : .seqs = out_seq,
536 :
537 : .cr_avail = cr_avail,
538 : .min_cr_avail = &min_cr_avail,
539 : .cr_decrement_amount = fd_ulong_if( out_cnt>0UL, 1UL, 0UL ),
540 : .out_reliable = out_reliable,
541 : .cons_seq = cons_seq,
542 : .in = in
543 : };
544 : #endif
545 :
546 0 : int charge_busy_before = 0;
547 : #ifdef STEM_CALLBACK_BEFORE_CREDIT
548 0 : STEM_CALLBACK_BEFORE_CREDIT( ctx, &stem, &charge_busy_before );
549 : #endif
550 :
551 : /* Check if we are backpressured. If so, count any transition into
552 : a backpressured regime and spin to wait for flow control credits
553 : to return. We don't do a fully atomic update here as it is only
554 : diagnostic and it will still be correct in the usual case where
555 : individual diagnostic counters aren't used by writers in
556 : different threads of execution. We only count the transition
557 : from not backpressured to backpressured. */
558 :
559 0 : int is_backpressured = min_cr_avail<burst;
560 : #ifdef STEM_CALLBACK_CHECK_CREDIT
561 0 : STEM_CALLBACK_CHECK_CREDIT( ctx, &stem, &charge_busy_before, &is_backpressured );
562 : #endif
563 0 : if( FD_UNLIKELY( is_backpressured ) ) {
564 0 : metric_backp_cnt += (ulong)!metric_in_backp;
565 0 : metric_in_backp = 1UL;
566 0 : FD_SPIN_PAUSE();
567 0 : metric_regime_ticks[2] += housekeeping_ticks;
568 0 : long next = fd_tickcount();
569 0 : metric_regime_ticks[5] += (ulong)(next - now);
570 0 : now = next;
571 0 : continue;
572 0 : }
573 0 : metric_in_backp = 0UL;
574 :
575 0 : int charge_busy_after = 0;
576 : #ifdef STEM_CALLBACK_AFTER_CREDIT
577 : int poll_in = 1;
578 0 : STEM_CALLBACK_AFTER_CREDIT( ctx, &stem, &poll_in, &charge_busy_after );
579 0 : if( FD_UNLIKELY( !poll_in ) ) {
580 0 : metric_regime_ticks[1] += housekeeping_ticks;
581 0 : long next = fd_tickcount();
582 0 : metric_regime_ticks[4] += (ulong)(next - now);
583 0 : now = next;
584 0 : continue;
585 0 : }
586 0 : #endif
587 :
588 : /* Select which in to poll next (randomized round robin) */
589 :
590 0 : if( FD_UNLIKELY( !in_cnt ) ) {
591 0 : int was_busy = charge_busy_before+charge_busy_after;
592 0 : metric_regime_ticks[0] += housekeeping_ticks;
593 0 : long next = fd_tickcount();
594 0 : if( FD_UNLIKELY( was_busy ) ) metric_regime_ticks[3] += (ulong)(next - now);
595 0 : else metric_regime_ticks[6] += (ulong)(next - now);
596 0 : now = next;
597 0 : continue;
598 0 : }
599 :
600 0 : ulong prefrag_ticks = 0UL;
601 : #if defined(STEM_CALLBACK_BEFORE_CREDIT) && defined(STEM_CALLBACK_AFTER_CREDIT)
602 0 : if( FD_LIKELY( charge_busy_before || charge_busy_after ) ) {
603 : #elif defined(STEM_CALLBACK_BEFORE_CREDIT)
604 0 : if( FD_LIKELY( charge_busy_before ) ) {
605 : #elif defined(STEM_CALLBACK_AFTER_CREDIT)
606 0 : if( FD_LIKELY( charge_busy_after ) ) {
607 0 : #endif
608 :
609 : #if defined(STEM_CALLBACK_BEFORE_CREDIT) || defined(STEM_CALLBACK_AFTER_CREDIT)
610 0 : long prefrag_next = fd_tickcount();
611 0 : prefrag_ticks = (ulong)(prefrag_next - now);
612 0 : now = prefrag_next;
613 0 : }
614 : #endif
615 :
616 0 : fd_stem_tile_in_t * this_in = &in[ in_seq ];
617 0 : in_seq++;
618 0 : if( in_seq>=in_cnt ) in_seq = 0UL; /* cmov */
619 :
620 : /* Check if this in has any new fragments to mux */
621 :
622 0 : ulong this_in_seq = this_in->seq;
623 0 : fd_frag_meta_t const * this_in_mline = this_in->mline; /* Already at appropriate line for this_in_seq */
624 :
625 0 : #if FD_HAS_AVX
626 0 : fd_frag_meta_v256_t yline = FD_VOLATILE_CONST( this_in_mline->avx );
627 0 : ulong seq_found = fd_frag_meta_avx_seq( yline );
628 0 : ulong sig = fd_frag_meta_avx_sig( yline );
629 : #elif FD_HAS_SSE
630 : __m128i seq_sig = fd_frag_meta_seq_sig_query( this_in_mline );
631 : ulong seq_found = fd_frag_meta_sse0_seq( seq_sig );
632 : ulong sig = fd_frag_meta_sse0_sig( seq_sig );
633 : #elif FD_HAS_ARM
634 : ulong seq_found = __atomic_load_n( &this_in_mline->seq, __ATOMIC_ACQUIRE );
635 : ulong sig;
636 : #else
637 : /* Without 128-bit atomic load, seq and sig might be read from
638 : different frags (due to overrun), which results in a before_frag
639 : and during_frag being issued with incorrect arguments, but not
640 : after_frag. */
641 : ulong seq_found = FD_VOLATILE_CONST( this_in_mline->seq );
642 : ulong sig = FD_VOLATILE_CONST( this_in_mline->sig );
643 : #endif
644 0 : long diff = fd_seq_diff( this_in_seq, seq_found );
645 0 : if( FD_UNLIKELY( diff ) ) { /* Caught up or overrun, optimize for new frag case */
646 0 : ulong * housekeeping_regime = &metric_regime_ticks[0];
647 0 : ulong * prefrag_regime = &metric_regime_ticks[3];
648 0 : ulong * finish_regime = &metric_regime_ticks[6];
649 0 : if( FD_UNLIKELY( diff<0L ) ) { /* Overrun (impossible if in is honoring our flow control) */
650 0 : this_in->seq = seq_found; /* Resume from here (probably reasonably current, could query in mcache sync directly instead) */
651 0 : housekeeping_regime = &metric_regime_ticks[1];
652 0 : prefrag_regime = &metric_regime_ticks[4];
653 0 : finish_regime = &metric_regime_ticks[7];
654 0 : this_in->accum[ FD_METRICS_COUNTER_LINK_LINK_POLLING_OVERRUN_OFF ]++;
655 0 : this_in->accum[ FD_METRICS_COUNTER_LINK_FRAG_POLLING_OVERRUN_OFF ] += (uint)(-diff);
656 :
657 : #ifdef STEM_CALLBACK_AFTER_POLL_OVERRUN
658 0 : STEM_CALLBACK_AFTER_POLL_OVERRUN( ctx );
659 : #endif
660 0 : }
661 :
662 : /* Don't bother with spin as polling multiple locations */
663 0 : *housekeeping_regime += housekeeping_ticks;
664 0 : *prefrag_regime += prefrag_ticks;
665 0 : long next = fd_tickcount();
666 0 : *finish_regime += (ulong)(next - now);
667 0 : now = next;
668 0 : continue;
669 0 : }
670 :
671 : #if FD_HAS_ARM
672 : /* arm requires a double take to avoid 'dmb ishld' on every frag */
673 : sig = __atomic_load_n( &this_in_mline->sig, __ATOMIC_ACQUIRE );
674 : ulong seq_confirm = __atomic_load_n( &this_in_mline->seq, __ATOMIC_ACQUIRE );
675 : if( FD_UNLIKELY( fd_seq_ne( seq_confirm, seq_found ) ) ) {
676 : metric_regime_ticks[1] += housekeeping_ticks;
677 : metric_regime_ticks[4] += prefrag_ticks;
678 : long next = fd_tickcount();
679 : metric_regime_ticks[7] += (ulong)(next - now);
680 : now = next;
681 : continue;
682 : }
683 : #endif
684 0 : (void)sig;
685 :
686 : #ifdef STEM_CALLBACK_BEFORE_FRAG
687 0 : int filter = STEM_CALLBACK_BEFORE_FRAG( ctx, (ulong)this_in->idx, seq_found, sig );
688 0 : if( FD_UNLIKELY( filter<0 ) ) {
689 0 : metric_regime_ticks[1] += housekeeping_ticks;
690 0 : metric_regime_ticks[4] += prefrag_ticks;
691 0 : long next = fd_tickcount();
692 0 : metric_regime_ticks[7] += (ulong)(next - now);
693 0 : now = next;
694 0 : continue;
695 0 : } else if( FD_UNLIKELY( filter>0 ) ) {
696 0 : this_in->accum[ FD_METRICS_COUNTER_LINK_FRAG_FILTERED_OFF ]++;
697 0 : this_in->accum[ FD_METRICS_COUNTER_LINK_FRAG_FILTERED_BYTES_OFF ] += (uint)this_in_mline->sz; /* TODO: This might be overrun ... ? Not loaded atomically */
698 :
699 : this_in_seq = fd_seq_inc( this_in_seq, 1UL );
700 : this_in->seq = this_in_seq;
701 : this_in->mline = this_in->mcache + fd_mcache_line_idx( this_in_seq, this_in->depth );
702 :
703 0 : metric_regime_ticks[1] += housekeeping_ticks;
704 0 : metric_regime_ticks[4] += prefrag_ticks;
705 0 : long next = fd_tickcount();
706 0 : metric_regime_ticks[7] += (ulong)(next - now);
707 0 : now = next;
708 0 : continue;
709 0 : }
710 0 : #endif
711 :
712 : /* We have a new fragment to mux. Try to load it. This attempt
713 : should always be successful if in producers are honoring our flow
714 : control. Since we can cheaply detect if there are
715 : misconfigurations (should be an L1 cache hit / predictable branch
716 : in the properly configured case), we do so anyway. */
717 0 : FD_COMPILER_MFENCE();
718 0 : #if FD_HAS_AVX
719 0 : ulong chunk = fd_frag_meta_avx_chunk ( yline ); (void)chunk;
720 0 : ulong sz = fd_frag_meta_avx_sz ( yline ); (void)sz;
721 0 : ulong ctl = fd_frag_meta_avx_ctl ( yline ); (void)ctl;
722 0 : ulong tsorig = fd_frag_meta_avx_tsorig( yline ); (void)tsorig;
723 0 : ulong tspub = fd_frag_meta_avx_tspub ( yline ); (void)tspub;
724 : #elif FD_HAS_ARM
725 : ulong ul2, ul3;
726 : fd_arm_ldp16( this_in_mline->ul+2, ul2, ul3 );
727 : ulong chunk = fd_frag_meta_ul2_chunk ( ul2 ); (void)chunk;
728 : ulong sz = fd_frag_meta_ul2_sz ( ul2 ); (void)sz;
729 : ulong ctl = fd_frag_meta_ul2_ctl ( ul2 ); (void)ctl;
730 : ulong tsorig = fd_frag_meta_ul3_tsorig( ul3 ); (void)tsorig;
731 : ulong tspub = fd_frag_meta_ul3_tspub ( ul3 ); (void)tspub;
732 : #else
733 : ulong chunk = (ulong)this_in_mline->chunk; (void)chunk;
734 : ulong sz = (ulong)this_in_mline->sz; (void)sz;
735 : ulong ctl = (ulong)this_in_mline->ctl; (void)ctl;
736 : ulong tsorig = (ulong)this_in_mline->tsorig; (void)tsorig;
737 : ulong tspub = (ulong)this_in_mline->tspub; (void)tspub;
738 : #endif
739 :
740 : #ifdef STEM_CALLBACK_DURING_FRAG
741 0 : STEM_CALLBACK_DURING_FRAG( ctx, (ulong)this_in->idx, seq_found, sig, chunk, sz, ctl );
742 : #endif
743 :
744 0 : FD_HW_MFENCE_LD();
745 0 : ulong seq_test = FD_VOLATILE_CONST( this_in_mline->seq );
746 0 : FD_COMPILER_MFENCE();
747 :
748 0 : if( FD_UNLIKELY( fd_seq_ne( seq_test, seq_found ) ) ) { /* Overrun while reading (impossible if this_in honoring our fctl) */
749 0 : this_in->seq = seq_test; /* Resume from here (probably reasonably current, could query in mcache sync instead) */
750 0 : fd_metrics_link_in( fd_metrics_base_tl, this_in->idx )[ FD_METRICS_COUNTER_LINK_LINK_READING_OVERRUN_OFF ]++; /* No local accum since extremely rare, faster to use smaller cache line */
751 0 : fd_metrics_link_in( fd_metrics_base_tl, this_in->idx )[ FD_METRICS_COUNTER_LINK_FRAG_READING_OVERRUN_OFF ] += (uint)fd_seq_diff( seq_test, seq_found ); /* No local accum since extremely rare, faster to use smaller cache line */
752 : /* Don't bother with spin as polling multiple locations */
753 0 : metric_regime_ticks[1] += housekeeping_ticks;
754 0 : metric_regime_ticks[4] += prefrag_ticks;
755 0 : long next = fd_tickcount();
756 0 : metric_regime_ticks[7] += (ulong)(next - now);
757 0 : now = next;
758 0 : continue;
759 0 : }
760 :
761 : #ifdef STEM_CALLBACK_RETURNABLE_FRAG
762 0 : int return_frag = STEM_CALLBACK_RETURNABLE_FRAG( ctx, (ulong)this_in->idx, seq_found, sig, chunk, sz, ctl, tsorig, tspub, &stem );
763 0 : if( FD_UNLIKELY( return_frag ) ) {
764 0 : metric_regime_ticks[1] += housekeeping_ticks;
765 0 : metric_regime_ticks[4] += prefrag_ticks;
766 0 : long next = fd_tickcount();
767 0 : metric_regime_ticks[7] += (ulong)(next - now);
768 0 : now = next;
769 0 : continue;
770 0 : }
771 0 : #endif
772 :
773 : #ifdef STEM_CALLBACK_AFTER_FRAG
774 0 : STEM_CALLBACK_AFTER_FRAG( ctx, (ulong)this_in->idx, seq_found, sig, sz, tsorig, tspub, &stem );
775 0 : #endif
776 :
777 : /* Windup for the next in poll and accumulate diagnostics */
778 :
779 0 : this_in_seq = fd_seq_inc( this_in_seq, 1UL );
780 0 : this_in->seq = this_in_seq;
781 0 : this_in->mline = this_in->mcache + fd_mcache_line_idx( this_in_seq, this_in->depth );
782 :
783 0 : this_in->accum[ FD_METRICS_COUNTER_LINK_FRAG_CONSUMED_OFF ]++;
784 0 : this_in->accum[ FD_METRICS_COUNTER_LINK_FRAG_CONSUMED_BYTES_OFF ] += (uint)sz;
785 :
786 0 : metric_regime_ticks[1] += housekeeping_ticks;
787 0 : metric_regime_ticks[4] += prefrag_ticks;
788 0 : long next = fd_tickcount();
789 0 : metric_regime_ticks[7] += (ulong)(next - now);
790 0 : now = next;
791 0 : }
792 0 : }
793 :
794 : FD_FN_UNUSED static void
795 : STEM_(run)( fd_topo_t * topo,
796 0 : fd_topo_tile_t * tile ) {
797 0 : const fd_frag_meta_t * in_mcache[ FD_TOPO_MAX_LINKS ];
798 0 : ulong * in_fseq[ FD_TOPO_MAX_TILE_IN_LINKS ];
799 :
800 0 : ulong polled_in_cnt = 0UL;
801 0 : for( ulong i=0UL; i<tile->in_cnt; i++ ) {
802 0 : if( FD_UNLIKELY( !tile->in_link_poll[ i ] ) ) continue;
803 :
804 0 : in_mcache[ polled_in_cnt ] = topo->links[ tile->in_link_id[ i ] ].mcache;
805 0 : FD_TEST( in_mcache[ polled_in_cnt ] );
806 0 : in_fseq[ polled_in_cnt ] = tile->in_link_fseq[ i ];
807 0 : FD_TEST( in_fseq[ polled_in_cnt ] );
808 0 : polled_in_cnt += 1;
809 0 : }
810 :
811 0 : fd_frag_meta_t * out_mcache[ FD_TOPO_MAX_LINKS ];
812 0 : for( ulong i=0UL; i<tile->out_cnt; i++ ) {
813 0 : out_mcache[ i ] = topo->links[ tile->out_link_id[ i ] ].mcache;
814 0 : FD_TEST( out_mcache[ i ] );
815 0 : }
816 :
817 0 : ulong reliable_cons_cnt = 0UL;
818 0 : ulong cons_out[ FD_TOPO_MAX_LINKS ];
819 0 : ulong * cons_fseq[ FD_TOPO_MAX_LINKS ];
820 0 : volatile ulong * cons_slow[ FD_TOPO_MAX_LINKS ];
821 0 : for( ulong i=0UL; i<topo->tile_cnt; i++ ) {
822 0 : fd_topo_tile_t * consumer_tile = &topo->tiles[ i ];
823 0 : ulong polled_in_idx = 0UL;
824 0 : for( ulong j=0UL; j<consumer_tile->in_cnt; j++ ) {
825 0 : int is_polled = consumer_tile->in_link_poll[ j ];
826 0 : for( ulong k=0UL; k<tile->out_cnt; k++ ) {
827 0 : if( FD_UNLIKELY( consumer_tile->in_link_id[ j ]==tile->out_link_id[ k ] && consumer_tile->in_link_reliable[ j ] ) ) {
828 0 : cons_out[ reliable_cons_cnt ] = k;
829 0 : cons_fseq[ reliable_cons_cnt ] = consumer_tile->in_link_fseq[ j ];
830 0 : FD_TEST( cons_fseq[ reliable_cons_cnt ] );
831 0 : cons_slow[ reliable_cons_cnt ] = fd_metrics_link_in( consumer_tile->metrics, polled_in_idx ) + FD_METRICS_COUNTER_LINK_SLOW_OFF;
832 0 : reliable_cons_cnt++;
833 : /* Need to test this, since each link may connect to many outs,
834 : you could construct a topology which has more than this
835 : consumers of links. */
836 0 : FD_TEST( reliable_cons_cnt<FD_TOPO_MAX_LINKS );
837 0 : }
838 0 : }
839 0 : if( FD_LIKELY( is_polled ) ) polled_in_idx++;
840 0 : }
841 0 : }
842 :
843 : /* The stem rng is only used internally for shuffling event/input
844 : polling ordering and for setting the housekeeping timer. It is
845 : never exposed to tile callbacks. As a result, no cryptographic
846 : quality is needed. fd_tickcount() provides per-run entropy,
847 : tile->id guarantees per-tile uniqueness, and fd_ulong_hash (a
848 : Murmur3 finalizer) gives near-perfect avalanche so even a 1-bit
849 : input difference flips ~50% of output bits. fd_rng_secure is not
850 : used here because STEM_(run) executes after the seccomp sandbox
851 : is applied. */
852 0 : fd_rng_t rng[1];
853 0 : FD_TEST( fd_rng_join( fd_rng_new( rng, (uint)fd_ulong_hash( (ulong)fd_tickcount() + tile->id ), 0UL ) ) );
854 :
855 0 : STEM_CALLBACK_CONTEXT_TYPE * ctx = (STEM_CALLBACK_CONTEXT_TYPE*)fd_ulong_align_up( (ulong)fd_topo_obj_laddr( topo, tile->tile_obj_id ), STEM_CALLBACK_CONTEXT_ALIGN );
856 :
857 0 : uchar __attribute__((aligned(FD_STEM_SCRATCH_ALIGN))) stem_scratch[ STEM_(scratch_footprint)( polled_in_cnt, tile->out_cnt, reliable_cons_cnt ) ];
858 :
859 0 : STEM_(run1)( polled_in_cnt,
860 0 : in_mcache,
861 0 : in_fseq,
862 0 : tile->out_cnt,
863 0 : out_mcache,
864 0 : reliable_cons_cnt,
865 0 : cons_out,
866 0 : cons_fseq,
867 0 : cons_slow,
868 0 : STEM_BURST,
869 0 : STEM_LAZY,
870 0 : rng,
871 0 : stem_scratch,
872 0 : ctx );
873 :
874 : #ifdef STEM_CALLBACK_METRICS_WRITE
875 : /* Write final metrics state before shutting down */
876 0 : FD_COMPILER_MFENCE();
877 0 : STEM_CALLBACK_METRICS_WRITE( ctx );
878 0 : FD_COMPILER_MFENCE();
879 : #endif
880 :
881 0 : if( FD_LIKELY( tile->allow_shutdown ) ) {
882 0 : for( ulong i=0UL; i<tile->in_cnt; i++ ) {
883 0 : if( FD_UNLIKELY( !tile->in_link_poll[ i ] || !tile->in_link_reliable[ i ] ) ) continue;
884 :
885 : /* Return infinite credits on any reliable consumer links so that
886 : producers now no longer expect us to consume. */
887 0 : ulong fseq_id = tile->in_link_fseq_obj_id[ i ];
888 0 : ulong * fseq = fd_fseq_join( fd_topo_obj_laddr( topo, fseq_id ) );
889 0 : FD_TEST( fseq );
890 0 : __atomic_store_n( fseq, STEM_SHUTDOWN_SEQ, __ATOMIC_RELEASE );
891 0 : }
892 0 : }
893 0 : }
894 :
895 : #undef STEM_NAME
896 : #undef STEM_
897 : #undef STEM_BURST
898 : #undef STEM_CALLBACK_CONTEXT_TYPE
899 : #undef STEM_CALLBACK_CONTEXT_ALIGN
900 : #undef STEM_LAZY
901 : #undef STEM_CALLBACK_SHOULD_SHUTDOWN
902 : #undef STEM_CALLBACK_DURING_HOUSEKEEPING
903 : #undef STEM_CALLBACK_METRICS_WRITE
904 : #undef STEM_CALLBACK_BEFORE_CREDIT
905 : #undef STEM_CALLBACK_AFTER_CREDIT
906 : #undef STEM_CALLBACK_BEFORE_FRAG
907 : #undef STEM_CALLBACK_DURING_FRAG
908 : #undef STEM_CALLBACK_RETURNABLE_FRAG
909 : #undef STEM_CALLBACK_AFTER_FRAG
910 : #undef STEM_CALLBACK_AFTER_POLL_OVERRUN
|
