/* $NetBSD: subr_pcq.c,v 1.20 2023/02/24 11:02:27 riastradh Exp $ */ /*- * Copyright (c) 2009, 2019 The NetBSD Foundation, Inc. * All rights reserved. * * This code is derived from software contributed to The NetBSD Foundation * by Andrew Doran. * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions * are met: * 1. Redistributions of source code must retain the above copyright * notice, this list of conditions and the following disclaimer. * 2. Redistributions in binary form must reproduce the above copyright * notice, this list of conditions and the following disclaimer in the * documentation and/or other materials provided with the distribution. * * THIS SOFTWARE IS PROVIDED BY THE NETBSD FOUNDATION, INC. AND CONTRIBUTORS * ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE FOUNDATION OR CONTRIBUTORS * BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE * POSSIBILITY OF SUCH DAMAGE. */ /* * Lockless producer/consumer queue. * * Summary of the producer algorithm in pcq_put (may run many in * parallel with each other and with a consumer): * * P1. initialize an item * * P2. atomic_cas(&pcq->pcq_pc) loop to advance the producer * pointer, reserving a space at c (fails if not enough space) * * P3. atomic_store_release(&pcq->pcq_items[c], item) to publish * the item in the space it reserved * * Summary of the consumer algorithm in pcq_get (must be serialized by * caller with other consumers, may run in parallel with any number of * producers): * * C1. atomic_load_relaxed(&pcq->pcq_pc) to get the consumer * pointer and a snapshot of the producer pointer, which may * point to null items or point to initialized items (fails if * no space reserved for published items yet) * * C2. atomic_load_consume(&pcq->pcq_items[c]) to get the next * unconsumed but potentially published item (fails if item * not published yet) * * C3. pcq->pcq_items[c] = NULL to consume the next unconsumed but * published item * * C4. membar_producer * * C5. atomic_cas(&pcq->pcq_pc) loop to advance the consumer * pointer * * C6. use the item * * Note that there is a weird bare membar_producer which is not matched * by membar_consumer. This is one of the rare cases of a memory * barrier on one side that is not matched by a memory barrier on * another side, but the ordering works out, with a somewhat more * involved proof. * * Some properties that need to be proved: * * Theorem 1. For pcq_put call that leads into pcq_get: * Initializing item at P1 is dependency-ordered before usage of * item at C6, so items placed by pcq_put can be safely used by * the caller of pcq_get. * * Proof sketch. * * Assume load/store P2 synchronizes with load/store C1 * (if not, pcq_get fails in `if (p == c) return NULL'). * * Assume store-release P3 synchronizes with load-consume * C2 (if not, pcq_get fails in `if (item == NULL) return * NULL'). * * Then: * * - P1 is sequenced before store-release P3 * - store-release P3 synchronizes with load-consume C2 * - load-consume C2 is dependency-ordered before C6 * * Hence transitively, P1 is dependency-ordered before C6, * QED. * * Theorem 2. For pcq_get call followed by pcq_put: Nulling out * location at store C3 happens before placing a new item in the * same location at store P3, so items are not lost. * * Proof sketch. * * Assume load/store C5 synchronizes with load/store P2 * (otherwise pcq_peek starts over the CAS loop or fails). * * Then: * * - store C3 is sequenced before membar_producer C4 * - membar_producer C4 is sequenced before load/store C5 * - load/store C5 synchronizes with load/store P2 at &pcq->pcq_pc * - P2 is sequenced before store-release P3 * * Hence transitively, store C3 happens before * store-release P3, QED. */ #include <sys/cdefs.h> __KERNEL_RCSID(0, "$NetBSD: subr_pcq.c,v 1.20 2023/02/24 11:02:27 riastradh Exp $"); #include <sys/param.h> #include <sys/types.h> #include <sys/atomic.h> #include <sys/kmem.h> #include <sys/pcq.h> /* * Internal producer-consumer queue structure. Note: providing a separate * cache-line both for pcq_t::pcq_pc and pcq_t::pcq_items. */ struct pcq { u_int pcq_nitems; uint8_t pcq_pad1[COHERENCY_UNIT - sizeof(u_int)]; volatile uint32_t pcq_pc; uint8_t pcq_pad2[COHERENCY_UNIT - sizeof(uint32_t)]; void * volatile pcq_items[]; }; /* * Producer (p) - stored in the lower 16 bits of pcq_t::pcq_pc. * Consumer (c) - in the higher 16 bits. * * We have a limitation of 16 bits i.e. 0xffff items in the queue. * The PCQ_MAXLEN constant is set accordingly. */ static inline void pcq_split(uint32_t v, u_int *p, u_int *c) { *p = v & 0xffff; *c = v >> 16; } static inline uint32_t pcq_combine(u_int p, u_int c) { return p | (c << 16); } static inline u_int pcq_advance(pcq_t *pcq, u_int pc) { if (__predict_false(++pc == pcq->pcq_nitems)) { return 0; } return pc; } /* * pcq_put: place an item at the end of the queue. */ bool pcq_put(pcq_t *pcq, void *item) { uint32_t v, nv; u_int op, p, c; KASSERT(item != NULL); do { v = atomic_load_relaxed(&pcq->pcq_pc); pcq_split(v, &op, &c); p = pcq_advance(pcq, op); if (p == c) { /* Queue is full. */ return false; } nv = pcq_combine(p, c); } while (atomic_cas_32(&pcq->pcq_pc, v, nv) != v); /* * Ensure that the update to pcq_pc is globally visible before the * data item. See pcq_get(). This also ensures that any changes * that the caller made to the data item are globally visible * before we put it onto the list. */ atomic_store_release(&pcq->pcq_items[op], item); /* * Synchronization activity to wake up the consumer will ensure * that the update to pcq_items[] is visible before the wakeup * arrives. So, we do not need an additional memory barrier here. */ return true; } /* * pcq_peek: return the next item from the queue without removal. */ void * pcq_peek(pcq_t *pcq) { const uint32_t v = atomic_load_relaxed(&pcq->pcq_pc); u_int p, c; pcq_split(v, &p, &c); /* See comment on race below in pcq_get(). */ return (p == c) ? NULL : atomic_load_consume(&pcq->pcq_items[c]); } /* * pcq_get: remove and return the next item for consumption or NULL if empty. * * => The caller must prevent concurrent gets from occurring. */ void * pcq_get(pcq_t *pcq) { uint32_t v, nv; u_int p, c; void *item; v = atomic_load_relaxed(&pcq->pcq_pc); pcq_split(v, &p, &c); if (p == c) { /* Queue is empty: nothing to return. */ return NULL; } item = atomic_load_consume(&pcq->pcq_items[c]); if (item == NULL) { /* * Raced with sender: we rely on a notification (e.g. softint * or wakeup) being generated after the producer's pcq_put(), * causing us to retry pcq_get() later. */ return NULL; } /* * We have exclusive access to this slot, so no need for * atomic_store_*. */ pcq->pcq_items[c] = NULL; c = pcq_advance(pcq, c); nv = pcq_combine(p, c); /* * Ensure that update to pcq_items[c] becomes globally visible * before the update to pcq_pc. If it were reordered to occur * after it, we could in theory wipe out a modification made * to pcq_items[c] by pcq_put(). * * No need for load-before-store ordering of membar_release * because the only load we need to ensure happens first is the * load of pcq->pcq_items[c], but that necessarily happens * before the store to pcq->pcq_items[c] to null it out because * it is at the same memory location. Yes, this is a bare * membar_producer with no matching membar_consumer. */ membar_producer(); while (__predict_false(atomic_cas_32(&pcq->pcq_pc, v, nv) != v)) { v = atomic_load_relaxed(&pcq->pcq_pc); pcq_split(v, &p, &c); c = pcq_advance(pcq, c); nv = pcq_combine(p, c); } return item; } pcq_t * pcq_create(size_t nitems, km_flag_t kmflags) { pcq_t *pcq; KASSERT(nitems > 0); KASSERT(nitems <= PCQ_MAXLEN); pcq = kmem_zalloc(offsetof(pcq_t, pcq_items[nitems]), kmflags); if (pcq != NULL) { pcq->pcq_nitems = nitems; } return pcq; } void pcq_destroy(pcq_t *pcq) { kmem_free(pcq, offsetof(pcq_t, pcq_items[pcq->pcq_nitems])); } size_t pcq_maxitems(pcq_t *pcq) { return pcq->pcq_nitems; }
