1 /*
2 * bcm_ring.h : Ring context abstraction
3 * The ring context tracks the WRITE and READ indices where elements may be
4 * produced and consumed respectively. All elements in the ring need to be
5 * fixed size.
6 *
7 * NOTE: A ring of size N, may only hold N-1 elements.
8 *
9 * Copyright (C) 2020, Broadcom.
10 *
11 * Unless you and Broadcom execute a separate written software license
12 * agreement governing use of this software, this software is licensed to you
13 * under the terms of the GNU General Public License version 2 (the "GPL"),
14 * available at http://www.broadcom.com/licenses/GPLv2.php, with the
15 * following added to such license:
16 *
17 * As a special exception, the copyright holders of this software give you
18 * permission to link this software with independent modules, and to copy and
19 * distribute the resulting executable under terms of your choice, provided that
20 * you also meet, for each linked independent module, the terms and conditions of
21 * the license of that module. An independent module is a module which is not
22 * derived from this software. The special exception does not apply to any
23 * modifications of the software.
24 *
25 *
26 * <<Broadcom-WL-IPTag/Dual:>>
27 */
28 #ifndef __bcm_ring_included__
29 #define __bcm_ring_included__
30 /*
31 * API Notes:
32 *
33 * Ring manipulation API allows for:
34 * Pending operations: Often before some work can be completed, it may be
35 * desired that several resources are available, e.g. space for production in
36 * a ring. Approaches such as, #1) reserve resources one by one and return them
37 * if another required resource is not available, or #2) employ a two pass
38 * algorithm of first testing whether all resources are available, have a
39 * an impact on performance critical code. The approach taken here is more akin
40 * to approach #2, where a test for resource availability essentially also
41 * provides the index for production in an un-committed state.
42 * The same approach is taken for the consumer side.
43 *
44 * - Pending production: Fetch the next index where a ring element may be
45 * produced. The caller may not commit the WRITE of the element.
46 * - Pending consumption: Fetch the next index where a ring element may be
47 * consumed. The caller may not commut the READ of the element.
48 *
49 * Producer side API:
50 * - bcm_ring_is_full : Test whether ring is full
51 * - bcm_ring_prod : Fetch index where an element may be produced (commit)
52 * - bcm_ring_prod_pend: Fetch index where an element may be produced (pending)
53 * - bcm_ring_prod_done: Commit a previous pending produce fetch
54 * - bcm_ring_prod_avail: Fetch total number free slots eligible for production
55 *
56 * Consumer side API:
57 * - bcm_ring_is_empty : Test whether ring is empty
58 * - bcm_ring_cons : Fetch index where an element may be consumed (commit)
59 * - bcm_ring_cons_pend: Fetch index where an element may be consumed (pending)
60 * - bcm_ring_cons_done: Commit a previous pending consume fetch
61 * - bcm_ring_cons_avail: Fetch total number elements eligible for consumption
62 *
63 * - bcm_ring_sync_read: Sync read offset in peer ring, from local ring
64 * - bcm_ring_sync_write: Sync write offset in peer ring, from local ring
65 *
66 * +----------------------------------------------------------------------------
67 *
68 * Design Notes:
69 * Following items are not tracked in a ring context (design decision)
70 * - width of a ring element.
71 * - depth of the ring.
72 * - base of the buffer, where the elements are stored.
73 * - count of number of free slots in the ring
74 *
75 * Implementation Notes:
76 * - When BCM_RING_DEBUG is enabled, need explicit bcm_ring_init().
77 * - BCM_RING_EMPTY and BCM_RING_FULL are (-1)
78 *
79 * +----------------------------------------------------------------------------
80 *
81 * Usage Notes:
82 * An application may incarnate a ring of some fixed sized elements, by defining
83 * - a ring data buffer to store the ring elements.
84 * - depth of the ring (max number of elements managed by ring context).
85 * Preferrably, depth may be represented as a constant.
86 * - width of a ring element: to be used in pointer arithmetic with the ring's
87 * data buffer base and an index to fetch the ring element.
88 *
89 * Use bcm_workq_t to instantiate a pair of workq constructs, one for the
90 * producer and the other for the consumer, both pointing to the same circular
91 * buffer. The producer may operate on it's own local workq and flush the write
92 * index to the consumer. Likewise the consumer may use its local workq and
93 * flush the read index to the producer. This way we do not repeatedly access
94 * the peer's context. The two peers may reside on different CPU cores with a
95 * private L1 data cache.
96 * +----------------------------------------------------------------------------
97 *
98 * -*- Mode: C; tab-width: 4; indent-tabs-mode: t; c-basic-offset: 4 -*-
99 * vim: set ts=4 noet sw=4 tw=80:
100 *
101 * +----------------------------------------------------------------------------
102 */
103
104 #ifdef ____cacheline_aligned
105 #define __ring_aligned ____cacheline_aligned
106 #else
107 #define __ring_aligned
108 #endif
109
110 /* Conditional compile for debug */
111 /* #define BCM_RING_DEBUG */
112
113 #define BCM_RING_EMPTY (-1)
114 #define BCM_RING_FULL (-1)
115 #define BCM_RING_NULL ((bcm_ring_t *)NULL)
116
117 #if defined(BCM_RING_DEBUG)
118 #define RING_ASSERT(exp) ASSERT(exp)
119 #define BCM_RING_IS_VALID(ring) (((ring) != BCM_RING_NULL) && \
120 ((ring)->self == (ring)))
121 #else /* ! BCM_RING_DEBUG */
122 #define RING_ASSERT(exp) do {} while (0)
123 #define BCM_RING_IS_VALID(ring) ((ring) != BCM_RING_NULL)
124 #endif /* ! BCM_RING_DEBUG */
125
126 #define BCM_RING_SIZE_IS_VALID(ring_size) ((ring_size) > 0)
127
128 /*
129 * +----------------------------------------------------------------------------
130 * Ring Context
131 * +----------------------------------------------------------------------------
132 */
133 typedef struct bcm_ring { /* Ring context */
134 #if defined(BCM_RING_DEBUG)
135 struct bcm_ring *self; /* ptr to self for IS VALID test */
136 #endif /* BCM_RING_DEBUG */
137 int write __ring_aligned; /* WRITE index in a circular ring */
138 int read __ring_aligned; /* READ index in a circular ring */
139 } bcm_ring_t;
140
141 static INLINE void bcm_ring_init(bcm_ring_t *ring);
142 static INLINE void bcm_ring_copy(bcm_ring_t *to, bcm_ring_t *from);
143 static INLINE bool bcm_ring_is_empty(const bcm_ring_t *ring);
144
145 static INLINE int __bcm_ring_next_write(const bcm_ring_t *ring, const int ring_size);
146
147 static INLINE bool __bcm_ring_full(const bcm_ring_t *ring, int next_write);
148 static INLINE bool bcm_ring_is_full(bcm_ring_t *ring, const int ring_size);
149
150 static INLINE void bcm_ring_prod_done(bcm_ring_t *ring, int write);
151 static INLINE int bcm_ring_prod_pend(const bcm_ring_t *ring, int *pend_write,
152 const int ring_size);
153 static INLINE int bcm_ring_prod(bcm_ring_t *ring, const int ring_size);
154
155 static INLINE void bcm_ring_cons_done(bcm_ring_t *ring, int read);
156 static INLINE int bcm_ring_cons_pend(const bcm_ring_t *ring, int *pend_read,
157 const int ring_size);
158 static INLINE int bcm_ring_cons(bcm_ring_t *ring, const int ring_size);
159
160 static INLINE void bcm_ring_sync_read(bcm_ring_t *peer, const bcm_ring_t *self);
161 static INLINE void bcm_ring_sync_write(bcm_ring_t *peer, const bcm_ring_t *self);
162
163 static INLINE int bcm_ring_prod_avail(const bcm_ring_t *ring,
164 const int ring_size);
165 static INLINE int bcm_ring_cons_avail(const bcm_ring_t *ring,
166 const int ring_size);
167 static INLINE void bcm_ring_cons_all(bcm_ring_t *ring);
168
169 /**
170 * bcm_ring_init - initialize a ring context.
171 * @ring: pointer to a ring context
172 */
173 static INLINE void
bcm_ring_init(bcm_ring_t * ring)174 bcm_ring_init(bcm_ring_t *ring)
175 {
176 ASSERT(ring != (bcm_ring_t *)NULL);
177 #if defined(BCM_RING_DEBUG)
178 ring->self = ring;
179 #endif /* BCM_RING_DEBUG */
180 ring->write = 0;
181 ring->read = 0;
182 }
183
184 /**
185 * bcm_ring_copy - copy construct a ring
186 * @to: pointer to the new ring context
187 * @from: pointer to orig ring context
188 */
189 static INLINE void
bcm_ring_copy(bcm_ring_t * to,bcm_ring_t * from)190 bcm_ring_copy(bcm_ring_t *to, bcm_ring_t *from)
191 {
192 bcm_ring_init(to);
193
194 to->write = from->write;
195 to->read = from->read;
196 }
197
198 /**
199 * bcm_ring_is_empty - "Boolean" test whether ring is empty.
200 * @ring: pointer to a ring context
201 *
202 * PS. does not return BCM_RING_EMPTY value.
203 */
204 static INLINE bool
bcm_ring_is_empty(const bcm_ring_t * ring)205 bcm_ring_is_empty(const bcm_ring_t *ring)
206 {
207 RING_ASSERT(BCM_RING_IS_VALID(ring));
208 return (ring->read == ring->write);
209 }
210
211 /**
212 * __bcm_ring_next_write - determine the index where the next write may occur
213 * (with wrap-around).
214 * @ring: pointer to a ring context
215 * @ring_size: size of the ring
216 *
217 * PRIVATE INTERNAL USE ONLY.
218 */
219 static INLINE int
__bcm_ring_next_write(const bcm_ring_t * ring,const int ring_size)220 __bcm_ring_next_write(const bcm_ring_t *ring, const int ring_size)
221 {
222 RING_ASSERT(BCM_RING_IS_VALID(ring) && BCM_RING_SIZE_IS_VALID(ring_size));
223 return ((ring->write + 1) % ring_size);
224 }
225
226 /**
227 * __bcm_ring_full - support function for ring full test.
228 * @ring: pointer to a ring context
229 * @next_write: next location in ring where an element is to be produced
230 *
231 * PRIVATE INTERNAL USE ONLY.
232 */
233 static INLINE bool
__bcm_ring_full(const bcm_ring_t * ring,int next_write)234 __bcm_ring_full(const bcm_ring_t *ring, int next_write)
235 {
236 return (next_write == ring->read);
237 }
238
239 /**
240 * bcm_ring_is_full - "Boolean" test whether a ring is full.
241 * @ring: pointer to a ring context
242 * @ring_size: size of the ring
243 *
244 * PS. does not return BCM_RING_FULL value.
245 */
246 static INLINE bool
bcm_ring_is_full(bcm_ring_t * ring,const int ring_size)247 bcm_ring_is_full(bcm_ring_t *ring, const int ring_size)
248 {
249 int next_write;
250 RING_ASSERT(BCM_RING_IS_VALID(ring) && BCM_RING_SIZE_IS_VALID(ring_size));
251 next_write = __bcm_ring_next_write(ring, ring_size);
252 return __bcm_ring_full(ring, next_write);
253 }
254
255 /**
256 * bcm_ring_prod_done - commit a previously pending index where production
257 * was requested.
258 * @ring: pointer to a ring context
259 * @write: index into ring upto where production was done.
260 * +----------------------------------------------------------------------------
261 */
262 static INLINE void
bcm_ring_prod_done(bcm_ring_t * ring,int write)263 bcm_ring_prod_done(bcm_ring_t *ring, int write)
264 {
265 RING_ASSERT(BCM_RING_IS_VALID(ring));
266 ring->write = write;
267 }
268
269 /**
270 * bcm_ring_prod_pend - Fetch in "pend" mode, the index where an element may be
271 * produced.
272 * @ring: pointer to a ring context
273 * @pend_write: next index, after the returned index
274 * @ring_size: size of the ring
275 */
276 static INLINE int
bcm_ring_prod_pend(const bcm_ring_t * ring,int * pend_write,const int ring_size)277 bcm_ring_prod_pend(const bcm_ring_t *ring, int *pend_write, const int ring_size)
278 {
279 int rtn;
280 RING_ASSERT(BCM_RING_IS_VALID(ring) && BCM_RING_SIZE_IS_VALID(ring_size));
281 *pend_write = __bcm_ring_next_write(ring, ring_size);
282 if (__bcm_ring_full(ring, *pend_write)) {
283 *pend_write = BCM_RING_FULL;
284 rtn = BCM_RING_FULL;
285 } else {
286 /* production is not committed, caller needs to explicitly commit */
287 rtn = ring->write;
288 }
289 return rtn;
290 }
291
292 /**
293 * bcm_ring_prod - Fetch and "commit" the next index where a ring element may
294 * be produced.
295 * @ring: pointer to a ring context
296 * @ring_size: size of the ring
297 */
298 static INLINE int
bcm_ring_prod(bcm_ring_t * ring,const int ring_size)299 bcm_ring_prod(bcm_ring_t *ring, const int ring_size)
300 {
301 int next_write, prod_write;
302 RING_ASSERT(BCM_RING_IS_VALID(ring) && BCM_RING_SIZE_IS_VALID(ring_size));
303
304 next_write = __bcm_ring_next_write(ring, ring_size);
305 if (__bcm_ring_full(ring, next_write)) {
306 prod_write = BCM_RING_FULL;
307 } else {
308 prod_write = ring->write;
309 bcm_ring_prod_done(ring, next_write); /* "commit" production */
310 }
311 return prod_write;
312 }
313
314 /**
315 * bcm_ring_cons_done - commit a previously pending read
316 * @ring: pointer to a ring context
317 * @read: index upto which elements have been consumed.
318 */
319 static INLINE void
bcm_ring_cons_done(bcm_ring_t * ring,int read)320 bcm_ring_cons_done(bcm_ring_t *ring, int read)
321 {
322 RING_ASSERT(BCM_RING_IS_VALID(ring));
323 ring->read = read;
324 }
325
326 /**
327 * bcm_ring_cons_pend - fetch in "pend" mode, the next index where a ring
328 * element may be consumed.
329 * @ring: pointer to a ring context
330 * @pend_read: index into ring upto which elements may be consumed.
331 * @ring_size: size of the ring
332 */
333 static INLINE int
bcm_ring_cons_pend(const bcm_ring_t * ring,int * pend_read,const int ring_size)334 bcm_ring_cons_pend(const bcm_ring_t *ring, int *pend_read, const int ring_size)
335 {
336 int rtn;
337 RING_ASSERT(BCM_RING_IS_VALID(ring) && BCM_RING_SIZE_IS_VALID(ring_size));
338 if (bcm_ring_is_empty(ring)) {
339 *pend_read = BCM_RING_EMPTY;
340 rtn = BCM_RING_EMPTY;
341 } else {
342 *pend_read = (ring->read + 1) % ring_size;
343 /* production is not committed, caller needs to explicitly commit */
344 rtn = ring->read;
345 }
346 return rtn;
347 }
348
349 /**
350 * bcm_ring_cons - fetch and "commit" the next index where a ring element may
351 * be consumed.
352 * @ring: pointer to a ring context
353 * @ring_size: size of the ring
354 */
355 static INLINE int
bcm_ring_cons(bcm_ring_t * ring,const int ring_size)356 bcm_ring_cons(bcm_ring_t *ring, const int ring_size)
357 {
358 int cons_read;
359 RING_ASSERT(BCM_RING_IS_VALID(ring) && BCM_RING_SIZE_IS_VALID(ring_size));
360 if (bcm_ring_is_empty(ring)) {
361 cons_read = BCM_RING_EMPTY;
362 } else {
363 cons_read = ring->read;
364 ring->read = (ring->read + 1) % ring_size; /* read is committed */
365 }
366 return cons_read;
367 }
368
369 /**
370 * bcm_ring_sync_read - on consumption, update peer's read index.
371 * @peer: pointer to peer's producer ring context
372 * @self: pointer to consumer's ring context
373 */
374 static INLINE void
bcm_ring_sync_read(bcm_ring_t * peer,const bcm_ring_t * self)375 bcm_ring_sync_read(bcm_ring_t *peer, const bcm_ring_t *self)
376 {
377 RING_ASSERT(BCM_RING_IS_VALID(peer));
378 RING_ASSERT(BCM_RING_IS_VALID(self));
379 peer->read = self->read; /* flush read update to peer producer */
380 }
381
382 /**
383 * bcm_ring_sync_write - on consumption, update peer's write index.
384 * @peer: pointer to peer's consumer ring context
385 * @self: pointer to producer's ring context
386 */
387 static INLINE void
bcm_ring_sync_write(bcm_ring_t * peer,const bcm_ring_t * self)388 bcm_ring_sync_write(bcm_ring_t *peer, const bcm_ring_t *self)
389 {
390 RING_ASSERT(BCM_RING_IS_VALID(peer));
391 RING_ASSERT(BCM_RING_IS_VALID(self));
392 peer->write = self->write; /* flush write update to peer consumer */
393 }
394
395 /**
396 * bcm_ring_prod_avail - fetch total number of available empty slots in the
397 * ring for production.
398 * @ring: pointer to a ring context
399 * @ring_size: size of the ring
400 */
401 static INLINE int
bcm_ring_prod_avail(const bcm_ring_t * ring,const int ring_size)402 bcm_ring_prod_avail(const bcm_ring_t *ring, const int ring_size)
403 {
404 int prod_avail;
405 RING_ASSERT(BCM_RING_IS_VALID(ring) && BCM_RING_SIZE_IS_VALID(ring_size));
406 if (ring->write >= ring->read) {
407 prod_avail = (ring_size - (ring->write - ring->read) - 1);
408 } else {
409 prod_avail = (ring->read - (ring->write + 1));
410 }
411 ASSERT(prod_avail < ring_size);
412 return prod_avail;
413 }
414
415 /**
416 * bcm_ring_cons_avail - fetch total number of available elements for consumption.
417 * @ring: pointer to a ring context
418 * @ring_size: size of the ring
419 */
420 static INLINE int
bcm_ring_cons_avail(const bcm_ring_t * ring,const int ring_size)421 bcm_ring_cons_avail(const bcm_ring_t *ring, const int ring_size)
422 {
423 int cons_avail;
424 RING_ASSERT(BCM_RING_IS_VALID(ring) && BCM_RING_SIZE_IS_VALID(ring_size));
425 if (ring->read == ring->write) {
426 cons_avail = 0;
427 } else if (ring->read > ring->write) {
428 cons_avail = ((ring_size - ring->read) + ring->write);
429 } else {
430 cons_avail = ring->write - ring->read;
431 }
432 ASSERT(cons_avail < ring_size);
433 return cons_avail;
434 }
435
436 /**
437 * bcm_ring_cons_all - set ring in state where all elements are consumed.
438 * @ring: pointer to a ring context
439 */
440 static INLINE void
bcm_ring_cons_all(bcm_ring_t * ring)441 bcm_ring_cons_all(bcm_ring_t *ring)
442 {
443 ring->read = ring->write;
444 }
445
446 /**
447 * Work Queue
448 * A work Queue is composed of a ring of work items, of a specified depth.
449 * It HAS-A bcm_ring object, comprising of a RD and WR offset, to implement a
450 * producer/consumer circular ring.
451 */
452
453 struct bcm_workq {
454 bcm_ring_t ring; /* Ring context abstraction */
455 struct bcm_workq *peer; /* Peer workq context */
456 void *buffer; /* Buffer storage for work items in workQ */
457 int ring_size; /* Depth of workQ */
458 } __ring_aligned;
459
460 typedef struct bcm_workq bcm_workq_t;
461
462 /* #define BCM_WORKQ_DEBUG */
463 #if defined(BCM_WORKQ_DEBUG)
464 #define WORKQ_ASSERT(exp) ASSERT(exp)
465 #else /* ! BCM_WORKQ_DEBUG */
466 #define WORKQ_ASSERT(exp) do {} while (0)
467 #endif /* ! BCM_WORKQ_DEBUG */
468
469 #define WORKQ_AUDIT(workq) \
470 WORKQ_ASSERT((workq) != BCM_WORKQ_NULL); \
471 WORKQ_ASSERT(WORKQ_PEER(workq) != BCM_WORKQ_NULL); \
472 WORKQ_ASSERT((workq)->buffer == WORKQ_PEER(workq)->buffer); \
473 WORKQ_ASSERT((workq)->ring_size == WORKQ_PEER(workq)->ring_size);
474
475 #define BCM_WORKQ_NULL ((bcm_workq_t *)NULL)
476
477 #define WORKQ_PEER(workq) ((workq)->peer)
478 #define WORKQ_RING(workq) (&((workq)->ring))
479 #define WORKQ_PEER_RING(workq) (&((workq)->peer->ring))
480
481 #define WORKQ_ELEMENT(__elem_type, __workq, __index) ({ \
482 WORKQ_ASSERT((__workq) != BCM_WORKQ_NULL); \
483 WORKQ_ASSERT((__index) < ((__workq)->ring_size)); \
484 ((__elem_type *)((__workq)->buffer)) + (__index); \
485 })
486
487 static INLINE void bcm_workq_init(bcm_workq_t *workq, bcm_workq_t *workq_peer,
488 void *buffer, int ring_size);
489
490 static INLINE bool bcm_workq_is_empty(const bcm_workq_t *workq_prod);
491
492 static INLINE void bcm_workq_prod_sync(bcm_workq_t *workq_prod);
493 static INLINE void bcm_workq_cons_sync(bcm_workq_t *workq_cons);
494
495 static INLINE void bcm_workq_prod_refresh(bcm_workq_t *workq_prod);
496 static INLINE void bcm_workq_cons_refresh(bcm_workq_t *workq_cons);
497
498 /**
499 * bcm_workq_init - initialize a workq
500 * @workq: pointer to a workq context
501 * @buffer: pointer to a pre-allocated circular buffer to serve as a ring
502 * @ring_size: size of the ring in terms of max number of elements.
503 */
504 static INLINE void
bcm_workq_init(bcm_workq_t * workq,bcm_workq_t * workq_peer,void * buffer,int ring_size)505 bcm_workq_init(bcm_workq_t *workq, bcm_workq_t *workq_peer,
506 void *buffer, int ring_size)
507 {
508 ASSERT(workq != BCM_WORKQ_NULL);
509 ASSERT(workq_peer != BCM_WORKQ_NULL);
510 ASSERT(buffer != NULL);
511 ASSERT(ring_size > 0);
512
513 WORKQ_PEER(workq) = workq_peer;
514 WORKQ_PEER(workq_peer) = workq;
515
516 bcm_ring_init(WORKQ_RING(workq));
517 bcm_ring_init(WORKQ_RING(workq_peer));
518
519 workq->buffer = workq_peer->buffer = buffer;
520 workq->ring_size = workq_peer->ring_size = ring_size;
521 }
522
523 /**
524 * bcm_workq_empty - test whether there is work
525 * @workq_prod: producer's workq
526 */
527 static INLINE bool
bcm_workq_is_empty(const bcm_workq_t * workq_prod)528 bcm_workq_is_empty(const bcm_workq_t *workq_prod)
529 {
530 return bcm_ring_is_empty(WORKQ_RING(workq_prod));
531 }
532
533 /**
534 * bcm_workq_prod_sync - Commit the producer write index to peer workq's ring
535 * @workq_prod: producer's workq whose write index must be synced to peer
536 */
537 static INLINE void
bcm_workq_prod_sync(bcm_workq_t * workq_prod)538 bcm_workq_prod_sync(bcm_workq_t *workq_prod)
539 {
540 WORKQ_AUDIT(workq_prod);
541
542 /* cons::write <--- prod::write */
543 bcm_ring_sync_write(WORKQ_PEER_RING(workq_prod), WORKQ_RING(workq_prod));
544 }
545
546 /**
547 * bcm_workq_cons_sync - Commit the consumer read index to the peer workq's ring
548 * @workq_cons: consumer's workq whose read index must be synced to peer
549 */
550 static INLINE void
bcm_workq_cons_sync(bcm_workq_t * workq_cons)551 bcm_workq_cons_sync(bcm_workq_t *workq_cons)
552 {
553 WORKQ_AUDIT(workq_cons);
554
555 /* prod::read <--- cons::read */
556 bcm_ring_sync_read(WORKQ_PEER_RING(workq_cons), WORKQ_RING(workq_cons));
557 }
558
559 /**
560 * bcm_workq_prod_refresh - Fetch the updated consumer's read index
561 * @workq_prod: producer's workq whose read index must be refreshed from peer
562 */
563 static INLINE void
bcm_workq_prod_refresh(bcm_workq_t * workq_prod)564 bcm_workq_prod_refresh(bcm_workq_t *workq_prod)
565 {
566 WORKQ_AUDIT(workq_prod);
567
568 /* prod::read <--- cons::read */
569 bcm_ring_sync_read(WORKQ_RING(workq_prod), WORKQ_PEER_RING(workq_prod));
570 }
571
572 /**
573 * bcm_workq_cons_refresh - Fetch the updated producer's write index
574 * @workq_cons: consumer's workq whose write index must be refreshed from peer
575 */
576 static INLINE void
bcm_workq_cons_refresh(bcm_workq_t * workq_cons)577 bcm_workq_cons_refresh(bcm_workq_t *workq_cons)
578 {
579 WORKQ_AUDIT(workq_cons);
580
581 /* cons::write <--- prod::write */
582 bcm_ring_sync_write(WORKQ_RING(workq_cons), WORKQ_PEER_RING(workq_cons));
583 }
584
585 #endif /* ! __bcm_ring_h_included__ */
586