sys/sys/buf_ring.h

   1 /**************************************************************************
   2  *
   3  * Copyright (c) 2007,2008 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         /*
  53          * Pad out to next L2 cache line
  54          */
  55         uint64_t                _pad0[13];
  56
  57         volatile uint32_t       br_cons_head;
  58         volatile uint32_t       br_cons_tail;
  59         int                     br_cons_size;
  60         int                     br_cons_mask;
  61
  62         /*
  63          * Pad out to next L2 cache line
  64          */
  65         uint64_t                _pad1[14];
  66 #ifdef DEBUG_BUFRING
  67         struct mtx              *br_lock;
  68 #endif
  69         void                    *br_ring[0];
  70 };
  71
  72
  73 static __inline int
  74 buf_ring_enqueue(struct buf_ring *br, void *buf)
  75 {
  76         uint32_t prod_head, prod_next;
  77         uint32_t cons_tail;
  78         int success;
  79 #ifdef DEBUG_BUFRING
  80         int i;
  81         for (i = br->br_cons_head; i != br->br_prod_head;
  82              i = ((i + 1) & br->br_cons_mask))
  83                 if(br->br_ring[i] == buf)
  84                         panic("buf=%p already enqueue at %d prod=%d cons=%d",
  85                             buf, i, br->br_prod_tail, br->br_cons_tail);
  86 #endif
  87         critical_enter();
  88         do {
  89                 prod_head = br->br_prod_head;
  90                 cons_tail = br->br_cons_tail;
  91
  92                 prod_next = (prod_head + 1) & br->br_prod_mask;
  93
  94                 if (prod_next == cons_tail) {
  95                         critical_exit();
  96                         return (ENOBUFS);
  97                 }
  98
  99                 success = atomic_cmpset_int(&br->br_prod_head, prod_head,
 100                     prod_next);
 101         } while (success == 0);
 102 #ifdef DEBUG_BUFRING
 103         if (br->br_ring[prod_head] != NULL)
 104                 panic("dangling value in enqueue");
 105 #endif
 106         br->br_ring[prod_head] = buf;
 107         wmb();
 108
 109         /*
 110          * If there are other enqueues in progress
 111          * that preceeded us, we need to wait for them
 112          * to complete
 113          */
 114         while (br->br_prod_tail != prod_head)
 115                 cpu_spinwait();
 116         br->br_prod_tail = prod_next;
 117         mb();
 118         critical_exit();
 119         return (0);
 120 }
 121
 122 /*
 123  * multi-consumer safe dequeue
 124  *
 125  */
 126 static __inline void *
 127 buf_ring_dequeue_mc(struct buf_ring *br)
 128 {
 129         uint32_t cons_head, cons_next;
 130         uint32_t prod_tail;
 131         void *buf;
 132         int success;
 133
 134         critical_enter();
 135         do {
 136                 cons_head = br->br_cons_head;
 137                 prod_tail = br->br_prod_tail;
 138
 139                 cons_next = (cons_head + 1) & br->br_cons_mask;
 140
 141                 if (cons_head == prod_tail) {
 142                         critical_exit();
 143                         return (NULL);
 144                 }
 145
 146                 success = atomic_cmpset_int(&br->br_cons_head, cons_head,
 147                     cons_next);
 148         } while (success == 0);
 149
 150         buf = br->br_ring[cons_head];
 151 #ifdef DEBUG_BUFRING
 152         br->br_ring[cons_head] = NULL;
 153 #endif
 154         mb();
 155
 156         /*
 157          * If there are other dequeues in progress
 158          * that preceeded us, we need to wait for them
 159          * to complete
 160          */
 161         while (br->br_cons_tail != cons_head)
 162                 cpu_spinwait();
 163
 164         br->br_cons_tail = cons_next;
 165         mb();
 166         critical_exit();
 167
 168         return (buf);
 169 }
 170
 171 /*
 172  * Single-Consumer dequeue for uses where dequeue
 173  * is protected by a lock
 174  */
 175 static __inline void *
 176 buf_ring_dequeue_sc(struct buf_ring *br)
 177 {
 178         uint32_t cons_head, cons_next;
 179         uint32_t prod_tail;
 180         void *buf;
 181
 182         critical_enter();
 183         cons_head = br->br_cons_head;
 184         prod_tail = br->br_prod_tail;
 185
 186         cons_next = (cons_head + 1) & br->br_cons_mask;
 187
 188         if (cons_head == prod_tail) {
 189                 critical_exit();
 190                 return (NULL);
 191         }
 192
 193         br->br_cons_head = cons_next;
 194         buf = br->br_ring[cons_head];
 195         mb();
 196
 197 #ifdef DEBUG_BUFRING
 198         br->br_ring[cons_head] = NULL;
 199         if (!mtx_owned(br->br_lock))
 200                 panic("lock not held on single consumer dequeue");
 201         if (br->br_cons_tail != cons_head)
 202                 panic("inconsistent list cons_tail=%d cons_head=%d",
 203                     br->br_cons_tail, cons_head);
 204 #endif
 205         br->br_cons_tail = cons_next;
 206         mb();
 207         critical_exit();
 208         return (buf);
 209 }
 210
 211 static __inline void *
 212 buf_ring_peek(struct buf_ring *br)
 213 {
 214
 215 #ifdef DEBUG_BUFRING
 216         if ((br->br_lock != NULL) && !mtx_owned(br->br_lock))
 217                 panic("lock not held on single consumer dequeue");
 218 #endif
 219         mb();
 220         if (br->br_cons_head == br->br_prod_tail)
 221                 return (NULL);
 222
 223         return (br->br_ring[br->br_cons_head]);
 224 }
 225
 226 static __inline int
 227 buf_ring_full(struct buf_ring *br)
 228 {
 229
 230         return (((br->br_prod_head + 1) & br->br_prod_mask) == br->br_cons_tail);
 231 }
 232
 233 static __inline int
 234 buf_ring_empty(struct buf_ring *br)
 235 {
 236
 237         return (br->br_cons_head == br->br_prod_tail);
 238 }
 239
 240 static __inline int
 241 buf_ring_count(struct buf_ring *br)
 242 {
 243
 244         return ((br->br_prod_size + br->br_prod_tail - br->br_cons_tail)
 245             & br->br_prod_mask);
 246 }
 247
 248 struct buf_ring *buf_ring_alloc(int count, struct malloc_type *type, int flags,
 249     struct mtx *);
 250 void buf_ring_free(struct buf_ring *br, struct malloc_type *type);
 251
 252
 253
 254 #endif