Line data Source code
1 : #ifndef HEADER_fd_src_disco_stem_fd_stem_h
2 : #define HEADER_fd_src_disco_stem_fd_stem_h
3 :
4 : #include "../fd_disco_base.h"
5 : #include "../sleep/fd_sleep.h"
6 :
7 0 : #define FD_STEM_SCRATCH_ALIGN (128UL)
8 :
9 : struct fd_stem_context {
10 : fd_frag_meta_t ** mcaches;
11 : ulong * seqs;
12 : ulong * depths;
13 :
14 : long now;
15 : long then;
16 :
17 : ulong * cr_avail;
18 : ulong * min_cr_avail;
19 : ulong cr_decrement_amount;
20 : int * out_reliable;
21 : ulong const * cons_seq;
22 : struct fd_stem_tile_in * in;
23 :
24 : fd_sleep_t * sleep;
25 : fd_sleep_wake_t const * wake;
26 : ushort const * wake_off;
27 : ulong * const * in_fseq;
28 : ulong const * in_producer;
29 : };
30 :
31 : typedef struct fd_stem_context fd_stem_context_t;
32 :
33 : struct fd_stem_sleep {
34 : fd_sleep_t * shmem;
35 :
36 : ulong tile_id;
37 : ulong const * waker_fseq; /* waker readiness word, NULL if not a client */
38 : ulong const * out_link_id; /* per out: link id (seq_mirror), the tile's own array */
39 : ulong in_link_id[ FD_SLEEP_IN_MAX ]; /* link id (seq_snap/sweep) */
40 : ulong in_producer[ FD_SLEEP_IN_MAX ]; /* producer tile id, rung when credits return (ULONG_MAX if none) */
41 : fd_sleep_wake_t wake[ FD_SLEEP_OUT_MAX*FD_SLEEP_BITS_CNT ]; /* flattened (word,mask) pairs */
42 : ushort wake_off[ FD_SLEEP_OUT_MAX+1UL ]; /* out_cnt+1 offsets into wake */
43 : };
44 :
45 : typedef struct fd_stem_sleep fd_stem_sleep_t;
46 :
47 : struct __attribute__((aligned(64))) fd_stem_tile_in {
48 : fd_frag_meta_t const * mcache; /* local join to this in's mcache */
49 : uint depth; /* == fd_mcache_depth( mcache ), depth of this in's cache (const) */
50 : uint idx; /* index of this in in the list of providers, [0, in_cnt) */
51 : ulong seq; /* sequence number of next frag expected from the upstream producer,
52 : updated when frag from this in is published */
53 : fd_frag_meta_t const * mline; /* == mcache + fd_mcache_line_idx( seq, depth ), location to poll next */
54 : ulong * fseq; /* local join to the fseq used to return flow control credits to the in */
55 : uint accum[6]; /* local diagnostic accumulators. These are drained during in housekeeping. */
56 : /* Assumes FD_FSEQ_DIAG_{PUB_CNT,PUB_SZ,FILT_CNT,FILT_SZ,OVRNP_CNT,OVRNP_FRAG_CNT} are 0:5 */
57 : };
58 :
59 : typedef struct fd_stem_tile_in fd_stem_tile_in_t;
60 :
61 : static inline ulong
62 : fd_stem_publish( fd_stem_context_t * stem,
63 : ulong out_idx,
64 : ulong sig,
65 : ulong chunk,
66 : ulong sz,
67 : ulong ctl,
68 : ulong tsorig,
69 0 : ulong tspub ) {
70 0 : fd_frag_meta_t * mcache = stem->mcaches[ out_idx ];
71 0 : ulong depth = stem->depths [ out_idx ];
72 0 : ulong * seqp = &stem->seqs [ out_idx ];
73 0 : ulong seq = *seqp;
74 0 : # if FD_HAS_AVX
75 0 : fd_mcache_publish_avx( mcache, depth, seq, sig, chunk, sz, ctl, tsorig, tspub );
76 : # elif FD_HAS_ARM
77 : fd_mcache_publish_arm( mcache, depth, seq, sig, chunk, sz, ctl, tsorig, tspub );
78 : # else
79 : fd_mcache_publish ( mcache, depth, seq, sig, chunk, sz, ctl, tsorig, tspub );
80 : # endif
81 0 : if( FD_LIKELY( stem->out_reliable[ out_idx ] ) ) {
82 0 : if( FD_UNLIKELY( stem->cr_avail[ out_idx ]<stem->cr_decrement_amount ) ) { /* Ensure producer BURST is set correctly */
83 0 : FD_LOG_ERR(( "BURST underprovisioned out_idx=%lu cr_avail=%lu min_cr_avail=%lu cr_decrement_amount=%lu", out_idx, stem->cr_avail[ out_idx ], *stem->min_cr_avail, stem->cr_decrement_amount ));
84 0 : }
85 0 : stem->cr_avail[ out_idx ] -= stem->cr_decrement_amount;
86 0 : *stem->min_cr_avail = fd_ulong_min( stem->cr_avail[ out_idx ], *stem->min_cr_avail );
87 0 : }
88 0 : *seqp = fd_seq_inc( seq, 1UL );
89 0 : if( FD_UNLIKELY( stem->sleep ) ) fd_sleep_wake_check( stem->sleep, stem->wake+stem->wake_off[ out_idx ], (ulong)(stem->wake_off[ out_idx+1UL ]-stem->wake_off[ out_idx ]) );
90 0 : return seq;
91 0 : }
92 :
93 : static inline ulong
94 : fd_stem_advance( fd_stem_context_t * stem,
95 0 : ulong out_idx ) {
96 0 : ulong * seqp = &stem->seqs[ out_idx ];
97 0 : ulong seq = *seqp;
98 0 : if( FD_LIKELY( stem->out_reliable[ out_idx ] ) ) {
99 0 : if( FD_UNLIKELY( stem->cr_avail[ out_idx ]<stem->cr_decrement_amount ) ) { /* Ensure producer BURST is set correctly */
100 0 : FD_LOG_ERR(( "BURST underprovisioned out_idx=%lu cr_avail=%lu min_cr_avail=%lu cr_decrement_amount=%lu", out_idx, stem->cr_avail[ out_idx ], *stem->min_cr_avail, stem->cr_decrement_amount ));
101 0 : }
102 0 : stem->cr_avail[ out_idx ] -= stem->cr_decrement_amount;
103 0 : *stem->min_cr_avail = fd_ulong_min( stem->cr_avail[ out_idx ], *stem->min_cr_avail );
104 0 : }
105 0 : *seqp = fd_seq_inc( seq, 1UL );
106 0 : if( FD_UNLIKELY( stem->sleep ) ) fd_sleep_wake_check( stem->sleep, stem->wake+stem->wake_off[ out_idx ], (ulong)(stem->wake_off[ out_idx+1UL ]-stem->wake_off[ out_idx ]) );
107 0 : return seq;
108 0 : }
109 :
110 : static inline void
111 : fd_stem_credit_return( fd_stem_context_t * stem,
112 : ulong in_idx,
113 0 : ulong seq ) {
114 0 : __atomic_store_n( stem->in_fseq[ in_idx ], seq, __ATOMIC_RELEASE );
115 0 : if( FD_LIKELY( !stem->sleep ) ) return;
116 0 : ulong producer = stem->in_producer[ in_idx ];
117 0 : if( FD_UNLIKELY( producer==ULONG_MAX ) ) return;
118 0 : __atomic_thread_fence( __ATOMIC_SEQ_CST );
119 0 : if( FD_UNLIKELY( FD_VOLATILE_CONST( stem->sleep->credit_bits[ producer>>6 ] ) & (1UL<<(producer&63UL)) ) ) fd_sleep_ring( stem->sleep, producer );
120 0 : }
121 :
122 : #endif /* HEADER_fd_src_disco_stem_fd_stem_h */
|
