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 18056 : 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 : /* fd_rdisp_mq_prio0_cnt returns the number of elements in the multiprq
100 : with the specified staging lane that have priority 0. */
101 : static inline ulong
102 : fd_rdisp_mq_prio0_cnt( fd_rdisp_mq_t const * mq,
103 : ulong staging_lane );
104 :
105 : /* fd_rdisp_mq_test_prio0 returns 1 if the element
106 : (acct_idx,staging_lane) has priority 0, and 0 otherwise. The element
107 : must be in the multiprq. */
108 : static inline int
109 : fd_rdisp_mq_test_prio0( fd_rdisp_mq_t const * mq,
110 : ulong acct_idx,
111 : ulong staging_lane );
112 :
113 : static inline void *
114 : fd_rdisp_mq_leave( fd_rdisp_mq_t * mq );
115 :
116 : static inline void *
117 : fd_rdisp_mq_delete( void * mem );
118 :
119 : /* Implementation details below. These would be in a .c file, but they
120 : are mostly pretty quick functions, such that inlining is probably
121 : a significant performance win.
122 :
123 : Probably the most straightforward implementation would be some kind
124 : of pool-based heap, with a map. However, the current pool-based heap
125 : does not allow deleting or changing priorities because it does not
126 : contain a parent pointer. Rather than that, we use fd_prq with an
127 : inverse index that gets updated as elements get moved. This leads to
128 : some strange macros, but they can be isolated to this file and
129 : tested. */
130 :
131 :
132 : struct fd_rdisp_mq_4u {
133 : /* The current position in prq, or UINT_MAX if it's not present. */
134 : uint per_lane[4];
135 : };
136 : typedef struct fd_rdisp_mq_4u fd_rdisp_mq_4u_t;
137 :
138 : struct fd_rdisp_mq_prqe {
139 : uint idx;
140 : float prio;
141 : };
142 : typedef struct fd_rdisp_mq_prqe fd_rdisp_mq_prqe_t;
143 :
144 :
145 : /* In this macro, heap (which is defined each place the macro expands)
146 : is a pointer to the field with offset 16B from private_t. The field
147 : before the private_t struct is a pointer to the start of the 4u_t
148 : array, with 4B*staging lane added. */
149 2888 : #define PRQ_TMP_ST(p, t) do { \
150 2888 : (p)[0] = (t); \
151 2888 : ((fd_rdisp_mq_4u_t *)((void * const *)heap)[-3])[(t).idx].per_lane[0] = (uint)((p)-heap); \
152 2888 : } while(0)
153 : #define PRQ_NAME fd_rdisp_mprq
154 4806 : #define PRQ_T fd_rdisp_mq_prqe_t
155 4806 : #define PRQ_TIMEOUT_T float
156 4806 : #define PRQ_TIMEOUT prio
157 : #include "../../util/tmpl/fd_prq.c"
158 : /* As an optimization, we store the prio0 elements at the end of the
159 : heap instead of in the heap. prq allocates 2 dummy elements at the
160 : end that are read but not written to, so we'll store the prio0
161 : elements there.
162 :
163 : Elements [0, prq_cnt) are handled by the prq
164 : Elements (max_writers_per_block-prio0_cnt, max_writers_per_block]
165 : are in the prio0 list.
166 :
167 : There's always one element in between the lists, since
168 : max_writers_per_block-prio0_cnt-prq_cnt >=0. */
169 :
170 : FD_STATIC_ASSERT( sizeof(fd_rdisp_mprq_private_t )==2*sizeof(ulong)+sizeof(fd_rdisp_mq_prqe_t), "prq details" );
171 : FD_STATIC_ASSERT( alignof(fd_rdisp_mprq_private_t)==8UL, "prq details" );
172 :
173 : struct fd_rdisp_mq_private {
174 : ulong acct_depth;
175 : ulong max_writers_per_block;
176 : ulong prio0_cnt[4];
177 :
178 : fd_rdisp_mq_4u_t * flat; /* indexed [0, acct_depth+1) */
179 : fd_rdisp_mq_prqe_t * prq[4];
180 : /* Then follows:
181 : the footprint for flat
182 : for i in [0, 4)
183 : a pointer to flat[0].per_lane[i]
184 : the footprint for prq[i]
185 : */
186 : };
187 : typedef struct fd_rdisp_mq_private fd_rdisp_mq_t;
188 :
189 : static inline ulong
190 : fd_rdisp_mq_footprint( ulong acct_depth,
191 5031 : ulong max_writers_per_block ) {
192 : /* The normal case is that max_writers_per_block < acct_depth, so this
193 : is normally a no-op. */
194 5031 : max_writers_per_block = fd_ulong_min( max_writers_per_block, acct_depth );
195 5031 : ulong l = FD_LAYOUT_INIT;
196 5031 : l = FD_LAYOUT_APPEND( l, alignof(fd_rdisp_mq_t), sizeof(fd_rdisp_mq_t) );
197 5031 : l = FD_LAYOUT_APPEND( l, alignof(fd_rdisp_mq_4u_t), (acct_depth+1UL)*sizeof(fd_rdisp_mq_4u_t) );
198 5031 : l = FD_LAYOUT_APPEND( l, 8UL, 4UL*(sizeof(void *)+fd_rdisp_mprq_footprint( max_writers_per_block )) );
199 5031 : return FD_LAYOUT_FINI( l, fd_rdisp_mq_align() );
200 5031 : }
201 :
202 : static inline void *
203 : fd_rdisp_mq_new( void * mem,
204 : ulong acct_depth,
205 1008 : ulong max_writers_per_block ) {
206 1008 : max_writers_per_block = fd_ulong_min( max_writers_per_block, acct_depth );
207 :
208 1008 : FD_SCRATCH_ALLOC_INIT( l, mem );
209 1008 : fd_rdisp_mq_t * mq = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_rdisp_mq_t), sizeof(fd_rdisp_mq_t) );
210 1008 : 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) );
211 :
212 0 : mq->acct_depth = acct_depth;
213 1008 : mq->max_writers_per_block = max_writers_per_block;
214 1008 : memset( mq->prio0_cnt, '\0', 4UL*sizeof(ulong) );
215 :
216 1008 : memset( u4, '\xFF', (acct_depth+1UL)*sizeof(fd_rdisp_mq_4u_t) );
217 :
218 5040 : for( ulong i=0UL; i<4UL; i++ ) {
219 4032 : void * _pq = FD_SCRATCH_ALLOC_APPEND( l, 8UL, sizeof(void *)+fd_rdisp_mprq_footprint( max_writers_per_block ) );
220 0 : fd_rdisp_mprq_new( (void *)(8UL + (ulong)_pq), max_writers_per_block );
221 4032 : }
222 1008 : return mem;
223 1008 : }
224 :
225 : static inline fd_rdisp_mq_t *
226 1008 : fd_rdisp_mq_join( void * mem ) {
227 1008 : fd_rdisp_mq_t * mq = (fd_rdisp_mq_t *)mem;
228 1008 : ulong acct_depth = mq->acct_depth;
229 1008 : ulong max_writers_per_block = mq->max_writers_per_block;
230 :
231 1008 : FD_SCRATCH_ALLOC_INIT( l, mem );
232 1008 : /* */ FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_rdisp_mq_t), sizeof(fd_rdisp_mq_t) );
233 1008 : 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) );
234 :
235 0 : mq->flat = u4;
236 5040 : for( ulong i=0UL; i<4UL; i++ ) {
237 4032 : void * _pq = FD_SCRATCH_ALLOC_APPEND( l, 8UL, sizeof(void *)+fd_rdisp_mprq_footprint( max_writers_per_block ) );
238 0 : *((uint **)_pq) = u4->per_lane+i;
239 4032 : mq->prq[i] = fd_rdisp_mprq_join( (void *)(sizeof(void *) + (ulong)_pq) );
240 4032 : }
241 1008 : return mq;
242 1008 : }
243 :
244 : static inline int
245 : fd_rdisp_mq_test( fd_rdisp_mq_t const * mq,
246 : ulong acct_idx,
247 6226 : ulong staging_lane ) {
248 6226 : return mq->flat[acct_idx].per_lane[staging_lane]!=UINT_MAX;
249 6226 : }
250 :
251 : static inline int
252 : fd_rdisp_mq_test_prio0( fd_rdisp_mq_t const * mq,
253 : ulong acct_idx,
254 224 : ulong staging_lane ) {
255 224 : return mq->flat[acct_idx].per_lane[staging_lane]>fd_rdisp_mprq_cnt( mq->prq[staging_lane] );
256 224 : }
257 :
258 : static inline ulong
259 : fd_rdisp_mq_prio0_cnt( fd_rdisp_mq_t const * mq,
260 0 : ulong staging_lane ) {
261 0 : return mq->prio0_cnt[staging_lane];
262 0 : }
263 :
264 : static inline void
265 : fd_rdisp_mq_insert( fd_rdisp_mq_t * mq,
266 : ulong acct_idx,
267 : ulong staging_lane,
268 2888 : float prio ) {
269 2888 : fd_rdisp_mq_prqe_t temp[1] = {{
270 2888 : .idx = (uint)acct_idx,
271 2888 : .prio = prio,
272 2888 : }};
273 2888 : if( FD_UNLIKELY( prio==0.0f ) ) {
274 0 : ulong fidx = mq->max_writers_per_block - (mq->prio0_cnt[staging_lane]++);
275 0 : mq->prq[staging_lane][fidx] = *temp;
276 0 : mq->flat[acct_idx].per_lane[staging_lane] = (uint)fidx;
277 2888 : } else {
278 2888 : fd_rdisp_mprq_insert( mq->prq[staging_lane], temp );
279 : /* flat is auto-populated by _insert */
280 2888 : }
281 2888 : }
282 :
283 : static inline void
284 : fd_rdisp_mq_adjust( fd_rdisp_mq_t * mq,
285 : ulong acct_idx,
286 : ulong staging_lane,
287 0 : float prio ) {
288 0 : ulong M = mq->max_writers_per_block;
289 0 : ulong p0_cnt = mq->prio0_cnt[staging_lane];
290 :
291 0 : fd_rdisp_mq_prqe_t * prq = mq->prq[staging_lane];
292 :
293 0 : uint fidx = mq->flat[acct_idx].per_lane[staging_lane];
294 0 : int pre_prio0 = fd_rdisp_mq_test_prio0( mq, acct_idx, staging_lane );
295 0 : int new_prio0 = prio==0.0f;
296 :
297 0 : if( FD_UNLIKELY( pre_prio0 & new_prio0 ) ) return; /* nothing to do */
298 0 : fd_rdisp_mq_prqe_t temp = prq[fidx];
299 :
300 0 : if( pre_prio0 ) {
301 : /* move tail to hole (they could be the same) */
302 0 : prq[fidx] = prq[M - (--p0_cnt)];
303 0 : mq->flat[prq[fidx].idx].per_lane[staging_lane] = fidx;
304 0 : } else {
305 0 : fd_rdisp_mprq_remove( prq, fidx );
306 0 : }
307 :
308 0 : temp.prio = prio;
309 :
310 0 : if( new_prio0 ) {
311 0 : ulong new_idx = M-(p0_cnt++);
312 0 : prq[new_idx] = temp;
313 0 : mq->flat[acct_idx].per_lane[staging_lane] = (uint)new_idx;
314 0 : } else {
315 0 : fd_rdisp_mprq_insert( prq, &temp );
316 0 : }
317 :
318 0 : mq->prio0_cnt[staging_lane] = p0_cnt;
319 0 : }
320 :
321 : static inline ulong
322 : fd_rdisp_mq_pop( fd_rdisp_mq_t * mq,
323 0 : ulong staging_lane ) {
324 0 : fd_rdisp_mq_prqe_t * prq = mq->prq[staging_lane];
325 :
326 0 : if( FD_LIKELY( mq->prio0_cnt[ staging_lane ] ) ) {
327 0 : ulong tail_idx = mq->max_writers_per_block - (--mq->prio0_cnt[staging_lane]);
328 0 : ulong to_return = prq[tail_idx].idx;
329 0 : mq->flat[to_return].per_lane[staging_lane] = UINT_MAX;
330 0 : return to_return;
331 0 : } else {
332 0 : ulong to_return = prq->idx;
333 0 : fd_rdisp_mprq_remove_min( prq );
334 0 : mq->flat[to_return].per_lane[staging_lane] = UINT_MAX;
335 0 : return to_return;
336 0 : }
337 0 : }
338 :
339 : static inline ulong
340 : fd_rdisp_mq_cnt( fd_rdisp_mq_t * mq,
341 2891 : ulong staging_lane ) {
342 2891 : return fd_rdisp_mprq_cnt( mq->prq[staging_lane] ) + mq->prio0_cnt[staging_lane];
343 2891 : }
344 :
345 0 : static inline void * fd_rdisp_mq_leave ( fd_rdisp_mq_t * mq ) { return (void *)mq; }
346 0 : static inline void * fd_rdisp_mq_delete( void * mem ) { return (void *)mem; }
347 :
348 : FD_PROTOTYPES_END
349 :
350 : #endif /* HEADER_fd_src_discof_replay_fd_rdisp_mq_h */
|
