bcmdhd.100.10.315.x/include/bcm_ring.h

   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) 1999-2019, 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  *      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.
  28  *
  29  *
  30  * <<Broadcom-WL-IPTag/Open:>>
  31  *
  32  * $Id: bcm_ring.h 700321 2017-05-18 16:09:07Z $
  33  */
  34 #ifndef __bcm_ring_included__
  35 #define __bcm_ring_included__
  36 /*
  37  * API Notes:
  38  *
  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.
  49  *
  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.
  54  *
  55  *  Producer side API:
  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
  61  *
  62  * Consumer side API:
  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
  68  *
  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
  71  *
  72  * +----------------------------------------------------------------------------
  73  *
  74  * Design Notes:
  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
  80  *
  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)
  84  *
  85  * +----------------------------------------------------------------------------
  86  *
  87  * Usage Notes:
  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.
  94  *
  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  * +----------------------------------------------------------------------------
 103  *
 104  * Copyright (C) 1999-2019, Broadcom.
 105  *
 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:
 111  *
 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.
 119  *
 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.
 123  *
 124  * $Id: bcm_ring.h 700321 2017-05-18 16:09:07Z $
 125  *
 126  * -*- Mode: C; tab-width: 4; indent-tabs-mode: t; c-basic-offset: 4 -*-
 127  * vim: set ts=4 noet sw=4 tw=80:
 128  *
 129  * +----------------------------------------------------------------------------
 130  */
 131
 132 #ifdef ____cacheline_aligned
 133 #define __ring_aligned                      ____cacheline_aligned
 134 #else
 135 #define __ring_aligned
 136 #endif // endif
 137
 138 /* Conditional compile for debug */
 139 /* #define BCM_RING_DEBUG */
 140
 141 #define BCM_RING_EMPTY                      (-1)
 142 #define BCM_RING_FULL                       (-1)
 143 #define BCM_RING_NULL                       ((bcm_ring_t *)NULL)
 144
 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 */
 153
 154 #define BCM_RING_SIZE_IS_VALID(ring_size)   ((ring_size) > 0)
 155
 156 /*
 157  * +----------------------------------------------------------------------------
 158  * Ring Context
 159  * +----------------------------------------------------------------------------
 160  */
 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 */
 167 } bcm_ring_t;
 168
 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);
 172
 173 static INLINE int  __bcm_ring_next_write(bcm_ring_t *ring, const int ring_size);
 174
 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);
 177
 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);
 182
 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);
 187
 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);
 190
 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);
 196
 197 /**
 198  * bcm_ring_init - initialize a ring context.
 199  * @ring: pointer to a ring context
 200  */
 201 static INLINE void
 202 bcm_ring_init(bcm_ring_t *ring)
 203 {
 204         ASSERT(ring != (bcm_ring_t *)NULL);
 205 #if defined(BCM_RING_DEBUG)
 206         ring->self = ring;
 207 #endif /* BCM_RING_DEBUG */
 208         ring->write = 0;
 209         ring->read = 0;
 210 }
 211
 212 /**
 213  * bcm_ring_copy - copy construct a ring
 214  * @to: pointer to the new ring context
 215  * @from: pointer to orig ring context
 216  */
 217 static INLINE void
 218 bcm_ring_copy(bcm_ring_t *to, bcm_ring_t *from)
 219 {
 220         bcm_ring_init(to);
 221
 222         to->write = from->write;
 223         to->read  = from->read;
 224 }
 225
 226 /**
 227  * bcm_ring_is_empty - "Boolean" test whether ring is empty.
 228  * @ring: pointer to a ring context
 229  *
 230  * PS. does not return BCM_RING_EMPTY value.
 231  */
 232 static INLINE bool
 233 bcm_ring_is_empty(bcm_ring_t *ring)
 234 {
 235         RING_ASSERT(BCM_RING_IS_VALID(ring));
 236         return (ring->read == ring->write);
 237 }
 238
 239 /**
 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
 244  *
 245  * PRIVATE INTERNAL USE ONLY.
 246  */
 247 static INLINE int
 248 __bcm_ring_next_write(bcm_ring_t *ring, const int ring_size)
 249 {
 250         RING_ASSERT(BCM_RING_IS_VALID(ring) && BCM_RING_SIZE_IS_VALID(ring_size));
 251         return ((ring->write + 1) % ring_size);
 252 }
 253
 254 /**
 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
 258  *
 259  * PRIVATE INTERNAL USE ONLY.
 260  */
 261 static INLINE bool
 262 __bcm_ring_full(bcm_ring_t *ring, int next_write)
 263 {
 264         return (next_write == ring->read);
 265 }
 266
 267 /**
 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
 271  *
 272  * PS. does not return BCM_RING_FULL value.
 273  */
 274 static INLINE bool
 275 bcm_ring_is_full(bcm_ring_t *ring, const int ring_size)
 276 {
 277         int next_write;
 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);
 281 }
 282
 283 /**
 284  * bcm_ring_prod_done - commit a previously pending index where production
 285  * was requested.
 286  * @ring: pointer to a ring context
 287  * @write: index into ring upto where production was done.
 288  * +----------------------------------------------------------------------------
 289  */
 290 static INLINE void
 291 bcm_ring_prod_done(bcm_ring_t *ring, int write)
 292 {
 293         RING_ASSERT(BCM_RING_IS_VALID(ring));
 294         ring->write = write;
 295 }
 296
 297 /**
 298  * bcm_ring_prod_pend - Fetch in "pend" mode, the index where an element may be
 299  * produced.
 300  * @ring: pointer to a ring context
 301  * @pend_write: next index, after the returned index
 302  * @ring_size: size of the ring
 303  */
 304 static INLINE int
 305 bcm_ring_prod_pend(bcm_ring_t *ring, int *pend_write, const int ring_size)
 306 {
 307         int rtn;
 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;
 312                 rtn = BCM_RING_FULL;
 313         } else {
 314                 /* production is not committed, caller needs to explicitly commit */
 315                 rtn = ring->write;
 316         }
 317         return rtn;
 318 }
 319
 320 /**
 321  * bcm_ring_prod - Fetch and "commit" the next index where a ring element may
 322  * be produced.
 323  * @ring: pointer to a ring context
 324  * @ring_size: size of the ring
 325  */
 326 static INLINE int
 327 bcm_ring_prod(bcm_ring_t *ring, const int ring_size)
 328 {
 329         int next_write, prod_write;
 330         RING_ASSERT(BCM_RING_IS_VALID(ring) && BCM_RING_SIZE_IS_VALID(ring_size));
 331
 332         next_write = __bcm_ring_next_write(ring, ring_size);
 333         if (__bcm_ring_full(ring, next_write)) {
 334                 prod_write = BCM_RING_FULL;
 335         } else {
 336                 prod_write = ring->write;
 337                 bcm_ring_prod_done(ring, next_write); /* "commit" production */
 338         }
 339         return prod_write;
 340 }
 341
 342 /**
 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.
 346  */
 347 static INLINE void
 348 bcm_ring_cons_done(bcm_ring_t *ring, int read)
 349 {
 350         RING_ASSERT(BCM_RING_IS_VALID(ring));
 351         ring->read = read;
 352 }
 353
 354 /**
 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
 360  */
 361 static INLINE int
 362 bcm_ring_cons_pend(bcm_ring_t *ring, int *pend_read, const int ring_size)
 363 {
 364         int rtn;
 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;
 369         } else {
 370                 *pend_read = (ring->read + 1) % ring_size;
 371                 /* production is not committed, caller needs to explicitly commit */
 372                 rtn = ring->read;
 373         }
 374         return rtn;
 375 }
 376
 377 /**
 378  * bcm_ring_cons - fetch and "commit" the next index where a ring element may
 379  * be consumed.
 380  * @ring: pointer to a ring context
 381  * @ring_size: size of the ring
 382  */
 383 static INLINE int
 384 bcm_ring_cons(bcm_ring_t *ring, const int ring_size)
 385 {
 386         int cons_read;
 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;
 390         } else {
 391                 cons_read = ring->read;
 392                 ring->read = (ring->read + 1) % ring_size; /* read is committed */
 393         }
 394         return cons_read;
 395 }
 396
 397 /**
 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
 401  */
 402 static INLINE void
 403 bcm_ring_sync_read(bcm_ring_t *peer, const bcm_ring_t *self)
 404 {
 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 */
 408 }
 409
 410 /**
 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
 414  */
 415 static INLINE void
 416 bcm_ring_sync_write(bcm_ring_t *peer, const bcm_ring_t *self)
 417 {
 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 */
 421 }
 422
 423 /**
 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
 428  */
 429 static INLINE int
 430 bcm_ring_prod_avail(const bcm_ring_t *ring, const int ring_size)
 431 {
 432         int prod_avail;
 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);
 436         } else {
 437                 prod_avail = (ring->read - (ring->write + 1));
 438         }
 439         ASSERT(prod_avail < ring_size);
 440         return prod_avail;
 441 }
 442
 443 /**
 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
 447  */
 448 static INLINE int
 449 bcm_ring_cons_avail(const bcm_ring_t *ring, const int ring_size)
 450 {
 451         int cons_avail;
 452         RING_ASSERT(BCM_RING_IS_VALID(ring) && BCM_RING_SIZE_IS_VALID(ring_size));
 453         if (ring->read == ring->write) {
 454                 cons_avail = 0;
 455         } else if (ring->read > ring->write) {
 456                 cons_avail = ((ring_size - ring->read) + ring->write);
 457         } else {
 458                 cons_avail = ring->write - ring->read;
 459         }
 460         ASSERT(cons_avail < ring_size);
 461         return cons_avail;
 462 }
 463
 464 /**
 465  * bcm_ring_cons_all - set ring in state where all elements are consumed.
 466  * @ring: pointer to a ring context
 467  */
 468 static INLINE void
 469 bcm_ring_cons_all(bcm_ring_t *ring)
 470 {
 471         ring->read = ring->write;
 472 }
 473
 474 /**
 475  * Work Queue
 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.
 479  */
 480
 481 struct bcm_workq {
 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 */
 486 } __ring_aligned;
 487
 488 typedef struct bcm_workq bcm_workq_t;
 489
 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 */
 496
 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);
 502
 503 #define BCM_WORKQ_NULL                  ((bcm_workq_t *)NULL)
 504
 505 #define WORKQ_PEER(workq)               ((workq)->peer)
 506 #define WORKQ_RING(workq)               (&((workq)->ring))
 507 #define WORKQ_PEER_RING(workq)          (&((workq)->peer->ring))
 508
 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); \
 513 })
 514
 515 static INLINE void bcm_workq_init(bcm_workq_t *workq, bcm_workq_t *workq_peer,
 516                                   void *buffer, int ring_size);
 517
 518 static INLINE bool bcm_workq_is_empty(bcm_workq_t *workq_prod);
 519
 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);
 522
 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);
 525
 526 /**
 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.
 531  */
 532 static INLINE void
 533 bcm_workq_init(bcm_workq_t *workq, bcm_workq_t *workq_peer,
 534                void *buffer, int ring_size)
 535 {
 536         ASSERT(workq != BCM_WORKQ_NULL);
 537         ASSERT(workq_peer != BCM_WORKQ_NULL);
 538         ASSERT(buffer != NULL);
 539         ASSERT(ring_size > 0);
 540
 541         WORKQ_PEER(workq) = workq_peer;
 542         WORKQ_PEER(workq_peer) = workq;
 543
 544         bcm_ring_init(WORKQ_RING(workq));
 545         bcm_ring_init(WORKQ_RING(workq_peer));
 546
 547         workq->buffer = workq_peer->buffer = buffer;
 548         workq->ring_size = workq_peer->ring_size = ring_size;
 549 }
 550
 551 /**
 552  * bcm_workq_empty - test whether there is work
 553  * @workq_prod: producer's workq
 554  */
 555 static INLINE bool
 556 bcm_workq_is_empty(bcm_workq_t *workq_prod)
 557 {
 558         return bcm_ring_is_empty(WORKQ_RING(workq_prod));
 559 }
 560
 561 /**
 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
 564  */
 565 static INLINE void
 566 bcm_workq_prod_sync(bcm_workq_t *workq_prod)
 567 {
 568         WORKQ_AUDIT(workq_prod);
 569
 570         /* cons::write <--- prod::write */
 571         bcm_ring_sync_write(WORKQ_PEER_RING(workq_prod), WORKQ_RING(workq_prod));
 572 }
 573
 574 /**
 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
 577  */
 578 static INLINE void
 579 bcm_workq_cons_sync(bcm_workq_t *workq_cons)
 580 {
 581         WORKQ_AUDIT(workq_cons);
 582
 583         /* prod::read <--- cons::read */
 584         bcm_ring_sync_read(WORKQ_PEER_RING(workq_cons), WORKQ_RING(workq_cons));
 585 }
 586
 587 /**
 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
 590  */
 591 static INLINE void
 592 bcm_workq_prod_refresh(bcm_workq_t *workq_prod)
 593 {
 594         WORKQ_AUDIT(workq_prod);
 595
 596         /* prod::read <--- cons::read */
 597         bcm_ring_sync_read(WORKQ_RING(workq_prod), WORKQ_PEER_RING(workq_prod));
 598 }
 599
 600 /**
 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
 603  */
 604 static INLINE void
 605 bcm_workq_cons_refresh(bcm_workq_t *workq_cons)
 606 {
 607         WORKQ_AUDIT(workq_cons);
 608
 609         /* cons::write <--- prod::write */
 610         bcm_ring_sync_write(WORKQ_RING(workq_cons), WORKQ_PEER_RING(workq_cons));
 611 }
 612
 613 #endif /* ! __bcm_ring_h_included__ */