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 : /* STEM_ALWAYS_PARK makes the tile park on its NEXT_DEADLINE even in
201 : performance mode, where nothing rings it, so its ins wait for the
202 : deadline too.
203 :
204 : STEM_NEVER_PARK keeps the tile spinning even if it has no fragments
205 : to process and would ordinarily try to park. This is typically for
206 : tiles which don't yet play nicely with the sleep infrastructure or do
207 : custom waiting. */
208 :
209 : #ifndef STEM_ALWAYS_PARK
210 : #define STEM_ALWAYS_PARK 0
211 : #endif
212 :
213 : #ifndef STEM_NEVER_PARK
214 : #define STEM_NEVER_PARK 0
215 : #endif
216 :
217 : #if STEM_ALWAYS_PARK && STEM_NEVER_PARK
218 : #error "STEM_ALWAYS_PARK and STEM_NEVER_PARK are exclusive"
219 : #endif
220 :
221 0 : #define STEM_SHUTDOWN_SEQ (ULONG_MAX-1UL)
222 :
223 : static inline void
224 : STEM_(credit_ring)( fd_stem_sleep_t const * sleep,
225 0 : ulong in_idx ) {
226 0 : ulong producer = sleep->in_producer[ in_idx ];
227 0 : if( FD_UNLIKELY( producer==ULONG_MAX ) ) return;
228 0 : __atomic_thread_fence( __ATOMIC_SEQ_CST );
229 0 : if( FD_UNLIKELY( FD_VOLATILE_CONST( sleep->shmem->credit_bits[ producer>>6 ] ) & (1UL<<(producer&63UL)) ) ) fd_sleep_ring( sleep->shmem, producer );
230 0 : }
231 :
232 : /* STEM_(in_dirty) is 1 if in has credits or diagnostics not yet
233 : published (in_update on a clean in rewrites the same values). */
234 :
235 : static inline int
236 0 : STEM_(in_dirty)( fd_stem_tile_in_t const * in ) {
237 0 : uint const * accum = in->accum;
238 0 : return ( __atomic_load_n( in->fseq, __ATOMIC_RELAXED )!=in->seq )
239 0 : | (int)(accum[0]|accum[1]|accum[2]|accum[3]|accum[4]|accum[5]);
240 0 : }
241 :
242 : static inline __attribute__((always_inline)) void
243 0 : STEM_(in_update)( fd_stem_tile_in_t * in ) {
244 0 : __atomic_store_n( in->fseq, in->seq, __ATOMIC_RELEASE );
245 :
246 0 : volatile ulong * metrics = fd_metrics_link_in( fd_metrics_base_tl, in->idx );
247 :
248 0 : uint * accum = in->accum;
249 0 : ulong a0 = (ulong)accum[0]; ulong a1 = (ulong)accum[1]; ulong a2 = (ulong)accum[2];
250 0 : ulong a3 = (ulong)accum[3]; ulong a4 = (ulong)accum[4]; ulong a5 = (ulong)accum[5];
251 0 : FD_COMPILER_MFENCE();
252 0 : metrics[0] += a0; metrics[1] += a1; metrics[2] += a2;
253 0 : metrics[3] += a3; metrics[4] += a4; metrics[5] += a5;
254 0 : FD_COMPILER_MFENCE();
255 0 : accum[0] = 0U; accum[1] = 0U; accum[2] = 0U;
256 0 : accum[3] = 0U; accum[4] = 0U; accum[5] = 0U;
257 0 : }
258 :
259 : /* STEM_(out_publish) publishes out_idx's sync word and seq_mirror.
260 : Callers skip an out whose seq has not moved (out_seq_pub), which
261 : also keeps the stem off outs another client publishes. The stem is
262 : the sole writer of its mirrors, so the store needs no guard load. */
263 :
264 : static inline void
265 : STEM_(out_publish)( fd_stem_sleep_t const * sleep,
266 : fd_frag_meta_t * mcache,
267 : ulong out_idx,
268 0 : ulong seq ) {
269 0 : fd_mcache_seq_update( fd_mcache_seq_laddr( mcache ), seq );
270 0 : if( sleep->shmem ) __atomic_store_n( &sleep->shmem->seq_mirror[ sleep->out_link_id[ out_idx ] ], seq, __ATOMIC_RELEASE );
271 0 : }
272 :
273 : static inline int
274 : STEM_(park)( STEM_CALLBACK_CONTEXT_TYPE * ctx,
275 : fd_stem_sleep_t const * sleep,
276 : fd_stem_tile_in_t * in,
277 : ulong in_cnt,
278 : fd_frag_meta_t ** out_mcache,
279 : ulong out_cnt,
280 : ulong const * out_seq,
281 : ulong * out_seq_pub,
282 : ulong cons_cnt,
283 : ulong const ** cons_fseq,
284 : ulong const * cons_seq,
285 : int backpressured,
286 : long deadline_hint,
287 : long cap_ticks,
288 : long min_ticks,
289 : double tick_per_ns,
290 0 : long now ) {
291 0 : (void)ctx;
292 :
293 0 : if( FD_LIKELY( !backpressured ) ) {
294 0 : for( ulong i=0UL; i<in_cnt; i++ ) {
295 : /* If there's a frag to read on some in, then we don't park. */
296 0 : if( FD_UNLIKELY( fd_seq_diff( in[ i ].seq, fd_frag_meta_seq_query( in[ i ].mline ) )<=0L ) ) return 0;
297 0 : }
298 0 : }
299 :
300 : /* Backpressured, waker readiness and already due work wait for
301 : after_credit, which runs once credits return and ring us. A
302 : before_credit tile still does them while backpressured. */
303 : #ifdef STEM_CALLBACK_BEFORE_CREDIT
304 0 : int bp_defer = 0;
305 : #else
306 0 : int bp_defer = backpressured;
307 0 : #endif
308 0 : if( FD_LIKELY( !bp_defer && sleep->waker_fseq ) ) {
309 0 : if( FD_UNLIKELY( fd_fseq_query( sleep->waker_fseq )==1UL ) ) return 0;
310 0 : }
311 :
312 : #ifdef STEM_CALLBACK_PREVENT_PARK
313 0 : if( FD_UNLIKELY( STEM_CALLBACK_PREVENT_PARK( ctx ) ) ) return 0;
314 0 : #endif
315 :
316 : #ifdef STEM_CALLBACK_SHOULD_SHUTDOWN
317 0 : if( FD_UNLIKELY( STEM_CALLBACK_SHOULD_SHUTDOWN( ctx ) ) ) return 0;
318 0 : #endif
319 :
320 :
321 0 : long due = deadline_hint;
322 : #ifdef STEM_CALLBACK_NEXT_DEADLINE
323 0 : long cb_due = STEM_CALLBACK_NEXT_DEADLINE( ctx );
324 0 : if( FD_UNLIKELY( bp_defer && cb_due<=now ) ) cb_due = LONG_MAX;
325 : due = fd_long_min( due, cb_due );
326 : #endif
327 0 : long deadline = fd_long_min( now+cap_ticks, due );
328 0 : if( FD_UNLIKELY( deadline-now<min_ticks ) ) return 0; /* already close enough to deadline, don't waste a syscall */
329 :
330 : /* We are now going to park ... flush all state before. Skip the
331 : ins and outs with nothing to flush: a clean in's producer was
332 : already rung when our fseq last moved. */
333 0 : for( ulong i=0UL; i<in_cnt; i++ ) {
334 0 : if( FD_LIKELY( !STEM_(in_dirty)( &in[ i ] ) ) ) continue;
335 0 : STEM_(in_update)( &in[ i ] );
336 0 : if( FD_LIKELY( sleep->shmem ) ) STEM_(credit_ring)( sleep, in[ i ].idx );
337 0 : }
338 0 : for( ulong o=0UL; o<out_cnt; o++ ) {
339 0 : if( FD_LIKELY( out_seq_pub[ o ]==out_seq[ o ] ) ) continue; /* idle, or never published by this stem */
340 0 : STEM_(out_publish)( sleep, out_mcache[ o ], o, out_seq[ o ] );
341 0 : out_seq_pub[ o ] = out_seq[ o ];
342 0 : }
343 :
344 0 : ulong _word;
345 0 : ulong * word;
346 0 : ulong my_w;
347 0 : ulong my_bit;
348 :
349 0 : int parked = 0;
350 0 : if( FD_LIKELY( sleep->shmem ) ) {
351 0 : sleep->shmem->tile[ sleep->tile_id ].deadline = (ulong)deadline;
352 0 : word = &sleep->shmem->tile[ sleep->tile_id ].word;
353 0 : my_w = sleep->tile_id>>6;
354 0 : my_bit = 1UL<<(sleep->tile_id&63UL);
355 0 : FD_VOLATILE( word[0] ) = 0UL;
356 :
357 : /* Set the parked bit so publishes to us ring us; when backpressured
358 : set the credit bit too, which swaps publish rings for credit
359 : return rings. */
360 0 : for( ulong i=0UL; i<in_cnt; i++ ) {
361 0 : ulong snap = in[ i ].seq;
362 0 : if( FD_UNLIKELY( backpressured ) ) snap = __atomic_load_n( &sleep->shmem->seq_mirror[ sleep->in_link_id[ in[ i ].idx ] ], __ATOMIC_ACQUIRE );
363 0 : sleep->shmem->seq_snap[ sleep->tile_id ][ in[ i ].idx ] = snap;
364 0 : }
365 0 : if( FD_UNLIKELY( backpressured ) ) __atomic_fetch_or( &sleep->shmem->credit_bits[ my_w ], my_bit, __ATOMIC_SEQ_CST );
366 0 : __atomic_fetch_or( &sleep->shmem->parked_bits[ my_w ], my_bit, __ATOMIC_SEQ_CST );
367 0 : parked = 1;
368 0 : } else {
369 : /* No shmem, so no one can wake us. We will just sleep until the
370 : deadline. This only happens for ALWAYS_PARK tiles when running
371 : in spin mode. */
372 0 : word = &_word;
373 0 : my_w = 0UL;
374 0 : my_bit = 0UL;
375 0 : FD_VOLATILE( word[0] ) = 0UL;
376 0 : __atomic_thread_fence( __ATOMIC_SEQ_CST );
377 0 : }
378 :
379 : /* Last chance checks to reduce race-window for a lost wake. If there
380 : is nothing to abort the park, then park. There's a short race
381 : between checking and parking, which cannot be avoided, and is
382 : bounded by the mwaitx tile which periodically sweeps and wakes us
383 : if we raced here. */
384 :
385 0 : int pending = 0;
386 0 : if( FD_LIKELY( parked && !backpressured ) ) {
387 0 : for( ulong i=0UL; i<in_cnt; i++ ) {
388 0 : pending |= fd_seq_diff( in[ i ].seq, fd_frag_meta_seq_query( in[ i ].mline ) )<=0L;
389 0 : }
390 0 : } else if( FD_LIKELY( parked ) ) {
391 : /* a credit return between the caller's snapshot and the credit
392 : bit went unrung; only its fseq shows it */
393 0 : for( ulong c=0UL; c<cons_cnt; c++ ) {
394 0 : pending |= __atomic_load_n( cons_fseq[ c ], __ATOMIC_ACQUIRE )!=cons_seq[ c ];
395 0 : }
396 0 : }
397 0 : if( FD_LIKELY( !bp_defer && sleep->waker_fseq ) ) pending |= fd_fseq_query( sleep->waker_fseq )==1UL;
398 : #ifdef STEM_CALLBACK_SHOULD_SHUTDOWN
399 0 : pending |= STEM_CALLBACK_SHOULD_SHUTDOWN( ctx );
400 : #endif
401 :
402 0 : if( FD_UNLIKELY( pending ) ) {
403 0 : FD_VOLATILE( word[0] ) = 1UL;
404 0 : if( FD_LIKELY( parked ) ) {
405 0 : __atomic_fetch_and( &sleep->shmem->parked_bits[ my_w ], ~my_bit, __ATOMIC_SEQ_CST );
406 0 : if( FD_UNLIKELY( backpressured ) ) __atomic_fetch_and( &sleep->shmem->credit_bits[ my_w ], ~my_bit, __ATOMIC_SEQ_CST );
407 0 : }
408 0 : FD_MCNT_INC( TILE, PARK_ATTEMPTED_PENDING, 1UL );
409 0 : return backpressured ? -1 : 0;
410 0 : }
411 :
412 : /* Only a due inside the cap arms a timer, the sweep enforces the cap */
413 0 : long timer = fd_long_if( parked & (due>deadline), LONG_MAX, deadline );
414 : #ifdef STEM_CALLBACK_PARK_WAIT
415 0 : int cause = STEM_CALLBACK_PARK_WAIT( ctx, word, timer, tick_per_ns );
416 : #else
417 : int cause = fd_sleep_park_wait( word, timer, tick_per_ns );
418 : #endif
419 :
420 : /* Park completed, either due to a ring or a deadline. Now clear the
421 : parked bit and publish the reason. There's a race here with the
422 : ringer but it's harmless, it might cause one more spurious wake. */
423 0 : FD_VOLATILE( word[0] ) = 1UL;
424 0 : if( FD_LIKELY( sleep->shmem && parked ) ) {
425 0 : __atomic_fetch_and( &sleep->shmem->parked_bits[ my_w ], ~my_bit, __ATOMIC_SEQ_CST );
426 0 : if( FD_UNLIKELY( backpressured ) ) __atomic_fetch_and( &sleep->shmem->credit_bits[ my_w ], ~my_bit, __ATOMIC_SEQ_CST );
427 0 : }
428 0 : if( FD_LIKELY( cause==FD_SLEEP_UNPARK_RING ) ) FD_MCNT_INC( TILE, PARK_ATTEMPTED_RING, 1UL );
429 0 : else FD_MCNT_INC( TILE, PARK_ATTEMPTED_DEADLINE, 1UL );
430 0 : return 1;
431 0 : }
432 :
433 : static inline __attribute__((always_inline)) void
434 : STEM_(cr_refresh)( ulong out_cnt,
435 : ulong const * out_depth,
436 : ulong const * out_seq,
437 : ulong * cr_avail,
438 : ulong * min_cr_avail,
439 : ulong cr_max,
440 : ulong cons_cnt,
441 : ulong const * cons_out,
442 : ulong const * cons_seq,
443 : volatile ulong ** cons_slow,
444 0 : ulong in_backp ) {
445 0 : if( FD_LIKELY( cons_cnt ) ) {
446 0 : ulong slowest_cons = ULONG_MAX;
447 0 : *min_cr_avail = cr_max;
448 0 : for( ulong out_idx=0; out_idx<out_cnt; out_idx++ ) {
449 0 : cr_avail[ out_idx ] = out_depth[ out_idx ];
450 0 : }
451 :
452 0 : for( ulong cons_idx=0UL; cons_idx<cons_cnt; cons_idx++ ) {
453 0 : ulong out_idx = cons_out[ cons_idx ];
454 :
455 : /* Read the fseq boot value (ULONG_MAX) as sequence 0, not -1,
456 : else the producer is one credit short until the consumer
457 : boots. */
458 0 : ulong cseq = fd_ulong_if( cons_seq[ cons_idx ]==ULONG_MAX, 0UL, cons_seq[ cons_idx ] );
459 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 );
460 :
461 : /* If a reliable consumer exits, they can set the credit return
462 : fseq to STEM_SHUTDOWN_SEQ to indicate they are no longer
463 : actively consuming. */
464 0 : cons_cr_avail = fd_ulong_if( cons_seq[ cons_idx ]==STEM_SHUTDOWN_SEQ, out_depth[ out_idx ], cons_cr_avail );
465 0 : slowest_cons = fd_ulong_if( cons_cr_avail<*min_cr_avail, cons_idx, slowest_cons );
466 :
467 0 : cr_avail[ out_idx ] = fd_ulong_min( cr_avail[ out_idx ], cons_cr_avail );
468 0 : *min_cr_avail = fd_ulong_min( cons_cr_avail, *min_cr_avail );
469 0 : }
470 :
471 : /* See notes above about use of quasi-atomic diagnostic accum.
472 : Skip a +=0: it would still take the consumer's line. */
473 0 : if( FD_UNLIKELY( in_backp && slowest_cons!=ULONG_MAX ) ) {
474 0 : FD_COMPILER_MFENCE();
475 0 : (*cons_slow[ slowest_cons ]) += in_backp;
476 0 : FD_COMPILER_MFENCE();
477 0 : }
478 0 : }
479 0 : }
480 :
481 : static inline void
482 : STEM_(park_attempt)( STEM_CALLBACK_CONTEXT_TYPE * ctx,
483 : fd_stem_sleep_t const * cfg,
484 : fd_stem_tile_in_t * in,
485 : ulong in_cnt,
486 : fd_frag_meta_t ** out_mcache,
487 : ulong out_cnt,
488 : ulong const * out_depth,
489 : ulong const * out_seq,
490 : ulong * out_seq_pub,
491 : ulong * cr_avail,
492 : ulong * min_cr_avail,
493 : ulong cr_max,
494 : ulong cons_cnt,
495 : ulong const ** cons_fseq,
496 : ulong * cons_seq,
497 : ulong const * cons_out,
498 : volatile ulong ** cons_slow,
499 : ulong event_cnt,
500 : ushort const * event_map,
501 : ulong * event_seq,
502 : ulong async_min,
503 : long hk_due,
504 : long cap_ticks,
505 : long min_ticks,
506 : double tick_per_ns,
507 : ulong * metric_regime_ticks,
508 : long * now,
509 : long * then,
510 : ulong regime,
511 : int backpressured,
512 0 : long deadline_hint ) {
513 0 : int slept = STEM_(park)( ctx, cfg, in, in_cnt, out_mcache, out_cnt, out_seq, out_seq_pub, cons_cnt, cons_fseq, cons_seq,
514 0 : backpressured, deadline_hint, cap_ticks, min_ticks, tick_per_ns, *now );
515 0 : if( FD_UNLIKELY( !slept ) ) return; /* found work */
516 :
517 0 : long next = fd_tickcount();
518 0 : long waited = next - *now;
519 0 : metric_regime_ticks[ regime ] += (ulong)waited;
520 0 : *now = next;
521 0 : if( FD_UNLIKELY( backpressured || waited>=(long)async_min ) ) {
522 : /* If we slept shorter than a housekeeping interval, we didn't miss
523 : any credit refresh housekeeping events, but otherwise we might
524 : have missed many, so refill them immediately here. A
525 : backpressure park is woken by the consumer's credit return, so
526 : reload regardless, and turn cons_seq into cr_avail now, or we
527 : re-park on the stale count until the timer. */
528 0 : for( ulong cons_idx=0UL; cons_idx<cons_cnt; cons_idx++ ) {
529 0 : ulong this_cons_seq = __atomic_load_n( cons_fseq[ cons_idx ], __ATOMIC_ACQUIRE );
530 0 : cons_seq[ cons_idx ] = this_cons_seq;
531 : #ifdef STEM_CALLBACK_RECV_CREDIT
532 0 : STEM_CALLBACK_RECV_CREDIT( ctx, cons_out[ cons_idx ], out_seq[ cons_out[ cons_idx ] ], this_cons_seq );
533 : #endif
534 0 : }
535 0 : if( FD_UNLIKELY( (*now-hk_due)>=0L ) ) {
536 : /* A lazy interval passed, full housekeeping is due now. */
537 0 : for( ulong k=0UL; k<event_cnt; k++ ) {
538 0 : if( event_map[ k ]==cons_cnt ) {
539 0 : *event_seq = k;
540 0 : break;
541 0 : }
542 0 : }
543 0 : *then = *now;
544 0 : } else {
545 : /* Credits only, nothing else is due yet. */
546 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, 0UL );
547 0 : *then = *now + (long)async_min;
548 0 : }
549 0 : }
550 0 : }
551 :
552 : FD_FN_PURE static inline ulong
553 0 : STEM_(scratch_align)( void ) {
554 0 : return FD_STEM_SCRATCH_ALIGN;
555 0 : }
556 :
557 : FD_FN_PURE static inline ulong
558 : STEM_(scratch_footprint)( ulong in_cnt,
559 : ulong out_cnt,
560 0 : ulong cons_cnt ) {
561 0 : ulong l = FD_LAYOUT_INIT;
562 0 : l = FD_LAYOUT_APPEND( l, alignof(fd_stem_tile_in_t), in_cnt*sizeof(fd_stem_tile_in_t) ); /* in */
563 0 : l = FD_LAYOUT_APPEND( l, alignof(ulong), out_cnt*sizeof(ulong) ); /* cr_avail */
564 0 : l = FD_LAYOUT_APPEND( l, alignof(ulong), out_cnt*sizeof(ulong) ); /* out_depth */
565 0 : l = FD_LAYOUT_APPEND( l, alignof(ulong), out_cnt*sizeof(ulong) ); /* out_seq */
566 0 : l = FD_LAYOUT_APPEND( l, alignof(ulong), out_cnt*sizeof(ulong) ); /* out_seq_pub */
567 0 : l = FD_LAYOUT_APPEND( l, alignof(int), out_cnt*sizeof(int) ); /* out_reliable */
568 0 : l = FD_LAYOUT_APPEND( l, alignof(ulong const *), cons_cnt*sizeof(ulong const *) ); /* cons_fseq */
569 0 : l = FD_LAYOUT_APPEND( l, alignof(ulong *), cons_cnt*sizeof(ulong *) ); /* cons_slow */
570 0 : l = FD_LAYOUT_APPEND( l, alignof(ulong), cons_cnt*sizeof(ulong) ); /* cons_out */
571 0 : l = FD_LAYOUT_APPEND( l, alignof(ulong), cons_cnt*sizeof(ulong) ); /* cons_seq */
572 0 : const ulong event_cnt = in_cnt + 1UL + cons_cnt;
573 0 : l = FD_LAYOUT_APPEND( l, alignof(ushort), event_cnt*sizeof(ushort) ); /* event_map */
574 0 : return FD_LAYOUT_FINI( l, STEM_(scratch_align)() );
575 0 : }
576 :
577 : #if !STEM_ALWAYS_PARK
578 : #define STEM_RUN1_NAME run1_spin
579 : #define STEM_SLEEP_PARKS 0
580 : #include "fd_stem_run1.c"
581 : #undef STEM_SLEEP_PARKS
582 : #undef STEM_RUN1_NAME
583 : #endif
584 :
585 : #if !STEM_NEVER_PARK
586 : #define STEM_RUN1_NAME run1_park
587 : #define STEM_SLEEP_PARKS 1
588 : #include "fd_stem_run1.c"
589 : #undef STEM_SLEEP_PARKS
590 : #undef STEM_RUN1_NAME
591 : #endif
592 :
593 : #if !STEM_ALWAYS_PARK
594 : static __attribute__((noinline)) void
595 : STEM_(run_spin)( ulong in_cnt,
596 : fd_frag_meta_t const ** in_mcache,
597 : ulong ** in_fseq,
598 : ulong out_cnt,
599 : fd_frag_meta_t ** out_mcache,
600 : ulong cons_cnt,
601 : ulong * _cons_out,
602 : ulong ** _cons_fseq,
603 : volatile ulong ** _cons_slow,
604 : ulong burst,
605 : long lazy,
606 : fd_rng_t * rng,
607 : void * scratch,
608 : STEM_CALLBACK_CONTEXT_TYPE * ctx,
609 0 : fd_stem_sleep_t const * sleep ) {
610 0 : STEM_(run1_spin)( in_cnt, in_mcache, in_fseq, out_cnt, out_mcache, cons_cnt, _cons_out, _cons_fseq, _cons_slow, burst, lazy, rng, scratch, ctx, sleep );
611 0 : }
612 : #endif
613 :
614 : #if !STEM_NEVER_PARK
615 : static __attribute__((noinline)) void
616 : STEM_(run_park)( ulong in_cnt,
617 : fd_frag_meta_t const ** in_mcache,
618 : ulong ** in_fseq,
619 : ulong out_cnt,
620 : fd_frag_meta_t ** out_mcache,
621 : ulong cons_cnt,
622 : ulong * _cons_out,
623 : ulong ** _cons_fseq,
624 : volatile ulong ** _cons_slow,
625 : ulong burst,
626 : long lazy,
627 : fd_rng_t * rng,
628 : void * scratch,
629 : STEM_CALLBACK_CONTEXT_TYPE * ctx,
630 0 : fd_stem_sleep_t const * sleep ) {
631 0 : STEM_(run1_park)( in_cnt, in_mcache, in_fseq, out_cnt, out_mcache, cons_cnt, _cons_out, _cons_fseq, _cons_slow, burst, lazy, rng, scratch, ctx, sleep );
632 0 : }
633 : #endif
634 :
635 : static inline void
636 : STEM_(run1)( ulong in_cnt,
637 : fd_frag_meta_t const ** in_mcache,
638 : ulong ** in_fseq,
639 : ulong out_cnt,
640 : fd_frag_meta_t ** out_mcache,
641 : ulong cons_cnt,
642 : ulong * _cons_out,
643 : ulong ** _cons_fseq,
644 : volatile ulong ** _cons_slow,
645 : ulong burst,
646 : long lazy,
647 : fd_rng_t * rng,
648 : void * scratch,
649 : STEM_CALLBACK_CONTEXT_TYPE * ctx,
650 0 : fd_stem_sleep_t const * sleep ) {
651 : #if STEM_ALWAYS_PARK
652 0 : STEM_(run_park)( in_cnt, in_mcache, in_fseq, out_cnt, out_mcache, cons_cnt, _cons_out, _cons_fseq, _cons_slow, burst, lazy, rng, scratch, ctx, sleep );
653 : #elif STEM_NEVER_PARK
654 0 : STEM_(run_spin)( in_cnt, in_mcache, in_fseq, out_cnt, out_mcache, cons_cnt, _cons_out, _cons_fseq, _cons_slow, burst, lazy, rng, scratch, ctx, sleep );
655 : #else
656 0 : if( sleep->shmem ) STEM_(run_park)( in_cnt, in_mcache, in_fseq, out_cnt, out_mcache, cons_cnt, _cons_out, _cons_fseq, _cons_slow, burst, lazy, rng, scratch, ctx, sleep );
657 0 : else STEM_(run_spin)( in_cnt, in_mcache, in_fseq, out_cnt, out_mcache, cons_cnt, _cons_out, _cons_fseq, _cons_slow, burst, lazy, rng, scratch, ctx, sleep );
658 : #endif
659 0 : }
660 :
661 : FD_FN_UNUSED static void
662 : STEM_(run)( fd_topo_t * topo,
663 0 : fd_topo_tile_t * tile ) {
664 0 : const fd_frag_meta_t * in_mcache[ FD_TOPO_MAX_LINKS ];
665 0 : ulong * in_fseq[ FD_TOPO_MAX_TILE_IN_LINKS ];
666 :
667 0 : ulong polled_in_cnt = 0UL;
668 0 : for( ulong i=0UL; i<tile->in_cnt; i++ ) {
669 0 : if( FD_UNLIKELY( !tile->in_link_poll[ i ] ) ) continue;
670 :
671 0 : in_mcache[ polled_in_cnt ] = topo->links[ tile->in_link_id[ i ] ].mcache;
672 0 : FD_TEST( in_mcache[ polled_in_cnt ] );
673 0 : in_fseq[ polled_in_cnt ] = tile->in_link_fseq[ i ];
674 0 : FD_TEST( in_fseq[ polled_in_cnt ] );
675 0 : polled_in_cnt += 1;
676 0 : }
677 :
678 0 : fd_frag_meta_t * out_mcache[ FD_TOPO_MAX_LINKS ];
679 0 : for( ulong i=0UL; i<tile->out_cnt; i++ ) {
680 0 : out_mcache[ i ] = topo->links[ tile->out_link_id[ i ] ].mcache;
681 0 : FD_TEST( out_mcache[ i ] );
682 0 : }
683 :
684 0 : fd_stem_sleep_t sleep[ 1 ] = {{ .shmem = NULL }};
685 0 : if( FD_UNLIKELY( topo->sleep_obj_id!=ULONG_MAX ) ) {
686 0 : sleep->shmem = fd_sleep_join( fd_topo_obj_laddr( topo, topo->sleep_obj_id ) );
687 0 : FD_TEST( sleep->shmem );
688 0 : sleep->tile_id = tile->id;
689 0 : sleep->waker_fseq = tile->is_waker_client ? fd_fseq_join( fd_topo_obj_laddr( topo, tile->waker_fseq_obj_id ) ) : NULL;
690 0 : FD_TEST( !tile->is_waker_client || sleep->waker_fseq );
691 0 : sleep->out_link_id = tile->out_link_id;
692 :
693 0 : ulong polled_idx = 0UL;
694 0 : for( ulong i=0UL; i<tile->in_cnt; i++ ) {
695 0 : if( FD_LIKELY( !tile->in_link_poll[ i ] ) ) continue;
696 0 : sleep->in_link_id [ polled_idx ] = tile->in_link_id[ i ];
697 0 : sleep->in_producer[ polled_idx ] = tile->in_link_reliable[ i ] ? fd_topo_find_link_producer( topo, &topo->links[ tile->in_link_id[ i ] ] ) : ULONG_MAX;
698 0 : polled_idx++;
699 0 : }
700 :
701 0 : ulong pair_cnt = 0UL;
702 0 : for( ulong i=0UL; i<tile->out_cnt; i++ ) {
703 0 : sleep->wake_off[ i ] = (ushort)pair_cnt;
704 0 : pair_cnt += fd_sleep_wake_table( sleep->wake+pair_cnt, topo, tile->out_link_id[ i ] );
705 0 : }
706 0 : sleep->wake_off[ tile->out_cnt ] = (ushort)pair_cnt;
707 0 : }
708 :
709 0 : ulong reliable_cons_cnt = 0UL;
710 0 : ulong cons_out[ FD_TOPO_MAX_LINKS ];
711 0 : ulong * cons_fseq[ FD_TOPO_MAX_LINKS ];
712 0 : volatile ulong * cons_slow[ FD_TOPO_MAX_LINKS ];
713 0 : for( ulong i=0UL; i<topo->tile_cnt; i++ ) {
714 0 : fd_topo_tile_t * consumer_tile = &topo->tiles[ i ];
715 0 : ulong polled_in_idx = 0UL;
716 0 : for( ulong j=0UL; j<consumer_tile->in_cnt; j++ ) {
717 0 : int is_polled = consumer_tile->in_link_poll[ j ];
718 0 : for( ulong k=0UL; k<tile->out_cnt; k++ ) {
719 0 : if( FD_UNLIKELY( consumer_tile->in_link_id[ j ]==tile->out_link_id[ k ] && consumer_tile->in_link_reliable[ j ] ) ) {
720 0 : cons_out[ reliable_cons_cnt ] = k;
721 0 : cons_fseq[ reliable_cons_cnt ] = consumer_tile->in_link_fseq[ j ];
722 0 : FD_TEST( cons_fseq[ reliable_cons_cnt ] );
723 0 : cons_slow[ reliable_cons_cnt ] = fd_metrics_link_in( consumer_tile->metrics, polled_in_idx ) + FD_METRICS_COUNTER_LINK_BACKPRESSURE_BLAMED_OFF;
724 0 : reliable_cons_cnt++;
725 : /* Need to test this, since each link may connect to many outs,
726 : you could construct a topology which has more than this
727 : consumers of links. */
728 0 : FD_TEST( reliable_cons_cnt<FD_TOPO_MAX_LINKS );
729 0 : }
730 0 : }
731 0 : if( FD_LIKELY( is_polled ) ) polled_in_idx++;
732 0 : }
733 0 : }
734 :
735 : /* The stem rng is only used internally for shuffling event/input
736 : polling ordering and for setting the housekeeping timer. It is
737 : never exposed to tile callbacks. As a result, no cryptographic
738 : quality is needed. fd_tickcount() provides per-run entropy,
739 : tile->id guarantees per-tile uniqueness, and fd_ulong_hash (a
740 : Murmur3 finalizer) gives near-perfect avalanche so even a 1-bit
741 : input difference flips ~50% of output bits. fd_rng_secure is not
742 : used here because STEM_(run) executes after the seccomp sandbox
743 : is applied. */
744 0 : fd_rng_t rng[1];
745 0 : FD_TEST( fd_rng_join( fd_rng_new( rng, (uint)fd_ulong_hash( (ulong)fd_tickcount() + tile->id ), 0UL ) ) );
746 :
747 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 );
748 :
749 0 : uchar __attribute__((aligned(FD_STEM_SCRATCH_ALIGN))) stem_scratch[ STEM_(scratch_footprint)( polled_in_cnt, tile->out_cnt, reliable_cons_cnt ) ];
750 :
751 0 : STEM_(run1)( polled_in_cnt,
752 0 : in_mcache,
753 0 : in_fseq,
754 0 : tile->out_cnt,
755 0 : out_mcache,
756 0 : reliable_cons_cnt,
757 0 : cons_out,
758 0 : cons_fseq,
759 0 : cons_slow,
760 0 : STEM_BURST,
761 0 : STEM_LAZY,
762 0 : rng,
763 0 : stem_scratch,
764 0 : ctx,
765 0 : sleep );
766 :
767 : #ifdef STEM_CALLBACK_METRICS_WRITE
768 : /* Write final metrics state before shutting down */
769 0 : FD_COMPILER_MFENCE();
770 0 : STEM_CALLBACK_METRICS_WRITE( ctx );
771 0 : FD_COMPILER_MFENCE();
772 : #endif
773 :
774 0 : if( FD_LIKELY( tile->allow_shutdown ) ) {
775 0 : for( ulong i=0UL; i<tile->in_cnt; i++ ) {
776 0 : if( FD_UNLIKELY( !tile->in_link_poll[ i ] || !tile->in_link_reliable[ i ] ) ) continue;
777 :
778 : /* Return infinite credits on any reliable consumer links so that
779 : producers now no longer expect us to consume. */
780 0 : ulong fseq_id = tile->in_link_fseq_obj_id[ i ];
781 0 : ulong * fseq = fd_fseq_join( fd_topo_obj_laddr( topo, fseq_id ) );
782 0 : FD_TEST( fseq );
783 0 : __atomic_store_n( fseq, STEM_SHUTDOWN_SEQ, __ATOMIC_RELEASE );
784 0 : }
785 :
786 0 : if( FD_UNLIKELY( sleep->shmem ) ) {
787 : /* Leave the sleep object as if we never parked, to prevent
788 : producers repeatedly trying to wake up. */
789 0 : FD_VOLATILE( sleep->shmem->tile[ sleep->tile_id ].word ) = 1UL;
790 0 : __atomic_fetch_and( &sleep->shmem->doorbell [ sleep->tile_id>>6 ], ~(1UL<<(sleep->tile_id&63UL)), __ATOMIC_SEQ_CST );
791 0 : __atomic_fetch_and( &sleep->shmem->parked_bits[ sleep->tile_id>>6 ], ~(1UL<<(sleep->tile_id&63UL)), __ATOMIC_SEQ_CST );
792 0 : }
793 0 : }
794 0 : }
795 :
796 : #undef STEM_NAME
797 : #undef STEM_
798 : #undef STEM_BURST
799 : #undef STEM_ALWAYS_PARK
800 : #undef STEM_NEVER_PARK
801 : #undef STEM_CALLBACK_CONTEXT_TYPE
802 : #undef STEM_CALLBACK_CONTEXT_ALIGN
803 : #undef STEM_LAZY
804 : #undef STEM_STICKY_POLL_MAX
805 : #undef STEM_CALLBACK_SHOULD_SHUTDOWN
806 : #undef STEM_CALLBACK_DURING_HOUSEKEEPING
807 : #undef STEM_CALLBACK_METRICS_WRITE
808 : #undef STEM_CALLBACK_NEXT_DEADLINE
809 : #undef STEM_CALLBACK_PREVENT_PARK
810 : #undef STEM_CALLBACK_PARK_WAIT
811 : #undef STEM_CALLBACK_RECV_CREDIT
812 : #undef STEM_CALLBACK_CHECK_CREDIT
813 : #undef STEM_CALLBACK_BEFORE_CREDIT
814 : #undef STEM_CALLBACK_AFTER_CREDIT
815 : #undef STEM_CALLBACK_BEFORE_FRAG
816 : #undef STEM_CALLBACK_DURING_FRAG
817 : #undef STEM_CALLBACK_RETURNABLE_FRAG
818 : #undef STEM_CALLBACK_AFTER_FRAG
819 : #undef STEM_CALLBACK_AFTER_POLL_OVERRUN
|
