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