Line data Source code
1 : #ifndef HEADER_fd_src_discof_replay_fd_rdisp_mq_h
2 : #define HEADER_fd_src_discof_replay_fd_rdisp_mq_h
3 :
4 : /* The multiprq (mq) is a strange combination of data structures
5 : specifically for use in fd_rdisp. It's factored out into its own
6 : file to abstract the details away and to facilitate testing. It
7 : supports 4 priority queues (corresponding to the 4 staging lanes),
8 : with each prq also having an optimization for top priority (prio=0)
9 : items. For each element (acct_idx, staging_lane_idx) it supports the
10 : following operations:
11 : * Is the element in the structure?
12 : * Insert the element with a specific priority
13 : * Adjust the element's priority
14 : Additionally, for a staging_lane_idx, it supports:
15 : * Return the element with the best priority (lowest value) and
16 : remove it from the structure.
17 : * Count the number of elements in the structure with the specified
18 : lane index.
19 :
20 : For each element, we require acct_idx in [1, acct_depth] and
21 : staging_lane_idx in [0, 4). Additionally, at most writers_per_block
22 : elements can be in the structure for each staging_lane_idx at a time.
23 : prio>=0 with 0 being the highest priority item. */
24 :
25 : struct fd_rdisp_mq_private;
26 : typedef struct fd_rdisp_mq_private fd_rdisp_mq_t;
27 :
28 : FD_PROTOTYPES_BEGIN
29 :
30 : /* fd_rdisp_mq_{align,footprint} return the required alignment and
31 : footprint in bytes of a region of memory to be formatted to be used
32 : as a multiprq. acct_depth is the maximum acct_idx (inclusive) that
33 : can be used for an element, and max_writers_per_block elements can be
34 : tracked in each staging lane. acct_depth in range [1, UINT_MAX] and
35 : max_writers_per_block in range [1, UINT_MAX-1]. If
36 : max_writers_per_block > acct_depth, the effective value used will be
37 : acct_depth instead. */
38 0 : static inline ulong fd_rdisp_mq_align( void ) { return 8UL; }
39 : static inline ulong fd_rdisp_mq_footprint( ulong acct_depth, ulong max_writers_per_block );
40 :
41 : /* fd_rdisp_mq_new formats a region of memory with the required
42 : footprint and depth for use as a multiprq. acct_depth and
43 : max_writers_per_block are as in footprint. The caller will not be
44 : joined upon return.
45 :
46 : fd_rdisp_mq_join joins the caller to the multiprq, making it ready
47 : for use. */
48 : static inline void *
49 : fd_rdisp_mq_new( void * mem,
50 : ulong acct_depth,
51 : ulong max_writers_per_block );
52 :
53 : static inline fd_rdisp_mq_t * fd_rdisp_mq_join( void * mem );
54 :
55 : /* fd_rdisp_mq_test returns 1 if the element (acct_idx, staging_lane)
56 : is in mq, and 0 otherwise. acct_idx and staging_lane must be in the
57 : requisite range and mq must be a valid local join. An element is in
58 : the multiprq if it has been inserted one more time than it has been
59 : popped. */
60 : static inline int
61 : fd_rdisp_mq_test( fd_rdisp_mq_t const * mq,
62 : ulong acct_idx,
63 : ulong staging_lane );
64 :
65 : /* fd_rdisp_mq_insert inserts the element (acct_idx, staging_lane) into
66 : the multiprq with priority prio. The element must not be in the
67 : multiprq prior to this call. prio>=0, with lower values getting
68 : popped first. */
69 : static inline void
70 : fd_rdisp_mq_insert( fd_rdisp_mq_t * mq,
71 : ulong acct_idx,
72 : ulong staging_lane,
73 : float prio );
74 :
75 : /* fd_rdisp_mq_adjust changes the priority of the specified element
76 : (acct_idx, staging_lane) to prio. The element must be in the
77 : multiprq prior to this call and remains so after the call. prio>=0.
78 : This is especially optimized for the prio==0 case. */
79 : static inline void
80 : fd_rdisp_mq_adjust( fd_rdisp_mq_t * mq,
81 : ulong acct_idx,
82 : ulong staging_lane,
83 : float prio );
84 :
85 : /* fd_rdisp_mq_pop returns and removes the best priority element with
86 : the specified staging lane. The caller promises that there is at
87 : least one element in the multiprq with the specified staging lane.
88 : Ties are broken arbitrarily. */
89 : static inline ulong
90 : fd_rdisp_mq_pop( fd_rdisp_mq_t * mq,
91 : ulong staging_lane );
92 :
93 : /* fd_rdisp_mq_cnt returns the number of elements in the multiprq with
94 : specified staging lane. */
95 : static inline ulong
96 : fd_rdisp_mq_cnt( fd_rdisp_mq_t * mq,
97 : ulong staging_lane );
98 :
99 : static inline void *
100 : fd_rdisp_mq_leave( fd_rdisp_mq_t * mq );
101 :
102 : static inline void *
103 : fd_rdisp_mq_delete( void * mem );
104 :
105 : /* Implementation details below. These would be in a .c file, but they
106 : are mostly pretty quick functions, such that inlining is probably
107 : a significant performance win.
108 :
109 : Probably the most straightforward implementation would be some kind
110 : of pool-based heap, with a map. However, the current pool-based heap
111 : does not allow deleting or changing priorities because it does not
112 : contain a parent pointer. Rather than that, we use fd_prq with an
113 : inverse index that gets updated as elements get moved. This leads to
114 : some strange macros, but they can be isolated to this file and
115 : tested. */
116 :
117 :
118 : struct fd_rdisp_mq_4u {
119 : /* The current position in prq, or UINT_MAX if it's not present. */
120 : uint per_lane[4];
121 : };
122 : typedef struct fd_rdisp_mq_4u fd_rdisp_mq_4u_t;
123 :
124 : struct fd_rdisp_mq_prqe {
125 : uint idx;
126 : float prio;
127 : };
128 : typedef struct fd_rdisp_mq_prqe fd_rdisp_mq_prqe_t;
129 :
130 :
131 : /* In this macro, heap (which is defined each place the macro expands)
132 : is a pointer to the field with offset 16B from private_t. The field
133 : before the private_t struct is a pointer to the start of the 4u_t
134 : array, with 4B*staging lane added. */
135 0 : #define PRQ_TMP_ST(p, t) do { \
136 0 : (p)[0] = (t); \
137 0 : ((fd_rdisp_mq_4u_t *)((void * const *)heap)[-3])[(t).idx].per_lane[0] = (uint)((p)-heap); \
138 0 : } while(0)
139 : #define PRQ_NAME fd_rdisp_mprq
140 0 : #define PRQ_T fd_rdisp_mq_prqe_t
141 0 : #define PRQ_TIMEOUT_T float
142 0 : #define PRQ_TIMEOUT prio
143 : #include "../../util/tmpl/fd_prq.c"
144 : /* As an optimization, we store the prio0 elements at the end of the
145 : heap instead of in the heap. prq allocates 2 dummy elements at the
146 : end that are read but not written to, so we'll store the prio0
147 : elements there.
148 :
149 : Elements [0, prq_cnt) are handled by the prq
150 : Elements (max_writers_per_block-prio0_cnt, max_writers_per_block]
151 : are in the prio0 list.
152 :
153 : There's always one element in between the lists, since
154 : max_writers_per_block-prio0_cnt-prq_cnt >=0. */
155 :
156 : FD_STATIC_ASSERT( sizeof(fd_rdisp_mprq_private_t )==2*sizeof(ulong)+sizeof(fd_rdisp_mq_prqe_t), "prq details" );
157 : FD_STATIC_ASSERT( alignof(fd_rdisp_mprq_private_t)==8UL, "prq details" );
158 :
159 : struct fd_rdisp_mq_private {
160 : ulong acct_depth;
161 : ulong max_writers_per_block;
162 : ulong prio0_cnt[4];
163 :
164 : fd_rdisp_mq_4u_t * flat; /* indexed [0, acct_depth+1) */
165 : fd_rdisp_mq_prqe_t * prq[4];
166 : /* Then follows:
167 : the footprint for flat
168 : for i in [0, 4)
169 : a pointer to flat[0].per_lane[i]
170 : the footprint for prq[i]
171 : */
172 : };
173 : typedef struct fd_rdisp_mq_private fd_rdisp_mq_t;
174 :
175 : static inline ulong
176 : fd_rdisp_mq_footprint( ulong acct_depth,
177 0 : ulong max_writers_per_block ) {
178 : /* The normal case is that max_writers_per_block < acct_depth, so this
179 : is normally a no-op. */
180 0 : max_writers_per_block = fd_ulong_min( max_writers_per_block, acct_depth );
181 0 : ulong l = FD_LAYOUT_INIT;
182 0 : l = FD_LAYOUT_APPEND( l, alignof(fd_rdisp_mq_t), sizeof(fd_rdisp_mq_t) );
183 0 : l = FD_LAYOUT_APPEND( l, alignof(fd_rdisp_mq_4u_t), (acct_depth+1UL)*sizeof(fd_rdisp_mq_4u_t) );
184 0 : l = FD_LAYOUT_APPEND( l, 8UL, 4UL*(sizeof(void *)+fd_rdisp_mprq_footprint( max_writers_per_block )) );
185 0 : return FD_LAYOUT_FINI( l, fd_rdisp_mq_align() );
186 0 : }
187 :
188 : static inline void *
189 : fd_rdisp_mq_new( void * mem,
190 : ulong acct_depth,
191 0 : ulong max_writers_per_block ) {
192 0 : max_writers_per_block = fd_ulong_min( max_writers_per_block, acct_depth );
193 :
194 0 : FD_SCRATCH_ALLOC_INIT( l, mem );
195 0 : fd_rdisp_mq_t * mq = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_rdisp_mq_t), sizeof(fd_rdisp_mq_t) );
196 0 : fd_rdisp_mq_4u_t * u4 = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_rdisp_mq_4u_t), (acct_depth+1UL)*sizeof(fd_rdisp_mq_4u_t) );
197 :
198 0 : mq->acct_depth = acct_depth;
199 0 : mq->max_writers_per_block = max_writers_per_block;
200 0 : memset( mq->prio0_cnt, '\0', 4UL*sizeof(ulong) );
201 :
202 0 : memset( u4, '\xFF', (acct_depth+1UL)*sizeof(fd_rdisp_mq_4u_t) );
203 :
204 0 : for( ulong i=0UL; i<4UL; i++ ) {
205 0 : void * _pq = FD_SCRATCH_ALLOC_APPEND( l, 8UL, sizeof(void *)+fd_rdisp_mprq_footprint( max_writers_per_block ) );
206 0 : fd_rdisp_mprq_new( (void *)(8UL + (ulong)_pq), max_writers_per_block );
207 0 : }
208 0 : return mem;
209 0 : }
210 :
211 : static inline fd_rdisp_mq_t *
212 0 : fd_rdisp_mq_join( void * mem ) {
213 0 : fd_rdisp_mq_t * mq = (fd_rdisp_mq_t *)mem;
214 0 : ulong acct_depth = mq->acct_depth;
215 0 : ulong max_writers_per_block = mq->max_writers_per_block;
216 :
217 0 : FD_SCRATCH_ALLOC_INIT( l, mem );
218 0 : /* */ FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_rdisp_mq_t), sizeof(fd_rdisp_mq_t) );
219 0 : fd_rdisp_mq_4u_t * u4 = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_rdisp_mq_4u_t), (acct_depth+1UL)*sizeof(fd_rdisp_mq_4u_t) );
220 :
221 0 : mq->flat = u4;
222 0 : for( ulong i=0UL; i<4UL; i++ ) {
223 0 : void * _pq = FD_SCRATCH_ALLOC_APPEND( l, 8UL, sizeof(void *)+fd_rdisp_mprq_footprint( max_writers_per_block ) );
224 0 : *((uint **)_pq) = u4->per_lane+i;
225 0 : mq->prq[i] = fd_rdisp_mprq_join( (void *)(sizeof(void *) + (ulong)_pq) );
226 0 : }
227 0 : return mq;
228 0 : }
229 :
230 : static inline int
231 : fd_rdisp_mq_test( fd_rdisp_mq_t const * mq,
232 : ulong acct_idx,
233 0 : ulong staging_lane ) {
234 0 : return mq->flat[acct_idx].per_lane[staging_lane]!=UINT_MAX;
235 0 : }
236 :
237 : static inline void
238 : fd_rdisp_mq_insert( fd_rdisp_mq_t * mq,
239 : ulong acct_idx,
240 : ulong staging_lane,
241 0 : float prio ) {
242 0 : fd_rdisp_mq_prqe_t temp[1] = {{
243 0 : .idx = (uint)acct_idx,
244 0 : .prio = prio,
245 0 : }};
246 0 : if( FD_UNLIKELY( prio==0.0f ) ) {
247 0 : ulong fidx = mq->max_writers_per_block - (mq->prio0_cnt[staging_lane]++);
248 0 : mq->prq[staging_lane][fidx] = *temp;
249 0 : mq->flat[acct_idx].per_lane[staging_lane] = (uint)fidx;
250 0 : } else {
251 0 : fd_rdisp_mprq_insert( mq->prq[staging_lane], temp );
252 : /* flat is auto-populated by _insert */
253 0 : }
254 0 : }
255 :
256 : static inline void
257 : fd_rdisp_mq_adjust( fd_rdisp_mq_t * mq,
258 : ulong acct_idx,
259 : ulong staging_lane,
260 0 : float prio ) {
261 0 : ulong M = mq->max_writers_per_block;
262 0 : ulong p0_cnt = mq->prio0_cnt[staging_lane];
263 :
264 0 : fd_rdisp_mq_prqe_t * prq = mq->prq[staging_lane];
265 :
266 0 : uint fidx = mq->flat[acct_idx].per_lane[staging_lane];
267 0 : int pre_prio0 = fidx>fd_rdisp_mprq_cnt( prq );
268 0 : int new_prio0 = prio==0.0f;
269 :
270 0 : if( FD_UNLIKELY( pre_prio0 & new_prio0 ) ) return; /* nothing to do */
271 0 : fd_rdisp_mq_prqe_t temp = prq[fidx];
272 :
273 0 : if( pre_prio0 ) {
274 : /* move tail to hole (they could be the same) */
275 0 : prq[fidx] = prq[M - (--p0_cnt)];
276 0 : mq->flat[prq[fidx].idx].per_lane[staging_lane] = fidx;
277 0 : } else {
278 0 : fd_rdisp_mprq_remove( prq, fidx );
279 0 : }
280 :
281 0 : temp.prio = prio;
282 :
283 0 : if( new_prio0 ) {
284 0 : ulong new_idx = M-(p0_cnt++);
285 0 : prq[new_idx] = temp;
286 0 : mq->flat[acct_idx].per_lane[staging_lane] = (uint)new_idx;
287 0 : } else {
288 0 : fd_rdisp_mprq_insert( prq, &temp );
289 0 : }
290 :
291 0 : mq->prio0_cnt[staging_lane] = p0_cnt;
292 0 : }
293 :
294 : static inline ulong
295 : fd_rdisp_mq_pop( fd_rdisp_mq_t * mq,
296 0 : ulong staging_lane ) {
297 0 : fd_rdisp_mq_prqe_t * prq = mq->prq[staging_lane];
298 :
299 0 : if( FD_LIKELY( mq->prio0_cnt[ staging_lane ] ) ) {
300 0 : ulong tail_idx = mq->max_writers_per_block - (--mq->prio0_cnt[staging_lane]);
301 0 : ulong to_return = prq[tail_idx].idx;
302 0 : mq->flat[to_return].per_lane[staging_lane] = UINT_MAX;
303 0 : return to_return;
304 0 : } else {
305 0 : ulong to_return = prq->idx;
306 0 : fd_rdisp_mprq_remove_min( prq );
307 0 : mq->flat[to_return].per_lane[staging_lane] = UINT_MAX;
308 0 : return to_return;
309 0 : }
310 0 : }
311 :
312 : static inline ulong
313 : fd_rdisp_mq_cnt( fd_rdisp_mq_t * mq,
314 0 : ulong staging_lane ) {
315 0 : return fd_rdisp_mprq_cnt( mq->prq[staging_lane] ) + mq->prio0_cnt[staging_lane];
316 0 : }
317 :
318 0 : static inline void * fd_rdisp_mq_leave ( fd_rdisp_mq_t * mq ) { return (void *)mq; }
319 0 : static inline void * fd_rdisp_mq_delete( void * mem ) { return (void *)mem; }
320 :
321 : FD_PROTOTYPES_END
322 :
323 : #endif /* HEADER_fd_src_discof_replay_fd_rdisp_mq_h */
|
