Line data Source code
1 : static inline __attribute__((always_inline)) void
2 : STEM_(STEM_RUN1_NAME)( ulong in_cnt,
3 : fd_frag_meta_t const ** in_mcache,
4 : ulong ** in_fseq,
5 : ulong out_cnt,
6 : fd_frag_meta_t ** out_mcache,
7 : ulong cons_cnt,
8 : ulong * _cons_out,
9 : ulong ** _cons_fseq,
10 : volatile ulong ** _cons_slow,
11 : ulong burst,
12 : long lazy,
13 : fd_rng_t * rng,
14 : void * scratch,
15 : STEM_CALLBACK_CONTEXT_TYPE * ctx,
16 0 : fd_stem_sleep_t const * sleep ) {
17 0 : (void)sleep; /* unread by a tile with no publish context and no park code */
18 : /* in frag stream state */
19 0 : ulong in_seq; /* current position in input poll sequence, in [0,in_cnt) */
20 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
21 : position in_seq in the in_idx polling sequence. The ordering of this array is continuously
22 : shuffled to avoid lighthousing effects in the output fragment stream at extreme fan-in and load */
23 :
24 : /* out frag stream state */
25 0 : ulong * out_depth; /* ==fd_mcache_depth( out_mcache[out_idx] ) for out_idx in [0, out_cnt) */
26 0 : ulong * out_seq; /* next mux frag sequence number to publish for out_idx in [0, out_cnt) ]*/
27 0 : ulong * out_seq_pub; /* out_seq[out_idx] as last published to the sync word and seq_mirror (0 before the first) */
28 0 : int * out_reliable; /* out_reliable[out_idx] is 1 if out_idx has at least one reliable consumer, else 0 */
29 :
30 : /* out flow control state */
31 0 : ulong * cr_avail; /* number of flow control credits available to publish downstream across all outs */
32 0 : ulong min_cr_avail; /* minimum number of flow control credits available to publish downstream */
33 0 : ulong const ** cons_fseq; /* cons_fseq[cons_idx] for cons_idx in [0,cons_cnt) is where to receive fctl credits from consumers */
34 0 : volatile ulong ** cons_slow; /* cons_slow[cons_idx] for cons_idx in [0,cons_cnt) is where to accumulate slow events */
35 0 : ulong * cons_out; /* cons_out[cons_idx] for cons_idx in [0,cons_ct) is which out the consumer consumes from */
36 0 : ulong * cons_seq; /* cons_seq [cons_idx] is the most recent observation of cons_fseq[cons_idx] */
37 :
38 : /* housekeeping state */
39 0 : ulong event_cnt; /* ==in_cnt+cons_cnt+1, total number of housekeeping events */
40 0 : ulong event_seq; /* current position in housekeeping event sequence, in [0,event_cnt) */
41 0 : ushort * event_map; /* current mapping of event_seq to event idx, event_map[ event_seq ] is next event to process */
42 0 : ulong async_min; /* minimum number of ticks between processing a housekeeping event, positive integer power of 2 */
43 :
44 : /* performance metrics */
45 0 : ulong metric_in_backp; /* is the run loop currently backpressured by one or more of the outs, in [0,1] */
46 0 : ulong metric_backp_cnt; /* Accumulates number of transitions of tile to backpressured between housekeeping events */
47 :
48 0 : ulong metric_regime_ticks[ FD_METRICS_ENUM_TILE_REGIME_CNT ]; /* How many ticks the tile has spent in each regime */
49 :
50 : #if STEM_SLEEP_PARKS
51 : double sleep_tick_per_ns = fd_tempo_tick_per_ns( NULL );
52 0 : long sleep_linger_ticks = (long)((double)FD_SLEEP_LINGER_NS *sleep_tick_per_ns);
53 0 : long sleep_cap_ticks = (long)((double)FD_SLEEP_PARK_CAP_NS*sleep_tick_per_ns);
54 0 : long sleep_min_ticks = (long)((double)FD_SLEEP_PARK_MIN_NS*sleep_tick_per_ns);
55 : ulong sleep_idle_streak = 0UL;
56 : #endif
57 :
58 0 : if( FD_UNLIKELY( !scratch ) ) FD_LOG_ERR(( "NULL scratch" ));
59 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)scratch, STEM_(scratch_align)() ) ) ) FD_LOG_ERR(( "misaligned scratch" ));
60 :
61 : /* in_backp==1, backp_cnt==0 indicates waiting for initial credits,
62 : cleared during first housekeeping if credits available */
63 0 : metric_in_backp = 1UL;
64 0 : metric_backp_cnt = 0UL;
65 0 : memset( metric_regime_ticks, 0, sizeof( metric_regime_ticks ) );
66 :
67 : /* in frag stream init */
68 :
69 0 : in_seq = 0UL; /* First in to poll */
70 : #ifdef STEM_STICKY_POLL_MAX
71 : fd_stem_tile_in_t * sticky_in = NULL;
72 : ulong sticky_rem = 0UL;
73 : #endif
74 :
75 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
76 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) );
77 :
78 0 : if( FD_UNLIKELY( !!in_cnt && !in_mcache ) ) FD_LOG_ERR(( "NULL in_mcache" ));
79 0 : if( FD_UNLIKELY( !!in_cnt && !in_fseq ) ) FD_LOG_ERR(( "NULL in_fseq" ));
80 0 : if( FD_UNLIKELY( in_cnt > UINT_MAX ) ) FD_LOG_ERR(( "in_cnt too large" ));
81 0 : for( ulong in_idx=0UL; in_idx<in_cnt; in_idx++ ) {
82 :
83 0 : if( FD_UNLIKELY( !in_mcache[ in_idx ] ) ) FD_LOG_ERR(( "NULL in_mcache[%lu]", in_idx ));
84 0 : if( FD_UNLIKELY( !in_fseq [ in_idx ] ) ) FD_LOG_ERR(( "NULL in_fseq[%lu]", in_idx ));
85 :
86 0 : fd_stem_tile_in_t * this_in = &in[ in_idx ];
87 :
88 0 : this_in->mcache = in_mcache[ in_idx ];
89 0 : this_in->fseq = in_fseq [ in_idx ];
90 :
91 0 : ulong depth = fd_mcache_depth( this_in->mcache );
92 0 : if( FD_UNLIKELY( depth > UINT_MAX ) ) FD_LOG_ERR(( "in_mcache[%lu] too deep", in_idx ));
93 0 : this_in->depth = (uint)depth;
94 0 : this_in->idx = (uint)in_idx;
95 0 : this_in->seq = 0UL;
96 0 : this_in->mline = this_in->mcache + fd_mcache_line_idx( this_in->seq, this_in->depth );
97 :
98 0 : this_in->accum[0] = 0U; this_in->accum[1] = 0U; this_in->accum[2] = 0U;
99 0 : this_in->accum[3] = 0U; this_in->accum[4] = 0U; this_in->accum[5] = 0U;
100 0 : }
101 :
102 : /* out frag stream init */
103 :
104 0 : cr_avail = (ulong *)FD_SCRATCH_ALLOC_APPEND( l, alignof(ulong), out_cnt*sizeof(ulong) );
105 0 : min_cr_avail = fd_ulong_if( cons_cnt>0UL, 0UL, ULONG_MAX );
106 :
107 0 : out_depth = (ulong *)FD_SCRATCH_ALLOC_APPEND( l, alignof(ulong), out_cnt*sizeof(ulong) );
108 0 : out_seq = (ulong *)FD_SCRATCH_ALLOC_APPEND( l, alignof(ulong), out_cnt*sizeof(ulong) );
109 0 : out_seq_pub = (ulong *)FD_SCRATCH_ALLOC_APPEND( l, alignof(ulong), out_cnt*sizeof(ulong) );
110 0 : out_reliable = (int *)FD_SCRATCH_ALLOC_APPEND( l, alignof(int), out_cnt*sizeof(int) );
111 :
112 0 : ulong cr_max = fd_ulong_if( !out_cnt || !cons_cnt, 128UL, ULONG_MAX );
113 :
114 0 : for( ulong out_idx=0UL; out_idx<out_cnt; out_idx++ ) {
115 :
116 0 : if( FD_UNLIKELY( !out_mcache[ out_idx ] ) ) FD_LOG_ERR(( "NULL out_mcache[%lu]", out_idx ));
117 :
118 0 : out_depth[ out_idx ] = fd_mcache_depth( out_mcache[ out_idx ] );
119 0 : out_seq[ out_idx ] = 0UL;
120 0 : out_seq_pub[ out_idx ] = 0UL; /* ==out_seq: an out never published is never flushed (another client may own it) */
121 :
122 0 : cr_avail[ out_idx ] = out_depth[ out_idx ];
123 0 : out_reliable[ out_idx ] = 0;
124 0 : }
125 :
126 0 : cons_fseq = (ulong const **)FD_SCRATCH_ALLOC_APPEND( l, alignof(ulong const *), cons_cnt*sizeof(ulong const *) );
127 0 : cons_slow = (volatile ulong **)FD_SCRATCH_ALLOC_APPEND( l, alignof(ulong *), cons_cnt*sizeof(ulong *) );
128 0 : cons_out = (ulong *) FD_SCRATCH_ALLOC_APPEND( l, alignof(ulong), cons_cnt*sizeof(ulong) );
129 0 : cons_seq = (ulong *) FD_SCRATCH_ALLOC_APPEND( l, alignof(ulong), cons_cnt*sizeof(ulong) );
130 :
131 0 : if( FD_UNLIKELY( !!cons_cnt && !_cons_fseq ) ) FD_LOG_ERR(( "NULL cons_fseq" ));
132 0 : if( FD_UNLIKELY( !!cons_cnt && !_cons_slow ) ) FD_LOG_ERR(( "NULL cons_slow" ));
133 0 : for( ulong cons_idx=0UL; cons_idx<cons_cnt; cons_idx++ ) {
134 0 : if( FD_UNLIKELY( !_cons_fseq[ cons_idx ] ) ) FD_LOG_ERR(( "NULL cons_fseq[%lu]", cons_idx ));
135 0 : if( FD_UNLIKELY( !_cons_slow[ cons_idx ] ) ) FD_LOG_ERR(( "NULL cons_slow[%lu]", cons_idx ));
136 0 : cons_fseq[ cons_idx ] = _cons_fseq[ cons_idx ];
137 0 : cons_out [ cons_idx ] = _cons_out [ cons_idx ];
138 0 : cons_slow[ cons_idx ] = _cons_slow[ cons_idx ];
139 0 : cons_seq [ cons_idx ] = __atomic_load_n( _cons_fseq[ cons_idx ], __ATOMIC_ACQUIRE );
140 :
141 0 : out_reliable[ cons_out[ cons_idx ] ] = 1;
142 0 : cr_max = fd_ulong_min( cr_max, out_depth[ cons_out[ cons_idx ] ] );
143 0 : }
144 :
145 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 ));
146 :
147 : /* housekeeping init */
148 :
149 0 : if( lazy<=0L ) lazy = fd_tempo_lazy_default( cr_max );
150 0 : if( FD_UNLIKELY( lazy>(long)1e9 ) ) FD_LOG_ERR(( "excessive stem lazy value: %li", lazy ));
151 0 : FD_LOG_INFO(( "Configuring housekeeping (lazy %li ns)", lazy ));
152 :
153 : /* Initialize the initial event sequence to immediately update
154 : cr_avail on the first run loop iteration and then update all the
155 : ins accordingly. */
156 :
157 0 : event_cnt = in_cnt + 1UL + cons_cnt;
158 0 : event_map = (ushort *)FD_SCRATCH_ALLOC_APPEND( l, alignof(ushort), event_cnt*sizeof(ushort) );
159 0 : event_seq = 0UL; event_map[ event_seq++ ] = (ushort)cons_cnt;
160 0 : for( ulong in_idx=0UL; in_idx< in_cnt; in_idx++ ) event_map[ event_seq++ ] = (ushort)(in_idx+cons_cnt+1UL);
161 0 : for( ulong cons_idx=0UL; cons_idx<cons_cnt; cons_idx++ ) event_map[ event_seq++ ] = (ushort)cons_idx;
162 0 : event_seq = 0UL;
163 :
164 0 : async_min = fd_tempo_async_min( lazy, event_cnt, (float)fd_tempo_tick_per_ns( NULL ) );
165 0 : if( FD_UNLIKELY( !async_min ) ) FD_LOG_ERR(( "bad lazy %lu %lu", (ulong)lazy, event_cnt ));
166 :
167 0 : FD_LOG_INFO(( "Running stem, cr_max = %lu", cr_max ));
168 0 : FD_MGAUGE_SET( TILE, STATE, 1UL );
169 0 : long then = fd_tickcount();
170 0 : long now = then;
171 :
172 : #if defined(STEM_CALLBACK_BEFORE_CREDIT) || defined(STEM_CALLBACK_AFTER_CREDIT) || defined(STEM_CALLBACK_AFTER_FRAG) || defined(STEM_CALLBACK_RETURNABLE_FRAG)
173 : fd_stem_context_t stem = {
174 : .mcaches = out_mcache,
175 : .depths = out_depth,
176 : .seqs = out_seq,
177 : .now = now,
178 : .then = then,
179 :
180 : .cr_avail = cr_avail,
181 : .min_cr_avail = &min_cr_avail,
182 : .cr_decrement_amount = fd_ulong_if( out_cnt>0UL, 1UL, 0UL ),
183 : .out_reliable = out_reliable,
184 : .cons_seq = cons_seq,
185 : .in = in,
186 :
187 : .sleep = sleep->shmem,
188 : .wake = sleep->wake,
189 : .wake_off = sleep->wake_off,
190 : .in_fseq = in_fseq,
191 : .in_producer = sleep->in_producer,
192 : };
193 : #endif
194 :
195 : #if STEM_SLEEP_PARKS
196 : long linger_start = then;
197 : long hk_ticks = (long)(async_min*event_cnt); /* ~lazy */
198 : long hk_due = then; /* a park wake past this runs housekeeping */
199 : #endif
200 0 : for(;;) {
201 :
202 : #ifdef STEM_CALLBACK_SHOULD_SHUTDOWN
203 0 : if( FD_UNLIKELY( STEM_CALLBACK_SHOULD_SHUTDOWN( ctx ) ) ) break;
204 0 : #endif
205 :
206 : /* Do housekeeping at a low rate in the background */
207 :
208 0 : ulong housekeeping_ticks = 0UL;
209 0 : if( FD_UNLIKELY( (now-then)>=0L ) ) {
210 0 : ulong event_idx = (ulong)event_map[ event_seq ];
211 :
212 : /* Do the next async event. event_idx:
213 : <out_cnt - receive credits from out event_idx
214 : ==out_cnt - housekeeping
215 : >out_cnt - send credits to in event_idx - out_cnt - 1.
216 : Branch hints and order are optimized for the case:
217 : out_cnt >~ in_cnt >~ 1. */
218 :
219 0 : if( FD_LIKELY( event_idx<cons_cnt ) ) { /* cons fctl for cons cons_idx */
220 0 : ulong cons_idx = event_idx;
221 :
222 : /* Receive flow control credits from this out. */
223 0 : ulong this_cons_seq = __atomic_load_n( cons_fseq[ cons_idx ], __ATOMIC_ACQUIRE );
224 0 : cons_seq[ cons_idx ] = this_cons_seq;
225 : #ifdef STEM_CALLBACK_RECV_CREDIT
226 0 : STEM_CALLBACK_RECV_CREDIT( ctx, cons_out[ cons_idx ], out_seq[ cons_out[ cons_idx ] ], this_cons_seq );
227 : #endif
228 :
229 0 : } else if( FD_LIKELY( event_idx>cons_cnt ) ) { /* in fctl for in in_idx */
230 0 : ulong in_idx = event_idx - cons_cnt - 1UL;
231 :
232 : /* Send flow control credits and drain flow control diagnostics
233 : for in_idx, unless both are already current. */
234 :
235 0 : if( FD_LIKELY( STEM_(in_dirty)( &in[ in_idx ] ) ) ) {
236 0 : STEM_(in_update)( &in[ in_idx ] );
237 0 : if( FD_UNLIKELY( sleep->shmem ) ) STEM_(credit_ring)( sleep, in[ in_idx ].idx );
238 0 : }
239 :
240 0 : } else { /* event_idx==cons_cnt, housekeeping event */
241 :
242 : #if STEM_SLEEP_PARKS
243 : hk_due = now + hk_ticks;
244 : #endif
245 :
246 : /* Update metrics counters to external viewers */
247 0 : FD_COMPILER_MFENCE();
248 0 : FD_MGAUGE_SET( TILE, HEARTBEAT_TIME, (ulong)fd_log_wallclock() );
249 0 : FD_MGAUGE_SET( TILE, IS_BACKPRESSURED, metric_in_backp );
250 0 : FD_MCNT_INC ( TILE, BACKPRESSURE, metric_backp_cnt );
251 0 : FD_MCNT_ENUM_COPY( TILE, REGIME_DURATION_NANOS, metric_regime_ticks );
252 : #ifdef STEM_CALLBACK_METRICS_WRITE
253 0 : STEM_CALLBACK_METRICS_WRITE( ctx );
254 : #endif
255 0 : FD_COMPILER_MFENCE();
256 0 : metric_backp_cnt = 0UL;
257 :
258 : /* Receive flow control credits */
259 0 : STEM_(cr_refresh)( out_cnt, out_depth, out_seq, cr_avail, &min_cr_avail, cr_max, cons_cnt, cons_out, cons_seq, cons_slow, metric_in_backp );
260 :
261 : /* Publish producer progress sync word, for the outs that moved
262 : (an out never published here may belong to another writer) */
263 0 : for( ulong out_idx=0UL; out_idx<out_cnt; out_idx++ ) {
264 0 : if( FD_UNLIKELY( out_seq_pub[ out_idx ]==out_seq[ out_idx ] ) ) continue;
265 0 : STEM_(out_publish)( sleep, out_mcache[ out_idx ], out_idx, out_seq[ out_idx ] );
266 0 : out_seq_pub[ out_idx ] = out_seq[ out_idx ];
267 0 : }
268 :
269 : #ifdef STEM_CALLBACK_DURING_HOUSEKEEPING
270 0 : STEM_CALLBACK_DURING_HOUSEKEEPING( ctx );
271 : #else
272 : (void)ctx;
273 : #endif
274 0 : }
275 :
276 : /* Select which event to do next (randomized round robin) and
277 : reload the housekeeping timer. */
278 :
279 0 : event_seq++;
280 0 : if( FD_UNLIKELY( event_seq>=event_cnt ) ) {
281 0 : event_seq = 0UL;
282 :
283 : /* Randomize the order of event processing for the next event
284 : event_cnt events to avoid lighthousing effects causing input
285 : credit starvation at extreme fan in/fan out, extreme in load
286 : and high credit return laziness. */
287 :
288 0 : ulong swap_idx = (ulong)fd_rng_uint_roll( rng, (uint)event_cnt );
289 0 : ushort map_tmp = event_map[ swap_idx ];
290 0 : event_map[ swap_idx ] = event_map[ 0 ];
291 0 : event_map[ 0 ] = map_tmp;
292 :
293 : /* We also do the same with the ins to prevent there being a
294 : correlated order frag origins from different inputs
295 : downstream at extreme fan in and extreme in load. */
296 :
297 0 : if( FD_LIKELY( in_cnt>1UL ) ) {
298 0 : swap_idx = (ulong)fd_rng_uint_roll( rng, (uint)in_cnt );
299 0 : fd_stem_tile_in_t in_tmp;
300 0 : in_tmp = in[ swap_idx ];
301 0 : in[ swap_idx ] = in[ 0 ];
302 0 : in[ 0 ] = in_tmp;
303 : #ifdef STEM_STICKY_POLL_MAX
304 : sticky_rem = 0UL; /* sticky_in is a slot, not a link */
305 : #endif
306 0 : }
307 0 : }
308 :
309 : /* Reload housekeeping timer */
310 0 : then = now + (long)fd_tempo_async_reload( rng, async_min );
311 0 : long next = fd_tickcount();
312 0 : housekeeping_ticks = (ulong)(next - now);
313 0 : now = next;
314 0 : }
315 :
316 : #if defined(STEM_CALLBACK_BEFORE_CREDIT) || defined(STEM_CALLBACK_AFTER_CREDIT) || defined(STEM_CALLBACK_AFTER_FRAG) || defined(STEM_CALLBACK_RETURNABLE_FRAG)
317 : stem.now = now;
318 : stem.then = then;
319 : #endif
320 :
321 0 : int charge_busy_before = 0;
322 : #ifdef STEM_CALLBACK_BEFORE_CREDIT
323 0 : STEM_CALLBACK_BEFORE_CREDIT( ctx, &stem, &charge_busy_before );
324 : #endif
325 :
326 : /* Check if we are backpressured. If so, count any transition into
327 : a backpressured regime and spin to wait for flow control credits
328 : to return. We don't do a fully atomic update here as it is only
329 : diagnostic and it will still be correct in the usual case where
330 : individual diagnostic counters aren't used by writers in
331 : different threads of execution. We only count the transition
332 : from not backpressured to backpressured. */
333 :
334 0 : int is_backpressured = min_cr_avail<burst;
335 : #ifdef STEM_CALLBACK_CHECK_CREDIT
336 0 : STEM_CALLBACK_CHECK_CREDIT( ctx, &stem, &charge_busy_before, &is_backpressured );
337 : #endif
338 0 : if( FD_UNLIKELY( is_backpressured ) ) {
339 0 : metric_backp_cnt += (ulong)!metric_in_backp;
340 0 : metric_in_backp = 1UL;
341 0 : FD_SPIN_PAUSE();
342 0 : metric_regime_ticks[2] += housekeeping_ticks;
343 0 : long next = fd_tickcount();
344 0 : metric_regime_ticks[5] += (ulong)(next - now);
345 0 : now = next;
346 : #if STEM_SLEEP_PARKS
347 : sleep_idle_streak++;
348 0 : if( FD_UNLIKELY( sleep_idle_streak>=in_cnt && (now-linger_start)>sleep_linger_ticks ) ) {
349 0 : sleep_idle_streak = 0UL;
350 0 : STEM_(park_attempt)( ctx, sleep, in, in_cnt, out_mcache, out_cnt, out_depth, out_seq, out_seq_pub, cr_avail, &min_cr_avail, cr_max,
351 0 : cons_cnt, cons_fseq, cons_seq, cons_out, cons_slow, event_cnt, event_map, &event_seq, async_min, hk_due,
352 0 : sleep_cap_ticks, sleep_min_ticks, sleep_tick_per_ns, metric_regime_ticks, &now, &then,
353 0 : FD_METRICS_ENUM_TILE_REGIME_V_BACKPRESSURE_SLEEPING_IDX, 1, then );
354 0 : }
355 : #endif
356 0 : continue;
357 0 : }
358 0 : metric_in_backp = 0UL;
359 :
360 0 : int charge_busy_after = 0;
361 : #ifdef STEM_CALLBACK_AFTER_CREDIT
362 : int poll_in = 1;
363 0 : STEM_CALLBACK_AFTER_CREDIT( ctx, &stem, &poll_in, &charge_busy_after );
364 0 : if( FD_UNLIKELY( !poll_in ) ) {
365 0 : metric_regime_ticks[1] += housekeeping_ticks;
366 0 : long next = fd_tickcount();
367 0 : metric_regime_ticks[4] += (ulong)(next - now);
368 0 : now = next;
369 : #if STEM_SLEEP_PARKS
370 : sleep_idle_streak = 0UL;
371 : linger_start = now;
372 : #endif
373 0 : continue;
374 0 : }
375 0 : #endif
376 :
377 : /* Select which in to poll next (randomized round robin) */
378 :
379 0 : if( FD_UNLIKELY( !in_cnt ) ) {
380 0 : int was_busy = charge_busy_before+charge_busy_after;
381 0 : metric_regime_ticks[0] += housekeeping_ticks;
382 0 : long next = fd_tickcount();
383 0 : if( FD_UNLIKELY( was_busy ) ) metric_regime_ticks[3] += (ulong)(next - now);
384 0 : else metric_regime_ticks[6] += (ulong)(next - now);
385 0 : now = next;
386 : #if STEM_SLEEP_PARKS
387 0 : if( FD_UNLIKELY( was_busy ) ) {
388 0 : sleep_idle_streak = 0UL;
389 0 : linger_start = now;
390 0 : } else {
391 0 : if( FD_UNLIKELY( (now-linger_start)>sleep_linger_ticks ) ) {
392 0 : sleep_idle_streak = 0UL;
393 0 : STEM_(park_attempt)( ctx, sleep, in, in_cnt, out_mcache, out_cnt, out_depth, out_seq, out_seq_pub, cr_avail, &min_cr_avail, cr_max,
394 0 : cons_cnt, cons_fseq, cons_seq, cons_out, cons_slow, event_cnt, event_map, &event_seq, async_min, hk_due,
395 0 : sleep_cap_ticks, sleep_min_ticks, sleep_tick_per_ns, metric_regime_ticks, &now, &then,
396 0 : FD_METRICS_ENUM_TILE_REGIME_V_CAUGHT_UP_SLEEPING_IDX, 0, LONG_MAX );
397 0 : }
398 0 : }
399 : #endif
400 0 : continue;
401 0 : }
402 :
403 0 : ulong prefrag_ticks = 0UL;
404 : #if defined(STEM_CALLBACK_BEFORE_CREDIT) && defined(STEM_CALLBACK_AFTER_CREDIT)
405 0 : if( FD_LIKELY( charge_busy_before || charge_busy_after ) ) {
406 : #elif defined(STEM_CALLBACK_BEFORE_CREDIT)
407 0 : if( FD_LIKELY( charge_busy_before ) ) {
408 : #elif defined(STEM_CALLBACK_AFTER_CREDIT)
409 0 : if( FD_LIKELY( charge_busy_after ) ) {
410 0 : #endif
411 :
412 : #if defined(STEM_CALLBACK_BEFORE_CREDIT) || defined(STEM_CALLBACK_AFTER_CREDIT)
413 0 : long prefrag_next = fd_tickcount();
414 0 : prefrag_ticks = (ulong)(prefrag_next - now);
415 0 : now = prefrag_next;
416 0 : }
417 : #endif
418 :
419 0 : fd_stem_tile_in_t * this_in;
420 : #ifdef STEM_STICKY_POLL_MAX
421 : int this_in_rr = 0;
422 0 : if( FD_LIKELY( sticky_rem ) ) {
423 0 : this_in = sticky_in;
424 0 : sticky_rem--;
425 0 : } else {
426 0 : this_in = &in[ in_seq ];
427 0 : this_in_rr = 1;
428 0 : in_seq++;
429 0 : if( in_seq>=in_cnt ) in_seq = 0UL; /* cmov */
430 0 : }
431 : #else
432 : this_in = &in[ in_seq ];
433 : in_seq++;
434 0 : if( in_seq>=in_cnt ) in_seq = 0UL; /* cmov */
435 : #endif
436 :
437 : /* Check if this in has any new fragments to mux */
438 :
439 0 : ulong this_in_seq = this_in->seq;
440 0 : fd_frag_meta_t const * this_in_mline = this_in->mline; /* Already at appropriate line for this_in_seq */
441 :
442 0 : #if FD_HAS_AVX
443 0 : fd_frag_meta_v256_t yline = FD_VOLATILE_CONST( this_in_mline->avx );
444 0 : ulong seq_found = fd_frag_meta_avx_seq( yline );
445 0 : ulong sig = fd_frag_meta_avx_sig( yline );
446 : #elif FD_HAS_SSE
447 : __m128i seq_sig = fd_frag_meta_seq_sig_query( this_in_mline );
448 : ulong seq_found = fd_frag_meta_sse0_seq( seq_sig );
449 : ulong sig = fd_frag_meta_sse0_sig( seq_sig );
450 : #elif FD_HAS_ARM || FD_HAS_RISCV
451 : ulong seq_found = __atomic_load_n( &this_in_mline->seq, __ATOMIC_ACQUIRE );
452 : ulong sig;
453 : #else
454 : /* Without 128-bit atomic load, seq and sig might be read from
455 : different frags (due to overrun), which results in a before_frag
456 : and during_frag being issued with incorrect arguments, but not
457 : after_frag. */
458 : ulong seq_found = FD_VOLATILE_CONST( this_in_mline->seq );
459 : ulong sig = FD_VOLATILE_CONST( this_in_mline->sig );
460 : #endif
461 0 : long diff = fd_seq_diff( this_in_seq, seq_found );
462 0 : if( FD_UNLIKELY( diff ) ) { /* Caught up or overrun, optimize for new frag case */
463 : #ifdef STEM_STICKY_POLL_MAX
464 : sticky_rem = 0UL;
465 : #endif
466 0 : ulong * housekeeping_regime = &metric_regime_ticks[0];
467 0 : ulong * prefrag_regime = &metric_regime_ticks[3];
468 0 : ulong * finish_regime = &metric_regime_ticks[6];
469 0 : if( FD_UNLIKELY( diff<0L ) ) { /* Overrun (impossible if in is honoring our flow control) */
470 0 : this_in->seq = seq_found; /* Resume from here (probably reasonably current, could query in mcache sync directly instead) */
471 0 : housekeeping_regime = &metric_regime_ticks[1];
472 0 : prefrag_regime = &metric_regime_ticks[4];
473 0 : finish_regime = &metric_regime_ticks[7];
474 0 : this_in->accum[ FD_METRICS_COUNTER_LINK_LINK_POLLING_OVERRUN_OFF ]++;
475 0 : this_in->accum[ FD_METRICS_COUNTER_LINK_FRAG_POLLING_OVERRUN_OFF ] += (uint)(-diff);
476 :
477 : #ifdef STEM_CALLBACK_AFTER_POLL_OVERRUN
478 0 : STEM_CALLBACK_AFTER_POLL_OVERRUN( ctx );
479 : #endif
480 0 : }
481 :
482 : /* Don't bother with spin as polling multiple locations */
483 0 : *housekeeping_regime += housekeeping_ticks;
484 0 : *prefrag_regime += prefrag_ticks;
485 0 : long next = fd_tickcount();
486 0 : *finish_regime += (ulong)(next - now);
487 0 : now = next;
488 : #if STEM_SLEEP_PARKS
489 0 : if( FD_UNLIKELY( (diff<0L) || (charge_busy_before+charge_busy_after)>0L ) ) {
490 0 : sleep_idle_streak = 0UL;
491 0 : linger_start = now;
492 0 : } else {
493 0 : sleep_idle_streak++;
494 0 : if( FD_UNLIKELY( sleep_idle_streak>=in_cnt && (now-linger_start)>sleep_linger_ticks ) ) {
495 0 : sleep_idle_streak = 0UL;
496 0 : STEM_(park_attempt)( ctx, sleep, in, in_cnt, out_mcache, out_cnt, out_depth, out_seq, out_seq_pub, cr_avail, &min_cr_avail, cr_max,
497 0 : cons_cnt, cons_fseq, cons_seq, cons_out, cons_slow, event_cnt, event_map, &event_seq, async_min, hk_due,
498 0 : sleep_cap_ticks, sleep_min_ticks, sleep_tick_per_ns, metric_regime_ticks, &now, &then,
499 0 : FD_METRICS_ENUM_TILE_REGIME_V_CAUGHT_UP_SLEEPING_IDX, 0, LONG_MAX );
500 0 : }
501 0 : }
502 : #endif
503 0 : continue;
504 0 : }
505 :
506 : #if FD_HAS_ARM || FD_HAS_RISCV
507 : /* On weakly ordered CPUs, acquire publication before reading sig,
508 : then confirm the sequence before allowing before_frag to filter. */
509 : sig = __atomic_load_n( &this_in_mline->sig, __ATOMIC_ACQUIRE );
510 : ulong seq_confirm = __atomic_load_n( &this_in_mline->seq, __ATOMIC_ACQUIRE );
511 : if( FD_UNLIKELY( fd_seq_ne( seq_confirm, seq_found ) ) ) {
512 : metric_regime_ticks[1] += housekeeping_ticks;
513 : metric_regime_ticks[4] += prefrag_ticks;
514 : long next = fd_tickcount();
515 : metric_regime_ticks[7] += (ulong)(next - now);
516 : now = next;
517 : continue;
518 : }
519 : #endif
520 0 : (void)sig;
521 :
522 : #ifdef STEM_CALLBACK_BEFORE_FRAG
523 0 : int filter = STEM_CALLBACK_BEFORE_FRAG( ctx, (ulong)this_in->idx, seq_found, sig );
524 0 : if( FD_UNLIKELY( filter<0 ) ) {
525 : #if STEM_SLEEP_PARKS
526 : sleep_idle_streak = 0UL;
527 : linger_start = now;
528 : #endif
529 0 : metric_regime_ticks[1] += housekeeping_ticks;
530 0 : metric_regime_ticks[4] += prefrag_ticks;
531 0 : long next = fd_tickcount();
532 0 : metric_regime_ticks[7] += (ulong)(next - now);
533 0 : now = next;
534 0 : continue;
535 0 : } else if( FD_UNLIKELY( filter>0 ) ) {
536 : #if STEM_SLEEP_PARKS
537 : linger_start = now;
538 : #endif
539 0 : this_in->accum[ FD_METRICS_COUNTER_LINK_FRAG_FILTERED_OFF ]++;
540 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 */
541 :
542 : this_in_seq = fd_seq_inc( this_in_seq, 1UL );
543 : this_in->seq = this_in_seq;
544 : this_in->mline = this_in->mcache + fd_mcache_line_idx( this_in_seq, this_in->depth );
545 :
546 0 : metric_regime_ticks[1] += housekeeping_ticks;
547 0 : metric_regime_ticks[4] += prefrag_ticks;
548 0 : long next = fd_tickcount();
549 0 : metric_regime_ticks[7] += (ulong)(next - now);
550 0 : now = next;
551 0 : continue;
552 0 : }
553 0 : #endif
554 :
555 : #if STEM_SLEEP_PARKS
556 0 : sleep_idle_streak = 0UL;
557 0 : linger_start = now;
558 0 : #endif
559 :
560 : /* We have a new fragment to mux. Try to load it. This attempt
561 : should always be successful if in producers are honoring our flow
562 : control. Since we can cheaply detect if there are
563 : misconfigurations (should be an L1 cache hit / predictable branch
564 : in the properly configured case), we do so anyway. */
565 0 : FD_COMPILER_MFENCE();
566 0 : #if FD_HAS_AVX
567 0 : ulong chunk = fd_frag_meta_avx_chunk ( yline ); (void)chunk;
568 0 : ulong sz = fd_frag_meta_avx_sz ( yline ); (void)sz;
569 0 : ulong ctl = fd_frag_meta_avx_ctl ( yline ); (void)ctl;
570 0 : ulong tsorig = fd_frag_meta_avx_tsorig( yline ); (void)tsorig;
571 0 : ulong tspub = fd_frag_meta_avx_tspub ( yline ); (void)tspub;
572 : #elif FD_HAS_ARM
573 : ulong ul2, ul3;
574 : fd_arm_ldp16( this_in_mline->ul+2, ul2, ul3 );
575 : ulong chunk = fd_frag_meta_ul2_chunk ( ul2 ); (void)chunk;
576 : ulong sz = fd_frag_meta_ul2_sz ( ul2 ); (void)sz;
577 : ulong ctl = fd_frag_meta_ul2_ctl ( ul2 ); (void)ctl;
578 : ulong tsorig = fd_frag_meta_ul3_tsorig( ul3 ); (void)tsorig;
579 : ulong tspub = fd_frag_meta_ul3_tspub ( ul3 ); (void)tspub;
580 : #else
581 : ulong chunk = (ulong)this_in_mline->chunk; (void)chunk;
582 : ulong sz = (ulong)this_in_mline->sz; (void)sz;
583 : ulong ctl = (ulong)this_in_mline->ctl; (void)ctl;
584 : ulong tsorig = (ulong)this_in_mline->tsorig; (void)tsorig;
585 : ulong tspub = (ulong)this_in_mline->tspub; (void)tspub;
586 : #endif
587 :
588 : #ifdef STEM_CALLBACK_DURING_FRAG
589 0 : STEM_CALLBACK_DURING_FRAG( ctx, (ulong)this_in->idx, seq_found, sig, chunk, sz, ctl );
590 : #endif
591 :
592 0 : FD_HW_MFENCE_LD();
593 0 : ulong seq_test = FD_VOLATILE_CONST( this_in_mline->seq );
594 0 : FD_COMPILER_MFENCE();
595 :
596 0 : if( FD_UNLIKELY( fd_seq_ne( seq_test, seq_found ) ) ) { /* Overrun while reading (impossible if this_in honoring our fctl) */
597 0 : this_in->seq = seq_test; /* Resume from here (probably reasonably current, could query in mcache sync instead) */
598 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 */
599 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 */
600 : /* Don't bother with spin as polling multiple locations */
601 0 : metric_regime_ticks[1] += housekeeping_ticks;
602 0 : metric_regime_ticks[4] += prefrag_ticks;
603 0 : long next = fd_tickcount();
604 0 : metric_regime_ticks[7] += (ulong)(next - now);
605 0 : now = next;
606 0 : continue;
607 0 : }
608 :
609 : #ifdef STEM_CALLBACK_RETURNABLE_FRAG
610 0 : int return_frag = STEM_CALLBACK_RETURNABLE_FRAG( ctx, (ulong)this_in->idx, seq_found, sig, chunk, sz, ctl, tsorig, tspub, &stem );
611 0 : if( FD_UNLIKELY( return_frag ) ) {
612 0 : metric_regime_ticks[1] += housekeeping_ticks;
613 0 : metric_regime_ticks[4] += prefrag_ticks;
614 0 : long next = fd_tickcount();
615 0 : metric_regime_ticks[7] += (ulong)(next - now);
616 0 : now = next;
617 0 : continue;
618 0 : }
619 0 : #endif
620 :
621 : #ifdef STEM_CALLBACK_AFTER_FRAG
622 0 : STEM_CALLBACK_AFTER_FRAG( ctx, (ulong)this_in->idx, seq_found, sig, sz, tsorig, tspub, &stem );
623 0 : #endif
624 :
625 : /* Windup for the next in poll and accumulate diagnostics */
626 :
627 0 : this_in_seq = fd_seq_inc( this_in_seq, 1UL );
628 0 : this_in->seq = this_in_seq;
629 0 : this_in->mline = this_in->mcache + fd_mcache_line_idx( this_in_seq, this_in->depth );
630 :
631 0 : this_in->accum[ FD_METRICS_COUNTER_LINK_FRAG_CONSUMED_OFF ]++;
632 0 : this_in->accum[ FD_METRICS_COUNTER_LINK_FRAG_CONSUMED_BYTES_OFF ] += (uint)sz;
633 :
634 : #ifdef STEM_STICKY_POLL_MAX
635 0 : sticky_in = this_in_rr ? this_in : sticky_in; /* cmov */
636 0 : sticky_rem = fd_ulong_if( this_in_rr, STEM_STICKY_POLL_MAX, sticky_rem );
637 : #endif
638 :
639 0 : metric_regime_ticks[1] += housekeeping_ticks;
640 0 : metric_regime_ticks[4] += prefrag_ticks;
641 0 : long next = fd_tickcount();
642 0 : metric_regime_ticks[7] += (ulong)(next - now);
643 0 : now = next;
644 0 : }
645 :
646 0 : for( ulong out_idx=0UL; out_idx<out_cnt; out_idx++ ) {
647 0 : if( FD_UNLIKELY( out_seq_pub[ out_idx ]==out_seq[ out_idx ] ) ) continue;
648 0 : STEM_(out_publish)( sleep, out_mcache[ out_idx ], out_idx, out_seq[ out_idx ] );
649 0 : out_seq_pub[ out_idx ] = out_seq[ out_idx ];
650 0 : }
651 0 : }
|
