1 /* $NetBSD: subr_pcq.c,v 1.6 2012/01/31 20:40:09 alnsn Exp $ */ 2 3 /*- 4 * Copyright (c) 2009 The NetBSD Foundation, Inc. 5 * All rights reserved. 6 * 7 * This code is derived from software contributed to The NetBSD Foundation 8 * by Andrew Doran. 9 * 10 * Redistribution and use in source and binary forms, with or without 11 * modification, are permitted provided that the following conditions 12 * are met: 13 * 1. Redistributions of source code must retain the above copyright 14 * notice, this list of conditions and the following disclaimer. 15 * 2. Redistributions in binary form must reproduce the above copyright 16 * notice, this list of conditions and the following disclaimer in the 17 * documentation and/or other materials provided with the distribution. 18 * 19 * THIS SOFTWARE IS PROVIDED BY THE NETBSD FOUNDATION, INC. AND CONTRIBUTORS 20 * ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED 21 * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR 22 * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE FOUNDATION OR CONTRIBUTORS 23 * BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 24 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 25 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 26 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 27 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) 28 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 29 * POSSIBILITY OF SUCH DAMAGE. 30 */ 31 32 /* 33 * Lockless producer/consumer queue. 34 */ 35 36 #include <sys/cdefs.h> 37 __KERNEL_RCSID(0, "$NetBSD: subr_pcq.c,v 1.6 2012/01/31 20:40:09 alnsn Exp $"); 38 39 #include <sys/param.h> 40 #include <sys/types.h> 41 #include <sys/atomic.h> 42 #include <sys/kmem.h> 43 44 #include <sys/pcq.h> 45 46 /* 47 * Internal producer-consumer queue structure. Note: providing a separate 48 * cache-line both for pcq_t::pcq_pc and pcq_t::pcq_items. 49 */ 50 struct pcq { 51 u_int pcq_nitems; 52 uint8_t pcq_pad1[COHERENCY_UNIT - sizeof(u_int)]; 53 volatile uint32_t pcq_pc; 54 uint8_t pcq_pad2[COHERENCY_UNIT - sizeof(uint32_t)]; 55 void * volatile pcq_items[]; 56 }; 57 58 /* 59 * Producer (p) - stored in the lower 16 bits of pcq_t::pcq_pc. 60 * Consumer (c) - in the higher 16 bits. 61 * 62 * We have a limitation of 16 bits i.e. 0xffff items in the queue. 63 */ 64 65 static inline void 66 pcq_split(uint32_t v, u_int *p, u_int *c) 67 { 68 69 *p = v & 0xffff; 70 *c = v >> 16; 71 } 72 73 static inline uint32_t 74 pcq_combine(u_int p, u_int c) 75 { 76 77 return p | (c << 16); 78 } 79 80 static inline u_int 81 pcq_advance(pcq_t *pcq, u_int pc) 82 { 83 84 if (__predict_false(++pc == pcq->pcq_nitems)) { 85 return 0; 86 } 87 return pc; 88 } 89 90 /* 91 * pcq_put: place an item at the end of the queue. 92 */ 93 bool 94 pcq_put(pcq_t *pcq, void *item) 95 { 96 uint32_t v, nv; 97 u_int op, p, c; 98 99 KASSERT(item != NULL); 100 101 do { 102 v = pcq->pcq_pc; 103 pcq_split(v, &op, &c); 104 p = pcq_advance(pcq, op); 105 if (p == c) { 106 /* Queue is full. */ 107 return false; 108 } 109 nv = pcq_combine(p, c); 110 } while (atomic_cas_32(&pcq->pcq_pc, v, nv) != v); 111 112 /* 113 * Ensure that the update to pcq_pc is globally visible before the 114 * data item. See pcq_get(). This also ensures that any changes 115 * that the caller made to the data item are globally visible 116 * before we put it onto the list. 117 */ 118 #ifndef _HAVE_ATOMIC_AS_MEMBAR 119 membar_producer(); 120 #endif 121 pcq->pcq_items[op] = item; 122 123 /* 124 * Synchronization activity to wake up the consumer will ensure 125 * that the update to pcq_items[] is visible before the wakeup 126 * arrives. So, we do not need an additonal memory barrier here. 127 */ 128 return true; 129 } 130 131 /* 132 * pcq_peek: return the next item from the queue without removal. 133 */ 134 void * 135 pcq_peek(pcq_t *pcq) 136 { 137 const uint32_t v = pcq->pcq_pc; 138 u_int p, c; 139 140 pcq_split(v, &p, &c); 141 142 /* See comment on race below in pcq_get(). */ 143 return (p == c) ? NULL : pcq->pcq_items[c]; 144 } 145 146 /* 147 * pcq_get: remove and return the next item for consumption or NULL if empty. 148 * 149 * => The caller must prevent concurrent gets from occuring. 150 */ 151 void * 152 pcq_get(pcq_t *pcq) 153 { 154 uint32_t v, nv; 155 u_int p, c; 156 void *item; 157 158 v = pcq->pcq_pc; 159 pcq_split(v, &p, &c); 160 if (p == c) { 161 /* Queue is empty: nothing to return. */ 162 return NULL; 163 } 164 item = pcq->pcq_items[c]; 165 if (item == NULL) { 166 /* 167 * Raced with sender: we rely on a notification (e.g. softint 168 * or wakeup) being generated after the producer's pcq_put(), 169 * causing us to retry pcq_get() later. 170 */ 171 return NULL; 172 } 173 pcq->pcq_items[c] = NULL; 174 c = pcq_advance(pcq, c); 175 nv = pcq_combine(p, c); 176 177 /* 178 * Ensure that update to pcq_items[] becomes globally visible 179 * before the update to pcq_pc. If it were reodered to occur 180 * after it, we could in theory wipe out a modification made 181 * to pcq_items[] by pcq_put(). 182 */ 183 #ifndef _HAVE_ATOMIC_AS_MEMBAR 184 membar_producer(); 185 #endif 186 while (__predict_false(atomic_cas_32(&pcq->pcq_pc, v, nv) != v)) { 187 v = pcq->pcq_pc; 188 pcq_split(v, &p, &c); 189 c = pcq_advance(pcq, c); 190 nv = pcq_combine(p, c); 191 } 192 return item; 193 } 194 195 pcq_t * 196 pcq_create(size_t nitems, km_flag_t kmflags) 197 { 198 pcq_t *pcq; 199 200 KASSERT(nitems > 0 || nitems <= 0xffff); 201 202 pcq = kmem_zalloc(offsetof(pcq_t, pcq_items[nitems]), kmflags); 203 if (pcq == NULL) { 204 return NULL; 205 } 206 pcq->pcq_nitems = nitems; 207 return pcq; 208 } 209 210 void 211 pcq_destroy(pcq_t *pcq) 212 { 213 214 kmem_free(pcq, offsetof(pcq_t, pcq_items[pcq->pcq_nitems])); 215 } 216 217 size_t 218 pcq_maxitems(pcq_t *pcq) 219 { 220 221 return pcq->pcq_nitems; 222 } 223