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
7 * NOTE: A ring of size N, may only hold N-1 elements.
9 * Copyright (C) 1999-2019, Broadcom.
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:
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.
25 * Notwithstanding the above, under no circumstances may you combine this
26 * software in any way with any other Broadcom software provided under a license
27 * other than the GPL, without Broadcom's express prior written consent.
30 * <<Broadcom-WL-IPTag/Open:>>
32 * $Id: bcm_ring.h 700321 2017-05-18 16:09:07Z $
34 #ifndef __bcm_ring_included__
35 #define __bcm_ring_included__
39 * Ring manipulation API allows for:
40 * Pending operations: Often before some work can be completed, it may be
41 * desired that several resources are available, e.g. space for production in
42 * a ring. Approaches such as, #1) reserve resources one by one and return them
43 * if another required resource is not available, or #2) employ a two pass
44 * algorithm of first testing whether all resources are available, have a
45 * an impact on performance critical code. The approach taken here is more akin
46 * to approach #2, where a test for resource availability essentially also
47 * provides the index for production in an un-committed state.
48 * The same approach is taken for the consumer side.
50 * - Pending production: Fetch the next index where a ring element may be
51 * produced. The caller may not commit the WRITE of the element.
52 * - Pending consumption: Fetch the next index where a ring element may be
53 * consumed. The caller may not commut the READ of the element.
56 * - bcm_ring_is_full : Test whether ring is full
57 * - bcm_ring_prod : Fetch index where an element may be produced (commit)
58 * - bcm_ring_prod_pend: Fetch index where an element may be produced (pending)
59 * - bcm_ring_prod_done: Commit a previous pending produce fetch
60 * - bcm_ring_prod_avail: Fetch total number free slots eligible for production
63 * - bcm_ring_is_empty : Test whether ring is empty
64 * - bcm_ring_cons : Fetch index where an element may be consumed (commit)
65 * - bcm_ring_cons_pend: Fetch index where an element may be consumed (pending)
66 * - bcm_ring_cons_done: Commit a previous pending consume fetch
67 * - bcm_ring_cons_avail: Fetch total number elements eligible for consumption
69 * - bcm_ring_sync_read: Sync read offset in peer ring, from local ring
70 * - bcm_ring_sync_write: Sync write offset in peer ring, from local ring
72 * +----------------------------------------------------------------------------
75 * Following items are not tracked in a ring context (design decision)
76 * - width of a ring element.
77 * - depth of the ring.
78 * - base of the buffer, where the elements are stored.
79 * - count of number of free slots in the ring
81 * Implementation Notes:
82 * - When BCM_RING_DEBUG is enabled, need explicit bcm_ring_init().
83 * - BCM_RING_EMPTY and BCM_RING_FULL are (-1)
85 * +----------------------------------------------------------------------------
88 * An application may incarnate a ring of some fixed sized elements, by defining
89 * - a ring data buffer to store the ring elements.
90 * - depth of the ring (max number of elements managed by ring context).
91 * Preferrably, depth may be represented as a constant.
92 * - width of a ring element: to be used in pointer arithmetic with the ring's
93 * data buffer base and an index to fetch the ring element.
95 * Use bcm_workq_t to instantiate a pair of workq constructs, one for the
96 * producer and the other for the consumer, both pointing to the same circular
97 * buffer. The producer may operate on it's own local workq and flush the write
98 * index to the consumer. Likewise the consumer may use its local workq and
99 * flush the read index to the producer. This way we do not repeatedly access
100 * the peer's context. The two peers may reside on different CPU cores with a
101 * private L1 data cache.
102 * +----------------------------------------------------------------------------
104 * Copyright (C) 1999-2019, Broadcom.
106 * Unless you and Broadcom execute a separate written software license
107 * agreement governing use of this software, this software is licensed to you
108 * under the terms of the GNU General Public License version 2 (the "GPL"),
109 * available at http://www.broadcom.com/licenses/GPLv2.php, with the
110 * following added to such license:
112 * As a special exception, the copyright holders of this software give you
113 * permission to link this software with independent modules, and to copy and
114 * distribute the resulting executable under terms of your choice, provided that
115 * you also meet, for each linked independent module, the terms and conditions of
116 * the license of that module. An independent module is a module which is not
117 * derived from this software. The special exception does not apply to any
118 * modifications of the software.
120 * Notwithstanding the above, under no circumstances may you combine this
121 * software in any way with any other Broadcom software provided under a license
122 * other than the GPL, without Broadcom's express prior written consent.
124 * $Id: bcm_ring.h 700321 2017-05-18 16:09:07Z $
126 * -*- Mode: C; tab-width: 4; indent-tabs-mode: t; c-basic-offset: 4 -*-
127 * vim: set ts=4 noet sw=4 tw=80:
129 * +----------------------------------------------------------------------------
132 #ifdef ____cacheline_aligned
133 #define __ring_aligned ____cacheline_aligned
135 #define __ring_aligned
138 /* Conditional compile for debug */
139 /* #define BCM_RING_DEBUG */
141 #define BCM_RING_EMPTY (-1)
142 #define BCM_RING_FULL (-1)
143 #define BCM_RING_NULL ((bcm_ring_t *)NULL)
145 #if defined(BCM_RING_DEBUG)
146 #define RING_ASSERT(exp) ASSERT(exp)
147 #define BCM_RING_IS_VALID(ring) (((ring) != BCM_RING_NULL) && \
148 ((ring)->self == (ring)))
149 #else /* ! BCM_RING_DEBUG */
150 #define RING_ASSERT(exp) do {} while (0)
151 #define BCM_RING_IS_VALID(ring) ((ring) != BCM_RING_NULL)
152 #endif /* ! BCM_RING_DEBUG */
154 #define BCM_RING_SIZE_IS_VALID(ring_size) ((ring_size) > 0)
157 * +----------------------------------------------------------------------------
159 * +----------------------------------------------------------------------------
161 typedef struct bcm_ring
{ /* Ring context */
162 #if defined(BCM_RING_DEBUG)
163 struct bcm_ring
*self
; /* ptr to self for IS VALID test */
164 #endif /* BCM_RING_DEBUG */
165 int write __ring_aligned
; /* WRITE index in a circular ring */
166 int read __ring_aligned
; /* READ index in a circular ring */
169 static INLINE
void bcm_ring_init(bcm_ring_t
*ring
);
170 static INLINE
void bcm_ring_copy(bcm_ring_t
*to
, bcm_ring_t
*from
);
171 static INLINE
bool bcm_ring_is_empty(bcm_ring_t
*ring
);
173 static INLINE
int __bcm_ring_next_write(bcm_ring_t
*ring
, const int ring_size
);
175 static INLINE
bool __bcm_ring_full(bcm_ring_t
*ring
, int next_write
);
176 static INLINE
bool bcm_ring_is_full(bcm_ring_t
*ring
, const int ring_size
);
178 static INLINE
void bcm_ring_prod_done(bcm_ring_t
*ring
, int write
);
179 static INLINE
int bcm_ring_prod_pend(bcm_ring_t
*ring
, int *pend_write
,
180 const int ring_size
);
181 static INLINE
int bcm_ring_prod(bcm_ring_t
*ring
, const int ring_size
);
183 static INLINE
void bcm_ring_cons_done(bcm_ring_t
*ring
, int read
);
184 static INLINE
int bcm_ring_cons_pend(bcm_ring_t
*ring
, int *pend_read
,
185 const int ring_size
);
186 static INLINE
int bcm_ring_cons(bcm_ring_t
*ring
, const int ring_size
);
188 static INLINE
void bcm_ring_sync_read(bcm_ring_t
*peer
, const bcm_ring_t
*self
);
189 static INLINE
void bcm_ring_sync_write(bcm_ring_t
*peer
, const bcm_ring_t
*self
);
191 static INLINE
int bcm_ring_prod_avail(const bcm_ring_t
*ring
,
192 const int ring_size
);
193 static INLINE
int bcm_ring_cons_avail(const bcm_ring_t
*ring
,
194 const int ring_size
);
195 static INLINE
void bcm_ring_cons_all(bcm_ring_t
*ring
);
198 * bcm_ring_init - initialize a ring context.
199 * @ring: pointer to a ring context
202 bcm_ring_init(bcm_ring_t
*ring
)
204 ASSERT(ring
!= (bcm_ring_t
*)NULL
);
205 #if defined(BCM_RING_DEBUG)
207 #endif /* BCM_RING_DEBUG */
213 * bcm_ring_copy - copy construct a ring
214 * @to: pointer to the new ring context
215 * @from: pointer to orig ring context
218 bcm_ring_copy(bcm_ring_t
*to
, bcm_ring_t
*from
)
222 to
->write
= from
->write
;
223 to
->read
= from
->read
;
227 * bcm_ring_is_empty - "Boolean" test whether ring is empty.
228 * @ring: pointer to a ring context
230 * PS. does not return BCM_RING_EMPTY value.
233 bcm_ring_is_empty(bcm_ring_t
*ring
)
235 RING_ASSERT(BCM_RING_IS_VALID(ring
));
236 return (ring
->read
== ring
->write
);
240 * __bcm_ring_next_write - determine the index where the next write may occur
241 * (with wrap-around).
242 * @ring: pointer to a ring context
243 * @ring_size: size of the ring
245 * PRIVATE INTERNAL USE ONLY.
248 __bcm_ring_next_write(bcm_ring_t
*ring
, const int ring_size
)
250 RING_ASSERT(BCM_RING_IS_VALID(ring
) && BCM_RING_SIZE_IS_VALID(ring_size
));
251 return ((ring
->write
+ 1) % ring_size
);
255 * __bcm_ring_full - support function for ring full test.
256 * @ring: pointer to a ring context
257 * @next_write: next location in ring where an element is to be produced
259 * PRIVATE INTERNAL USE ONLY.
262 __bcm_ring_full(bcm_ring_t
*ring
, int next_write
)
264 return (next_write
== ring
->read
);
268 * bcm_ring_is_full - "Boolean" test whether a ring is full.
269 * @ring: pointer to a ring context
270 * @ring_size: size of the ring
272 * PS. does not return BCM_RING_FULL value.
275 bcm_ring_is_full(bcm_ring_t
*ring
, const int ring_size
)
278 RING_ASSERT(BCM_RING_IS_VALID(ring
) && BCM_RING_SIZE_IS_VALID(ring_size
));
279 next_write
= __bcm_ring_next_write(ring
, ring_size
);
280 return __bcm_ring_full(ring
, next_write
);
284 * bcm_ring_prod_done - commit a previously pending index where production
286 * @ring: pointer to a ring context
287 * @write: index into ring upto where production was done.
288 * +----------------------------------------------------------------------------
291 bcm_ring_prod_done(bcm_ring_t
*ring
, int write
)
293 RING_ASSERT(BCM_RING_IS_VALID(ring
));
298 * bcm_ring_prod_pend - Fetch in "pend" mode, the index where an element may be
300 * @ring: pointer to a ring context
301 * @pend_write: next index, after the returned index
302 * @ring_size: size of the ring
305 bcm_ring_prod_pend(bcm_ring_t
*ring
, int *pend_write
, const int ring_size
)
308 RING_ASSERT(BCM_RING_IS_VALID(ring
) && BCM_RING_SIZE_IS_VALID(ring_size
));
309 *pend_write
= __bcm_ring_next_write(ring
, ring_size
);
310 if (__bcm_ring_full(ring
, *pend_write
)) {
311 *pend_write
= BCM_RING_FULL
;
314 /* production is not committed, caller needs to explicitly commit */
321 * bcm_ring_prod - Fetch and "commit" the next index where a ring element may
323 * @ring: pointer to a ring context
324 * @ring_size: size of the ring
327 bcm_ring_prod(bcm_ring_t
*ring
, const int ring_size
)
329 int next_write
, prod_write
;
330 RING_ASSERT(BCM_RING_IS_VALID(ring
) && BCM_RING_SIZE_IS_VALID(ring_size
));
332 next_write
= __bcm_ring_next_write(ring
, ring_size
);
333 if (__bcm_ring_full(ring
, next_write
)) {
334 prod_write
= BCM_RING_FULL
;
336 prod_write
= ring
->write
;
337 bcm_ring_prod_done(ring
, next_write
); /* "commit" production */
343 * bcm_ring_cons_done - commit a previously pending read
344 * @ring: pointer to a ring context
345 * @read: index upto which elements have been consumed.
348 bcm_ring_cons_done(bcm_ring_t
*ring
, int read
)
350 RING_ASSERT(BCM_RING_IS_VALID(ring
));
355 * bcm_ring_cons_pend - fetch in "pend" mode, the next index where a ring
356 * element may be consumed.
357 * @ring: pointer to a ring context
358 * @pend_read: index into ring upto which elements may be consumed.
359 * @ring_size: size of the ring
362 bcm_ring_cons_pend(bcm_ring_t
*ring
, int *pend_read
, const int ring_size
)
365 RING_ASSERT(BCM_RING_IS_VALID(ring
) && BCM_RING_SIZE_IS_VALID(ring_size
));
366 if (bcm_ring_is_empty(ring
)) {
367 *pend_read
= BCM_RING_EMPTY
;
368 rtn
= BCM_RING_EMPTY
;
370 *pend_read
= (ring
->read
+ 1) % ring_size
;
371 /* production is not committed, caller needs to explicitly commit */
378 * bcm_ring_cons - fetch and "commit" the next index where a ring element may
380 * @ring: pointer to a ring context
381 * @ring_size: size of the ring
384 bcm_ring_cons(bcm_ring_t
*ring
, const int ring_size
)
387 RING_ASSERT(BCM_RING_IS_VALID(ring
) && BCM_RING_SIZE_IS_VALID(ring_size
));
388 if (bcm_ring_is_empty(ring
)) {
389 cons_read
= BCM_RING_EMPTY
;
391 cons_read
= ring
->read
;
392 ring
->read
= (ring
->read
+ 1) % ring_size
; /* read is committed */
398 * bcm_ring_sync_read - on consumption, update peer's read index.
399 * @peer: pointer to peer's producer ring context
400 * @self: pointer to consumer's ring context
403 bcm_ring_sync_read(bcm_ring_t
*peer
, const bcm_ring_t
*self
)
405 RING_ASSERT(BCM_RING_IS_VALID(peer
));
406 RING_ASSERT(BCM_RING_IS_VALID(self
));
407 peer
->read
= self
->read
; /* flush read update to peer producer */
411 * bcm_ring_sync_write - on consumption, update peer's write index.
412 * @peer: pointer to peer's consumer ring context
413 * @self: pointer to producer's ring context
416 bcm_ring_sync_write(bcm_ring_t
*peer
, const bcm_ring_t
*self
)
418 RING_ASSERT(BCM_RING_IS_VALID(peer
));
419 RING_ASSERT(BCM_RING_IS_VALID(self
));
420 peer
->write
= self
->write
; /* flush write update to peer consumer */
424 * bcm_ring_prod_avail - fetch total number of available empty slots in the
425 * ring for production.
426 * @ring: pointer to a ring context
427 * @ring_size: size of the ring
430 bcm_ring_prod_avail(const bcm_ring_t
*ring
, const int ring_size
)
433 RING_ASSERT(BCM_RING_IS_VALID(ring
) && BCM_RING_SIZE_IS_VALID(ring_size
));
434 if (ring
->write
>= ring
->read
) {
435 prod_avail
= (ring_size
- (ring
->write
- ring
->read
) - 1);
437 prod_avail
= (ring
->read
- (ring
->write
+ 1));
439 ASSERT(prod_avail
< ring_size
);
444 * bcm_ring_cons_avail - fetch total number of available elements for consumption.
445 * @ring: pointer to a ring context
446 * @ring_size: size of the ring
449 bcm_ring_cons_avail(const bcm_ring_t
*ring
, const int ring_size
)
452 RING_ASSERT(BCM_RING_IS_VALID(ring
) && BCM_RING_SIZE_IS_VALID(ring_size
));
453 if (ring
->read
== ring
->write
) {
455 } else if (ring
->read
> ring
->write
) {
456 cons_avail
= ((ring_size
- ring
->read
) + ring
->write
);
458 cons_avail
= ring
->write
- ring
->read
;
460 ASSERT(cons_avail
< ring_size
);
465 * bcm_ring_cons_all - set ring in state where all elements are consumed.
466 * @ring: pointer to a ring context
469 bcm_ring_cons_all(bcm_ring_t
*ring
)
471 ring
->read
= ring
->write
;
476 * A work Queue is composed of a ring of work items, of a specified depth.
477 * It HAS-A bcm_ring object, comprising of a RD and WR offset, to implement a
478 * producer/consumer circular ring.
482 bcm_ring_t ring
; /* Ring context abstraction */
483 struct bcm_workq
*peer
; /* Peer workq context */
484 void *buffer
; /* Buffer storage for work items in workQ */
485 int ring_size
; /* Depth of workQ */
488 typedef struct bcm_workq bcm_workq_t
;
490 /* #define BCM_WORKQ_DEBUG */
491 #if defined(BCM_WORKQ_DEBUG)
492 #define WORKQ_ASSERT(exp) ASSERT(exp)
493 #else /* ! BCM_WORKQ_DEBUG */
494 #define WORKQ_ASSERT(exp) do {} while (0)
495 #endif /* ! BCM_WORKQ_DEBUG */
497 #define WORKQ_AUDIT(workq) \
498 WORKQ_ASSERT((workq) != BCM_WORKQ_NULL); \
499 WORKQ_ASSERT(WORKQ_PEER(workq) != BCM_WORKQ_NULL); \
500 WORKQ_ASSERT((workq)->buffer == WORKQ_PEER(workq)->buffer); \
501 WORKQ_ASSERT((workq)->ring_size == WORKQ_PEER(workq)->ring_size);
503 #define BCM_WORKQ_NULL ((bcm_workq_t *)NULL)
505 #define WORKQ_PEER(workq) ((workq)->peer)
506 #define WORKQ_RING(workq) (&((workq)->ring))
507 #define WORKQ_PEER_RING(workq) (&((workq)->peer->ring))
509 #define WORKQ_ELEMENT(__elem_type, __workq, __index) ({ \
510 WORKQ_ASSERT((__workq) != BCM_WORKQ_NULL); \
511 WORKQ_ASSERT((__index) < ((__workq)->ring_size)); \
512 ((__elem_type *)((__workq)->buffer)) + (__index); \
515 static INLINE
void bcm_workq_init(bcm_workq_t
*workq
, bcm_workq_t
*workq_peer
,
516 void *buffer
, int ring_size
);
518 static INLINE
bool bcm_workq_is_empty(bcm_workq_t
*workq_prod
);
520 static INLINE
void bcm_workq_prod_sync(bcm_workq_t
*workq_prod
);
521 static INLINE
void bcm_workq_cons_sync(bcm_workq_t
*workq_cons
);
523 static INLINE
void bcm_workq_prod_refresh(bcm_workq_t
*workq_prod
);
524 static INLINE
void bcm_workq_cons_refresh(bcm_workq_t
*workq_cons
);
527 * bcm_workq_init - initialize a workq
528 * @workq: pointer to a workq context
529 * @buffer: pointer to a pre-allocated circular buffer to serve as a ring
530 * @ring_size: size of the ring in terms of max number of elements.
533 bcm_workq_init(bcm_workq_t
*workq
, bcm_workq_t
*workq_peer
,
534 void *buffer
, int ring_size
)
536 ASSERT(workq
!= BCM_WORKQ_NULL
);
537 ASSERT(workq_peer
!= BCM_WORKQ_NULL
);
538 ASSERT(buffer
!= NULL
);
539 ASSERT(ring_size
> 0);
541 WORKQ_PEER(workq
) = workq_peer
;
542 WORKQ_PEER(workq_peer
) = workq
;
544 bcm_ring_init(WORKQ_RING(workq
));
545 bcm_ring_init(WORKQ_RING(workq_peer
));
547 workq
->buffer
= workq_peer
->buffer
= buffer
;
548 workq
->ring_size
= workq_peer
->ring_size
= ring_size
;
552 * bcm_workq_empty - test whether there is work
553 * @workq_prod: producer's workq
556 bcm_workq_is_empty(bcm_workq_t
*workq_prod
)
558 return bcm_ring_is_empty(WORKQ_RING(workq_prod
));
562 * bcm_workq_prod_sync - Commit the producer write index to peer workq's ring
563 * @workq_prod: producer's workq whose write index must be synced to peer
566 bcm_workq_prod_sync(bcm_workq_t
*workq_prod
)
568 WORKQ_AUDIT(workq_prod
);
570 /* cons::write <--- prod::write */
571 bcm_ring_sync_write(WORKQ_PEER_RING(workq_prod
), WORKQ_RING(workq_prod
));
575 * bcm_workq_cons_sync - Commit the consumer read index to the peer workq's ring
576 * @workq_cons: consumer's workq whose read index must be synced to peer
579 bcm_workq_cons_sync(bcm_workq_t
*workq_cons
)
581 WORKQ_AUDIT(workq_cons
);
583 /* prod::read <--- cons::read */
584 bcm_ring_sync_read(WORKQ_PEER_RING(workq_cons
), WORKQ_RING(workq_cons
));
588 * bcm_workq_prod_refresh - Fetch the updated consumer's read index
589 * @workq_prod: producer's workq whose read index must be refreshed from peer
592 bcm_workq_prod_refresh(bcm_workq_t
*workq_prod
)
594 WORKQ_AUDIT(workq_prod
);
596 /* prod::read <--- cons::read */
597 bcm_ring_sync_read(WORKQ_RING(workq_prod
), WORKQ_PEER_RING(workq_prod
));
601 * bcm_workq_cons_refresh - Fetch the updated producer's write index
602 * @workq_cons: consumer's workq whose write index must be refreshed from peer
605 bcm_workq_cons_refresh(bcm_workq_t
*workq_cons
)
607 WORKQ_AUDIT(workq_cons
);
609 /* cons::write <--- prod::write */
610 bcm_ring_sync_write(WORKQ_RING(workq_cons
), WORKQ_PEER_RING(workq_cons
));
613 #endif /* ! __bcm_ring_h_included__ */