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