sys/sys/buf_ring.h

   1 /**************************************************************************
   2  *
   3  * Copyright (c) 2007-2009 Kip Macy kmacy@freebsd.org
   4  * All rights reserved.
   5  *
   6  * Redistribution and use in source and binary forms, with or without
   7  * modification, are permitted provided that the following conditions are met:
   8  *
   9  * 1. Redistributions of source code must retain the above copyright notice,
  10  *    this list of conditions and the following disclaimer.
  11  *
  12  * 2. The name of Kip Macy nor the names of other
  13  *    contributors may be used to endorse or promote products derived from
  14  *    this software without specific prior written permission.
  15  *
  16  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
  17  * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  18  * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  19  * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
  20  * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
  21  * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
  22  * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
  23  * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
  24  * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
  25  * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
  26  * POSSIBILITY OF SUCH DAMAGE.
  27  *
  28  * $FreeBSD$
  29  *
  30  ***************************************************************************/
  31
  32 #ifndef _SYS_BUF_RING_H_
  33 #define _SYS_BUF_RING_H_
  34
  35 #include <machine/cpu.h>
  36
  37 #if defined(INVARIANTS) && !defined(DEBUG_BUFRING)
  38 #define DEBUG_BUFRING 1
  39 #endif
  40
  41 #ifdef DEBUG_BUFRING
  42 #include <sys/lock.h>
  43 #include <sys/mutex.h>
  44 #endif
  45
  46 struct buf_ring {
  47         volatile uint32_t       br_prod_head;
  48         volatile uint32_t       br_prod_tail;
  49         int                     br_prod_size;
  50         int                     br_prod_mask;
  51         uint64_t                br_drops;
  52         uint64_t                br_prod_bufs;
  53         uint64_t                br_prod_bytes;
  54         /*
  55          * Pad out to next L2 cache line
  56          */
  57         uint64_t                _pad0[11];
  58
  59         volatile uint32_t       br_cons_head;
  60         volatile uint32_t       br_cons_tail;
  61         int                     br_cons_size;
  62         int                     br_cons_mask;
  63
  64         /*
  65          * Pad out to next L2 cache line
  66          */
  67         uint64_t                _pad1[14];
  68 #ifdef DEBUG_BUFRING
  69         struct mtx              *br_lock;
  70 #endif
  71         void                    *br_ring[0];
  72 };
  73
  74 /*
  75  * multi-producer safe lock-free ring buffer enqueue
  76  *
  77  */
  78 static __inline int
  79 buf_ring_enqueue_bytes(struct buf_ring *br, void *buf, int nbytes)
  80 {
  81         uint32_t prod_head, prod_next;
  82         uint32_t cons_tail;
  83         int success;
  84 #ifdef DEBUG_BUFRING
  85         int i;
  86         for (i = br->br_cons_head; i != br->br_prod_head;
  87              i = ((i + 1) & br->br_cons_mask))
  88                 if(br->br_ring[i] == buf)
  89                         panic("buf=%p already enqueue at %d prod=%d cons=%d",
  90                             buf, i, br->br_prod_tail, br->br_cons_tail);
  91 #endif
  92         critical_enter();
  93         do {
  94                 prod_head = br->br_prod_head;
  95                 cons_tail = br->br_cons_tail;
  96
  97                 prod_next = (prod_head + 1) & br->br_prod_mask;
  98
  99                 if (prod_next == cons_tail) {
 100                         critical_exit();
 101                         return (ENOBUFS);
 102                 }
 103
 104                 success = atomic_cmpset_int(&br->br_prod_head, prod_head,
 105                     prod_next);
 106         } while (success == 0);
 107 #ifdef DEBUG_BUFRING
 108         if (br->br_ring[prod_head] != NULL)
 109                 panic("dangling value in enqueue");
 110 #endif
 111         br->br_ring[prod_head] = buf;
 112         wmb();
 113
 114         /*
 115          * If there are other enqueues in progress
 116          * that preceeded us, we need to wait for them
 117          * to complete
 118          */
 119         while (br->br_prod_tail != prod_head)
 120                 cpu_spinwait();
 121         br->br_prod_bufs++;
 122         br->br_prod_bytes += nbytes;
 123         br->br_prod_tail = prod_next;
 124         critical_exit();
 125         return (0);
 126 }
 127
 128 static __inline int
 129 buf_ring_enqueue(struct buf_ring *br, void *buf)
 130 {
 131
 132         return (buf_ring_enqueue_bytes(br, buf, 0));
 133 }
 134
 135 /*
 136  * multi-consumer safe dequeue
 137  *
 138  */
 139 static __inline void *
 140 buf_ring_dequeue_mc(struct buf_ring *br)
 141 {
 142         uint32_t cons_head, cons_next;
 143         uint32_t prod_tail;
 144         void *buf;
 145         int success;
 146
 147         critical_enter();
 148         do {
 149                 cons_head = br->br_cons_head;
 150                 prod_tail = br->br_prod_tail;
 151
 152                 cons_next = (cons_head + 1) & br->br_cons_mask;
 153
 154                 if (cons_head == prod_tail) {
 155                         critical_exit();
 156                         return (NULL);
 157                 }
 158
 159                 success = atomic_cmpset_int(&br->br_cons_head, cons_head,
 160                     cons_next);
 161         } while (success == 0);
 162
 163         buf = br->br_ring[cons_head];
 164 #ifdef DEBUG_BUFRING
 165         br->br_ring[cons_head] = NULL;
 166 #endif
 167         rmb();
 168
 169         /*
 170          * If there are other dequeues in progress
 171          * that preceeded us, we need to wait for them
 172          * to complete
 173          */
 174         while (br->br_cons_tail != cons_head)
 175                 cpu_spinwait();
 176
 177         br->br_cons_tail = cons_next;
 178         critical_exit();
 179
 180         return (buf);
 181 }
 182
 183 /*
 184  * single-consumer dequeue
 185  * use where dequeue is protected by a lock
 186  * e.g. a network driver's tx queue lock
 187  */
 188 static __inline void *
 189 buf_ring_dequeue_sc(struct buf_ring *br)
 190 {
 191         uint32_t cons_head, cons_next, cons_next_next;
 192         uint32_t prod_tail;
 193         void *buf;
 194
 195         cons_head = br->br_cons_head;
 196         prod_tail = br->br_prod_tail;
 197
 198         cons_next = (cons_head + 1) & br->br_cons_mask;
 199         cons_next_next = (cons_head + 2) & br->br_cons_mask;
 200
 201         if (cons_head == prod_tail)
 202                 return (NULL);
 203
 204 #ifdef PREFETCH_DEFINED
 205         if (cons_next != prod_tail) {
 206                 prefetch(br->br_ring[cons_next]);
 207                 if (cons_next_next != prod_tail)
 208                         prefetch(br->br_ring[cons_next_next]);
 209         }
 210 #endif
 211         br->br_cons_head = cons_next;
 212         buf = br->br_ring[cons_head];
 213
 214 #ifdef DEBUG_BUFRING
 215         br->br_ring[cons_head] = NULL;
 216         if (!mtx_owned(br->br_lock))
 217                 panic("lock not held on single consumer dequeue");
 218         if (br->br_cons_tail != cons_head)
 219                 panic("inconsistent list cons_tail=%d cons_head=%d",
 220                     br->br_cons_tail, cons_head);
 221 #endif
 222         br->br_cons_tail = cons_next;
 223         return (buf);
 224 }
 225
 226 /*
 227  * return a pointer to the first entry in the ring
 228  * without modifying it, or NULL if the ring is empty
 229  * race-prone if not protected by a lock
 230  */
 231 static __inline void *
 232 buf_ring_peek(struct buf_ring *br)
 233 {
 234
 235 #ifdef DEBUG_BUFRING
 236         if ((br->br_lock != NULL) && !mtx_owned(br->br_lock))
 237                 panic("lock not held on single consumer dequeue");
 238 #endif
 239         /*
 240          * I believe it is safe to not have a memory barrier
 241          * here because we control cons and tail is worst case
 242          * a lagging indicator so we worst case we might
 243          * return NULL immediately after a buffer has been enqueued
 244          */
 245         if (br->br_cons_head == br->br_prod_tail)
 246                 return (NULL);
 247
 248         return (br->br_ring[br->br_cons_head]);
 249 }
 250
 251 static __inline int
 252 buf_ring_full(struct buf_ring *br)
 253 {
 254
 255         return (((br->br_prod_head + 1) & br->br_prod_mask) == br->br_cons_tail);
 256 }
 257
 258 static __inline int
 259 buf_ring_empty(struct buf_ring *br)
 260 {
 261
 262         return (br->br_cons_head == br->br_prod_tail);
 263 }
 264
 265 static __inline int
 266 buf_ring_count(struct buf_ring *br)
 267 {
 268
 269         return ((br->br_prod_size + br->br_prod_tail - br->br_cons_tail)
 270             & br->br_prod_mask);
 271 }
 272
 273 struct buf_ring *buf_ring_alloc(int count, struct malloc_type *type, int flags,
 274     struct mtx *);
 275 void buf_ring_free(struct buf_ring *br, struct malloc_type *type);
 276
 277
 278
 279 #endif