 @@ -1,209 +1,222 @@
-/*	$NetBSD: subr_pcq.c,v 1.3 2008/11/11 21:45:33 rmind Exp $	*/
+/*	$NetBSD: subr_pcq.c,v 1.4 2012/01/22 02:55:47 rmind Exp $	*/
 /*-
- * Copyright (c) 2008 The NetBSD Foundation, Inc.
+ * Copyright (c) 2009 The NetBSD Foundation, Inc.
  * All rights reserved.
+ *
  * This code is derived from software contributed to The NetBSD Foundation
- * by Matt Thomas <matt@3am-software.com>
+ * 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.
  */
 #include <sys/cdefs.h>
-__KERNEL_RCSID(0, "$NetBSD: subr_pcq.c,v 1.3 2008/11/11 21:45:33 rmind Exp $");
+__KERNEL_RCSID(0, "$NetBSD: subr_pcq.c,v 1.4 2012/01/22 02:55:47 rmind Exp $");
 #include <sys/param.h>
 #include <sys/types.h>
 #include <sys/atomic.h>
 #include <sys/errno.h>
 #include <sys/kmem.h>
 #include <sys/pcq.h>
-typedef void * volatile pcq_entry_t;
+/*
  * Internal producer-consumer queue structure.  Note: providing a separate
  * cache-line both for pcq_t::pcq_pc and pcq_t::pcq_items.
  */
 struct pcq {
-	pcq_entry_t *pcq_consumer;
+	u_int			pcq_nitems;
-	pcq_entry_t *pcq_producer;
+	uint8_t			pcq_pad1[COHERENCY_UNIT - sizeof(u_int)];
-	pcq_entry_t *pcq_limit;
+	volatile uint32_t	pcq_pc;
-	pcq_entry_t pcq_base[];
+	uint8_t			pcq_pad2[COHERENCY_UNIT - sizeof(uint32_t)];
 	void * volatile		pcq_items[];
 };
-static inline pcq_entry_t *
+/*
-pcq_advance(pcq_t *pcq, pcq_entry_t *ptr)
+ * 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.
  */
-	if (__predict_false(++ptr == pcq->pcq_limit))
+static inline void
-		return pcq->pcq_base;
+pcq_split(uint32_t v, u_int *p, u_int *c)
+{
-	return ptr;
+	*p = v & 0xffff;
 	*c = v >> 16;
+}
-bool
+static inline uint32_t
-pcq_put(pcq_t *pcq, void *item)
+pcq_combine(u_int p, u_int c)
+{
 	pcq_entry_t *producer;
 	KASSERT(item != NULL);
 	/*
 	 * Get our starting point,  While we are doing this, it is
 	 * imperative that pcq->pcq_base/pcq->pcq_limit not change
 	 * in value.  If you need to resize a pcq, init a new pcq
 	 * with the right size and swap pointers to it.
 	 */
 	membar_consumer();	/* see updates to pcq_producer */
 	producer = pcq->pcq_producer;
 	for (;;) {
 		/*
 		 * Preadvance so we reduce the window on updates.
 		 */
 		pcq_entry_t * const new_producer = pcq_advance(pcq, producer);
-		/*
+	return p | (c << 16);
 		 * Try to fill an empty slot
 		 */
 		if (NULL == atomic_cas_ptr(producer, NULL, item)) {
 			/*
 			 * We need to use atomic_cas_ptr since another thread
 			 * might have inserted between these two cas operations
 			 * and we don't want to overwrite a producer that's
 			 * more up-to-date.
 			 */
 			atomic_cas_ptr(&pcq->pcq_producer,
 			    __UNVOLATILE(producer),
 			    __UNVOLATILE(new_producer));
 			/*
 			 * Tell them we were able to enqueue it.
 			 */
 #ifndef __HAVE_ATOMIC_AS_MEMBAR
 			membar_producer();
 #endif
 			return true;
-		/*
+static inline u_int
-		 * If we've reached the consumer, we've filled all the
+pcq_advance(pcq_t *pcq, u_int pc)
 		 * slots and there's no more room so return false.
 		 */
 #ifndef __HAVE_ATOMIC_AS_MEMBAR
 		membar_consumer();	/* see updates to pcq_consumer */
 #endif
 		if (producer == pcq->pcq_consumer)
 			return false;
-		/*
+	if (__predict_false(++pc == pcq->pcq_nitems)) {
-		 * Let's see if the next slot is free...
+		return 0;
 		 */
 		producer = new_producer;
+	}
 	return pc;
+}
 /*
- * It's assumed that the enclosing structure that contains the pcq will
+ * pcq_put: place an item at the end of the queue.
  * provide appropriate locking to prevent concurrent gets from occurring.
  */
-void *
+bool
-pcq_get(pcq_t *pcq)
+pcq_put(pcq_t *pcq, void *item)
+{
-	pcq_entry_t * const consumer = pcq->pcq_consumer;
+	uint32_t v, nv;
-	void *item;
+	u_int op, p, c;
-	/*
+	KASSERT(item != NULL);
 	 * Updates to pcq_consumer doesn't matter since we control it but we
 	 * want to make sure that any stores to what it references have
 	 * completed.
 	 */
 	membar_consumer();
-	/*
+	do {
-	 * If there's nothing to return, just return.
+		v = pcq->pcq_pc;
-	 */
+		pcq_split(v, &op, &c);
-	if ((item = *consumer) == NULL)
+		p = pcq_advance(pcq, op);
-		return NULL;
+		if (p == c) {
 			/* Queue is full. */
 			return false;
+		}
 		nv = pcq_combine(p, c);
 	} while (atomic_cas_32(&pcq->pcq_pc, v, nv) != v);
 	/*
-	 * Update the consumer and free the slot.
+	 * Ensure that the update to pcq_pc is globally visible before the
-	 * Update the consumer pointer first so when producer == consumer
+	 * data item.  See pcq_get().  This also ensures that any changes
-	 * the right thing happens.
+	 * that the caller made to the data item are globally visible
 	 * before we put it onto the list.
 	 * 1) until the slot set to NULL, pcq_put will fail since
 	 *    the slot != NULL && producer == consumer.
 	 * 2) consumer is advanced but the slot is still not NULL,
 	 *    pcq_put will advance by one, see that producer == consumer,
 	 *    and fail.
 	 * 4) Once the slot is set to NULL, the producer can fill the slot
 	 *    and advance the producer.
 	 * and then we are back to 1.
 	 */
-	pcq->pcq_consumer = pcq_advance(pcq, consumer);
+#ifndef _HAVE_ATOMIC_AS_MEMBAR
 	membar_producer();
 	*consumer = NULL;
 	membar_producer();
 #endif
 	pcq->pcq_items[op] = item;
-	return 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 additonal 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 = pcq->pcq_pc;
 	u_int p, c;
-	membar_consumer();	/* see updates to *pcq_consumer */
+	pcq_split(v, &p, &c);
 	return *pcq->pcq_consumer;
 	/* See comment on race below in pcq_get(). */
 	return (p == c) ? NULL : pcq->pcq_items[c];
+}
-size_t
+/*
-pcq_maxitems(pcq_t *pcq)
+ * pcq_get: remove and return the next item for consumption or NULL if empty.
+ *
  * => The caller must prevent concurrent gets from occuring.
  */
 void *
 pcq_get(pcq_t *pcq)
+{
 	uint32_t v, nv;
 	u_int p, c;
 	void *item;
-	return pcq->pcq_limit - pcq->pcq_base;
+	v = pcq->pcq_pc;
 	pcq_split(v, &p, &c);
 	if (p == c) {
 		/* Queue is empty: nothing to return. */
 		return NULL;
+	}
 	item = 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;
+	}
 	pcq->pcq_items[c] = NULL;
 	c = pcq_advance(pcq, c);
 	nv = pcq_combine(p, c);
 	/*
 	 * Ensure that update to pcq_items[] becomes globally visible
 	 * before the update to pcq_pc.  If it were reodered to occur
 	 * after it, we could in theory wipe out a modification made
 	 * to pcq_items[] by pcq_put().
 	 */
 #ifndef _HAVE_ATOMIC_AS_MEMBAR
 	membar_producer();
 #endif
 	while (__predict_false(atomic_cas_32(&pcq->pcq_pc, v, nv) != v)) {
 		v = 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 maxitems, km_flag_t kmflags)
+pcq_create(size_t nitems, km_flag_t kmflags)
+{
 	pcq_t *pcq;
-	KASSERT(maxitems > 0);
+	KASSERT(nitems > 0 || nitems <= 0xffff);
-	pcq = kmem_zalloc(offsetof(pcq_t, pcq_base[maxitems]), kmflags);
+	pcq = kmem_zalloc(offsetof(pcq_t, pcq_items[nitems]), kmflags);
-	if (__predict_false(pcq == NULL))
+	if (pcq == NULL) {
 		return NULL;
+	}
-	pcq->pcq_limit = pcq->pcq_base + maxitems;
+	pcq->pcq_nitems = nitems;
 	pcq->pcq_producer = pcq->pcq_base;
 	pcq->pcq_consumer = pcq->pcq_producer;
 	return pcq;
+}
 void
 pcq_destroy(pcq_t *pcq)
+{
-	KASSERT(*pcq->pcq_consumer == NULL);
+	kmem_free(pcq, offsetof(pcq_t, pcq_items[pcq->pcq_nitems]));
+}
 size_t
 pcq_maxitems(pcq_t *pcq)
+{
-	kmem_free(pcq, (uintptr_t)pcq->pcq_limit - (uintptr_t)pcq);
+	return pcq->pcq_nitems;
+}

 @@ -1,14 +1,14 @@
-.\"     $NetBSD: pcq.9,v 1.5 2010/12/02 12:54:13 wiz Exp $
+.\"     $NetBSD: pcq.9,v 1.6 2012/01/22 02:55:47 rmind Exp $
 .\"
 .\" Copyright (c) 2010 The NetBSD Foundation, Inc.
 .\" All rights reserved.
 .\"
 .\" This code is derived from software contributed to The NetBSD Foundation
 .\" by Matt Thomas.
 .\"
 .\" 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
 @@ -17,27 +17,27 @@
 .\"
 .\" 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.
 .\"
-.Dd January 8, 2010
+.Dd January 22, 2012
 .Dt PCQ 9
 .Os
 .Sh NAME
 .Nm pcq
 .Nd producer/consumer queue
 .Sh SYNOPSIS
 .In sys/pcq.h
 .Ft pcq_t *
 .Fn pcq_create "size_t maxlen" "km_flags_t kmflags"
 .Ft void
 .Fn pcq_destroy "pcq_t *pcq"
 .Ft void *
 .Fn pcq_get "pcq_t *pcq"
 @@ -85,31 +85,31 @@ items at one time.
 should be either
 .Dv KM_SLEEP ,
 if
 .Fn pcq_create
 is allowed to sleep until resources are available, or
 .Dv KM_NOSLEEP
 if it should return
 .Dv NULL
 immediately, if resources are unavailable.
 .It Fn pcq_destroy "pcq"
 Free the resources held by
 .Fa pcq .
 .It Fn pcq_get "pcq"
-Remove the next item to be consumed from the queue and return
+Remove the next item to be consumed from the queue and return it.
 it.
 If the queue is empty,
 return
 .Dv NULL .
 The caller must prevent concurrent gets from occuring.
 .It Fn pcq_maxitems "pcq"
 Return the maximum number of items that the queue can store at
 any one time.
 .It Fn pcq_peek "pcq"
 Return the next item to be consumed from the queue but do not remove
 it from the queue.
 If the queue is empty,
 return
 .Dv NULL .
 .It Fn pcq_put "pcq" "item"
 Place an item at the end of the queue.
 If there is no room in the queue for the item, return
 .Dv false ;
 @@ -122,18 +122,16 @@ The item must not have the value of
 The
 .Nm
 interface is implemented within the file
 .Pa sys/kern/subr_pcq.c .
 .\" .Sh EXAMPLES
 .Sh SEE ALSO
 .Xr atomic_ops 3 ,
 .Xr queue 9
 .Sh HISTORY
 The
 .Nm
 interface first appeared in
 .Nx 6.0 .
 .Sh AUTHORS
 .An Matt Thomas Aq matt@NetBSD.org
 .\" .Sh CAVEATS
 .\" .Sh BUGS
 .\" .Sh SECURITY CONSIDERATIONS