sys/kern/subr_pcq.c

   1 /*      $NetBSD$        */
   2
   3 /*-
   4  * Copyright (c) 2008 The NetBSD Foundation, Inc.
   5  * All rights reserved.
   6  *
   7  * This code is derived from software contributed to The NetBSD Foundation
   8  * by Matt Thomas <matt@3am-software.com>
   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 #include <sys/cdefs.h>
  33 __KERNEL_RCSID(0, "$NetBSD: subr_pcq.c,v 1.2 2008/11/11 20:37:15 snj Exp $");
  34
  35 #include <sys/param.h>
  36 #include <sys/types.h>
  37 #include <sys/atomic.h>
  38 #include <sys/errno.h>
  39 #include <sys/kmem.h>
  40
  41 #include <sys/pcq.h>
  42
  43 typedef void * volatile pcq_entry_t;
  44
  45 struct pcq {
  46         pcq_entry_t *pcq_consumer;
  47         pcq_entry_t *pcq_producer;
  48         pcq_entry_t *pcq_limit;
  49         pcq_entry_t pcq_base[];
  50 };
  51
  52 static inline pcq_entry_t *
  53 pcq_advance(pcq_t *pcq, pcq_entry_t *ptr)
  54 {
  55
  56         if (__predict_false(++ptr == pcq->pcq_limit))
  57                 return pcq->pcq_base;
  58
  59         return ptr;
  60 }
  61
  62 bool
  63 pcq_put(pcq_t *pcq, void *item)
  64 {
  65         pcq_entry_t *producer;
  66
  67         KASSERT(item != NULL);
  68
  69         /*
  70          * Get our starting point,  While we are doing this, it is
  71          * imperative that pcq->pcq_base/pcq->pcq_limit not change
  72          * in value.  If you need to resize a pcq, init a new pcq
  73          * with the right size and swap pointers to it.
  74          */
  75         membar_consumer();      /* see updates to pcq_producer */
  76         producer = pcq->pcq_producer;
  77         for (;;) {
  78                 /*
  79                  * Preadvance so we reduce the window on updates.
  80                  */
  81                 pcq_entry_t * const new_producer = pcq_advance(pcq, producer);
  82
  83                 /*
  84                  * Try to fill an empty slot
  85                  */
  86                 if (NULL == atomic_cas_ptr(producer, NULL, item)) {
  87                         /*
  88                          * We need to use atomic_cas_ptr since another thread
  89                          * might have inserted between these two cas operations
  90                          * and we don't want to overwrite a producer that's
  91                          * more up-to-date.
  92                          */
  93                         atomic_cas_ptr(&pcq->pcq_producer,
  94                             __UNVOLATILE(producer),
  95                             __UNVOLATILE(new_producer));
  96                         /*
  97                          * Tell them we were able to enqueue it.
  98                          */
  99 #ifndef __HAVE_ATOMIC_AS_MEMBAR
 100                         membar_producer();
 101 #endif
 102                         return true;
 103                 }
 104
 105                 /*
 106                  * If we've reached the consumer, we've filled all the
 107                  * slots and there's no more room so return false.
 108                  */
 109 #ifndef __HAVE_ATOMIC_AS_MEMBAR
 110                 membar_consumer();      /* see updates to pcq_consumer */
 111 #endif
 112                 if (producer == pcq->pcq_consumer)
 113                         return false;
 114
 115                 /*
 116                  * Let's see if the next slot is free...
 117                  */
 118                 producer = new_producer;
 119         }
 120 }
 121
 122 /*
 123  * It's assumed that the enclosing structure that contains the pcq will
 124  * provide appropriate locking to prevent concurrent gets from occurring.
 125  */
 126 void *
 127 pcq_get(pcq_t *pcq)
 128 {
 129         pcq_entry_t * const consumer = pcq->pcq_consumer;
 130         void *item;
 131
 132         /*
 133          * Updates to pcq_consumer doesn't matter since we control it but we
 134          * want to make sure that any stores to what it references have
 135          * completed.
 136          */
 137         membar_consumer();
 138
 139         /*
 140          * If there's nothing to return, just return.
 141          */
 142         if ((item = *consumer) == NULL)
 143                 return NULL;
 144
 145         /*
 146          * Update the consumer and free the slot.
 147          * Update the consumer pointer first so when producer == consumer
 148          * the right thing happens.
 149          *
 150          * 1) until the slot set to NULL, pcq_put will fail since
 151          *    the slot != NULL && producer == consumer.
 152          * 2) consumer is advanced but the slot is still not NULL,
 153          *    pcq_put will advance by one, see that producer == consumer,
 154          *    and fail.
 155          * 4) Once the slot is set to NULL, the producer can fill the slot
 156          *    and advance the producer.
 157          *
 158          * and then we are back to 1.
 159          */
 160         pcq->pcq_consumer = pcq_advance(pcq, consumer);
 161         membar_producer();
 162
 163         *consumer = NULL;
 164         membar_producer();
 165
 166         return item;
 167 }
 168
 169 void *
 170 pcq_peek(pcq_t *pcq)
 171 {
 172
 173         membar_consumer();      /* see updates to *pcq_consumer */
 174         return *pcq->pcq_consumer;
 175 }
 176
 177 size_t
 178 pcq_maxitems(pcq_t *pcq)
 179 {
 180
 181         return pcq->pcq_limit - pcq->pcq_base;
 182 }
 183
 184 pcq_t *
 185 pcq_create(size_t maxitems, km_flag_t kmflags)
 186 {
 187         pcq_t *pcq;
 188
 189         KASSERT(maxitems > 0);
 190
 191         pcq = kmem_zalloc(offsetof(pcq_t, pcq_base[maxitems]), kmflags);
 192         if (__predict_false(pcq == NULL))
 193                 return NULL;
 194
 195         pcq->pcq_limit = pcq->pcq_base + maxitems;
 196         pcq->pcq_producer = pcq->pcq_base;
 197         pcq->pcq_consumer = pcq->pcq_producer;
 198
 199         return pcq;
 200 }
 201
 202 void
 203 pcq_destroy(pcq_t *pcq)
 204 {
 205
 206         KASSERT(*pcq->pcq_consumer == NULL);
 207
 208         kmem_free(pcq, (uintptr_t)pcq->pcq_limit - (uintptr_t)pcq);
 209 }