* [dpdk-dev] [PATCH RFCv2 0/4] generalise rte_ring to allow different datatypes
@ 2017-01-24 10:39 Bruce Richardson
2017-01-24 10:39 ` [dpdk-dev] [PATCH RFCv2 1/4] ring: create common ring files Bruce Richardson
` (3 more replies)
0 siblings, 4 replies; 5+ messages in thread
From: Bruce Richardson @ 2017-01-24 10:39 UTC (permalink / raw)
To: olivier.matz; +Cc: dev, Bruce Richardson
Following on from the previous RFC, this generalises the rte_ring
structure using C constructs rather than using macros. The idea
here is to have the size of the data objects passed in to all common
functions and then switching the code paths, where necessary, based on
those size parameters. For the standard rte_ring APIs, these work on
void pointers, so the extra parameter is always specified as
sizeof(void *). A new event_ring type is also added as a patch to test
out creating and using rings with different size objects.
Comments welcome on this RFC, though I still hope to try out some
other options, as I think there may be still better ways to achieve
the same end.
Bruce Richardson (4):
ring: create common ring files
ring: separate common and rte_ring specific functions
ring: allow common ring to use 8 or 16 byte values
ring: add new event ring class
app/test/Makefile | 1 +
app/test/test_event_ring.c | 83 ++++
lib/librte_ring/Makefile | 8 +-
lib/librte_ring/rte_common_ring.c | 386 ++++++++++++++++
lib/librte_ring/rte_common_ring.h | 925 ++++++++++++++++++++++++++++++++++++++
lib/librte_ring/rte_event_ring.c | 87 ++++
lib/librte_ring/rte_event_ring.h | 696 ++++++++++++++++++++++++++++
lib/librte_ring/rte_ring.c | 320 +------------
lib/librte_ring/rte_ring.h | 606 +------------------------
9 files changed, 2210 insertions(+), 902 deletions(-)
create mode 100644 app/test/test_event_ring.c
create mode 100644 lib/librte_ring/rte_common_ring.c
create mode 100644 lib/librte_ring/rte_common_ring.h
create mode 100644 lib/librte_ring/rte_event_ring.c
create mode 100644 lib/librte_ring/rte_event_ring.h
--
2.9.3
^ permalink raw reply [flat|nested] 5+ messages in thread
* [dpdk-dev] [PATCH RFCv2 1/4] ring: create common ring files
2017-01-24 10:39 [dpdk-dev] [PATCH RFCv2 0/4] generalise rte_ring to allow different datatypes Bruce Richardson
@ 2017-01-24 10:39 ` Bruce Richardson
2017-01-24 10:39 ` [dpdk-dev] [PATCH RFCv2 2/4] ring: separate common and rte_ring specific functions Bruce Richardson
` (2 subsequent siblings)
3 siblings, 0 replies; 5+ messages in thread
From: Bruce Richardson @ 2017-01-24 10:39 UTC (permalink / raw)
To: olivier.matz; +Cc: dev, Bruce Richardson
Create rte_common_ring.[ch] files which will be modified to contain
generic ring implementation code to be shared across multiple ring
implementations for different sizes and types of data.
For now, these are exact copies of the original rte_ring files.
Signed-off-by: Bruce Richardson <bruce.richardson@intel.com>
---
lib/librte_ring/Makefile | 5 +-
lib/librte_ring/{rte_ring.c => rte_common_ring.c} | 0
lib/librte_ring/rte_common_ring.h | 1269 ++++++++++++++++++++
lib/librte_ring/rte_ring.h | 1270 +--------------------
4 files changed, 1273 insertions(+), 1271 deletions(-)
rename lib/librte_ring/{rte_ring.c => rte_common_ring.c} (100%)
create mode 100644 lib/librte_ring/rte_common_ring.h
mode change 100644 => 120000 lib/librte_ring/rte_ring.h
diff --git a/lib/librte_ring/Makefile b/lib/librte_ring/Makefile
index 4b1112e..1e2396e 100644
--- a/lib/librte_ring/Makefile
+++ b/lib/librte_ring/Makefile
@@ -41,10 +41,11 @@ EXPORT_MAP := rte_ring_version.map
LIBABIVER := 1
# all source are stored in SRCS-y
-SRCS-$(CONFIG_RTE_LIBRTE_RING) := rte_ring.c
+SRCS-$(CONFIG_RTE_LIBRTE_RING) += rte_common_ring.c
# install includes
-SYMLINK-$(CONFIG_RTE_LIBRTE_RING)-include := rte_ring.h
+SYMLINK-$(CONFIG_RTE_LIBRTE_RING)-include += rte_ring.h
+SYMLINK-$(CONFIG_RTE_LIBRTE_RING)-include += rte_common_ring.h
DEPDIRS-$(CONFIG_RTE_LIBRTE_RING) += lib/librte_eal
diff --git a/lib/librte_ring/rte_ring.c b/lib/librte_ring/rte_common_ring.c
similarity index 100%
rename from lib/librte_ring/rte_ring.c
rename to lib/librte_ring/rte_common_ring.c
diff --git a/lib/librte_ring/rte_common_ring.h b/lib/librte_ring/rte_common_ring.h
new file mode 100644
index 0000000..e359aff
--- /dev/null
+++ b/lib/librte_ring/rte_common_ring.h
@@ -0,0 +1,1269 @@
+/*-
+ * BSD LICENSE
+ *
+ * Copyright(c) 2010-2014 Intel Corporation. All rights reserved.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ * * 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.
+ * * Neither the name of Intel Corporation nor the names of its
+ * contributors may be used to endorse or promote products derived
+ * from this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS 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 COPYRIGHT
+ * OWNER 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.
+ */
+
+/*
+ * Derived from FreeBSD's bufring.h
+ *
+ **************************************************************************
+ *
+ * Copyright (c) 2007-2009 Kip Macy kmacy@freebsd.org
+ * All rights reserved.
+ *
+ * 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. The name of Kip Macy nor the names of other
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS 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 COPYRIGHT OWNER 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.
+ *
+ ***************************************************************************/
+
+#ifndef _RTE_RING_H_
+#define _RTE_RING_H_
+
+/**
+ * @file
+ * RTE Ring
+ *
+ * The Ring Manager is a fixed-size queue, implemented as a table of
+ * pointers. Head and tail pointers are modified atomically, allowing
+ * concurrent access to it. It has the following features:
+ *
+ * - FIFO (First In First Out)
+ * - Maximum size is fixed; the pointers are stored in a table.
+ * - Lockless implementation.
+ * - Multi- or single-consumer dequeue.
+ * - Multi- or single-producer enqueue.
+ * - Bulk dequeue.
+ * - Bulk enqueue.
+ *
+ * Note: the ring implementation is not preemptable. A lcore must not
+ * be interrupted by another task that uses the same ring.
+ *
+ */
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include <stdio.h>
+#include <stdint.h>
+#include <sys/queue.h>
+#include <errno.h>
+#include <rte_common.h>
+#include <rte_memory.h>
+#include <rte_lcore.h>
+#include <rte_atomic.h>
+#include <rte_branch_prediction.h>
+#include <rte_memzone.h>
+
+#define RTE_TAILQ_RING_NAME "RTE_RING"
+
+enum rte_ring_queue_behavior {
+ RTE_RING_QUEUE_FIXED = 0, /* Enq/Deq a fixed number of items from a ring */
+ RTE_RING_QUEUE_VARIABLE /* Enq/Deq as many items as possible from ring */
+};
+
+#ifdef RTE_LIBRTE_RING_DEBUG
+/**
+ * A structure that stores the ring statistics (per-lcore).
+ */
+struct rte_ring_debug_stats {
+ uint64_t enq_success_bulk; /**< Successful enqueues number. */
+ uint64_t enq_success_objs; /**< Objects successfully enqueued. */
+ uint64_t enq_quota_bulk; /**< Successful enqueues above watermark. */
+ uint64_t enq_quota_objs; /**< Objects enqueued above watermark. */
+ uint64_t enq_fail_bulk; /**< Failed enqueues number. */
+ uint64_t enq_fail_objs; /**< Objects that failed to be enqueued. */
+ uint64_t deq_success_bulk; /**< Successful dequeues number. */
+ uint64_t deq_success_objs; /**< Objects successfully dequeued. */
+ uint64_t deq_fail_bulk; /**< Failed dequeues number. */
+ uint64_t deq_fail_objs; /**< Objects that failed to be dequeued. */
+} __rte_cache_aligned;
+#endif
+
+#define RTE_RING_MZ_PREFIX "RG_"
+/**< The maximum length of a ring name. */
+#define RTE_RING_NAMESIZE (RTE_MEMZONE_NAMESIZE - \
+ sizeof(RTE_RING_MZ_PREFIX) + 1)
+
+#ifndef RTE_RING_PAUSE_REP_COUNT
+#define RTE_RING_PAUSE_REP_COUNT 0 /**< Yield after pause num of times, no yield
+ * if RTE_RING_PAUSE_REP not defined. */
+#endif
+
+struct rte_memzone; /* forward declaration, so as not to require memzone.h */
+
+/**
+ * An RTE ring structure.
+ *
+ * The producer and the consumer have a head and a tail index. The particularity
+ * of these index is that they are not between 0 and size(ring). These indexes
+ * are between 0 and 2^32, and we mask their value when we access the ring[]
+ * field. Thanks to this assumption, we can do subtractions between 2 index
+ * values in a modulo-32bit base: that's why the overflow of the indexes is not
+ * a problem.
+ */
+struct rte_ring {
+ /*
+ * Note: this field kept the RTE_MEMZONE_NAMESIZE size due to ABI
+ * compatibility requirements, it could be changed to RTE_RING_NAMESIZE
+ * next time the ABI changes
+ */
+ char name[RTE_MEMZONE_NAMESIZE]; /**< Name of the ring. */
+ int flags; /**< Flags supplied at creation. */
+ const struct rte_memzone *memzone;
+ /**< Memzone, if any, containing the rte_ring */
+
+ /** Ring producer status. */
+ struct prod {
+ uint32_t watermark; /**< Maximum items before EDQUOT. */
+ uint32_t sp_enqueue; /**< True, if single producer. */
+ uint32_t size; /**< Size of ring. */
+ uint32_t mask; /**< Mask (size-1) of ring. */
+ volatile uint32_t head; /**< Producer head. */
+ volatile uint32_t tail; /**< Producer tail. */
+ } prod __rte_cache_aligned;
+
+ /** Ring consumer status. */
+ struct cons {
+ uint32_t sc_dequeue; /**< True, if single consumer. */
+ uint32_t size; /**< Size of the ring. */
+ uint32_t mask; /**< Mask (size-1) of ring. */
+ volatile uint32_t head; /**< Consumer head. */
+ volatile uint32_t tail; /**< Consumer tail. */
+#ifdef RTE_RING_SPLIT_PROD_CONS
+ } cons __rte_cache_aligned;
+#else
+ } cons;
+#endif
+
+#ifdef RTE_LIBRTE_RING_DEBUG
+ struct rte_ring_debug_stats stats[RTE_MAX_LCORE];
+#endif
+
+ void *ring[] __rte_cache_aligned; /**< Memory space of ring starts here.
+ * not volatile so need to be careful
+ * about compiler re-ordering */
+};
+
+#define RING_F_SP_ENQ 0x0001 /**< The default enqueue is "single-producer". */
+#define RING_F_SC_DEQ 0x0002 /**< The default dequeue is "single-consumer". */
+#define RTE_RING_QUOT_EXCEED (1 << 31) /**< Quota exceed for burst ops */
+#define RTE_RING_SZ_MASK (unsigned)(0x0fffffff) /**< Ring size mask */
+
+/**
+ * @internal When debug is enabled, store ring statistics.
+ * @param r
+ * A pointer to the ring.
+ * @param name
+ * The name of the statistics field to increment in the ring.
+ * @param n
+ * The number to add to the object-oriented statistics.
+ */
+#ifdef RTE_LIBRTE_RING_DEBUG
+#define __RING_STAT_ADD(r, name, n) do { \
+ unsigned __lcore_id = rte_lcore_id(); \
+ if (__lcore_id < RTE_MAX_LCORE) { \
+ r->stats[__lcore_id].name##_objs += n; \
+ r->stats[__lcore_id].name##_bulk += 1; \
+ } \
+ } while(0)
+#else
+#define __RING_STAT_ADD(r, name, n) do {} while(0)
+#endif
+
+/**
+ * Calculate the memory size needed for a ring
+ *
+ * This function returns the number of bytes needed for a ring, given
+ * the number of elements in it. This value is the sum of the size of
+ * the structure rte_ring and the size of the memory needed by the
+ * objects pointers. The value is aligned to a cache line size.
+ *
+ * @param count
+ * The number of elements in the ring (must be a power of 2).
+ * @return
+ * - The memory size needed for the ring on success.
+ * - -EINVAL if count is not a power of 2.
+ */
+ssize_t rte_ring_get_memsize(unsigned count);
+
+/**
+ * Initialize a ring structure.
+ *
+ * Initialize a ring structure in memory pointed by "r". The size of the
+ * memory area must be large enough to store the ring structure and the
+ * object table. It is advised to use rte_ring_get_memsize() to get the
+ * appropriate size.
+ *
+ * The ring size is set to *count*, which must be a power of two. Water
+ * marking is disabled by default. The real usable ring size is
+ * *count-1* instead of *count* to differentiate a free ring from an
+ * empty ring.
+ *
+ * The ring is not added in RTE_TAILQ_RING global list. Indeed, the
+ * memory given by the caller may not be shareable among dpdk
+ * processes.
+ *
+ * @param r
+ * The pointer to the ring structure followed by the objects table.
+ * @param name
+ * The name of the ring.
+ * @param count
+ * The number of elements in the ring (must be a power of 2).
+ * @param flags
+ * An OR of the following:
+ * - RING_F_SP_ENQ: If this flag is set, the default behavior when
+ * using ``rte_ring_enqueue()`` or ``rte_ring_enqueue_bulk()``
+ * is "single-producer". Otherwise, it is "multi-producers".
+ * - RING_F_SC_DEQ: If this flag is set, the default behavior when
+ * using ``rte_ring_dequeue()`` or ``rte_ring_dequeue_bulk()``
+ * is "single-consumer". Otherwise, it is "multi-consumers".
+ * @return
+ * 0 on success, or a negative value on error.
+ */
+int rte_ring_init(struct rte_ring *r, const char *name, unsigned count,
+ unsigned flags);
+
+/**
+ * Create a new ring named *name* in memory.
+ *
+ * This function uses ``memzone_reserve()`` to allocate memory. Then it
+ * calls rte_ring_init() to initialize an empty ring.
+ *
+ * The new ring size is set to *count*, which must be a power of
+ * two. Water marking is disabled by default. The real usable ring size
+ * is *count-1* instead of *count* to differentiate a free ring from an
+ * empty ring.
+ *
+ * The ring is added in RTE_TAILQ_RING list.
+ *
+ * @param name
+ * The name of the ring.
+ * @param count
+ * The size of the ring (must be a power of 2).
+ * @param socket_id
+ * The *socket_id* argument is the socket identifier in case of
+ * NUMA. The value can be *SOCKET_ID_ANY* if there is no NUMA
+ * constraint for the reserved zone.
+ * @param flags
+ * An OR of the following:
+ * - RING_F_SP_ENQ: If this flag is set, the default behavior when
+ * using ``rte_ring_enqueue()`` or ``rte_ring_enqueue_bulk()``
+ * is "single-producer". Otherwise, it is "multi-producers".
+ * - RING_F_SC_DEQ: If this flag is set, the default behavior when
+ * using ``rte_ring_dequeue()`` or ``rte_ring_dequeue_bulk()``
+ * is "single-consumer". Otherwise, it is "multi-consumers".
+ * @return
+ * On success, the pointer to the new allocated ring. NULL on error with
+ * rte_errno set appropriately. Possible errno values include:
+ * - E_RTE_NO_CONFIG - function could not get pointer to rte_config structure
+ * - E_RTE_SECONDARY - function was called from a secondary process instance
+ * - EINVAL - count provided is not a power of 2
+ * - ENOSPC - the maximum number of memzones has already been allocated
+ * - EEXIST - a memzone with the same name already exists
+ * - ENOMEM - no appropriate memory area found in which to create memzone
+ */
+struct rte_ring *rte_ring_create(const char *name, unsigned count,
+ int socket_id, unsigned flags);
+/**
+ * De-allocate all memory used by the ring.
+ *
+ * @param r
+ * Ring to free
+ */
+void rte_ring_free(struct rte_ring *r);
+
+/**
+ * Change the high water mark.
+ *
+ * If *count* is 0, water marking is disabled. Otherwise, it is set to the
+ * *count* value. The *count* value must be greater than 0 and less
+ * than the ring size.
+ *
+ * This function can be called at any time (not necessarily at
+ * initialization).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param count
+ * The new water mark value.
+ * @return
+ * - 0: Success; water mark changed.
+ * - -EINVAL: Invalid water mark value.
+ */
+int rte_ring_set_water_mark(struct rte_ring *r, unsigned count);
+
+/**
+ * Dump the status of the ring to a file.
+ *
+ * @param f
+ * A pointer to a file for output
+ * @param r
+ * A pointer to the ring structure.
+ */
+void rte_ring_dump(FILE *f, const struct rte_ring *r);
+
+/* the actual enqueue of pointers on the ring.
+ * Placed here since identical code needed in both
+ * single and multi producer enqueue functions */
+#define ENQUEUE_PTRS() do { \
+ const uint32_t size = r->prod.size; \
+ uint32_t idx = prod_head & mask; \
+ if (likely(idx + n < size)) { \
+ for (i = 0; i < (n & ((~(unsigned)0x3))); i+=4, idx+=4) { \
+ r->ring[idx] = obj_table[i]; \
+ r->ring[idx+1] = obj_table[i+1]; \
+ r->ring[idx+2] = obj_table[i+2]; \
+ r->ring[idx+3] = obj_table[i+3]; \
+ } \
+ switch (n & 0x3) { \
+ case 3: r->ring[idx++] = obj_table[i++]; \
+ case 2: r->ring[idx++] = obj_table[i++]; \
+ case 1: r->ring[idx++] = obj_table[i++]; \
+ } \
+ } else { \
+ for (i = 0; idx < size; i++, idx++)\
+ r->ring[idx] = obj_table[i]; \
+ for (idx = 0; i < n; i++, idx++) \
+ r->ring[idx] = obj_table[i]; \
+ } \
+} while(0)
+
+/* the actual copy of pointers on the ring to obj_table.
+ * Placed here since identical code needed in both
+ * single and multi consumer dequeue functions */
+#define DEQUEUE_PTRS() do { \
+ uint32_t idx = cons_head & mask; \
+ const uint32_t size = r->cons.size; \
+ if (likely(idx + n < size)) { \
+ for (i = 0; i < (n & (~(unsigned)0x3)); i+=4, idx+=4) {\
+ obj_table[i] = r->ring[idx]; \
+ obj_table[i+1] = r->ring[idx+1]; \
+ obj_table[i+2] = r->ring[idx+2]; \
+ obj_table[i+3] = r->ring[idx+3]; \
+ } \
+ switch (n & 0x3) { \
+ case 3: obj_table[i++] = r->ring[idx++]; \
+ case 2: obj_table[i++] = r->ring[idx++]; \
+ case 1: obj_table[i++] = r->ring[idx++]; \
+ } \
+ } else { \
+ for (i = 0; idx < size; i++, idx++) \
+ obj_table[i] = r->ring[idx]; \
+ for (idx = 0; i < n; i++, idx++) \
+ obj_table[i] = r->ring[idx]; \
+ } \
+} while (0)
+
+/**
+ * @internal Enqueue several objects on the ring (multi-producers safe).
+ *
+ * This function uses a "compare and set" instruction to move the
+ * producer index atomically.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects).
+ * @param n
+ * The number of objects to add in the ring from the obj_table.
+ * @param behavior
+ * RTE_RING_QUEUE_FIXED: Enqueue a fixed number of items from a ring
+ * RTE_RING_QUEUE_VARIABLE: Enqueue as many items a possible from ring
+ * @return
+ * Depend on the behavior value
+ * if behavior = RTE_RING_QUEUE_FIXED
+ * - 0: Success; objects enqueue.
+ * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
+ * high water mark is exceeded.
+ * - -ENOBUFS: Not enough room in the ring to enqueue, no object is enqueued.
+ * if behavior = RTE_RING_QUEUE_VARIABLE
+ * - n: Actual number of objects enqueued.
+ */
+static inline int __attribute__((always_inline))
+__rte_ring_mp_do_enqueue(struct rte_ring *r, void * const *obj_table,
+ unsigned n, enum rte_ring_queue_behavior behavior)
+{
+ uint32_t prod_head, prod_next;
+ uint32_t cons_tail, free_entries;
+ const unsigned max = n;
+ int success;
+ unsigned i, rep = 0;
+ uint32_t mask = r->prod.mask;
+ int ret;
+
+ /* Avoid the unnecessary cmpset operation below, which is also
+ * potentially harmful when n equals 0. */
+ if (n == 0)
+ return 0;
+
+ /* move prod.head atomically */
+ do {
+ /* Reset n to the initial burst count */
+ n = max;
+
+ prod_head = r->prod.head;
+ cons_tail = r->cons.tail;
+ /* The subtraction is done between two unsigned 32bits value
+ * (the result is always modulo 32 bits even if we have
+ * prod_head > cons_tail). So 'free_entries' is always between 0
+ * and size(ring)-1. */
+ free_entries = (mask + cons_tail - prod_head);
+
+ /* check that we have enough room in ring */
+ if (unlikely(n > free_entries)) {
+ if (behavior == RTE_RING_QUEUE_FIXED) {
+ __RING_STAT_ADD(r, enq_fail, n);
+ return -ENOBUFS;
+ }
+ else {
+ /* No free entry available */
+ if (unlikely(free_entries == 0)) {
+ __RING_STAT_ADD(r, enq_fail, n);
+ return 0;
+ }
+
+ n = free_entries;
+ }
+ }
+
+ prod_next = prod_head + n;
+ success = rte_atomic32_cmpset(&r->prod.head, prod_head,
+ prod_next);
+ } while (unlikely(success == 0));
+
+ /* write entries in ring */
+ ENQUEUE_PTRS();
+ rte_smp_wmb();
+
+ /* if we exceed the watermark */
+ if (unlikely(((mask + 1) - free_entries + n) > r->prod.watermark)) {
+ ret = (behavior == RTE_RING_QUEUE_FIXED) ? -EDQUOT :
+ (int)(n | RTE_RING_QUOT_EXCEED);
+ __RING_STAT_ADD(r, enq_quota, n);
+ }
+ else {
+ ret = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : n;
+ __RING_STAT_ADD(r, enq_success, n);
+ }
+
+ /*
+ * If there are other enqueues in progress that preceded us,
+ * we need to wait for them to complete
+ */
+ while (unlikely(r->prod.tail != prod_head)) {
+ rte_pause();
+
+ /* Set RTE_RING_PAUSE_REP_COUNT to avoid spin too long waiting
+ * for other thread finish. It gives pre-empted thread a chance
+ * to proceed and finish with ring dequeue operation. */
+ if (RTE_RING_PAUSE_REP_COUNT &&
+ ++rep == RTE_RING_PAUSE_REP_COUNT) {
+ rep = 0;
+ sched_yield();
+ }
+ }
+ r->prod.tail = prod_next;
+ return ret;
+}
+
+/**
+ * @internal Enqueue several objects on a ring (NOT multi-producers safe).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects).
+ * @param n
+ * The number of objects to add in the ring from the obj_table.
+ * @param behavior
+ * RTE_RING_QUEUE_FIXED: Enqueue a fixed number of items from a ring
+ * RTE_RING_QUEUE_VARIABLE: Enqueue as many items a possible from ring
+ * @return
+ * Depend on the behavior value
+ * if behavior = RTE_RING_QUEUE_FIXED
+ * - 0: Success; objects enqueue.
+ * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
+ * high water mark is exceeded.
+ * - -ENOBUFS: Not enough room in the ring to enqueue, no object is enqueued.
+ * if behavior = RTE_RING_QUEUE_VARIABLE
+ * - n: Actual number of objects enqueued.
+ */
+static inline int __attribute__((always_inline))
+__rte_ring_sp_do_enqueue(struct rte_ring *r, void * const *obj_table,
+ unsigned n, enum rte_ring_queue_behavior behavior)
+{
+ uint32_t prod_head, cons_tail;
+ uint32_t prod_next, free_entries;
+ unsigned i;
+ uint32_t mask = r->prod.mask;
+ int ret;
+
+ prod_head = r->prod.head;
+ cons_tail = r->cons.tail;
+ /* The subtraction is done between two unsigned 32bits value
+ * (the result is always modulo 32 bits even if we have
+ * prod_head > cons_tail). So 'free_entries' is always between 0
+ * and size(ring)-1. */
+ free_entries = mask + cons_tail - prod_head;
+
+ /* check that we have enough room in ring */
+ if (unlikely(n > free_entries)) {
+ if (behavior == RTE_RING_QUEUE_FIXED) {
+ __RING_STAT_ADD(r, enq_fail, n);
+ return -ENOBUFS;
+ }
+ else {
+ /* No free entry available */
+ if (unlikely(free_entries == 0)) {
+ __RING_STAT_ADD(r, enq_fail, n);
+ return 0;
+ }
+
+ n = free_entries;
+ }
+ }
+
+ prod_next = prod_head + n;
+ r->prod.head = prod_next;
+
+ /* write entries in ring */
+ ENQUEUE_PTRS();
+ rte_smp_wmb();
+
+ /* if we exceed the watermark */
+ if (unlikely(((mask + 1) - free_entries + n) > r->prod.watermark)) {
+ ret = (behavior == RTE_RING_QUEUE_FIXED) ? -EDQUOT :
+ (int)(n | RTE_RING_QUOT_EXCEED);
+ __RING_STAT_ADD(r, enq_quota, n);
+ }
+ else {
+ ret = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : n;
+ __RING_STAT_ADD(r, enq_success, n);
+ }
+
+ r->prod.tail = prod_next;
+ return ret;
+}
+
+/**
+ * @internal Dequeue several objects from a ring (multi-consumers safe). When
+ * the request objects are more than the available objects, only dequeue the
+ * actual number of objects
+ *
+ * This function uses a "compare and set" instruction to move the
+ * consumer index atomically.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects) that will be filled.
+ * @param n
+ * The number of objects to dequeue from the ring to the obj_table.
+ * @param behavior
+ * RTE_RING_QUEUE_FIXED: Dequeue a fixed number of items from a ring
+ * RTE_RING_QUEUE_VARIABLE: Dequeue as many items a possible from ring
+ * @return
+ * Depend on the behavior value
+ * if behavior = RTE_RING_QUEUE_FIXED
+ * - 0: Success; objects dequeued.
+ * - -ENOENT: Not enough entries in the ring to dequeue; no object is
+ * dequeued.
+ * if behavior = RTE_RING_QUEUE_VARIABLE
+ * - n: Actual number of objects dequeued.
+ */
+
+static inline int __attribute__((always_inline))
+__rte_ring_mc_do_dequeue(struct rte_ring *r, void **obj_table,
+ unsigned n, enum rte_ring_queue_behavior behavior)
+{
+ uint32_t cons_head, prod_tail;
+ uint32_t cons_next, entries;
+ const unsigned max = n;
+ int success;
+ unsigned i, rep = 0;
+ uint32_t mask = r->prod.mask;
+
+ /* Avoid the unnecessary cmpset operation below, which is also
+ * potentially harmful when n equals 0. */
+ if (n == 0)
+ return 0;
+
+ /* move cons.head atomically */
+ do {
+ /* Restore n as it may change every loop */
+ n = max;
+
+ cons_head = r->cons.head;
+ prod_tail = r->prod.tail;
+ /* The subtraction is done between two unsigned 32bits value
+ * (the result is always modulo 32 bits even if we have
+ * cons_head > prod_tail). So 'entries' is always between 0
+ * and size(ring)-1. */
+ entries = (prod_tail - cons_head);
+
+ /* Set the actual entries for dequeue */
+ if (n > entries) {
+ if (behavior == RTE_RING_QUEUE_FIXED) {
+ __RING_STAT_ADD(r, deq_fail, n);
+ return -ENOENT;
+ }
+ else {
+ if (unlikely(entries == 0)){
+ __RING_STAT_ADD(r, deq_fail, n);
+ return 0;
+ }
+
+ n = entries;
+ }
+ }
+
+ cons_next = cons_head + n;
+ success = rte_atomic32_cmpset(&r->cons.head, cons_head,
+ cons_next);
+ } while (unlikely(success == 0));
+
+ /* copy in table */
+ DEQUEUE_PTRS();
+ rte_smp_rmb();
+
+ /*
+ * If there are other dequeues in progress that preceded us,
+ * we need to wait for them to complete
+ */
+ while (unlikely(r->cons.tail != cons_head)) {
+ rte_pause();
+
+ /* Set RTE_RING_PAUSE_REP_COUNT to avoid spin too long waiting
+ * for other thread finish. It gives pre-empted thread a chance
+ * to proceed and finish with ring dequeue operation. */
+ if (RTE_RING_PAUSE_REP_COUNT &&
+ ++rep == RTE_RING_PAUSE_REP_COUNT) {
+ rep = 0;
+ sched_yield();
+ }
+ }
+ __RING_STAT_ADD(r, deq_success, n);
+ r->cons.tail = cons_next;
+
+ return behavior == RTE_RING_QUEUE_FIXED ? 0 : n;
+}
+
+/**
+ * @internal Dequeue several objects from a ring (NOT multi-consumers safe).
+ * When the request objects are more than the available objects, only dequeue
+ * the actual number of objects
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects) that will be filled.
+ * @param n
+ * The number of objects to dequeue from the ring to the obj_table.
+ * @param behavior
+ * RTE_RING_QUEUE_FIXED: Dequeue a fixed number of items from a ring
+ * RTE_RING_QUEUE_VARIABLE: Dequeue as many items a possible from ring
+ * @return
+ * Depend on the behavior value
+ * if behavior = RTE_RING_QUEUE_FIXED
+ * - 0: Success; objects dequeued.
+ * - -ENOENT: Not enough entries in the ring to dequeue; no object is
+ * dequeued.
+ * if behavior = RTE_RING_QUEUE_VARIABLE
+ * - n: Actual number of objects dequeued.
+ */
+static inline int __attribute__((always_inline))
+__rte_ring_sc_do_dequeue(struct rte_ring *r, void **obj_table,
+ unsigned n, enum rte_ring_queue_behavior behavior)
+{
+ uint32_t cons_head, prod_tail;
+ uint32_t cons_next, entries;
+ unsigned i;
+ uint32_t mask = r->prod.mask;
+
+ cons_head = r->cons.head;
+ prod_tail = r->prod.tail;
+ /* The subtraction is done between two unsigned 32bits value
+ * (the result is always modulo 32 bits even if we have
+ * cons_head > prod_tail). So 'entries' is always between 0
+ * and size(ring)-1. */
+ entries = prod_tail - cons_head;
+
+ if (n > entries) {
+ if (behavior == RTE_RING_QUEUE_FIXED) {
+ __RING_STAT_ADD(r, deq_fail, n);
+ return -ENOENT;
+ }
+ else {
+ if (unlikely(entries == 0)){
+ __RING_STAT_ADD(r, deq_fail, n);
+ return 0;
+ }
+
+ n = entries;
+ }
+ }
+
+ cons_next = cons_head + n;
+ r->cons.head = cons_next;
+
+ /* copy in table */
+ DEQUEUE_PTRS();
+ rte_smp_rmb();
+
+ __RING_STAT_ADD(r, deq_success, n);
+ r->cons.tail = cons_next;
+ return behavior == RTE_RING_QUEUE_FIXED ? 0 : n;
+}
+
+/**
+ * Enqueue several objects on the ring (multi-producers safe).
+ *
+ * This function uses a "compare and set" instruction to move the
+ * producer index atomically.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects).
+ * @param n
+ * The number of objects to add in the ring from the obj_table.
+ * @return
+ * - 0: Success; objects enqueue.
+ * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
+ * high water mark is exceeded.
+ * - -ENOBUFS: Not enough room in the ring to enqueue, no object is enqueued.
+ */
+static inline int __attribute__((always_inline))
+rte_ring_mp_enqueue_bulk(struct rte_ring *r, void * const *obj_table,
+ unsigned n)
+{
+ return __rte_ring_mp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
+}
+
+/**
+ * Enqueue several objects on a ring (NOT multi-producers safe).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects).
+ * @param n
+ * The number of objects to add in the ring from the obj_table.
+ * @return
+ * - 0: Success; objects enqueued.
+ * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
+ * high water mark is exceeded.
+ * - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
+ */
+static inline int __attribute__((always_inline))
+rte_ring_sp_enqueue_bulk(struct rte_ring *r, void * const *obj_table,
+ unsigned n)
+{
+ return __rte_ring_sp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
+}
+
+/**
+ * Enqueue several objects on a ring.
+ *
+ * This function calls the multi-producer or the single-producer
+ * version depending on the default behavior that was specified at
+ * ring creation time (see flags).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects).
+ * @param n
+ * The number of objects to add in the ring from the obj_table.
+ * @return
+ * - 0: Success; objects enqueued.
+ * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
+ * high water mark is exceeded.
+ * - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
+ */
+static inline int __attribute__((always_inline))
+rte_ring_enqueue_bulk(struct rte_ring *r, void * const *obj_table,
+ unsigned n)
+{
+ if (r->prod.sp_enqueue)
+ return rte_ring_sp_enqueue_bulk(r, obj_table, n);
+ else
+ return rte_ring_mp_enqueue_bulk(r, obj_table, n);
+}
+
+/**
+ * Enqueue one object on a ring (multi-producers safe).
+ *
+ * This function uses a "compare and set" instruction to move the
+ * producer index atomically.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj
+ * A pointer to the object to be added.
+ * @return
+ * - 0: Success; objects enqueued.
+ * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
+ * high water mark is exceeded.
+ * - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
+ */
+static inline int __attribute__((always_inline))
+rte_ring_mp_enqueue(struct rte_ring *r, void *obj)
+{
+ return rte_ring_mp_enqueue_bulk(r, &obj, 1);
+}
+
+/**
+ * Enqueue one object on a ring (NOT multi-producers safe).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj
+ * A pointer to the object to be added.
+ * @return
+ * - 0: Success; objects enqueued.
+ * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
+ * high water mark is exceeded.
+ * - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
+ */
+static inline int __attribute__((always_inline))
+rte_ring_sp_enqueue(struct rte_ring *r, void *obj)
+{
+ return rte_ring_sp_enqueue_bulk(r, &obj, 1);
+}
+
+/**
+ * Enqueue one object on a ring.
+ *
+ * This function calls the multi-producer or the single-producer
+ * version, depending on the default behaviour that was specified at
+ * ring creation time (see flags).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj
+ * A pointer to the object to be added.
+ * @return
+ * - 0: Success; objects enqueued.
+ * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
+ * high water mark is exceeded.
+ * - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
+ */
+static inline int __attribute__((always_inline))
+rte_ring_enqueue(struct rte_ring *r, void *obj)
+{
+ if (r->prod.sp_enqueue)
+ return rte_ring_sp_enqueue(r, obj);
+ else
+ return rte_ring_mp_enqueue(r, obj);
+}
+
+/**
+ * Dequeue several objects from a ring (multi-consumers safe).
+ *
+ * This function uses a "compare and set" instruction to move the
+ * consumer index atomically.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects) that will be filled.
+ * @param n
+ * The number of objects to dequeue from the ring to the obj_table.
+ * @return
+ * - 0: Success; objects dequeued.
+ * - -ENOENT: Not enough entries in the ring to dequeue; no object is
+ * dequeued.
+ */
+static inline int __attribute__((always_inline))
+rte_ring_mc_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n)
+{
+ return __rte_ring_mc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
+}
+
+/**
+ * Dequeue several objects from a ring (NOT multi-consumers safe).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects) that will be filled.
+ * @param n
+ * The number of objects to dequeue from the ring to the obj_table,
+ * must be strictly positive.
+ * @return
+ * - 0: Success; objects dequeued.
+ * - -ENOENT: Not enough entries in the ring to dequeue; no object is
+ * dequeued.
+ */
+static inline int __attribute__((always_inline))
+rte_ring_sc_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n)
+{
+ return __rte_ring_sc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
+}
+
+/**
+ * Dequeue several objects from a ring.
+ *
+ * This function calls the multi-consumers or the single-consumer
+ * version, depending on the default behaviour that was specified at
+ * ring creation time (see flags).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects) that will be filled.
+ * @param n
+ * The number of objects to dequeue from the ring to the obj_table.
+ * @return
+ * - 0: Success; objects dequeued.
+ * - -ENOENT: Not enough entries in the ring to dequeue, no object is
+ * dequeued.
+ */
+static inline int __attribute__((always_inline))
+rte_ring_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n)
+{
+ if (r->cons.sc_dequeue)
+ return rte_ring_sc_dequeue_bulk(r, obj_table, n);
+ else
+ return rte_ring_mc_dequeue_bulk(r, obj_table, n);
+}
+
+/**
+ * Dequeue one object from a ring (multi-consumers safe).
+ *
+ * This function uses a "compare and set" instruction to move the
+ * consumer index atomically.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_p
+ * A pointer to a void * pointer (object) that will be filled.
+ * @return
+ * - 0: Success; objects dequeued.
+ * - -ENOENT: Not enough entries in the ring to dequeue; no object is
+ * dequeued.
+ */
+static inline int __attribute__((always_inline))
+rte_ring_mc_dequeue(struct rte_ring *r, void **obj_p)
+{
+ return rte_ring_mc_dequeue_bulk(r, obj_p, 1);
+}
+
+/**
+ * Dequeue one object from a ring (NOT multi-consumers safe).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_p
+ * A pointer to a void * pointer (object) that will be filled.
+ * @return
+ * - 0: Success; objects dequeued.
+ * - -ENOENT: Not enough entries in the ring to dequeue, no object is
+ * dequeued.
+ */
+static inline int __attribute__((always_inline))
+rte_ring_sc_dequeue(struct rte_ring *r, void **obj_p)
+{
+ return rte_ring_sc_dequeue_bulk(r, obj_p, 1);
+}
+
+/**
+ * Dequeue one object from a ring.
+ *
+ * This function calls the multi-consumers or the single-consumer
+ * version depending on the default behaviour that was specified at
+ * ring creation time (see flags).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_p
+ * A pointer to a void * pointer (object) that will be filled.
+ * @return
+ * - 0: Success, objects dequeued.
+ * - -ENOENT: Not enough entries in the ring to dequeue, no object is
+ * dequeued.
+ */
+static inline int __attribute__((always_inline))
+rte_ring_dequeue(struct rte_ring *r, void **obj_p)
+{
+ if (r->cons.sc_dequeue)
+ return rte_ring_sc_dequeue(r, obj_p);
+ else
+ return rte_ring_mc_dequeue(r, obj_p);
+}
+
+/**
+ * Test if a ring is full.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @return
+ * - 1: The ring is full.
+ * - 0: The ring is not full.
+ */
+static inline int
+rte_ring_full(const struct rte_ring *r)
+{
+ uint32_t prod_tail = r->prod.tail;
+ uint32_t cons_tail = r->cons.tail;
+ return ((cons_tail - prod_tail - 1) & r->prod.mask) == 0;
+}
+
+/**
+ * Test if a ring is empty.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @return
+ * - 1: The ring is empty.
+ * - 0: The ring is not empty.
+ */
+static inline int
+rte_ring_empty(const struct rte_ring *r)
+{
+ uint32_t prod_tail = r->prod.tail;
+ uint32_t cons_tail = r->cons.tail;
+ return !!(cons_tail == prod_tail);
+}
+
+/**
+ * Return the number of entries in a ring.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @return
+ * The number of entries in the ring.
+ */
+static inline unsigned
+rte_ring_count(const struct rte_ring *r)
+{
+ uint32_t prod_tail = r->prod.tail;
+ uint32_t cons_tail = r->cons.tail;
+ return (prod_tail - cons_tail) & r->prod.mask;
+}
+
+/**
+ * Return the number of free entries in a ring.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @return
+ * The number of free entries in the ring.
+ */
+static inline unsigned
+rte_ring_free_count(const struct rte_ring *r)
+{
+ uint32_t prod_tail = r->prod.tail;
+ uint32_t cons_tail = r->cons.tail;
+ return (cons_tail - prod_tail - 1) & r->prod.mask;
+}
+
+/**
+ * Dump the status of all rings on the console
+ *
+ * @param f
+ * A pointer to a file for output
+ */
+void rte_ring_list_dump(FILE *f);
+
+/**
+ * Search a ring from its name
+ *
+ * @param name
+ * The name of the ring.
+ * @return
+ * The pointer to the ring matching the name, or NULL if not found,
+ * with rte_errno set appropriately. Possible rte_errno values include:
+ * - ENOENT - required entry not available to return.
+ */
+struct rte_ring *rte_ring_lookup(const char *name);
+
+/**
+ * Enqueue several objects on the ring (multi-producers safe).
+ *
+ * This function uses a "compare and set" instruction to move the
+ * producer index atomically.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects).
+ * @param n
+ * The number of objects to add in the ring from the obj_table.
+ * @return
+ * - n: Actual number of objects enqueued.
+ */
+static inline unsigned __attribute__((always_inline))
+rte_ring_mp_enqueue_burst(struct rte_ring *r, void * const *obj_table,
+ unsigned n)
+{
+ return __rte_ring_mp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
+}
+
+/**
+ * Enqueue several objects on a ring (NOT multi-producers safe).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects).
+ * @param n
+ * The number of objects to add in the ring from the obj_table.
+ * @return
+ * - n: Actual number of objects enqueued.
+ */
+static inline unsigned __attribute__((always_inline))
+rte_ring_sp_enqueue_burst(struct rte_ring *r, void * const *obj_table,
+ unsigned n)
+{
+ return __rte_ring_sp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
+}
+
+/**
+ * Enqueue several objects on a ring.
+ *
+ * This function calls the multi-producer or the single-producer
+ * version depending on the default behavior that was specified at
+ * ring creation time (see flags).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects).
+ * @param n
+ * The number of objects to add in the ring from the obj_table.
+ * @return
+ * - n: Actual number of objects enqueued.
+ */
+static inline unsigned __attribute__((always_inline))
+rte_ring_enqueue_burst(struct rte_ring *r, void * const *obj_table,
+ unsigned n)
+{
+ if (r->prod.sp_enqueue)
+ return rte_ring_sp_enqueue_burst(r, obj_table, n);
+ else
+ return rte_ring_mp_enqueue_burst(r, obj_table, n);
+}
+
+/**
+ * Dequeue several objects from a ring (multi-consumers safe). When the request
+ * objects are more than the available objects, only dequeue the actual number
+ * of objects
+ *
+ * This function uses a "compare and set" instruction to move the
+ * consumer index atomically.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects) that will be filled.
+ * @param n
+ * The number of objects to dequeue from the ring to the obj_table.
+ * @return
+ * - n: Actual number of objects dequeued, 0 if ring is empty
+ */
+static inline unsigned __attribute__((always_inline))
+rte_ring_mc_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n)
+{
+ return __rte_ring_mc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
+}
+
+/**
+ * Dequeue several objects from a ring (NOT multi-consumers safe).When the
+ * request objects are more than the available objects, only dequeue the
+ * actual number of objects
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects) that will be filled.
+ * @param n
+ * The number of objects to dequeue from the ring to the obj_table.
+ * @return
+ * - n: Actual number of objects dequeued, 0 if ring is empty
+ */
+static inline unsigned __attribute__((always_inline))
+rte_ring_sc_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n)
+{
+ return __rte_ring_sc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
+}
+
+/**
+ * Dequeue multiple objects from a ring up to a maximum number.
+ *
+ * This function calls the multi-consumers or the single-consumer
+ * version, depending on the default behaviour that was specified at
+ * ring creation time (see flags).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects) that will be filled.
+ * @param n
+ * The number of objects to dequeue from the ring to the obj_table.
+ * @return
+ * - Number of objects dequeued
+ */
+static inline unsigned __attribute__((always_inline))
+rte_ring_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n)
+{
+ if (r->cons.sc_dequeue)
+ return rte_ring_sc_dequeue_burst(r, obj_table, n);
+ else
+ return rte_ring_mc_dequeue_burst(r, obj_table, n);
+}
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* _RTE_RING_H_ */
diff --git a/lib/librte_ring/rte_ring.h b/lib/librte_ring/rte_ring.h
deleted file mode 100644
index e359aff..0000000
--- a/lib/librte_ring/rte_ring.h
+++ /dev/null
@@ -1,1269 +0,0 @@
-/*-
- * BSD LICENSE
- *
- * Copyright(c) 2010-2014 Intel Corporation. All rights reserved.
- * All rights reserved.
- *
- * Redistribution and use in source and binary forms, with or without
- * modification, are permitted provided that the following conditions
- * are met:
- *
- * * Redistributions of source code must retain the above copyright
- * notice, this list of conditions and the following disclaimer.
- * * 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.
- * * Neither the name of Intel Corporation nor the names of its
- * contributors may be used to endorse or promote products derived
- * from this software without specific prior written permission.
- *
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS 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 COPYRIGHT
- * OWNER 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.
- */
-
-/*
- * Derived from FreeBSD's bufring.h
- *
- **************************************************************************
- *
- * Copyright (c) 2007-2009 Kip Macy kmacy@freebsd.org
- * All rights reserved.
- *
- * 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. The name of Kip Macy nor the names of other
- * contributors may be used to endorse or promote products derived from
- * this software without specific prior written permission.
- *
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS 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 COPYRIGHT OWNER 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.
- *
- ***************************************************************************/
-
-#ifndef _RTE_RING_H_
-#define _RTE_RING_H_
-
-/**
- * @file
- * RTE Ring
- *
- * The Ring Manager is a fixed-size queue, implemented as a table of
- * pointers. Head and tail pointers are modified atomically, allowing
- * concurrent access to it. It has the following features:
- *
- * - FIFO (First In First Out)
- * - Maximum size is fixed; the pointers are stored in a table.
- * - Lockless implementation.
- * - Multi- or single-consumer dequeue.
- * - Multi- or single-producer enqueue.
- * - Bulk dequeue.
- * - Bulk enqueue.
- *
- * Note: the ring implementation is not preemptable. A lcore must not
- * be interrupted by another task that uses the same ring.
- *
- */
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-#include <stdio.h>
-#include <stdint.h>
-#include <sys/queue.h>
-#include <errno.h>
-#include <rte_common.h>
-#include <rte_memory.h>
-#include <rte_lcore.h>
-#include <rte_atomic.h>
-#include <rte_branch_prediction.h>
-#include <rte_memzone.h>
-
-#define RTE_TAILQ_RING_NAME "RTE_RING"
-
-enum rte_ring_queue_behavior {
- RTE_RING_QUEUE_FIXED = 0, /* Enq/Deq a fixed number of items from a ring */
- RTE_RING_QUEUE_VARIABLE /* Enq/Deq as many items as possible from ring */
-};
-
-#ifdef RTE_LIBRTE_RING_DEBUG
-/**
- * A structure that stores the ring statistics (per-lcore).
- */
-struct rte_ring_debug_stats {
- uint64_t enq_success_bulk; /**< Successful enqueues number. */
- uint64_t enq_success_objs; /**< Objects successfully enqueued. */
- uint64_t enq_quota_bulk; /**< Successful enqueues above watermark. */
- uint64_t enq_quota_objs; /**< Objects enqueued above watermark. */
- uint64_t enq_fail_bulk; /**< Failed enqueues number. */
- uint64_t enq_fail_objs; /**< Objects that failed to be enqueued. */
- uint64_t deq_success_bulk; /**< Successful dequeues number. */
- uint64_t deq_success_objs; /**< Objects successfully dequeued. */
- uint64_t deq_fail_bulk; /**< Failed dequeues number. */
- uint64_t deq_fail_objs; /**< Objects that failed to be dequeued. */
-} __rte_cache_aligned;
-#endif
-
-#define RTE_RING_MZ_PREFIX "RG_"
-/**< The maximum length of a ring name. */
-#define RTE_RING_NAMESIZE (RTE_MEMZONE_NAMESIZE - \
- sizeof(RTE_RING_MZ_PREFIX) + 1)
-
-#ifndef RTE_RING_PAUSE_REP_COUNT
-#define RTE_RING_PAUSE_REP_COUNT 0 /**< Yield after pause num of times, no yield
- * if RTE_RING_PAUSE_REP not defined. */
-#endif
-
-struct rte_memzone; /* forward declaration, so as not to require memzone.h */
-
-/**
- * An RTE ring structure.
- *
- * The producer and the consumer have a head and a tail index. The particularity
- * of these index is that they are not between 0 and size(ring). These indexes
- * are between 0 and 2^32, and we mask their value when we access the ring[]
- * field. Thanks to this assumption, we can do subtractions between 2 index
- * values in a modulo-32bit base: that's why the overflow of the indexes is not
- * a problem.
- */
-struct rte_ring {
- /*
- * Note: this field kept the RTE_MEMZONE_NAMESIZE size due to ABI
- * compatibility requirements, it could be changed to RTE_RING_NAMESIZE
- * next time the ABI changes
- */
- char name[RTE_MEMZONE_NAMESIZE]; /**< Name of the ring. */
- int flags; /**< Flags supplied at creation. */
- const struct rte_memzone *memzone;
- /**< Memzone, if any, containing the rte_ring */
-
- /** Ring producer status. */
- struct prod {
- uint32_t watermark; /**< Maximum items before EDQUOT. */
- uint32_t sp_enqueue; /**< True, if single producer. */
- uint32_t size; /**< Size of ring. */
- uint32_t mask; /**< Mask (size-1) of ring. */
- volatile uint32_t head; /**< Producer head. */
- volatile uint32_t tail; /**< Producer tail. */
- } prod __rte_cache_aligned;
-
- /** Ring consumer status. */
- struct cons {
- uint32_t sc_dequeue; /**< True, if single consumer. */
- uint32_t size; /**< Size of the ring. */
- uint32_t mask; /**< Mask (size-1) of ring. */
- volatile uint32_t head; /**< Consumer head. */
- volatile uint32_t tail; /**< Consumer tail. */
-#ifdef RTE_RING_SPLIT_PROD_CONS
- } cons __rte_cache_aligned;
-#else
- } cons;
-#endif
-
-#ifdef RTE_LIBRTE_RING_DEBUG
- struct rte_ring_debug_stats stats[RTE_MAX_LCORE];
-#endif
-
- void *ring[] __rte_cache_aligned; /**< Memory space of ring starts here.
- * not volatile so need to be careful
- * about compiler re-ordering */
-};
-
-#define RING_F_SP_ENQ 0x0001 /**< The default enqueue is "single-producer". */
-#define RING_F_SC_DEQ 0x0002 /**< The default dequeue is "single-consumer". */
-#define RTE_RING_QUOT_EXCEED (1 << 31) /**< Quota exceed for burst ops */
-#define RTE_RING_SZ_MASK (unsigned)(0x0fffffff) /**< Ring size mask */
-
-/**
- * @internal When debug is enabled, store ring statistics.
- * @param r
- * A pointer to the ring.
- * @param name
- * The name of the statistics field to increment in the ring.
- * @param n
- * The number to add to the object-oriented statistics.
- */
-#ifdef RTE_LIBRTE_RING_DEBUG
-#define __RING_STAT_ADD(r, name, n) do { \
- unsigned __lcore_id = rte_lcore_id(); \
- if (__lcore_id < RTE_MAX_LCORE) { \
- r->stats[__lcore_id].name##_objs += n; \
- r->stats[__lcore_id].name##_bulk += 1; \
- } \
- } while(0)
-#else
-#define __RING_STAT_ADD(r, name, n) do {} while(0)
-#endif
-
-/**
- * Calculate the memory size needed for a ring
- *
- * This function returns the number of bytes needed for a ring, given
- * the number of elements in it. This value is the sum of the size of
- * the structure rte_ring and the size of the memory needed by the
- * objects pointers. The value is aligned to a cache line size.
- *
- * @param count
- * The number of elements in the ring (must be a power of 2).
- * @return
- * - The memory size needed for the ring on success.
- * - -EINVAL if count is not a power of 2.
- */
-ssize_t rte_ring_get_memsize(unsigned count);
-
-/**
- * Initialize a ring structure.
- *
- * Initialize a ring structure in memory pointed by "r". The size of the
- * memory area must be large enough to store the ring structure and the
- * object table. It is advised to use rte_ring_get_memsize() to get the
- * appropriate size.
- *
- * The ring size is set to *count*, which must be a power of two. Water
- * marking is disabled by default. The real usable ring size is
- * *count-1* instead of *count* to differentiate a free ring from an
- * empty ring.
- *
- * The ring is not added in RTE_TAILQ_RING global list. Indeed, the
- * memory given by the caller may not be shareable among dpdk
- * processes.
- *
- * @param r
- * The pointer to the ring structure followed by the objects table.
- * @param name
- * The name of the ring.
- * @param count
- * The number of elements in the ring (must be a power of 2).
- * @param flags
- * An OR of the following:
- * - RING_F_SP_ENQ: If this flag is set, the default behavior when
- * using ``rte_ring_enqueue()`` or ``rte_ring_enqueue_bulk()``
- * is "single-producer". Otherwise, it is "multi-producers".
- * - RING_F_SC_DEQ: If this flag is set, the default behavior when
- * using ``rte_ring_dequeue()`` or ``rte_ring_dequeue_bulk()``
- * is "single-consumer". Otherwise, it is "multi-consumers".
- * @return
- * 0 on success, or a negative value on error.
- */
-int rte_ring_init(struct rte_ring *r, const char *name, unsigned count,
- unsigned flags);
-
-/**
- * Create a new ring named *name* in memory.
- *
- * This function uses ``memzone_reserve()`` to allocate memory. Then it
- * calls rte_ring_init() to initialize an empty ring.
- *
- * The new ring size is set to *count*, which must be a power of
- * two. Water marking is disabled by default. The real usable ring size
- * is *count-1* instead of *count* to differentiate a free ring from an
- * empty ring.
- *
- * The ring is added in RTE_TAILQ_RING list.
- *
- * @param name
- * The name of the ring.
- * @param count
- * The size of the ring (must be a power of 2).
- * @param socket_id
- * The *socket_id* argument is the socket identifier in case of
- * NUMA. The value can be *SOCKET_ID_ANY* if there is no NUMA
- * constraint for the reserved zone.
- * @param flags
- * An OR of the following:
- * - RING_F_SP_ENQ: If this flag is set, the default behavior when
- * using ``rte_ring_enqueue()`` or ``rte_ring_enqueue_bulk()``
- * is "single-producer". Otherwise, it is "multi-producers".
- * - RING_F_SC_DEQ: If this flag is set, the default behavior when
- * using ``rte_ring_dequeue()`` or ``rte_ring_dequeue_bulk()``
- * is "single-consumer". Otherwise, it is "multi-consumers".
- * @return
- * On success, the pointer to the new allocated ring. NULL on error with
- * rte_errno set appropriately. Possible errno values include:
- * - E_RTE_NO_CONFIG - function could not get pointer to rte_config structure
- * - E_RTE_SECONDARY - function was called from a secondary process instance
- * - EINVAL - count provided is not a power of 2
- * - ENOSPC - the maximum number of memzones has already been allocated
- * - EEXIST - a memzone with the same name already exists
- * - ENOMEM - no appropriate memory area found in which to create memzone
- */
-struct rte_ring *rte_ring_create(const char *name, unsigned count,
- int socket_id, unsigned flags);
-/**
- * De-allocate all memory used by the ring.
- *
- * @param r
- * Ring to free
- */
-void rte_ring_free(struct rte_ring *r);
-
-/**
- * Change the high water mark.
- *
- * If *count* is 0, water marking is disabled. Otherwise, it is set to the
- * *count* value. The *count* value must be greater than 0 and less
- * than the ring size.
- *
- * This function can be called at any time (not necessarily at
- * initialization).
- *
- * @param r
- * A pointer to the ring structure.
- * @param count
- * The new water mark value.
- * @return
- * - 0: Success; water mark changed.
- * - -EINVAL: Invalid water mark value.
- */
-int rte_ring_set_water_mark(struct rte_ring *r, unsigned count);
-
-/**
- * Dump the status of the ring to a file.
- *
- * @param f
- * A pointer to a file for output
- * @param r
- * A pointer to the ring structure.
- */
-void rte_ring_dump(FILE *f, const struct rte_ring *r);
-
-/* the actual enqueue of pointers on the ring.
- * Placed here since identical code needed in both
- * single and multi producer enqueue functions */
-#define ENQUEUE_PTRS() do { \
- const uint32_t size = r->prod.size; \
- uint32_t idx = prod_head & mask; \
- if (likely(idx + n < size)) { \
- for (i = 0; i < (n & ((~(unsigned)0x3))); i+=4, idx+=4) { \
- r->ring[idx] = obj_table[i]; \
- r->ring[idx+1] = obj_table[i+1]; \
- r->ring[idx+2] = obj_table[i+2]; \
- r->ring[idx+3] = obj_table[i+3]; \
- } \
- switch (n & 0x3) { \
- case 3: r->ring[idx++] = obj_table[i++]; \
- case 2: r->ring[idx++] = obj_table[i++]; \
- case 1: r->ring[idx++] = obj_table[i++]; \
- } \
- } else { \
- for (i = 0; idx < size; i++, idx++)\
- r->ring[idx] = obj_table[i]; \
- for (idx = 0; i < n; i++, idx++) \
- r->ring[idx] = obj_table[i]; \
- } \
-} while(0)
-
-/* the actual copy of pointers on the ring to obj_table.
- * Placed here since identical code needed in both
- * single and multi consumer dequeue functions */
-#define DEQUEUE_PTRS() do { \
- uint32_t idx = cons_head & mask; \
- const uint32_t size = r->cons.size; \
- if (likely(idx + n < size)) { \
- for (i = 0; i < (n & (~(unsigned)0x3)); i+=4, idx+=4) {\
- obj_table[i] = r->ring[idx]; \
- obj_table[i+1] = r->ring[idx+1]; \
- obj_table[i+2] = r->ring[idx+2]; \
- obj_table[i+3] = r->ring[idx+3]; \
- } \
- switch (n & 0x3) { \
- case 3: obj_table[i++] = r->ring[idx++]; \
- case 2: obj_table[i++] = r->ring[idx++]; \
- case 1: obj_table[i++] = r->ring[idx++]; \
- } \
- } else { \
- for (i = 0; idx < size; i++, idx++) \
- obj_table[i] = r->ring[idx]; \
- for (idx = 0; i < n; i++, idx++) \
- obj_table[i] = r->ring[idx]; \
- } \
-} while (0)
-
-/**
- * @internal Enqueue several objects on the ring (multi-producers safe).
- *
- * This function uses a "compare and set" instruction to move the
- * producer index atomically.
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_table
- * A pointer to a table of void * pointers (objects).
- * @param n
- * The number of objects to add in the ring from the obj_table.
- * @param behavior
- * RTE_RING_QUEUE_FIXED: Enqueue a fixed number of items from a ring
- * RTE_RING_QUEUE_VARIABLE: Enqueue as many items a possible from ring
- * @return
- * Depend on the behavior value
- * if behavior = RTE_RING_QUEUE_FIXED
- * - 0: Success; objects enqueue.
- * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
- * high water mark is exceeded.
- * - -ENOBUFS: Not enough room in the ring to enqueue, no object is enqueued.
- * if behavior = RTE_RING_QUEUE_VARIABLE
- * - n: Actual number of objects enqueued.
- */
-static inline int __attribute__((always_inline))
-__rte_ring_mp_do_enqueue(struct rte_ring *r, void * const *obj_table,
- unsigned n, enum rte_ring_queue_behavior behavior)
-{
- uint32_t prod_head, prod_next;
- uint32_t cons_tail, free_entries;
- const unsigned max = n;
- int success;
- unsigned i, rep = 0;
- uint32_t mask = r->prod.mask;
- int ret;
-
- /* Avoid the unnecessary cmpset operation below, which is also
- * potentially harmful when n equals 0. */
- if (n == 0)
- return 0;
-
- /* move prod.head atomically */
- do {
- /* Reset n to the initial burst count */
- n = max;
-
- prod_head = r->prod.head;
- cons_tail = r->cons.tail;
- /* The subtraction is done between two unsigned 32bits value
- * (the result is always modulo 32 bits even if we have
- * prod_head > cons_tail). So 'free_entries' is always between 0
- * and size(ring)-1. */
- free_entries = (mask + cons_tail - prod_head);
-
- /* check that we have enough room in ring */
- if (unlikely(n > free_entries)) {
- if (behavior == RTE_RING_QUEUE_FIXED) {
- __RING_STAT_ADD(r, enq_fail, n);
- return -ENOBUFS;
- }
- else {
- /* No free entry available */
- if (unlikely(free_entries == 0)) {
- __RING_STAT_ADD(r, enq_fail, n);
- return 0;
- }
-
- n = free_entries;
- }
- }
-
- prod_next = prod_head + n;
- success = rte_atomic32_cmpset(&r->prod.head, prod_head,
- prod_next);
- } while (unlikely(success == 0));
-
- /* write entries in ring */
- ENQUEUE_PTRS();
- rte_smp_wmb();
-
- /* if we exceed the watermark */
- if (unlikely(((mask + 1) - free_entries + n) > r->prod.watermark)) {
- ret = (behavior == RTE_RING_QUEUE_FIXED) ? -EDQUOT :
- (int)(n | RTE_RING_QUOT_EXCEED);
- __RING_STAT_ADD(r, enq_quota, n);
- }
- else {
- ret = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : n;
- __RING_STAT_ADD(r, enq_success, n);
- }
-
- /*
- * If there are other enqueues in progress that preceded us,
- * we need to wait for them to complete
- */
- while (unlikely(r->prod.tail != prod_head)) {
- rte_pause();
-
- /* Set RTE_RING_PAUSE_REP_COUNT to avoid spin too long waiting
- * for other thread finish. It gives pre-empted thread a chance
- * to proceed and finish with ring dequeue operation. */
- if (RTE_RING_PAUSE_REP_COUNT &&
- ++rep == RTE_RING_PAUSE_REP_COUNT) {
- rep = 0;
- sched_yield();
- }
- }
- r->prod.tail = prod_next;
- return ret;
-}
-
-/**
- * @internal Enqueue several objects on a ring (NOT multi-producers safe).
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_table
- * A pointer to a table of void * pointers (objects).
- * @param n
- * The number of objects to add in the ring from the obj_table.
- * @param behavior
- * RTE_RING_QUEUE_FIXED: Enqueue a fixed number of items from a ring
- * RTE_RING_QUEUE_VARIABLE: Enqueue as many items a possible from ring
- * @return
- * Depend on the behavior value
- * if behavior = RTE_RING_QUEUE_FIXED
- * - 0: Success; objects enqueue.
- * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
- * high water mark is exceeded.
- * - -ENOBUFS: Not enough room in the ring to enqueue, no object is enqueued.
- * if behavior = RTE_RING_QUEUE_VARIABLE
- * - n: Actual number of objects enqueued.
- */
-static inline int __attribute__((always_inline))
-__rte_ring_sp_do_enqueue(struct rte_ring *r, void * const *obj_table,
- unsigned n, enum rte_ring_queue_behavior behavior)
-{
- uint32_t prod_head, cons_tail;
- uint32_t prod_next, free_entries;
- unsigned i;
- uint32_t mask = r->prod.mask;
- int ret;
-
- prod_head = r->prod.head;
- cons_tail = r->cons.tail;
- /* The subtraction is done between two unsigned 32bits value
- * (the result is always modulo 32 bits even if we have
- * prod_head > cons_tail). So 'free_entries' is always between 0
- * and size(ring)-1. */
- free_entries = mask + cons_tail - prod_head;
-
- /* check that we have enough room in ring */
- if (unlikely(n > free_entries)) {
- if (behavior == RTE_RING_QUEUE_FIXED) {
- __RING_STAT_ADD(r, enq_fail, n);
- return -ENOBUFS;
- }
- else {
- /* No free entry available */
- if (unlikely(free_entries == 0)) {
- __RING_STAT_ADD(r, enq_fail, n);
- return 0;
- }
-
- n = free_entries;
- }
- }
-
- prod_next = prod_head + n;
- r->prod.head = prod_next;
-
- /* write entries in ring */
- ENQUEUE_PTRS();
- rte_smp_wmb();
-
- /* if we exceed the watermark */
- if (unlikely(((mask + 1) - free_entries + n) > r->prod.watermark)) {
- ret = (behavior == RTE_RING_QUEUE_FIXED) ? -EDQUOT :
- (int)(n | RTE_RING_QUOT_EXCEED);
- __RING_STAT_ADD(r, enq_quota, n);
- }
- else {
- ret = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : n;
- __RING_STAT_ADD(r, enq_success, n);
- }
-
- r->prod.tail = prod_next;
- return ret;
-}
-
-/**
- * @internal Dequeue several objects from a ring (multi-consumers safe). When
- * the request objects are more than the available objects, only dequeue the
- * actual number of objects
- *
- * This function uses a "compare and set" instruction to move the
- * consumer index atomically.
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_table
- * A pointer to a table of void * pointers (objects) that will be filled.
- * @param n
- * The number of objects to dequeue from the ring to the obj_table.
- * @param behavior
- * RTE_RING_QUEUE_FIXED: Dequeue a fixed number of items from a ring
- * RTE_RING_QUEUE_VARIABLE: Dequeue as many items a possible from ring
- * @return
- * Depend on the behavior value
- * if behavior = RTE_RING_QUEUE_FIXED
- * - 0: Success; objects dequeued.
- * - -ENOENT: Not enough entries in the ring to dequeue; no object is
- * dequeued.
- * if behavior = RTE_RING_QUEUE_VARIABLE
- * - n: Actual number of objects dequeued.
- */
-
-static inline int __attribute__((always_inline))
-__rte_ring_mc_do_dequeue(struct rte_ring *r, void **obj_table,
- unsigned n, enum rte_ring_queue_behavior behavior)
-{
- uint32_t cons_head, prod_tail;
- uint32_t cons_next, entries;
- const unsigned max = n;
- int success;
- unsigned i, rep = 0;
- uint32_t mask = r->prod.mask;
-
- /* Avoid the unnecessary cmpset operation below, which is also
- * potentially harmful when n equals 0. */
- if (n == 0)
- return 0;
-
- /* move cons.head atomically */
- do {
- /* Restore n as it may change every loop */
- n = max;
-
- cons_head = r->cons.head;
- prod_tail = r->prod.tail;
- /* The subtraction is done between two unsigned 32bits value
- * (the result is always modulo 32 bits even if we have
- * cons_head > prod_tail). So 'entries' is always between 0
- * and size(ring)-1. */
- entries = (prod_tail - cons_head);
-
- /* Set the actual entries for dequeue */
- if (n > entries) {
- if (behavior == RTE_RING_QUEUE_FIXED) {
- __RING_STAT_ADD(r, deq_fail, n);
- return -ENOENT;
- }
- else {
- if (unlikely(entries == 0)){
- __RING_STAT_ADD(r, deq_fail, n);
- return 0;
- }
-
- n = entries;
- }
- }
-
- cons_next = cons_head + n;
- success = rte_atomic32_cmpset(&r->cons.head, cons_head,
- cons_next);
- } while (unlikely(success == 0));
-
- /* copy in table */
- DEQUEUE_PTRS();
- rte_smp_rmb();
-
- /*
- * If there are other dequeues in progress that preceded us,
- * we need to wait for them to complete
- */
- while (unlikely(r->cons.tail != cons_head)) {
- rte_pause();
-
- /* Set RTE_RING_PAUSE_REP_COUNT to avoid spin too long waiting
- * for other thread finish. It gives pre-empted thread a chance
- * to proceed and finish with ring dequeue operation. */
- if (RTE_RING_PAUSE_REP_COUNT &&
- ++rep == RTE_RING_PAUSE_REP_COUNT) {
- rep = 0;
- sched_yield();
- }
- }
- __RING_STAT_ADD(r, deq_success, n);
- r->cons.tail = cons_next;
-
- return behavior == RTE_RING_QUEUE_FIXED ? 0 : n;
-}
-
-/**
- * @internal Dequeue several objects from a ring (NOT multi-consumers safe).
- * When the request objects are more than the available objects, only dequeue
- * the actual number of objects
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_table
- * A pointer to a table of void * pointers (objects) that will be filled.
- * @param n
- * The number of objects to dequeue from the ring to the obj_table.
- * @param behavior
- * RTE_RING_QUEUE_FIXED: Dequeue a fixed number of items from a ring
- * RTE_RING_QUEUE_VARIABLE: Dequeue as many items a possible from ring
- * @return
- * Depend on the behavior value
- * if behavior = RTE_RING_QUEUE_FIXED
- * - 0: Success; objects dequeued.
- * - -ENOENT: Not enough entries in the ring to dequeue; no object is
- * dequeued.
- * if behavior = RTE_RING_QUEUE_VARIABLE
- * - n: Actual number of objects dequeued.
- */
-static inline int __attribute__((always_inline))
-__rte_ring_sc_do_dequeue(struct rte_ring *r, void **obj_table,
- unsigned n, enum rte_ring_queue_behavior behavior)
-{
- uint32_t cons_head, prod_tail;
- uint32_t cons_next, entries;
- unsigned i;
- uint32_t mask = r->prod.mask;
-
- cons_head = r->cons.head;
- prod_tail = r->prod.tail;
- /* The subtraction is done between two unsigned 32bits value
- * (the result is always modulo 32 bits even if we have
- * cons_head > prod_tail). So 'entries' is always between 0
- * and size(ring)-1. */
- entries = prod_tail - cons_head;
-
- if (n > entries) {
- if (behavior == RTE_RING_QUEUE_FIXED) {
- __RING_STAT_ADD(r, deq_fail, n);
- return -ENOENT;
- }
- else {
- if (unlikely(entries == 0)){
- __RING_STAT_ADD(r, deq_fail, n);
- return 0;
- }
-
- n = entries;
- }
- }
-
- cons_next = cons_head + n;
- r->cons.head = cons_next;
-
- /* copy in table */
- DEQUEUE_PTRS();
- rte_smp_rmb();
-
- __RING_STAT_ADD(r, deq_success, n);
- r->cons.tail = cons_next;
- return behavior == RTE_RING_QUEUE_FIXED ? 0 : n;
-}
-
-/**
- * Enqueue several objects on the ring (multi-producers safe).
- *
- * This function uses a "compare and set" instruction to move the
- * producer index atomically.
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_table
- * A pointer to a table of void * pointers (objects).
- * @param n
- * The number of objects to add in the ring from the obj_table.
- * @return
- * - 0: Success; objects enqueue.
- * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
- * high water mark is exceeded.
- * - -ENOBUFS: Not enough room in the ring to enqueue, no object is enqueued.
- */
-static inline int __attribute__((always_inline))
-rte_ring_mp_enqueue_bulk(struct rte_ring *r, void * const *obj_table,
- unsigned n)
-{
- return __rte_ring_mp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
-}
-
-/**
- * Enqueue several objects on a ring (NOT multi-producers safe).
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_table
- * A pointer to a table of void * pointers (objects).
- * @param n
- * The number of objects to add in the ring from the obj_table.
- * @return
- * - 0: Success; objects enqueued.
- * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
- * high water mark is exceeded.
- * - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
- */
-static inline int __attribute__((always_inline))
-rte_ring_sp_enqueue_bulk(struct rte_ring *r, void * const *obj_table,
- unsigned n)
-{
- return __rte_ring_sp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
-}
-
-/**
- * Enqueue several objects on a ring.
- *
- * This function calls the multi-producer or the single-producer
- * version depending on the default behavior that was specified at
- * ring creation time (see flags).
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_table
- * A pointer to a table of void * pointers (objects).
- * @param n
- * The number of objects to add in the ring from the obj_table.
- * @return
- * - 0: Success; objects enqueued.
- * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
- * high water mark is exceeded.
- * - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
- */
-static inline int __attribute__((always_inline))
-rte_ring_enqueue_bulk(struct rte_ring *r, void * const *obj_table,
- unsigned n)
-{
- if (r->prod.sp_enqueue)
- return rte_ring_sp_enqueue_bulk(r, obj_table, n);
- else
- return rte_ring_mp_enqueue_bulk(r, obj_table, n);
-}
-
-/**
- * Enqueue one object on a ring (multi-producers safe).
- *
- * This function uses a "compare and set" instruction to move the
- * producer index atomically.
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj
- * A pointer to the object to be added.
- * @return
- * - 0: Success; objects enqueued.
- * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
- * high water mark is exceeded.
- * - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
- */
-static inline int __attribute__((always_inline))
-rte_ring_mp_enqueue(struct rte_ring *r, void *obj)
-{
- return rte_ring_mp_enqueue_bulk(r, &obj, 1);
-}
-
-/**
- * Enqueue one object on a ring (NOT multi-producers safe).
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj
- * A pointer to the object to be added.
- * @return
- * - 0: Success; objects enqueued.
- * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
- * high water mark is exceeded.
- * - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
- */
-static inline int __attribute__((always_inline))
-rte_ring_sp_enqueue(struct rte_ring *r, void *obj)
-{
- return rte_ring_sp_enqueue_bulk(r, &obj, 1);
-}
-
-/**
- * Enqueue one object on a ring.
- *
- * This function calls the multi-producer or the single-producer
- * version, depending on the default behaviour that was specified at
- * ring creation time (see flags).
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj
- * A pointer to the object to be added.
- * @return
- * - 0: Success; objects enqueued.
- * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
- * high water mark is exceeded.
- * - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
- */
-static inline int __attribute__((always_inline))
-rte_ring_enqueue(struct rte_ring *r, void *obj)
-{
- if (r->prod.sp_enqueue)
- return rte_ring_sp_enqueue(r, obj);
- else
- return rte_ring_mp_enqueue(r, obj);
-}
-
-/**
- * Dequeue several objects from a ring (multi-consumers safe).
- *
- * This function uses a "compare and set" instruction to move the
- * consumer index atomically.
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_table
- * A pointer to a table of void * pointers (objects) that will be filled.
- * @param n
- * The number of objects to dequeue from the ring to the obj_table.
- * @return
- * - 0: Success; objects dequeued.
- * - -ENOENT: Not enough entries in the ring to dequeue; no object is
- * dequeued.
- */
-static inline int __attribute__((always_inline))
-rte_ring_mc_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n)
-{
- return __rte_ring_mc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
-}
-
-/**
- * Dequeue several objects from a ring (NOT multi-consumers safe).
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_table
- * A pointer to a table of void * pointers (objects) that will be filled.
- * @param n
- * The number of objects to dequeue from the ring to the obj_table,
- * must be strictly positive.
- * @return
- * - 0: Success; objects dequeued.
- * - -ENOENT: Not enough entries in the ring to dequeue; no object is
- * dequeued.
- */
-static inline int __attribute__((always_inline))
-rte_ring_sc_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n)
-{
- return __rte_ring_sc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
-}
-
-/**
- * Dequeue several objects from a ring.
- *
- * This function calls the multi-consumers or the single-consumer
- * version, depending on the default behaviour that was specified at
- * ring creation time (see flags).
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_table
- * A pointer to a table of void * pointers (objects) that will be filled.
- * @param n
- * The number of objects to dequeue from the ring to the obj_table.
- * @return
- * - 0: Success; objects dequeued.
- * - -ENOENT: Not enough entries in the ring to dequeue, no object is
- * dequeued.
- */
-static inline int __attribute__((always_inline))
-rte_ring_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n)
-{
- if (r->cons.sc_dequeue)
- return rte_ring_sc_dequeue_bulk(r, obj_table, n);
- else
- return rte_ring_mc_dequeue_bulk(r, obj_table, n);
-}
-
-/**
- * Dequeue one object from a ring (multi-consumers safe).
- *
- * This function uses a "compare and set" instruction to move the
- * consumer index atomically.
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_p
- * A pointer to a void * pointer (object) that will be filled.
- * @return
- * - 0: Success; objects dequeued.
- * - -ENOENT: Not enough entries in the ring to dequeue; no object is
- * dequeued.
- */
-static inline int __attribute__((always_inline))
-rte_ring_mc_dequeue(struct rte_ring *r, void **obj_p)
-{
- return rte_ring_mc_dequeue_bulk(r, obj_p, 1);
-}
-
-/**
- * Dequeue one object from a ring (NOT multi-consumers safe).
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_p
- * A pointer to a void * pointer (object) that will be filled.
- * @return
- * - 0: Success; objects dequeued.
- * - -ENOENT: Not enough entries in the ring to dequeue, no object is
- * dequeued.
- */
-static inline int __attribute__((always_inline))
-rte_ring_sc_dequeue(struct rte_ring *r, void **obj_p)
-{
- return rte_ring_sc_dequeue_bulk(r, obj_p, 1);
-}
-
-/**
- * Dequeue one object from a ring.
- *
- * This function calls the multi-consumers or the single-consumer
- * version depending on the default behaviour that was specified at
- * ring creation time (see flags).
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_p
- * A pointer to a void * pointer (object) that will be filled.
- * @return
- * - 0: Success, objects dequeued.
- * - -ENOENT: Not enough entries in the ring to dequeue, no object is
- * dequeued.
- */
-static inline int __attribute__((always_inline))
-rte_ring_dequeue(struct rte_ring *r, void **obj_p)
-{
- if (r->cons.sc_dequeue)
- return rte_ring_sc_dequeue(r, obj_p);
- else
- return rte_ring_mc_dequeue(r, obj_p);
-}
-
-/**
- * Test if a ring is full.
- *
- * @param r
- * A pointer to the ring structure.
- * @return
- * - 1: The ring is full.
- * - 0: The ring is not full.
- */
-static inline int
-rte_ring_full(const struct rte_ring *r)
-{
- uint32_t prod_tail = r->prod.tail;
- uint32_t cons_tail = r->cons.tail;
- return ((cons_tail - prod_tail - 1) & r->prod.mask) == 0;
-}
-
-/**
- * Test if a ring is empty.
- *
- * @param r
- * A pointer to the ring structure.
- * @return
- * - 1: The ring is empty.
- * - 0: The ring is not empty.
- */
-static inline int
-rte_ring_empty(const struct rte_ring *r)
-{
- uint32_t prod_tail = r->prod.tail;
- uint32_t cons_tail = r->cons.tail;
- return !!(cons_tail == prod_tail);
-}
-
-/**
- * Return the number of entries in a ring.
- *
- * @param r
- * A pointer to the ring structure.
- * @return
- * The number of entries in the ring.
- */
-static inline unsigned
-rte_ring_count(const struct rte_ring *r)
-{
- uint32_t prod_tail = r->prod.tail;
- uint32_t cons_tail = r->cons.tail;
- return (prod_tail - cons_tail) & r->prod.mask;
-}
-
-/**
- * Return the number of free entries in a ring.
- *
- * @param r
- * A pointer to the ring structure.
- * @return
- * The number of free entries in the ring.
- */
-static inline unsigned
-rte_ring_free_count(const struct rte_ring *r)
-{
- uint32_t prod_tail = r->prod.tail;
- uint32_t cons_tail = r->cons.tail;
- return (cons_tail - prod_tail - 1) & r->prod.mask;
-}
-
-/**
- * Dump the status of all rings on the console
- *
- * @param f
- * A pointer to a file for output
- */
-void rte_ring_list_dump(FILE *f);
-
-/**
- * Search a ring from its name
- *
- * @param name
- * The name of the ring.
- * @return
- * The pointer to the ring matching the name, or NULL if not found,
- * with rte_errno set appropriately. Possible rte_errno values include:
- * - ENOENT - required entry not available to return.
- */
-struct rte_ring *rte_ring_lookup(const char *name);
-
-/**
- * Enqueue several objects on the ring (multi-producers safe).
- *
- * This function uses a "compare and set" instruction to move the
- * producer index atomically.
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_table
- * A pointer to a table of void * pointers (objects).
- * @param n
- * The number of objects to add in the ring from the obj_table.
- * @return
- * - n: Actual number of objects enqueued.
- */
-static inline unsigned __attribute__((always_inline))
-rte_ring_mp_enqueue_burst(struct rte_ring *r, void * const *obj_table,
- unsigned n)
-{
- return __rte_ring_mp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
-}
-
-/**
- * Enqueue several objects on a ring (NOT multi-producers safe).
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_table
- * A pointer to a table of void * pointers (objects).
- * @param n
- * The number of objects to add in the ring from the obj_table.
- * @return
- * - n: Actual number of objects enqueued.
- */
-static inline unsigned __attribute__((always_inline))
-rte_ring_sp_enqueue_burst(struct rte_ring *r, void * const *obj_table,
- unsigned n)
-{
- return __rte_ring_sp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
-}
-
-/**
- * Enqueue several objects on a ring.
- *
- * This function calls the multi-producer or the single-producer
- * version depending on the default behavior that was specified at
- * ring creation time (see flags).
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_table
- * A pointer to a table of void * pointers (objects).
- * @param n
- * The number of objects to add in the ring from the obj_table.
- * @return
- * - n: Actual number of objects enqueued.
- */
-static inline unsigned __attribute__((always_inline))
-rte_ring_enqueue_burst(struct rte_ring *r, void * const *obj_table,
- unsigned n)
-{
- if (r->prod.sp_enqueue)
- return rte_ring_sp_enqueue_burst(r, obj_table, n);
- else
- return rte_ring_mp_enqueue_burst(r, obj_table, n);
-}
-
-/**
- * Dequeue several objects from a ring (multi-consumers safe). When the request
- * objects are more than the available objects, only dequeue the actual number
- * of objects
- *
- * This function uses a "compare and set" instruction to move the
- * consumer index atomically.
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_table
- * A pointer to a table of void * pointers (objects) that will be filled.
- * @param n
- * The number of objects to dequeue from the ring to the obj_table.
- * @return
- * - n: Actual number of objects dequeued, 0 if ring is empty
- */
-static inline unsigned __attribute__((always_inline))
-rte_ring_mc_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n)
-{
- return __rte_ring_mc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
-}
-
-/**
- * Dequeue several objects from a ring (NOT multi-consumers safe).When the
- * request objects are more than the available objects, only dequeue the
- * actual number of objects
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_table
- * A pointer to a table of void * pointers (objects) that will be filled.
- * @param n
- * The number of objects to dequeue from the ring to the obj_table.
- * @return
- * - n: Actual number of objects dequeued, 0 if ring is empty
- */
-static inline unsigned __attribute__((always_inline))
-rte_ring_sc_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n)
-{
- return __rte_ring_sc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
-}
-
-/**
- * Dequeue multiple objects from a ring up to a maximum number.
- *
- * This function calls the multi-consumers or the single-consumer
- * version, depending on the default behaviour that was specified at
- * ring creation time (see flags).
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_table
- * A pointer to a table of void * pointers (objects) that will be filled.
- * @param n
- * The number of objects to dequeue from the ring to the obj_table.
- * @return
- * - Number of objects dequeued
- */
-static inline unsigned __attribute__((always_inline))
-rte_ring_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n)
-{
- if (r->cons.sc_dequeue)
- return rte_ring_sc_dequeue_burst(r, obj_table, n);
- else
- return rte_ring_mc_dequeue_burst(r, obj_table, n);
-}
-
-#ifdef __cplusplus
-}
-#endif
-
-#endif /* _RTE_RING_H_ */
diff --git a/lib/librte_ring/rte_ring.h b/lib/librte_ring/rte_ring.h
new file mode 120000
index 0000000..54dad23
--- /dev/null
+++ b/lib/librte_ring/rte_ring.h
@@ -0,0 +1 @@
+rte_common_ring.h
\ No newline at end of file
--
2.9.3
^ permalink raw reply [flat|nested] 5+ messages in thread
* [dpdk-dev] [PATCH RFCv2 2/4] ring: separate common and rte_ring specific functions
2017-01-24 10:39 [dpdk-dev] [PATCH RFCv2 0/4] generalise rte_ring to allow different datatypes Bruce Richardson
2017-01-24 10:39 ` [dpdk-dev] [PATCH RFCv2 1/4] ring: create common ring files Bruce Richardson
@ 2017-01-24 10:39 ` Bruce Richardson
2017-01-24 10:39 ` [dpdk-dev] [PATCH RFCv2 3/4] ring: allow common ring to use 8 or 16 byte values Bruce Richardson
2017-01-24 10:39 ` [dpdk-dev] [PATCH RFCv2 4/4] ring: add new event ring class Bruce Richardson
3 siblings, 0 replies; 5+ messages in thread
From: Bruce Richardson @ 2017-01-24 10:39 UTC (permalink / raw)
To: olivier.matz; +Cc: dev, Bruce Richardson
Provide a separate rte_ring implementation which just calls into the
common ring code. This allows us to generalise the common ring code
without affecting the API/ABI of the rte_ring. The common functions
are now all renamed to have an rte_common_ring prefix.
Signed-off-by: Bruce Richardson <bruce.richardson@intel.com>
---
lib/librte_ring/Makefile | 1 +
lib/librte_ring/rte_common_ring.c | 57 ++--
lib/librte_ring/rte_common_ring.h | 463 ++-----------------------
lib/librte_ring/rte_ring.c | 86 +++++
lib/librte_ring/rte_ring.h | 692 +++++++++++++++++++++++++++++++++++++-
5 files changed, 832 insertions(+), 467 deletions(-)
create mode 100644 lib/librte_ring/rte_ring.c
mode change 120000 => 100644 lib/librte_ring/rte_ring.h
diff --git a/lib/librte_ring/Makefile b/lib/librte_ring/Makefile
index 1e2396e..5cebb29 100644
--- a/lib/librte_ring/Makefile
+++ b/lib/librte_ring/Makefile
@@ -42,6 +42,7 @@ LIBABIVER := 1
# all source are stored in SRCS-y
SRCS-$(CONFIG_RTE_LIBRTE_RING) += rte_common_ring.c
+SRCS-$(CONFIG_RTE_LIBRTE_RING) += rte_ring.c
# install includes
SYMLINK-$(CONFIG_RTE_LIBRTE_RING)-include += rte_ring.h
diff --git a/lib/librte_ring/rte_common_ring.c b/lib/librte_ring/rte_common_ring.c
index ca0a108..a0c4b5a 100644
--- a/lib/librte_ring/rte_common_ring.c
+++ b/lib/librte_ring/rte_common_ring.c
@@ -89,19 +89,19 @@
#include "rte_ring.h"
-TAILQ_HEAD(rte_ring_list, rte_tailq_entry);
+TAILQ_HEAD(rte_common_ring_list, rte_tailq_entry);
-static struct rte_tailq_elem rte_ring_tailq = {
+static struct rte_tailq_elem rte_common_ring_tailq = {
.name = RTE_TAILQ_RING_NAME,
};
-EAL_REGISTER_TAILQ(rte_ring_tailq)
+EAL_REGISTER_TAILQ(rte_common_ring_tailq)
/* true if x is a power of 2 */
#define POWEROF2(x) ((((x)-1) & (x)) == 0)
/* return the size of memory occupied by a ring */
ssize_t
-rte_ring_get_memsize(unsigned count)
+rte_common_ring_get_memsize(unsigned count)
{
ssize_t sz;
@@ -119,7 +119,7 @@ rte_ring_get_memsize(unsigned count)
}
int
-rte_ring_init(struct rte_ring *r, const char *name, unsigned count,
+rte_common_ring_init(struct rte_ring *r, const char *name, unsigned count,
unsigned flags)
{
int ret;
@@ -134,7 +134,7 @@ rte_ring_init(struct rte_ring *r, const char *name, unsigned count,
RTE_BUILD_BUG_ON((offsetof(struct rte_ring, prod) &
RTE_CACHE_LINE_MASK) != 0);
#ifdef RTE_LIBRTE_RING_DEBUG
- RTE_BUILD_BUG_ON((sizeof(struct rte_ring_debug_stats) &
+ RTE_BUILD_BUG_ON((sizeof(struct rte_common_ring_debug_stats) &
RTE_CACHE_LINE_MASK) != 0);
RTE_BUILD_BUG_ON((offsetof(struct rte_ring, stats) &
RTE_CACHE_LINE_MASK) != 0);
@@ -159,7 +159,7 @@ rte_ring_init(struct rte_ring *r, const char *name, unsigned count,
/* create the ring */
struct rte_ring *
-rte_ring_create(const char *name, unsigned count, int socket_id,
+rte_common_ring_create(const char *name, unsigned count, int socket_id,
unsigned flags)
{
char mz_name[RTE_MEMZONE_NAMESIZE];
@@ -168,12 +168,12 @@ rte_ring_create(const char *name, unsigned count, int socket_id,
const struct rte_memzone *mz;
ssize_t ring_size;
int mz_flags = 0;
- struct rte_ring_list* ring_list = NULL;
+ struct rte_common_ring_list* ring_list = NULL;
int ret;
- ring_list = RTE_TAILQ_CAST(rte_ring_tailq.head, rte_ring_list);
+ ring_list = RTE_TAILQ_CAST(rte_common_ring_tailq.head, rte_common_ring_list);
- ring_size = rte_ring_get_memsize(count);
+ ring_size = rte_common_ring_get_memsize(count);
if (ring_size < 0) {
rte_errno = ring_size;
return NULL;
@@ -203,7 +203,7 @@ rte_ring_create(const char *name, unsigned count, int socket_id,
r = mz->addr;
/* no need to check return value here, we already checked the
* arguments above */
- rte_ring_init(r, name, count, flags);
+ rte_common_ring_init(r, name, count, flags);
te->data = (void *) r;
r->memzone = mz;
@@ -221,20 +221,20 @@ rte_ring_create(const char *name, unsigned count, int socket_id,
/* free the ring */
void
-rte_ring_free(struct rte_ring *r)
+rte_common_ring_free(struct rte_ring *r)
{
- struct rte_ring_list *ring_list = NULL;
+ struct rte_common_ring_list *ring_list = NULL;
struct rte_tailq_entry *te;
if (r == NULL)
return;
/*
- * Ring was not created with rte_ring_create,
+ * Ring was not created with rte_common_ring_create,
* therefore, there is no memzone to free.
*/
if (r->memzone == NULL) {
- RTE_LOG(ERR, RING, "Cannot free ring (not created with rte_ring_create()");
+ RTE_LOG(ERR, RING, "Cannot free ring (not created with rte_common_ring_create()");
return;
}
@@ -243,7 +243,7 @@ rte_ring_free(struct rte_ring *r)
return;
}
- ring_list = RTE_TAILQ_CAST(rte_ring_tailq.head, rte_ring_list);
+ ring_list = RTE_TAILQ_CAST(rte_common_ring_tailq.head, rte_common_ring_list);
rte_rwlock_write_lock(RTE_EAL_TAILQ_RWLOCK);
/* find out tailq entry */
@@ -269,7 +269,7 @@ rte_ring_free(struct rte_ring *r)
* disabled
*/
int
-rte_ring_set_water_mark(struct rte_ring *r, unsigned count)
+rte_common_ring_set_water_mark(struct rte_ring *r, unsigned count)
{
if (count >= r->prod.size)
return -EINVAL;
@@ -284,10 +284,10 @@ rte_ring_set_water_mark(struct rte_ring *r, unsigned count)
/* dump the status of the ring on the console */
void
-rte_ring_dump(FILE *f, const struct rte_ring *r)
+rte_common_ring_dump(FILE *f, const struct rte_ring *r)
{
#ifdef RTE_LIBRTE_RING_DEBUG
- struct rte_ring_debug_stats sum;
+ struct rte_common_ring_debug_stats sum;
unsigned lcore_id;
#endif
@@ -298,8 +298,8 @@ rte_ring_dump(FILE *f, const struct rte_ring *r)
fprintf(f, " ch=%"PRIu32"\n", r->cons.head);
fprintf(f, " pt=%"PRIu32"\n", r->prod.tail);
fprintf(f, " ph=%"PRIu32"\n", r->prod.head);
- fprintf(f, " used=%u\n", rte_ring_count(r));
- fprintf(f, " avail=%u\n", rte_ring_free_count(r));
+ fprintf(f, " used=%u\n", rte_common_ring_count(r));
+ fprintf(f, " avail=%u\n", rte_common_ring_free_count(r));
if (r->prod.watermark == r->prod.size)
fprintf(f, " watermark=0\n");
else
@@ -338,17 +338,17 @@ rte_ring_dump(FILE *f, const struct rte_ring *r)
/* dump the status of all rings on the console */
void
-rte_ring_list_dump(FILE *f)
+rte_common_ring_list_dump(FILE *f)
{
const struct rte_tailq_entry *te;
- struct rte_ring_list *ring_list;
+ struct rte_common_ring_list *ring_list;
- ring_list = RTE_TAILQ_CAST(rte_ring_tailq.head, rte_ring_list);
+ ring_list = RTE_TAILQ_CAST(rte_common_ring_tailq.head, rte_common_ring_list);
rte_rwlock_read_lock(RTE_EAL_TAILQ_RWLOCK);
TAILQ_FOREACH(te, ring_list, next) {
- rte_ring_dump(f, (struct rte_ring *) te->data);
+ rte_common_ring_dump(f, (struct rte_ring *) te->data);
}
rte_rwlock_read_unlock(RTE_EAL_TAILQ_RWLOCK);
@@ -356,13 +356,14 @@ rte_ring_list_dump(FILE *f)
/* search a ring from its name */
struct rte_ring *
-rte_ring_lookup(const char *name)
+rte_common_ring_lookup(const char *name)
{
struct rte_tailq_entry *te;
struct rte_ring *r = NULL;
- struct rte_ring_list *ring_list;
+ struct rte_common_ring_list *ring_list;
- ring_list = RTE_TAILQ_CAST(rte_ring_tailq.head, rte_ring_list);
+ ring_list = RTE_TAILQ_CAST(rte_common_ring_tailq.head,
+ rte_common_ring_list);
rte_rwlock_read_lock(RTE_EAL_TAILQ_RWLOCK);
diff --git a/lib/librte_ring/rte_common_ring.h b/lib/librte_ring/rte_common_ring.h
index e359aff..f2c1c46 100644
--- a/lib/librte_ring/rte_common_ring.h
+++ b/lib/librte_ring/rte_common_ring.h
@@ -63,8 +63,8 @@
*
***************************************************************************/
-#ifndef _RTE_RING_H_
-#define _RTE_RING_H_
+#ifndef _RTE_COMMON_RING_H_
+#define _RTE_COMMON_RING_H_
/**
* @file
@@ -232,14 +232,14 @@ struct rte_ring {
* - The memory size needed for the ring on success.
* - -EINVAL if count is not a power of 2.
*/
-ssize_t rte_ring_get_memsize(unsigned count);
+ssize_t rte_common_ring_get_memsize(unsigned count);
/**
* Initialize a ring structure.
*
* Initialize a ring structure in memory pointed by "r". The size of the
* memory area must be large enough to store the ring structure and the
- * object table. It is advised to use rte_ring_get_memsize() to get the
+ * object table. It is advised to use rte_common_ring_get_memsize() to get the
* appropriate size.
*
* The ring size is set to *count*, which must be a power of two. Water
@@ -260,22 +260,22 @@ ssize_t rte_ring_get_memsize(unsigned count);
* @param flags
* An OR of the following:
* - RING_F_SP_ENQ: If this flag is set, the default behavior when
- * using ``rte_ring_enqueue()`` or ``rte_ring_enqueue_bulk()``
+ * using ``rte_common_ring_enqueue()`` or ``rte_common_ring_enqueue_bulk()``
* is "single-producer". Otherwise, it is "multi-producers".
* - RING_F_SC_DEQ: If this flag is set, the default behavior when
- * using ``rte_ring_dequeue()`` or ``rte_ring_dequeue_bulk()``
+ * using ``rte_common_ring_dequeue()`` or ``rte_common_ring_dequeue_bulk()``
* is "single-consumer". Otherwise, it is "multi-consumers".
* @return
* 0 on success, or a negative value on error.
*/
-int rte_ring_init(struct rte_ring *r, const char *name, unsigned count,
+int rte_common_ring_init(struct rte_ring *r, const char *name, unsigned count,
unsigned flags);
/**
* Create a new ring named *name* in memory.
*
* This function uses ``memzone_reserve()`` to allocate memory. Then it
- * calls rte_ring_init() to initialize an empty ring.
+ * calls rte_common_ring_init() to initialize an empty ring.
*
* The new ring size is set to *count*, which must be a power of
* two. Water marking is disabled by default. The real usable ring size
@@ -295,10 +295,10 @@ int rte_ring_init(struct rte_ring *r, const char *name, unsigned count,
* @param flags
* An OR of the following:
* - RING_F_SP_ENQ: If this flag is set, the default behavior when
- * using ``rte_ring_enqueue()`` or ``rte_ring_enqueue_bulk()``
+ * using ``rte_common_ring_enqueue()`` or ``rte_common_ring_enqueue_bulk()``
* is "single-producer". Otherwise, it is "multi-producers".
* - RING_F_SC_DEQ: If this flag is set, the default behavior when
- * using ``rte_ring_dequeue()`` or ``rte_ring_dequeue_bulk()``
+ * using ``rte_common_ring_dequeue()`` or ``rte_common_ring_dequeue_bulk()``
* is "single-consumer". Otherwise, it is "multi-consumers".
* @return
* On success, the pointer to the new allocated ring. NULL on error with
@@ -310,7 +310,7 @@ int rte_ring_init(struct rte_ring *r, const char *name, unsigned count,
* - EEXIST - a memzone with the same name already exists
* - ENOMEM - no appropriate memory area found in which to create memzone
*/
-struct rte_ring *rte_ring_create(const char *name, unsigned count,
+struct rte_ring *rte_common_ring_create(const char *name, unsigned count,
int socket_id, unsigned flags);
/**
* De-allocate all memory used by the ring.
@@ -318,7 +318,7 @@ struct rte_ring *rte_ring_create(const char *name, unsigned count,
* @param r
* Ring to free
*/
-void rte_ring_free(struct rte_ring *r);
+void rte_common_ring_free(struct rte_ring *r);
/**
* Change the high water mark.
@@ -338,7 +338,7 @@ void rte_ring_free(struct rte_ring *r);
* - 0: Success; water mark changed.
* - -EINVAL: Invalid water mark value.
*/
-int rte_ring_set_water_mark(struct rte_ring *r, unsigned count);
+int rte_common_ring_set_water_mark(struct rte_ring *r, unsigned count);
/**
* Dump the status of the ring to a file.
@@ -348,7 +348,7 @@ int rte_ring_set_water_mark(struct rte_ring *r, unsigned count);
* @param r
* A pointer to the ring structure.
*/
-void rte_ring_dump(FILE *f, const struct rte_ring *r);
+void rte_common_ring_dump(FILE *f, const struct rte_ring *r);
/* the actual enqueue of pointers on the ring.
* Placed here since identical code needed in both
@@ -428,7 +428,7 @@ void rte_ring_dump(FILE *f, const struct rte_ring *r);
* - n: Actual number of objects enqueued.
*/
static inline int __attribute__((always_inline))
-__rte_ring_mp_do_enqueue(struct rte_ring *r, void * const *obj_table,
+__rte_common_ring_mp_do_enqueue(struct rte_ring *r, void * const *obj_table,
unsigned n, enum rte_ring_queue_behavior behavior)
{
uint32_t prod_head, prod_next;
@@ -537,7 +537,7 @@ __rte_ring_mp_do_enqueue(struct rte_ring *r, void * const *obj_table,
* - n: Actual number of objects enqueued.
*/
static inline int __attribute__((always_inline))
-__rte_ring_sp_do_enqueue(struct rte_ring *r, void * const *obj_table,
+__rte_common_ring_sp_do_enqueue(struct rte_ring *r, void * const *obj_table,
unsigned n, enum rte_ring_queue_behavior behavior)
{
uint32_t prod_head, cons_tail;
@@ -621,7 +621,7 @@ __rte_ring_sp_do_enqueue(struct rte_ring *r, void * const *obj_table,
*/
static inline int __attribute__((always_inline))
-__rte_ring_mc_do_dequeue(struct rte_ring *r, void **obj_table,
+__rte_common_ring_mc_do_dequeue(struct rte_ring *r, void **obj_table,
unsigned n, enum rte_ring_queue_behavior behavior)
{
uint32_t cons_head, prod_tail;
@@ -720,7 +720,7 @@ __rte_ring_mc_do_dequeue(struct rte_ring *r, void **obj_table,
* - n: Actual number of objects dequeued.
*/
static inline int __attribute__((always_inline))
-__rte_ring_sc_do_dequeue(struct rte_ring *r, void **obj_table,
+__rte_common_ring_sc_do_dequeue(struct rte_ring *r, void **obj_table,
unsigned n, enum rte_ring_queue_behavior behavior)
{
uint32_t cons_head, prod_tail;
@@ -764,284 +764,6 @@ __rte_ring_sc_do_dequeue(struct rte_ring *r, void **obj_table,
}
/**
- * Enqueue several objects on the ring (multi-producers safe).
- *
- * This function uses a "compare and set" instruction to move the
- * producer index atomically.
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_table
- * A pointer to a table of void * pointers (objects).
- * @param n
- * The number of objects to add in the ring from the obj_table.
- * @return
- * - 0: Success; objects enqueue.
- * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
- * high water mark is exceeded.
- * - -ENOBUFS: Not enough room in the ring to enqueue, no object is enqueued.
- */
-static inline int __attribute__((always_inline))
-rte_ring_mp_enqueue_bulk(struct rte_ring *r, void * const *obj_table,
- unsigned n)
-{
- return __rte_ring_mp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
-}
-
-/**
- * Enqueue several objects on a ring (NOT multi-producers safe).
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_table
- * A pointer to a table of void * pointers (objects).
- * @param n
- * The number of objects to add in the ring from the obj_table.
- * @return
- * - 0: Success; objects enqueued.
- * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
- * high water mark is exceeded.
- * - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
- */
-static inline int __attribute__((always_inline))
-rte_ring_sp_enqueue_bulk(struct rte_ring *r, void * const *obj_table,
- unsigned n)
-{
- return __rte_ring_sp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
-}
-
-/**
- * Enqueue several objects on a ring.
- *
- * This function calls the multi-producer or the single-producer
- * version depending on the default behavior that was specified at
- * ring creation time (see flags).
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_table
- * A pointer to a table of void * pointers (objects).
- * @param n
- * The number of objects to add in the ring from the obj_table.
- * @return
- * - 0: Success; objects enqueued.
- * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
- * high water mark is exceeded.
- * - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
- */
-static inline int __attribute__((always_inline))
-rte_ring_enqueue_bulk(struct rte_ring *r, void * const *obj_table,
- unsigned n)
-{
- if (r->prod.sp_enqueue)
- return rte_ring_sp_enqueue_bulk(r, obj_table, n);
- else
- return rte_ring_mp_enqueue_bulk(r, obj_table, n);
-}
-
-/**
- * Enqueue one object on a ring (multi-producers safe).
- *
- * This function uses a "compare and set" instruction to move the
- * producer index atomically.
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj
- * A pointer to the object to be added.
- * @return
- * - 0: Success; objects enqueued.
- * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
- * high water mark is exceeded.
- * - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
- */
-static inline int __attribute__((always_inline))
-rte_ring_mp_enqueue(struct rte_ring *r, void *obj)
-{
- return rte_ring_mp_enqueue_bulk(r, &obj, 1);
-}
-
-/**
- * Enqueue one object on a ring (NOT multi-producers safe).
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj
- * A pointer to the object to be added.
- * @return
- * - 0: Success; objects enqueued.
- * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
- * high water mark is exceeded.
- * - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
- */
-static inline int __attribute__((always_inline))
-rte_ring_sp_enqueue(struct rte_ring *r, void *obj)
-{
- return rte_ring_sp_enqueue_bulk(r, &obj, 1);
-}
-
-/**
- * Enqueue one object on a ring.
- *
- * This function calls the multi-producer or the single-producer
- * version, depending on the default behaviour that was specified at
- * ring creation time (see flags).
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj
- * A pointer to the object to be added.
- * @return
- * - 0: Success; objects enqueued.
- * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
- * high water mark is exceeded.
- * - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
- */
-static inline int __attribute__((always_inline))
-rte_ring_enqueue(struct rte_ring *r, void *obj)
-{
- if (r->prod.sp_enqueue)
- return rte_ring_sp_enqueue(r, obj);
- else
- return rte_ring_mp_enqueue(r, obj);
-}
-
-/**
- * Dequeue several objects from a ring (multi-consumers safe).
- *
- * This function uses a "compare and set" instruction to move the
- * consumer index atomically.
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_table
- * A pointer to a table of void * pointers (objects) that will be filled.
- * @param n
- * The number of objects to dequeue from the ring to the obj_table.
- * @return
- * - 0: Success; objects dequeued.
- * - -ENOENT: Not enough entries in the ring to dequeue; no object is
- * dequeued.
- */
-static inline int __attribute__((always_inline))
-rte_ring_mc_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n)
-{
- return __rte_ring_mc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
-}
-
-/**
- * Dequeue several objects from a ring (NOT multi-consumers safe).
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_table
- * A pointer to a table of void * pointers (objects) that will be filled.
- * @param n
- * The number of objects to dequeue from the ring to the obj_table,
- * must be strictly positive.
- * @return
- * - 0: Success; objects dequeued.
- * - -ENOENT: Not enough entries in the ring to dequeue; no object is
- * dequeued.
- */
-static inline int __attribute__((always_inline))
-rte_ring_sc_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n)
-{
- return __rte_ring_sc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
-}
-
-/**
- * Dequeue several objects from a ring.
- *
- * This function calls the multi-consumers or the single-consumer
- * version, depending on the default behaviour that was specified at
- * ring creation time (see flags).
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_table
- * A pointer to a table of void * pointers (objects) that will be filled.
- * @param n
- * The number of objects to dequeue from the ring to the obj_table.
- * @return
- * - 0: Success; objects dequeued.
- * - -ENOENT: Not enough entries in the ring to dequeue, no object is
- * dequeued.
- */
-static inline int __attribute__((always_inline))
-rte_ring_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n)
-{
- if (r->cons.sc_dequeue)
- return rte_ring_sc_dequeue_bulk(r, obj_table, n);
- else
- return rte_ring_mc_dequeue_bulk(r, obj_table, n);
-}
-
-/**
- * Dequeue one object from a ring (multi-consumers safe).
- *
- * This function uses a "compare and set" instruction to move the
- * consumer index atomically.
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_p
- * A pointer to a void * pointer (object) that will be filled.
- * @return
- * - 0: Success; objects dequeued.
- * - -ENOENT: Not enough entries in the ring to dequeue; no object is
- * dequeued.
- */
-static inline int __attribute__((always_inline))
-rte_ring_mc_dequeue(struct rte_ring *r, void **obj_p)
-{
- return rte_ring_mc_dequeue_bulk(r, obj_p, 1);
-}
-
-/**
- * Dequeue one object from a ring (NOT multi-consumers safe).
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_p
- * A pointer to a void * pointer (object) that will be filled.
- * @return
- * - 0: Success; objects dequeued.
- * - -ENOENT: Not enough entries in the ring to dequeue, no object is
- * dequeued.
- */
-static inline int __attribute__((always_inline))
-rte_ring_sc_dequeue(struct rte_ring *r, void **obj_p)
-{
- return rte_ring_sc_dequeue_bulk(r, obj_p, 1);
-}
-
-/**
- * Dequeue one object from a ring.
- *
- * This function calls the multi-consumers or the single-consumer
- * version depending on the default behaviour that was specified at
- * ring creation time (see flags).
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_p
- * A pointer to a void * pointer (object) that will be filled.
- * @return
- * - 0: Success, objects dequeued.
- * - -ENOENT: Not enough entries in the ring to dequeue, no object is
- * dequeued.
- */
-static inline int __attribute__((always_inline))
-rte_ring_dequeue(struct rte_ring *r, void **obj_p)
-{
- if (r->cons.sc_dequeue)
- return rte_ring_sc_dequeue(r, obj_p);
- else
- return rte_ring_mc_dequeue(r, obj_p);
-}
-
-/**
* Test if a ring is full.
*
* @param r
@@ -1051,7 +773,7 @@ rte_ring_dequeue(struct rte_ring *r, void **obj_p)
* - 0: The ring is not full.
*/
static inline int
-rte_ring_full(const struct rte_ring *r)
+rte_common_ring_full(const struct rte_ring *r)
{
uint32_t prod_tail = r->prod.tail;
uint32_t cons_tail = r->cons.tail;
@@ -1068,7 +790,7 @@ rte_ring_full(const struct rte_ring *r)
* - 0: The ring is not empty.
*/
static inline int
-rte_ring_empty(const struct rte_ring *r)
+rte_common_ring_empty(const struct rte_ring *r)
{
uint32_t prod_tail = r->prod.tail;
uint32_t cons_tail = r->cons.tail;
@@ -1084,7 +806,7 @@ rte_ring_empty(const struct rte_ring *r)
* The number of entries in the ring.
*/
static inline unsigned
-rte_ring_count(const struct rte_ring *r)
+rte_common_ring_count(const struct rte_ring *r)
{
uint32_t prod_tail = r->prod.tail;
uint32_t cons_tail = r->cons.tail;
@@ -1100,7 +822,7 @@ rte_ring_count(const struct rte_ring *r)
* The number of free entries in the ring.
*/
static inline unsigned
-rte_ring_free_count(const struct rte_ring *r)
+rte_common_ring_free_count(const struct rte_ring *r)
{
uint32_t prod_tail = r->prod.tail;
uint32_t cons_tail = r->cons.tail;
@@ -1113,7 +835,7 @@ rte_ring_free_count(const struct rte_ring *r)
* @param f
* A pointer to a file for output
*/
-void rte_ring_list_dump(FILE *f);
+void rte_common_ring_list_dump(FILE *f);
/**
* Search a ring from its name
@@ -1125,145 +847,10 @@ void rte_ring_list_dump(FILE *f);
* with rte_errno set appropriately. Possible rte_errno values include:
* - ENOENT - required entry not available to return.
*/
-struct rte_ring *rte_ring_lookup(const char *name);
-
-/**
- * Enqueue several objects on the ring (multi-producers safe).
- *
- * This function uses a "compare and set" instruction to move the
- * producer index atomically.
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_table
- * A pointer to a table of void * pointers (objects).
- * @param n
- * The number of objects to add in the ring from the obj_table.
- * @return
- * - n: Actual number of objects enqueued.
- */
-static inline unsigned __attribute__((always_inline))
-rte_ring_mp_enqueue_burst(struct rte_ring *r, void * const *obj_table,
- unsigned n)
-{
- return __rte_ring_mp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
-}
-
-/**
- * Enqueue several objects on a ring (NOT multi-producers safe).
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_table
- * A pointer to a table of void * pointers (objects).
- * @param n
- * The number of objects to add in the ring from the obj_table.
- * @return
- * - n: Actual number of objects enqueued.
- */
-static inline unsigned __attribute__((always_inline))
-rte_ring_sp_enqueue_burst(struct rte_ring *r, void * const *obj_table,
- unsigned n)
-{
- return __rte_ring_sp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
-}
-
-/**
- * Enqueue several objects on a ring.
- *
- * This function calls the multi-producer or the single-producer
- * version depending on the default behavior that was specified at
- * ring creation time (see flags).
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_table
- * A pointer to a table of void * pointers (objects).
- * @param n
- * The number of objects to add in the ring from the obj_table.
- * @return
- * - n: Actual number of objects enqueued.
- */
-static inline unsigned __attribute__((always_inline))
-rte_ring_enqueue_burst(struct rte_ring *r, void * const *obj_table,
- unsigned n)
-{
- if (r->prod.sp_enqueue)
- return rte_ring_sp_enqueue_burst(r, obj_table, n);
- else
- return rte_ring_mp_enqueue_burst(r, obj_table, n);
-}
-
-/**
- * Dequeue several objects from a ring (multi-consumers safe). When the request
- * objects are more than the available objects, only dequeue the actual number
- * of objects
- *
- * This function uses a "compare and set" instruction to move the
- * consumer index atomically.
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_table
- * A pointer to a table of void * pointers (objects) that will be filled.
- * @param n
- * The number of objects to dequeue from the ring to the obj_table.
- * @return
- * - n: Actual number of objects dequeued, 0 if ring is empty
- */
-static inline unsigned __attribute__((always_inline))
-rte_ring_mc_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n)
-{
- return __rte_ring_mc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
-}
-
-/**
- * Dequeue several objects from a ring (NOT multi-consumers safe).When the
- * request objects are more than the available objects, only dequeue the
- * actual number of objects
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_table
- * A pointer to a table of void * pointers (objects) that will be filled.
- * @param n
- * The number of objects to dequeue from the ring to the obj_table.
- * @return
- * - n: Actual number of objects dequeued, 0 if ring is empty
- */
-static inline unsigned __attribute__((always_inline))
-rte_ring_sc_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n)
-{
- return __rte_ring_sc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
-}
-
-/**
- * Dequeue multiple objects from a ring up to a maximum number.
- *
- * This function calls the multi-consumers or the single-consumer
- * version, depending on the default behaviour that was specified at
- * ring creation time (see flags).
- *
- * @param r
- * A pointer to the ring structure.
- * @param obj_table
- * A pointer to a table of void * pointers (objects) that will be filled.
- * @param n
- * The number of objects to dequeue from the ring to the obj_table.
- * @return
- * - Number of objects dequeued
- */
-static inline unsigned __attribute__((always_inline))
-rte_ring_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n)
-{
- if (r->cons.sc_dequeue)
- return rte_ring_sc_dequeue_burst(r, obj_table, n);
- else
- return rte_ring_mc_dequeue_burst(r, obj_table, n);
-}
+struct rte_ring *rte_common_ring_lookup(const char *name);
#ifdef __cplusplus
}
#endif
-#endif /* _RTE_RING_H_ */
+#endif /* _RTE_COMMON_RING_H_ */
diff --git a/lib/librte_ring/rte_ring.c b/lib/librte_ring/rte_ring.c
new file mode 100644
index 0000000..16ddc39
--- /dev/null
+++ b/lib/librte_ring/rte_ring.c
@@ -0,0 +1,86 @@
+/*-
+ * BSD LICENSE
+ *
+ * Copyright(c) 2017 Intel Corporation. All rights reserved.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ * * 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.
+ * * Neither the name of Intel Corporation nor the names of its
+ * contributors may be used to endorse or promote products derived
+ * from this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS 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 COPYRIGHT
+ * OWNER 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.
+ */
+
+#include "rte_ring.h"
+
+ssize_t
+rte_ring_get_memsize(unsigned count)
+{
+ return rte_common_ring_get_memsize(count);
+}
+
+int
+rte_ring_init(struct rte_ring *r, const char *name, unsigned count,
+ unsigned flags)
+{
+ return rte_common_ring_init(r, name, count, flags);
+}
+
+
+struct rte_ring *
+rte_ring_create(const char *name, unsigned count, int socket_id,
+ unsigned flags)
+{
+ return rte_common_ring_create(name, count, socket_id, flags);
+}
+
+void
+rte_ring_free(struct rte_ring *r)
+{
+ return rte_common_ring_free(r);
+}
+
+int
+rte_ring_set_water_mark(struct rte_ring *r, unsigned count)
+{
+ return rte_common_ring_set_water_mark(r, count);
+}
+
+void
+rte_ring_dump(FILE *f, const struct rte_ring *r)
+{
+ return rte_common_ring_dump(f, r);
+}
+
+void
+rte_ring_list_dump(FILE *f)
+{
+ rte_common_ring_list_dump(f);
+}
+
+struct rte_ring *
+rte_ring_lookup(const char *name)
+{
+ return rte_common_ring_lookup(name);
+}
+
diff --git a/lib/librte_ring/rte_ring.h b/lib/librte_ring/rte_ring.h
deleted file mode 120000
index 54dad23..0000000
--- a/lib/librte_ring/rte_ring.h
+++ /dev/null
@@ -1 +0,0 @@
-rte_common_ring.h
\ No newline at end of file
diff --git a/lib/librte_ring/rte_ring.h b/lib/librte_ring/rte_ring.h
new file mode 100644
index 0000000..993796f
--- /dev/null
+++ b/lib/librte_ring/rte_ring.h
@@ -0,0 +1,691 @@
+/*-
+ * BSD LICENSE
+ *
+ * Copyright(c) 2017 Intel Corporation. All rights reserved.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ * * 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.
+ * * Neither the name of Intel Corporation nor the names of its
+ * contributors may be used to endorse or promote products derived
+ * from this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS 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 COPYRIGHT
+ * OWNER 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.
+ */
+
+#ifndef _RTE_RING_H_
+#define _RTE_RING_H_
+
+/**
+ * @file
+ * RTE Ring
+ *
+ * The Ring Manager is a fixed-size queue, implemented as a table of
+ * pointers. Head and tail pointers are modified atomically, allowing
+ * concurrent access to it. It has the following features:
+ *
+ * - FIFO (First In First Out)
+ * - Maximum size is fixed; the pointers are stored in a table.
+ * - Lockless implementation.
+ * - Multi- or single-consumer dequeue.
+ * - Multi- or single-producer enqueue.
+ * - Bulk dequeue.
+ * - Bulk enqueue.
+ *
+ * Note: the ring implementation is not preemptable. A lcore must not
+ * be interrupted by another task that uses the same ring.
+ *
+ */
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include "rte_common_ring.h"
+
+/**
+ * Calculate the memory size needed for a ring
+ *
+ * This function returns the number of bytes needed for a ring, given
+ * the number of elements in it. This value is the sum of the size of
+ * the structure rte_ring and the size of the memory needed by the
+ * objects pointers. The value is aligned to a cache line size.
+ *
+ * @param count
+ * The number of elements in the ring (must be a power of 2).
+ * @return
+ * - The memory size needed for the ring on success.
+ * - -EINVAL if count is not a power of 2.
+ */
+ssize_t rte_ring_get_memsize(unsigned count);
+
+/**
+ * Initialize a ring structure.
+ *
+ * Initialize a ring structure in memory pointed by "r". The size of the
+ * memory area must be large enough to store the ring structure and the
+ * object table. It is advised to use rte_ring_get_memsize() to get the
+ * appropriate size.
+ *
+ * The ring size is set to *count*, which must be a power of two. Water
+ * marking is disabled by default. The real usable ring size is
+ * *count-1* instead of *count* to differentiate a free ring from an
+ * empty ring.
+ *
+ * The ring is not added in RTE_TAILQ_RING global list. Indeed, the
+ * memory given by the caller may not be shareable among dpdk
+ * processes.
+ *
+ * @param r
+ * The pointer to the ring structure followed by the objects table.
+ * @param name
+ * The name of the ring.
+ * @param count
+ * The number of elements in the ring (must be a power of 2).
+ * @param flags
+ * An OR of the following:
+ * - RING_F_SP_ENQ: If this flag is set, the default behavior when
+ * using ``rte_ring_enqueue()`` or ``rte_ring_enqueue_bulk()``
+ * is "single-producer". Otherwise, it is "multi-producers".
+ * - RING_F_SC_DEQ: If this flag is set, the default behavior when
+ * using ``rte_ring_dequeue()`` or ``rte_ring_dequeue_bulk()``
+ * is "single-consumer". Otherwise, it is "multi-consumers".
+ * @return
+ * 0 on success, or a negative value on error.
+ */
+int rte_ring_init(struct rte_ring *r, const char *name, unsigned count,
+ unsigned flags);
+
+/**
+ * Create a new ring named *name* in memory.
+ *
+ * This function uses ``memzone_reserve()`` to allocate memory. Then it
+ * calls rte_ring_init() to initialize an empty ring.
+ *
+ * The new ring size is set to *count*, which must be a power of
+ * two. Water marking is disabled by default. The real usable ring size
+ * is *count-1* instead of *count* to differentiate a free ring from an
+ * empty ring.
+ *
+ * The ring is added in RTE_TAILQ_RING list.
+ *
+ * @param name
+ * The name of the ring.
+ * @param count
+ * The size of the ring (must be a power of 2).
+ * @param socket_id
+ * The *socket_id* argument is the socket identifier in case of
+ * NUMA. The value can be *SOCKET_ID_ANY* if there is no NUMA
+ * constraint for the reserved zone.
+ * @param flags
+ * An OR of the following:
+ * - RING_F_SP_ENQ: If this flag is set, the default behavior when
+ * using ``rte_ring_enqueue()`` or ``rte_ring_enqueue_bulk()``
+ * is "single-producer". Otherwise, it is "multi-producers".
+ * - RING_F_SC_DEQ: If this flag is set, the default behavior when
+ * using ``rte_ring_dequeue()`` or ``rte_ring_dequeue_bulk()``
+ * is "single-consumer". Otherwise, it is "multi-consumers".
+ * @return
+ * On success, the pointer to the new allocated ring. NULL on error with
+ * rte_errno set appropriately. Possible errno values include:
+ * - E_RTE_NO_CONFIG - function could not get pointer to rte_config structure
+ * - E_RTE_SECONDARY - function was called from a secondary process instance
+ * - EINVAL - count provided is not a power of 2
+ * - ENOSPC - the maximum number of memzones has already been allocated
+ * - EEXIST - a memzone with the same name already exists
+ * - ENOMEM - no appropriate memory area found in which to create memzone
+ */
+struct rte_ring *rte_ring_create(const char *name, unsigned count,
+ int socket_id, unsigned flags);
+/**
+ * De-allocate all memory used by the ring.
+ *
+ * @param r
+ * Ring to free
+ */
+void rte_ring_free(struct rte_ring *r);
+
+/**
+ * Change the high water mark.
+ *
+ * If *count* is 0, water marking is disabled. Otherwise, it is set to the
+ * *count* value. The *count* value must be greater than 0 and less
+ * than the ring size.
+ *
+ * This function can be called at any time (not necessarily at
+ * initialization).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param count
+ * The new water mark value.
+ * @return
+ * - 0: Success; water mark changed.
+ * - -EINVAL: Invalid water mark value.
+ */
+int rte_ring_set_water_mark(struct rte_ring *r, unsigned count);
+
+/**
+ * Dump the status of the ring to a file.
+ *
+ * @param f
+ * A pointer to a file for output
+ * @param r
+ * A pointer to the ring structure.
+ */
+void rte_ring_dump(FILE *f, const struct rte_ring *r);
+
+/**
+ * Enqueue several objects on the ring (multi-producers safe).
+ *
+ * This function uses a "compare and set" instruction to move the
+ * producer index atomically.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects).
+ * @param n
+ * The number of objects to add in the ring from the obj_table.
+ * @return
+ * - 0: Success; objects enqueue.
+ * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
+ * high water mark is exceeded.
+ * - -ENOBUFS: Not enough room in the ring to enqueue, no object is enqueued.
+ */
+static inline int __attribute__((always_inline))
+rte_ring_mp_enqueue_bulk(struct rte_ring *r, void * const *obj_table,
+ unsigned n)
+{
+ return __rte_common_ring_mp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
+}
+
+/**
+ * Enqueue several objects on a ring (NOT multi-producers safe).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects).
+ * @param n
+ * The number of objects to add in the ring from the obj_table.
+ * @return
+ * - 0: Success; objects enqueued.
+ * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
+ * high water mark is exceeded.
+ * - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
+ */
+static inline int __attribute__((always_inline))
+rte_ring_sp_enqueue_bulk(struct rte_ring *r, void * const *obj_table,
+ unsigned n)
+{
+ return __rte_common_ring_sp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
+}
+
+/**
+ * Enqueue several objects on a ring.
+ *
+ * This function calls the multi-producer or the single-producer
+ * version depending on the default behavior that was specified at
+ * ring creation time (see flags).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects).
+ * @param n
+ * The number of objects to add in the ring from the obj_table.
+ * @return
+ * - 0: Success; objects enqueued.
+ * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
+ * high water mark is exceeded.
+ * - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
+ */
+static inline int __attribute__((always_inline))
+rte_ring_enqueue_bulk(struct rte_ring *r, void * const *obj_table,
+ unsigned n)
+{
+ if (r->prod.sp_enqueue)
+ return rte_ring_sp_enqueue_bulk(r, obj_table, n);
+ else
+ return rte_ring_mp_enqueue_bulk(r, obj_table, n);
+}
+
+/**
+ * Enqueue one object on a ring (multi-producers safe).
+ *
+ * This function uses a "compare and set" instruction to move the
+ * producer index atomically.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj
+ * A pointer to the object to be added.
+ * @return
+ * - 0: Success; objects enqueued.
+ * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
+ * high water mark is exceeded.
+ * - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
+ */
+static inline int __attribute__((always_inline))
+rte_ring_mp_enqueue(struct rte_ring *r, void *obj)
+{
+ return rte_ring_mp_enqueue_bulk(r, &obj, 1);
+}
+
+/**
+ * Enqueue one object on a ring (NOT multi-producers safe).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj
+ * A pointer to the object to be added.
+ * @return
+ * - 0: Success; objects enqueued.
+ * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
+ * high water mark is exceeded.
+ * - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
+ */
+static inline int __attribute__((always_inline))
+rte_ring_sp_enqueue(struct rte_ring *r, void *obj)
+{
+ return rte_ring_sp_enqueue_bulk(r, &obj, 1);
+}
+
+/**
+ * Enqueue one object on a ring.
+ *
+ * This function calls the multi-producer or the single-producer
+ * version, depending on the default behaviour that was specified at
+ * ring creation time (see flags).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj
+ * A pointer to the object to be added.
+ * @return
+ * - 0: Success; objects enqueued.
+ * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
+ * high water mark is exceeded.
+ * - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
+ */
+static inline int __attribute__((always_inline))
+rte_ring_enqueue(struct rte_ring *r, void *obj)
+{
+ if (r->prod.sp_enqueue)
+ return rte_ring_sp_enqueue(r, obj);
+ else
+ return rte_ring_mp_enqueue(r, obj);
+}
+
+/**
+ * Dequeue several objects from a ring (multi-consumers safe).
+ *
+ * This function uses a "compare and set" instruction to move the
+ * consumer index atomically.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects) that will be filled.
+ * @param n
+ * The number of objects to dequeue from the ring to the obj_table.
+ * @return
+ * - 0: Success; objects dequeued.
+ * - -ENOENT: Not enough entries in the ring to dequeue; no object is
+ * dequeued.
+ */
+static inline int __attribute__((always_inline))
+rte_ring_mc_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n)
+{
+ return __rte_common_ring_mc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
+}
+
+/**
+ * Dequeue several objects from a ring (NOT multi-consumers safe).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects) that will be filled.
+ * @param n
+ * The number of objects to dequeue from the ring to the obj_table,
+ * must be strictly positive.
+ * @return
+ * - 0: Success; objects dequeued.
+ * - -ENOENT: Not enough entries in the ring to dequeue; no object is
+ * dequeued.
+ */
+static inline int __attribute__((always_inline))
+rte_ring_sc_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n)
+{
+ return __rte_common_ring_sc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
+}
+
+/**
+ * Dequeue several objects from a ring.
+ *
+ * This function calls the multi-consumers or the single-consumer
+ * version, depending on the default behaviour that was specified at
+ * ring creation time (see flags).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects) that will be filled.
+ * @param n
+ * The number of objects to dequeue from the ring to the obj_table.
+ * @return
+ * - 0: Success; objects dequeued.
+ * - -ENOENT: Not enough entries in the ring to dequeue, no object is
+ * dequeued.
+ */
+static inline int __attribute__((always_inline))
+rte_ring_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n)
+{
+ if (r->cons.sc_dequeue)
+ return rte_ring_sc_dequeue_bulk(r, obj_table, n);
+ else
+ return rte_ring_mc_dequeue_bulk(r, obj_table, n);
+}
+
+/**
+ * Dequeue one object from a ring (multi-consumers safe).
+ *
+ * This function uses a "compare and set" instruction to move the
+ * consumer index atomically.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_p
+ * A pointer to a void * pointer (object) that will be filled.
+ * @return
+ * - 0: Success; objects dequeued.
+ * - -ENOENT: Not enough entries in the ring to dequeue; no object is
+ * dequeued.
+ */
+static inline int __attribute__((always_inline))
+rte_ring_mc_dequeue(struct rte_ring *r, void **obj_p)
+{
+ return rte_ring_mc_dequeue_bulk(r, obj_p, 1);
+}
+
+/**
+ * Dequeue one object from a ring (NOT multi-consumers safe).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_p
+ * A pointer to a void * pointer (object) that will be filled.
+ * @return
+ * - 0: Success; objects dequeued.
+ * - -ENOENT: Not enough entries in the ring to dequeue, no object is
+ * dequeued.
+ */
+static inline int __attribute__((always_inline))
+rte_ring_sc_dequeue(struct rte_ring *r, void **obj_p)
+{
+ return rte_ring_sc_dequeue_bulk(r, obj_p, 1);
+}
+
+/**
+ * Dequeue one object from a ring.
+ *
+ * This function calls the multi-consumers or the single-consumer
+ * version depending on the default behaviour that was specified at
+ * ring creation time (see flags).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_p
+ * A pointer to a void * pointer (object) that will be filled.
+ * @return
+ * - 0: Success, objects dequeued.
+ * - -ENOENT: Not enough entries in the ring to dequeue, no object is
+ * dequeued.
+ */
+static inline int __attribute__((always_inline))
+rte_ring_dequeue(struct rte_ring *r, void **obj_p)
+{
+ if (r->cons.sc_dequeue)
+ return rte_ring_sc_dequeue(r, obj_p);
+ else
+ return rte_ring_mc_dequeue(r, obj_p);
+}
+
+/**
+ * Test if a ring is full.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @return
+ * - 1: The ring is full.
+ * - 0: The ring is not full.
+ */
+static inline int
+rte_ring_full(const struct rte_ring *r)
+{
+ return rte_common_ring_full(r);
+}
+
+/**
+ * Test if a ring is empty.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @return
+ * - 1: The ring is empty.
+ * - 0: The ring is not empty.
+ */
+static inline int
+rte_ring_empty(const struct rte_ring *r)
+{
+ return rte_common_ring_empty(r);
+}
+
+/**
+ * Return the number of entries in a ring.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @return
+ * The number of entries in the ring.
+ */
+static inline unsigned
+rte_ring_count(const struct rte_ring *r)
+{
+ return rte_common_ring_count(r);
+}
+
+/**
+ * Return the number of free entries in a ring.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @return
+ * The number of free entries in the ring.
+ */
+static inline unsigned
+rte_ring_free_count(const struct rte_ring *r)
+{
+ return rte_common_ring_free_count(r);
+}
+
+/**
+ * Dump the status of all rings on the console
+ *
+ * @param f
+ * A pointer to a file for output
+ */
+void rte_ring_list_dump(FILE *f);
+
+/**
+ * Search a ring from its name
+ *
+ * @param name
+ * The name of the ring.
+ * @return
+ * The pointer to the ring matching the name, or NULL if not found,
+ * with rte_errno set appropriately. Possible rte_errno values include:
+ * - ENOENT - required entry not available to return.
+ */
+struct rte_ring *rte_ring_lookup(const char *name);
+
+/**
+ * Enqueue several objects on the ring (multi-producers safe).
+ *
+ * This function uses a "compare and set" instruction to move the
+ * producer index atomically.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects).
+ * @param n
+ * The number of objects to add in the ring from the obj_table.
+ * @return
+ * - n: Actual number of objects enqueued.
+ */
+static inline unsigned __attribute__((always_inline))
+rte_ring_mp_enqueue_burst(struct rte_ring *r, void * const *obj_table,
+ unsigned n)
+{
+ return __rte_common_ring_mp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
+}
+
+/**
+ * Enqueue several objects on a ring (NOT multi-producers safe).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects).
+ * @param n
+ * The number of objects to add in the ring from the obj_table.
+ * @return
+ * - n: Actual number of objects enqueued.
+ */
+static inline unsigned __attribute__((always_inline))
+rte_ring_sp_enqueue_burst(struct rte_ring *r, void * const *obj_table,
+ unsigned n)
+{
+ return __rte_common_ring_sp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
+}
+
+/**
+ * Enqueue several objects on a ring.
+ *
+ * This function calls the multi-producer or the single-producer
+ * version depending on the default behavior that was specified at
+ * ring creation time (see flags).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects).
+ * @param n
+ * The number of objects to add in the ring from the obj_table.
+ * @return
+ * - n: Actual number of objects enqueued.
+ */
+static inline unsigned __attribute__((always_inline))
+rte_ring_enqueue_burst(struct rte_ring *r, void * const *obj_table,
+ unsigned n)
+{
+ if (r->prod.sp_enqueue)
+ return rte_ring_sp_enqueue_burst(r, obj_table, n);
+ else
+ return rte_ring_mp_enqueue_burst(r, obj_table, n);
+}
+
+/**
+ * Dequeue several objects from a ring (multi-consumers safe). When the request
+ * objects are more than the available objects, only dequeue the actual number
+ * of objects
+ *
+ * This function uses a "compare and set" instruction to move the
+ * consumer index atomically.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects) that will be filled.
+ * @param n
+ * The number of objects to dequeue from the ring to the obj_table.
+ * @return
+ * - n: Actual number of objects dequeued, 0 if ring is empty
+ */
+static inline unsigned __attribute__((always_inline))
+rte_ring_mc_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n)
+{
+ return __rte_common_ring_mc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
+}
+
+/**
+ * Dequeue several objects from a ring (NOT multi-consumers safe).When the
+ * request objects are more than the available objects, only dequeue the
+ * actual number of objects
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects) that will be filled.
+ * @param n
+ * The number of objects to dequeue from the ring to the obj_table.
+ * @return
+ * - n: Actual number of objects dequeued, 0 if ring is empty
+ */
+static inline unsigned __attribute__((always_inline))
+rte_ring_sc_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n)
+{
+ return __rte_common_ring_sc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
+}
+
+/**
+ * Dequeue multiple objects from a ring up to a maximum number.
+ *
+ * This function calls the multi-consumers or the single-consumer
+ * version, depending on the default behaviour that was specified at
+ * ring creation time (see flags).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects) that will be filled.
+ * @param n
+ * The number of objects to dequeue from the ring to the obj_table.
+ * @return
+ * - Number of objects dequeued
+ */
+static inline unsigned __attribute__((always_inline))
+rte_ring_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n)
+{
+ if (r->cons.sc_dequeue)
+ return rte_ring_sc_dequeue_burst(r, obj_table, n);
+ else
+ return rte_ring_mc_dequeue_burst(r, obj_table, n);
+}
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* _RTE_RING_H_ */
--
2.9.3
^ permalink raw reply [flat|nested] 5+ messages in thread
* [dpdk-dev] [PATCH RFCv2 3/4] ring: allow common ring to use 8 or 16 byte values
2017-01-24 10:39 [dpdk-dev] [PATCH RFCv2 0/4] generalise rte_ring to allow different datatypes Bruce Richardson
2017-01-24 10:39 ` [dpdk-dev] [PATCH RFCv2 1/4] ring: create common ring files Bruce Richardson
2017-01-24 10:39 ` [dpdk-dev] [PATCH RFCv2 2/4] ring: separate common and rte_ring specific functions Bruce Richardson
@ 2017-01-24 10:39 ` Bruce Richardson
2017-01-24 10:39 ` [dpdk-dev] [PATCH RFCv2 4/4] ring: add new event ring class Bruce Richardson
3 siblings, 0 replies; 5+ messages in thread
From: Bruce Richardson @ 2017-01-24 10:39 UTC (permalink / raw)
To: olivier.matz; +Cc: dev, Bruce Richardson
Change the common ring enqueue/dequeue functions to support enqueuing and
dequeuing 16B values. Add the element size as parameter to all common ring
functions that need it, and pass that as parameter from the rte_ring
functions.
Signed-off-by: Bruce Richardson <bruce.richardson@intel.com>
---
lib/librte_ring/rte_common_ring.c | 14 ++--
lib/librte_ring/rte_common_ring.h | 135 ++++++++++++++++++++++++++++----------
lib/librte_ring/rte_ring.c | 7 +-
lib/librte_ring/rte_ring.h | 16 ++---
4 files changed, 122 insertions(+), 50 deletions(-)
diff --git a/lib/librte_ring/rte_common_ring.c b/lib/librte_ring/rte_common_ring.c
index a0c4b5a..eb04de4 100644
--- a/lib/librte_ring/rte_common_ring.c
+++ b/lib/librte_ring/rte_common_ring.c
@@ -101,7 +101,7 @@ EAL_REGISTER_TAILQ(rte_common_ring_tailq)
/* return the size of memory occupied by a ring */
ssize_t
-rte_common_ring_get_memsize(unsigned count)
+rte_common_ring_get_memsize(unsigned count, unsigned int elem_sz)
{
ssize_t sz;
@@ -113,14 +113,14 @@ rte_common_ring_get_memsize(unsigned count)
return -EINVAL;
}
- sz = sizeof(struct rte_ring) + count * sizeof(void *);
+ sz = sizeof(struct rte_ring) + count * elem_sz;
sz = RTE_ALIGN(sz, RTE_CACHE_LINE_SIZE);
return sz;
}
int
rte_common_ring_init(struct rte_ring *r, const char *name, unsigned count,
- unsigned flags)
+ unsigned flags, unsigned int elem_sz)
{
int ret;
@@ -146,6 +146,7 @@ rte_common_ring_init(struct rte_ring *r, const char *name, unsigned count,
if (ret < 0 || ret >= (int)sizeof(r->name))
return -ENAMETOOLONG;
r->flags = flags;
+ r->elem_sz = elem_sz;
r->prod.watermark = count;
r->prod.sp_enqueue = !!(flags & RING_F_SP_ENQ);
r->cons.sc_dequeue = !!(flags & RING_F_SC_DEQ);
@@ -160,7 +161,7 @@ rte_common_ring_init(struct rte_ring *r, const char *name, unsigned count,
/* create the ring */
struct rte_ring *
rte_common_ring_create(const char *name, unsigned count, int socket_id,
- unsigned flags)
+ unsigned flags, unsigned int elem_sz)
{
char mz_name[RTE_MEMZONE_NAMESIZE];
struct rte_ring *r;
@@ -173,7 +174,7 @@ rte_common_ring_create(const char *name, unsigned count, int socket_id,
ring_list = RTE_TAILQ_CAST(rte_common_ring_tailq.head, rte_common_ring_list);
- ring_size = rte_common_ring_get_memsize(count);
+ ring_size = rte_common_ring_get_memsize(count, elem_sz);
if (ring_size < 0) {
rte_errno = ring_size;
return NULL;
@@ -203,7 +204,7 @@ rte_common_ring_create(const char *name, unsigned count, int socket_id,
r = mz->addr;
/* no need to check return value here, we already checked the
* arguments above */
- rte_common_ring_init(r, name, count, flags);
+ rte_common_ring_init(r, name, count, flags, elem_sz);
te->data = (void *) r;
r->memzone = mz;
@@ -293,6 +294,7 @@ rte_common_ring_dump(FILE *f, const struct rte_ring *r)
fprintf(f, "ring <%s>@%p\n", r->name, r);
fprintf(f, " flags=%x\n", r->flags);
+ fprintf(f, " elem_sz=%u\n", r->elem_sz);
fprintf(f, " size=%"PRIu32"\n", r->prod.size);
fprintf(f, " ct=%"PRIu32"\n", r->cons.tail);
fprintf(f, " ch=%"PRIu32"\n", r->cons.head);
diff --git a/lib/librte_ring/rte_common_ring.h b/lib/librte_ring/rte_common_ring.h
index f2c1c46..314d53b 100644
--- a/lib/librte_ring/rte_common_ring.h
+++ b/lib/librte_ring/rte_common_ring.h
@@ -101,6 +101,7 @@ extern "C" {
#include <rte_atomic.h>
#include <rte_branch_prediction.h>
#include <rte_memzone.h>
+#include <rte_vect.h>
#define RTE_TAILQ_RING_NAME "RTE_RING"
@@ -157,6 +158,7 @@ struct rte_ring {
*/
char name[RTE_MEMZONE_NAMESIZE]; /**< Name of the ring. */
int flags; /**< Flags supplied at creation. */
+ unsigned int elem_sz; /**< Size of a ring entry */
const struct rte_memzone *memzone;
/**< Memzone, if any, containing the rte_ring */
@@ -232,7 +234,7 @@ struct rte_ring {
* - The memory size needed for the ring on success.
* - -EINVAL if count is not a power of 2.
*/
-ssize_t rte_common_ring_get_memsize(unsigned count);
+ssize_t rte_common_ring_get_memsize(unsigned count, unsigned int elem_sz);
/**
* Initialize a ring structure.
@@ -269,7 +271,7 @@ ssize_t rte_common_ring_get_memsize(unsigned count);
* 0 on success, or a negative value on error.
*/
int rte_common_ring_init(struct rte_ring *r, const char *name, unsigned count,
- unsigned flags);
+ unsigned flags, unsigned int elem_sz);
/**
* Create a new ring named *name* in memory.
@@ -311,7 +313,8 @@ int rte_common_ring_init(struct rte_ring *r, const char *name, unsigned count,
* - ENOMEM - no appropriate memory area found in which to create memzone
*/
struct rte_ring *rte_common_ring_create(const char *name, unsigned count,
- int socket_id, unsigned flags);
+ int socket_id, unsigned flags,
+ unsigned int elem_sz);
/**
* De-allocate all memory used by the ring.
*
@@ -354,25 +357,50 @@ void rte_common_ring_dump(FILE *f, const struct rte_ring *r);
* Placed here since identical code needed in both
* single and multi producer enqueue functions */
#define ENQUEUE_PTRS() do { \
+ void * const *objs = obj_table; \
const uint32_t size = r->prod.size; \
uint32_t idx = prod_head & mask; \
if (likely(idx + n < size)) { \
for (i = 0; i < (n & ((~(unsigned)0x3))); i+=4, idx+=4) { \
- r->ring[idx] = obj_table[i]; \
- r->ring[idx+1] = obj_table[i+1]; \
- r->ring[idx+2] = obj_table[i+2]; \
- r->ring[idx+3] = obj_table[i+3]; \
+ r->ring[idx] = objs[i]; \
+ r->ring[idx+1] = objs[i+1]; \
+ r->ring[idx+2] = objs[i+2]; \
+ r->ring[idx+3] = objs[i+3]; \
} \
switch (n & 0x3) { \
- case 3: r->ring[idx++] = obj_table[i++]; \
- case 2: r->ring[idx++] = obj_table[i++]; \
- case 1: r->ring[idx++] = obj_table[i++]; \
+ case 3: r->ring[idx++] = objs[i++]; \
+ case 2: r->ring[idx++] = objs[i++]; \
+ case 1: r->ring[idx++] = objs[i++]; \
} \
} else { \
for (i = 0; idx < size; i++, idx++)\
- r->ring[idx] = obj_table[i]; \
+ r->ring[idx] = objs[i]; \
for (idx = 0; i < n; i++, idx++) \
- r->ring[idx] = obj_table[i]; \
+ r->ring[idx] = objs[i]; \
+ } \
+} while(0)
+#define ENQUEUE_16B() do { \
+ rte_xmm_t *ring = (void *)r->ring; \
+ const rte_xmm_t *objs = obj_table; \
+ const uint32_t size = r->prod.size; \
+ uint32_t idx = prod_head & mask; \
+ if (likely(idx + n < size)) { \
+ for (i = 0; i < (n & ((~(unsigned)0x3))); i+=4, idx+=4) { \
+ ring[idx] = objs[i]; \
+ ring[idx+1] = objs[i+1]; \
+ ring[idx+2] = objs[i+2]; \
+ ring[idx+3] = objs[i+3]; \
+ } \
+ switch (n & 0x3) { \
+ case 3: ring[idx++] = objs[i++]; \
+ case 2: ring[idx++] = objs[i++]; \
+ case 1: ring[idx++] = objs[i++]; \
+ } \
+ } else { \
+ for (i = 0; idx < size; i++, idx++)\
+ ring[idx] = objs[i]; \
+ for (idx = 0; i < n; i++, idx++) \
+ ring[idx] = objs[i]; \
} \
} while(0)
@@ -380,25 +408,50 @@ void rte_common_ring_dump(FILE *f, const struct rte_ring *r);
* Placed here since identical code needed in both
* single and multi consumer dequeue functions */
#define DEQUEUE_PTRS() do { \
+ void **objs = obj_table; \
+ uint32_t idx = cons_head & mask; \
+ const uint32_t size = r->cons.size; \
+ if (likely(idx + n < size)) { \
+ for (i = 0; i < (n & (~(unsigned)0x3)); i+=4, idx+=4) {\
+ objs[i] = r->ring[idx]; \
+ objs[i+1] = r->ring[idx+1]; \
+ objs[i+2] = r->ring[idx+2]; \
+ objs[i+3] = r->ring[idx+3]; \
+ } \
+ switch (n & 0x3) { \
+ case 3: objs[i++] = r->ring[idx++]; \
+ case 2: objs[i++] = r->ring[idx++]; \
+ case 1: objs[i++] = r->ring[idx++]; \
+ } \
+ } else { \
+ for (i = 0; idx < size; i++, idx++) \
+ objs[i] = r->ring[idx]; \
+ for (idx = 0; i < n; i++, idx++) \
+ objs[i] = r->ring[idx]; \
+ } \
+} while (0)
+#define DEQUEUE_16B() do { \
+ rte_xmm_t *ring = (void *)r->ring; \
+ rte_xmm_t *objs = obj_table; \
uint32_t idx = cons_head & mask; \
const uint32_t size = r->cons.size; \
if (likely(idx + n < size)) { \
for (i = 0; i < (n & (~(unsigned)0x3)); i+=4, idx+=4) {\
- obj_table[i] = r->ring[idx]; \
- obj_table[i+1] = r->ring[idx+1]; \
- obj_table[i+2] = r->ring[idx+2]; \
- obj_table[i+3] = r->ring[idx+3]; \
+ objs[i] = ring[idx]; \
+ objs[i+1] = ring[idx+1]; \
+ objs[i+2] = ring[idx+2]; \
+ objs[i+3] = ring[idx+3]; \
} \
switch (n & 0x3) { \
- case 3: obj_table[i++] = r->ring[idx++]; \
- case 2: obj_table[i++] = r->ring[idx++]; \
- case 1: obj_table[i++] = r->ring[idx++]; \
+ case 3: objs[i++] = ring[idx++]; \
+ case 2: objs[i++] = ring[idx++]; \
+ case 1: objs[i++] = ring[idx++]; \
} \
} else { \
for (i = 0; idx < size; i++, idx++) \
- obj_table[i] = r->ring[idx]; \
+ objs[i] = ring[idx]; \
for (idx = 0; i < n; i++, idx++) \
- obj_table[i] = r->ring[idx]; \
+ objs[i] = ring[idx]; \
} \
} while (0)
@@ -428,8 +481,9 @@ void rte_common_ring_dump(FILE *f, const struct rte_ring *r);
* - n: Actual number of objects enqueued.
*/
static inline int __attribute__((always_inline))
-__rte_common_ring_mp_do_enqueue(struct rte_ring *r, void * const *obj_table,
- unsigned n, enum rte_ring_queue_behavior behavior)
+__rte_common_ring_mp_do_enqueue(struct rte_ring *r, const void *obj_table,
+ unsigned n, unsigned int elem_sz,
+ enum rte_ring_queue_behavior behavior)
{
uint32_t prod_head, prod_next;
uint32_t cons_tail, free_entries;
@@ -480,7 +534,10 @@ __rte_common_ring_mp_do_enqueue(struct rte_ring *r, void * const *obj_table,
} while (unlikely(success == 0));
/* write entries in ring */
- ENQUEUE_PTRS();
+ switch (elem_sz) {
+ case sizeof(void *): ENQUEUE_PTRS(); break;
+ case sizeof(rte_xmm_t): ENQUEUE_16B(); break;
+ }
rte_smp_wmb();
/* if we exceed the watermark */
@@ -537,8 +594,9 @@ __rte_common_ring_mp_do_enqueue(struct rte_ring *r, void * const *obj_table,
* - n: Actual number of objects enqueued.
*/
static inline int __attribute__((always_inline))
-__rte_common_ring_sp_do_enqueue(struct rte_ring *r, void * const *obj_table,
- unsigned n, enum rte_ring_queue_behavior behavior)
+__rte_common_ring_sp_do_enqueue(struct rte_ring *r, const void *obj_table,
+ unsigned n, unsigned int elem_sz,
+ enum rte_ring_queue_behavior behavior)
{
uint32_t prod_head, cons_tail;
uint32_t prod_next, free_entries;
@@ -575,7 +633,10 @@ __rte_common_ring_sp_do_enqueue(struct rte_ring *r, void * const *obj_table,
r->prod.head = prod_next;
/* write entries in ring */
- ENQUEUE_PTRS();
+ switch (elem_sz) {
+ case sizeof(void *): ENQUEUE_PTRS(); break;
+ case sizeof(rte_xmm_t): ENQUEUE_16B(); break;
+ }
rte_smp_wmb();
/* if we exceed the watermark */
@@ -621,8 +682,9 @@ __rte_common_ring_sp_do_enqueue(struct rte_ring *r, void * const *obj_table,
*/
static inline int __attribute__((always_inline))
-__rte_common_ring_mc_do_dequeue(struct rte_ring *r, void **obj_table,
- unsigned n, enum rte_ring_queue_behavior behavior)
+__rte_common_ring_mc_do_dequeue(struct rte_ring *r, void *obj_table,
+ unsigned n, unsigned int elem_sz,
+ enum rte_ring_queue_behavior behavior)
{
uint32_t cons_head, prod_tail;
uint32_t cons_next, entries;
@@ -671,7 +733,10 @@ __rte_common_ring_mc_do_dequeue(struct rte_ring *r, void **obj_table,
} while (unlikely(success == 0));
/* copy in table */
- DEQUEUE_PTRS();
+ switch (elem_sz) {
+ case sizeof(void *): DEQUEUE_PTRS(); break;
+ case sizeof(rte_xmm_t): DEQUEUE_16B(); break;
+ }
rte_smp_rmb();
/*
@@ -720,8 +785,9 @@ __rte_common_ring_mc_do_dequeue(struct rte_ring *r, void **obj_table,
* - n: Actual number of objects dequeued.
*/
static inline int __attribute__((always_inline))
-__rte_common_ring_sc_do_dequeue(struct rte_ring *r, void **obj_table,
- unsigned n, enum rte_ring_queue_behavior behavior)
+__rte_common_ring_sc_do_dequeue(struct rte_ring *r, void *obj_table,
+ unsigned n, unsigned int elem_sz,
+ enum rte_ring_queue_behavior behavior)
{
uint32_t cons_head, prod_tail;
uint32_t cons_next, entries;
@@ -755,7 +821,10 @@ __rte_common_ring_sc_do_dequeue(struct rte_ring *r, void **obj_table,
r->cons.head = cons_next;
/* copy in table */
- DEQUEUE_PTRS();
+ switch (elem_sz) {
+ case sizeof(void *): DEQUEUE_PTRS(); break;
+ case sizeof(rte_xmm_t): DEQUEUE_16B(); break;
+ }
rte_smp_rmb();
__RING_STAT_ADD(r, deq_success, n);
diff --git a/lib/librte_ring/rte_ring.c b/lib/librte_ring/rte_ring.c
index 16ddc39..0fd20f3 100644
--- a/lib/librte_ring/rte_ring.c
+++ b/lib/librte_ring/rte_ring.c
@@ -36,14 +36,14 @@
ssize_t
rte_ring_get_memsize(unsigned count)
{
- return rte_common_ring_get_memsize(count);
+ return rte_common_ring_get_memsize(count, sizeof(void *));
}
int
rte_ring_init(struct rte_ring *r, const char *name, unsigned count,
unsigned flags)
{
- return rte_common_ring_init(r, name, count, flags);
+ return rte_common_ring_init(r, name, count, flags, sizeof(void *));
}
@@ -51,7 +51,8 @@ struct rte_ring *
rte_ring_create(const char *name, unsigned count, int socket_id,
unsigned flags)
{
- return rte_common_ring_create(name, count, socket_id, flags);
+ return rte_common_ring_create(name, count, socket_id, flags,
+ sizeof(void *));
}
void
diff --git a/lib/librte_ring/rte_ring.h b/lib/librte_ring/rte_ring.h
index 993796f..3b934da 100644
--- a/lib/librte_ring/rte_ring.h
+++ b/lib/librte_ring/rte_ring.h
@@ -215,7 +215,7 @@ static inline int __attribute__((always_inline))
rte_ring_mp_enqueue_bulk(struct rte_ring *r, void * const *obj_table,
unsigned n)
{
- return __rte_common_ring_mp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
+ return __rte_common_ring_mp_do_enqueue(r, obj_table, n, sizeof(void *), RTE_RING_QUEUE_FIXED);
}
/**
@@ -237,7 +237,7 @@ static inline int __attribute__((always_inline))
rte_ring_sp_enqueue_bulk(struct rte_ring *r, void * const *obj_table,
unsigned n)
{
- return __rte_common_ring_sp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
+ return __rte_common_ring_sp_do_enqueue(r, obj_table, n, sizeof(void *), RTE_RING_QUEUE_FIXED);
}
/**
@@ -356,7 +356,7 @@ rte_ring_enqueue(struct rte_ring *r, void *obj)
static inline int __attribute__((always_inline))
rte_ring_mc_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n)
{
- return __rte_common_ring_mc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
+ return __rte_common_ring_mc_do_dequeue(r, obj_table, n, sizeof(void *), RTE_RING_QUEUE_FIXED);
}
/**
@@ -377,7 +377,7 @@ rte_ring_mc_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n)
static inline int __attribute__((always_inline))
rte_ring_sc_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned n)
{
- return __rte_common_ring_sc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_FIXED);
+ return __rte_common_ring_sc_do_dequeue(r, obj_table, n, sizeof(void *), RTE_RING_QUEUE_FIXED);
}
/**
@@ -568,7 +568,7 @@ static inline unsigned __attribute__((always_inline))
rte_ring_mp_enqueue_burst(struct rte_ring *r, void * const *obj_table,
unsigned n)
{
- return __rte_common_ring_mp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
+ return __rte_common_ring_mp_do_enqueue(r, obj_table, n, sizeof(void *), RTE_RING_QUEUE_VARIABLE);
}
/**
@@ -587,7 +587,7 @@ static inline unsigned __attribute__((always_inline))
rte_ring_sp_enqueue_burst(struct rte_ring *r, void * const *obj_table,
unsigned n)
{
- return __rte_common_ring_sp_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
+ return __rte_common_ring_sp_do_enqueue(r, obj_table, n, sizeof(void *), RTE_RING_QUEUE_VARIABLE);
}
/**
@@ -636,7 +636,7 @@ rte_ring_enqueue_burst(struct rte_ring *r, void * const *obj_table,
static inline unsigned __attribute__((always_inline))
rte_ring_mc_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n)
{
- return __rte_common_ring_mc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
+ return __rte_common_ring_mc_do_dequeue(r, obj_table, n, sizeof(void *), RTE_RING_QUEUE_VARIABLE);
}
/**
@@ -656,7 +656,7 @@ rte_ring_mc_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n)
static inline unsigned __attribute__((always_inline))
rte_ring_sc_dequeue_burst(struct rte_ring *r, void **obj_table, unsigned n)
{
- return __rte_common_ring_sc_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE);
+ return __rte_common_ring_sc_do_dequeue(r, obj_table, n, sizeof(void *), RTE_RING_QUEUE_VARIABLE);
}
/**
--
2.9.3
^ permalink raw reply [flat|nested] 5+ messages in thread
* [dpdk-dev] [PATCH RFCv2 4/4] ring: add new event ring class
2017-01-24 10:39 [dpdk-dev] [PATCH RFCv2 0/4] generalise rte_ring to allow different datatypes Bruce Richardson
` (2 preceding siblings ...)
2017-01-24 10:39 ` [dpdk-dev] [PATCH RFCv2 3/4] ring: allow common ring to use 8 or 16 byte values Bruce Richardson
@ 2017-01-24 10:39 ` Bruce Richardson
3 siblings, 0 replies; 5+ messages in thread
From: Bruce Richardson @ 2017-01-24 10:39 UTC (permalink / raw)
To: olivier.matz; +Cc: dev, Bruce Richardson
using a placeholder struct type of the correct (16B) size, create an
event ring type and unit test it to verify the generalized ring
implementation works.
Signed-off-by: Bruce Richardson <bruce.richardson@intel.com>
---
app/test/Makefile | 1 +
app/test/test_event_ring.c | 83 +++++
lib/librte_ring/Makefile | 2 +
lib/librte_ring/rte_event_ring.c | 87 +++++
lib/librte_ring/rte_event_ring.h | 696 +++++++++++++++++++++++++++++++++++++++
5 files changed, 869 insertions(+)
create mode 100644 app/test/test_event_ring.c
create mode 100644 lib/librte_ring/rte_event_ring.c
create mode 100644 lib/librte_ring/rte_event_ring.h
diff --git a/app/test/Makefile b/app/test/Makefile
index 9de301f..c0ab2ba 100644
--- a/app/test/Makefile
+++ b/app/test/Makefile
@@ -96,6 +96,7 @@ SRCS-y += test_memory.c
SRCS-y += test_memzone.c
SRCS-y += test_ring.c
+SRCS-y += test_event_ring.c
SRCS-y += test_ring_perf.c
SRCS-y += test_pmd_perf.c
diff --git a/app/test/test_event_ring.c b/app/test/test_event_ring.c
new file mode 100644
index 0000000..94f85cd
--- /dev/null
+++ b/app/test/test_event_ring.c
@@ -0,0 +1,83 @@
+/*-
+ * BSD LICENSE
+ *
+ * Copyright(c) 2017 Intel Corporation. All rights reserved.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ * * 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.
+ * * Neither the name of Intel Corporation nor the names of its
+ * contributors may be used to endorse or promote products derived
+ * from this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS 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 COPYRIGHT
+ * OWNER 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.
+ */
+
+#include <rte_random.h>
+#include <rte_event_ring.h>
+#include "test.h"
+
+#define BURST_SIZE 24 /* not a power of two so we can test wrap around better */
+#define RING_SIZE 128
+#define ENQ_ITERATIONS 48
+
+#define ERR_OUT() do { \
+ printf("Error %s:%d\n", __FILE__, __LINE__); \
+ return -1; \
+} while (0)
+
+static int
+test_event_ring(void)
+{
+ struct rte_event in_events[BURST_SIZE];
+ struct rte_event out_events[BURST_SIZE];
+ unsigned int i;
+
+ struct rte_ring *ering = rte_event_ring_create("TEST_RING", RING_SIZE,
+ rte_socket_id(), RING_F_SP_ENQ|RING_F_SC_DEQ);
+ if (ering == NULL)
+ ERR_OUT();
+
+ for (i = 0; i < BURST_SIZE; i++)
+ in_events[i].event_metadata = rte_rand();
+
+ for (i = 0; i < ENQ_ITERATIONS; i++)
+ if (rte_event_ring_enqueue_bulk(ering, in_events,
+ RTE_DIM(in_events)) != 0) {
+ unsigned j;
+
+ if (rte_event_ring_dequeue_burst(ering, out_events,
+ RTE_DIM(out_events)) != RTE_DIM(out_events))
+ ERR_OUT();
+ for (j = 0; j < RTE_DIM(out_events); j++)
+ if (out_events[j].event_metadata !=
+ in_events[j].event_metadata)
+ ERR_OUT();
+ /* retry, now that we've made space */
+ if (rte_event_ring_enqueue_bulk(ering, in_events,
+ RTE_DIM(in_events)) != 0)
+ ERR_OUT();
+ }
+
+ return 0;
+}
+
+REGISTER_TEST_COMMAND(event_ring_autotest, test_event_ring);
diff --git a/lib/librte_ring/Makefile b/lib/librte_ring/Makefile
index 5cebb29..fb1c9e5 100644
--- a/lib/librte_ring/Makefile
+++ b/lib/librte_ring/Makefile
@@ -43,10 +43,12 @@ LIBABIVER := 1
# all source are stored in SRCS-y
SRCS-$(CONFIG_RTE_LIBRTE_RING) += rte_common_ring.c
SRCS-$(CONFIG_RTE_LIBRTE_RING) += rte_ring.c
+SRCS-$(CONFIG_RTE_LIBRTE_RING) += rte_event_ring.c
# install includes
SYMLINK-$(CONFIG_RTE_LIBRTE_RING)-include += rte_ring.h
SYMLINK-$(CONFIG_RTE_LIBRTE_RING)-include += rte_common_ring.h
+SYMLINK-$(CONFIG_RTE_LIBRTE_RING)-include += rte_event_ring.h
DEPDIRS-$(CONFIG_RTE_LIBRTE_RING) += lib/librte_eal
diff --git a/lib/librte_ring/rte_event_ring.c b/lib/librte_ring/rte_event_ring.c
new file mode 100644
index 0000000..db053bc
--- /dev/null
+++ b/lib/librte_ring/rte_event_ring.c
@@ -0,0 +1,87 @@
+/*-
+ * BSD LICENSE
+ *
+ * Copyright(c) 2017 Intel Corporation. All rights reserved.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ * * 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.
+ * * Neither the name of Intel Corporation nor the names of its
+ * contributors may be used to endorse or promote products derived
+ * from this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS 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 COPYRIGHT
+ * OWNER 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.
+ */
+
+#include "rte_event_ring.h"
+
+ssize_t
+rte_event_ring_get_memsize(unsigned count)
+{
+ return rte_common_ring_get_memsize(count, sizeof(struct rte_event));
+}
+
+int
+rte_event_ring_init(struct rte_ring *r, const char *name, unsigned count,
+ unsigned flags)
+{
+ return rte_common_ring_init(r, name, count, flags, sizeof(struct rte_event));
+}
+
+
+struct rte_ring *
+rte_event_ring_create(const char *name, unsigned count, int socket_id,
+ unsigned flags)
+{
+ return rte_common_ring_create(name, count, socket_id, flags,
+ sizeof(struct rte_event));
+}
+
+void
+rte_event_ring_free(struct rte_ring *r)
+{
+ return rte_common_ring_free(r);
+}
+
+int
+rte_event_ring_set_water_mark(struct rte_ring *r, unsigned count)
+{
+ return rte_common_ring_set_water_mark(r, count);
+}
+
+void
+rte_event_ring_dump(FILE *f, const struct rte_ring *r)
+{
+ return rte_common_ring_dump(f, r);
+}
+
+void
+rte_event_ring_list_dump(FILE *f)
+{
+ rte_common_ring_list_dump(f);
+}
+
+struct rte_ring *
+rte_event_ring_lookup(const char *name)
+{
+ return rte_common_ring_lookup(name);
+}
+
diff --git a/lib/librte_ring/rte_event_ring.h b/lib/librte_ring/rte_event_ring.h
new file mode 100644
index 0000000..a175800
--- /dev/null
+++ b/lib/librte_ring/rte_event_ring.h
@@ -0,0 +1,696 @@
+/*-
+ * BSD LICENSE
+ *
+ * Copyright(c) 2017 Intel Corporation. All rights reserved.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ * * 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.
+ * * Neither the name of Intel Corporation nor the names of its
+ * contributors may be used to endorse or promote products derived
+ * from this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS 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 COPYRIGHT
+ * OWNER 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.
+ */
+
+#ifndef _RTE_EVENT_RING_H_
+#define _RTE_EVENT_RING_H_
+
+/**
+ * @file
+ * RTE Event Ring
+ *
+ * The Ring Manager is a fixed-size event queue, implemented as a table of
+ * events. Head and tail pointers are modified atomically, allowing
+ * concurrent access to it. It has the following features:
+ *
+ * - FIFO (First In First Out)
+ * - Maximum size is fixed; the pointers are stored in a table.
+ * - Lockless implementation.
+ * - Multi- or single-consumer dequeue.
+ * - Multi- or single-producer enqueue.
+ * - Bulk dequeue.
+ * - Bulk enqueue.
+ *
+ * Note: the ring implementation is not preemptable. A lcore must not
+ * be interrupted by another task that uses the same ring.
+ *
+ */
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include "rte_common_ring.h"
+
+struct rte_event {
+ uint64_t event_metadata;
+ struct rte_mbuf *mbuf;
+};
+
+/**
+ * Calculate the memory size needed for a ring
+ *
+ * This function returns the number of bytes needed for a ring, given
+ * the number of elements in it. This value is the sum of the size of
+ * the structure rte_ring and the size of the memory needed by the
+ * objects pointers. The value is aligned to a cache line size.
+ *
+ * @param count
+ * The number of elements in the ring (must be a power of 2).
+ * @return
+ * - The memory size needed for the ring on success.
+ * - -EINVAL if count is not a power of 2.
+ */
+ssize_t rte_event_ring_get_memsize(unsigned count);
+
+/**
+ * Initialize a ring structure.
+ *
+ * Initialize a ring structure in memory pointed by "r". The size of the
+ * memory area must be large enough to store the ring structure and the
+ * object table. It is advised to use rte_event_ring_get_memsize() to get the
+ * appropriate size.
+ *
+ * The ring size is set to *count*, which must be a power of two. Water
+ * marking is disabled by default. The real usable ring size is
+ * *count-1* instead of *count* to differentiate a free ring from an
+ * empty ring.
+ *
+ * The ring is not added in RTE_TAILQ_RING global list. Indeed, the
+ * memory given by the caller may not be shareable among dpdk
+ * processes.
+ *
+ * @param r
+ * The pointer to the ring structure followed by the objects table.
+ * @param name
+ * The name of the ring.
+ * @param count
+ * The number of elements in the ring (must be a power of 2).
+ * @param flags
+ * An OR of the following:
+ * - RING_F_SP_ENQ: If this flag is set, the default behavior when
+ * using ``rte_event_ring_enqueue()`` or ``rte_event_ring_enqueue_bulk()``
+ * is "single-producer". Otherwise, it is "multi-producers".
+ * - RING_F_SC_DEQ: If this flag is set, the default behavior when
+ * using ``rte_event_ring_dequeue()`` or ``rte_event_ring_dequeue_bulk()``
+ * is "single-consumer". Otherwise, it is "multi-consumers".
+ * @return
+ * 0 on success, or a negative value on error.
+ */
+int rte_event_ring_init(struct rte_ring *r, const char *name, unsigned count,
+ unsigned flags);
+
+/**
+ * Create a new ring named *name* in memory.
+ *
+ * This function uses ``memzone_reserve()`` to allocate memory. Then it
+ * calls rte_event_ring_init() to initialize an empty ring.
+ *
+ * The new ring size is set to *count*, which must be a power of
+ * two. Water marking is disabled by default. The real usable ring size
+ * is *count-1* instead of *count* to differentiate a free ring from an
+ * empty ring.
+ *
+ * The ring is added in RTE_TAILQ_RING list.
+ *
+ * @param name
+ * The name of the ring.
+ * @param count
+ * The size of the ring (must be a power of 2).
+ * @param socket_id
+ * The *socket_id* argument is the socket identifier in case of
+ * NUMA. The value can be *SOCKET_ID_ANY* if there is no NUMA
+ * constraint for the reserved zone.
+ * @param flags
+ * An OR of the following:
+ * - RING_F_SP_ENQ: If this flag is set, the default behavior when
+ * using ``rte_event_ring_enqueue()`` or ``rte_event_ring_enqueue_bulk()``
+ * is "single-producer". Otherwise, it is "multi-producers".
+ * - RING_F_SC_DEQ: If this flag is set, the default behavior when
+ * using ``rte_event_ring_dequeue()`` or ``rte_event_ring_dequeue_bulk()``
+ * is "single-consumer". Otherwise, it is "multi-consumers".
+ * @return
+ * On success, the pointer to the new allocated ring. NULL on error with
+ * rte_errno set appropriately. Possible errno values include:
+ * - E_RTE_NO_CONFIG - function could not get pointer to rte_config structure
+ * - E_RTE_SECONDARY - function was called from a secondary process instance
+ * - EINVAL - count provided is not a power of 2
+ * - ENOSPC - the maximum number of memzones has already been allocated
+ * - EEXIST - a memzone with the same name already exists
+ * - ENOMEM - no appropriate memory area found in which to create memzone
+ */
+struct rte_ring *rte_event_ring_create(const char *name, unsigned count,
+ int socket_id, unsigned flags);
+/**
+ * De-allocate all memory used by the ring.
+ *
+ * @param r
+ * Ring to free
+ */
+void rte_event_ring_free(struct rte_ring *r);
+
+/**
+ * Change the high water mark.
+ *
+ * If *count* is 0, water marking is disabled. Otherwise, it is set to the
+ * *count* value. The *count* value must be greater than 0 and less
+ * than the ring size.
+ *
+ * This function can be called at any time (not necessarily at
+ * initialization).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param count
+ * The new water mark value.
+ * @return
+ * - 0: Success; water mark changed.
+ * - -EINVAL: Invalid water mark value.
+ */
+int rte_event_ring_set_water_mark(struct rte_ring *r, unsigned count);
+
+/**
+ * Dump the status of the ring to a file.
+ *
+ * @param f
+ * A pointer to a file for output
+ * @param r
+ * A pointer to the ring structure.
+ */
+void rte_event_ring_dump(FILE *f, const struct rte_ring *r);
+
+/**
+ * Enqueue several objects on the ring (multi-producers safe).
+ *
+ * This function uses a "compare and set" instruction to move the
+ * producer index atomically.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects).
+ * @param n
+ * The number of objects to add in the ring from the obj_table.
+ * @return
+ * - 0: Success; objects enqueue.
+ * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
+ * high water mark is exceeded.
+ * - -ENOBUFS: Not enough room in the ring to enqueue, no object is enqueued.
+ */
+static inline int __attribute__((always_inline))
+rte_event_ring_mp_enqueue_bulk(struct rte_ring *r, const struct rte_event * const obj_table,
+ unsigned n)
+{
+ return __rte_common_ring_mp_do_enqueue(r, obj_table, n, sizeof(struct rte_event), RTE_RING_QUEUE_FIXED);
+}
+
+/**
+ * Enqueue several objects on a ring (NOT multi-producers safe).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects).
+ * @param n
+ * The number of objects to add in the ring from the obj_table.
+ * @return
+ * - 0: Success; objects enqueued.
+ * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
+ * high water mark is exceeded.
+ * - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
+ */
+static inline int __attribute__((always_inline))
+rte_event_ring_sp_enqueue_bulk(struct rte_ring *r, const struct rte_event * const obj_table,
+ unsigned n)
+{
+ return __rte_common_ring_sp_do_enqueue(r, obj_table, n, sizeof(struct rte_event), RTE_RING_QUEUE_FIXED);
+}
+
+/**
+ * Enqueue several objects on a ring.
+ *
+ * This function calls the multi-producer or the single-producer
+ * version depending on the default behavior that was specified at
+ * ring creation time (see flags).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects).
+ * @param n
+ * The number of objects to add in the ring from the obj_table.
+ * @return
+ * - 0: Success; objects enqueued.
+ * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
+ * high water mark is exceeded.
+ * - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
+ */
+static inline int __attribute__((always_inline))
+rte_event_ring_enqueue_bulk(struct rte_ring *r, const struct rte_event * const obj_table,
+ unsigned n)
+{
+ if (r->prod.sp_enqueue)
+ return rte_event_ring_sp_enqueue_bulk(r, obj_table, n);
+ else
+ return rte_event_ring_mp_enqueue_bulk(r, obj_table, n);
+}
+
+/**
+ * Enqueue one object on a ring (multi-producers safe).
+ *
+ * This function uses a "compare and set" instruction to move the
+ * producer index atomically.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj
+ * A pointer to the object to be added.
+ * @return
+ * - 0: Success; objects enqueued.
+ * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
+ * high water mark is exceeded.
+ * - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
+ */
+static inline int __attribute__((always_inline))
+rte_event_ring_mp_enqueue(struct rte_ring *r, const struct rte_event obj)
+{
+ return rte_event_ring_mp_enqueue_bulk(r, &obj, 1);
+}
+
+/**
+ * Enqueue one object on a ring (NOT multi-producers safe).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj
+ * A pointer to the object to be added.
+ * @return
+ * - 0: Success; objects enqueued.
+ * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
+ * high water mark is exceeded.
+ * - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
+ */
+static inline int __attribute__((always_inline))
+rte_event_ring_sp_enqueue(struct rte_ring *r, const struct rte_event obj)
+{
+ return rte_event_ring_sp_enqueue_bulk(r, &obj, 1);
+}
+
+/**
+ * Enqueue one object on a ring.
+ *
+ * This function calls the multi-producer or the single-producer
+ * version, depending on the default behaviour that was specified at
+ * ring creation time (see flags).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj
+ * A pointer to the object to be added.
+ * @return
+ * - 0: Success; objects enqueued.
+ * - -EDQUOT: Quota exceeded. The objects have been enqueued, but the
+ * high water mark is exceeded.
+ * - -ENOBUFS: Not enough room in the ring to enqueue; no object is enqueued.
+ */
+static inline int __attribute__((always_inline))
+rte_event_ring_enqueue(struct rte_ring *r, const struct rte_event obj)
+{
+ if (r->prod.sp_enqueue)
+ return rte_event_ring_sp_enqueue(r, obj);
+ else
+ return rte_event_ring_mp_enqueue(r, obj);
+}
+
+/**
+ * Dequeue several objects from a ring (multi-consumers safe).
+ *
+ * This function uses a "compare and set" instruction to move the
+ * consumer index atomically.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects) that will be filled.
+ * @param n
+ * The number of objects to dequeue from the ring to the obj_table.
+ * @return
+ * - 0: Success; objects dequeued.
+ * - -ENOENT: Not enough entries in the ring to dequeue; no object is
+ * dequeued.
+ */
+static inline int __attribute__((always_inline))
+rte_event_ring_mc_dequeue_bulk(struct rte_ring *r, struct rte_event * const obj_table, unsigned n)
+{
+ return __rte_common_ring_mc_do_dequeue(r, obj_table, n, sizeof(struct rte_event), RTE_RING_QUEUE_FIXED);
+}
+
+/**
+ * Dequeue several objects from a ring (NOT multi-consumers safe).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects) that will be filled.
+ * @param n
+ * The number of objects to dequeue from the ring to the obj_table,
+ * must be strictly positive.
+ * @return
+ * - 0: Success; objects dequeued.
+ * - -ENOENT: Not enough entries in the ring to dequeue; no object is
+ * dequeued.
+ */
+static inline int __attribute__((always_inline))
+rte_event_ring_sc_dequeue_bulk(struct rte_ring *r, struct rte_event * const obj_table, unsigned n)
+{
+ return __rte_common_ring_sc_do_dequeue(r, obj_table, n, sizeof(struct rte_event), RTE_RING_QUEUE_FIXED);
+}
+
+/**
+ * Dequeue several objects from a ring.
+ *
+ * This function calls the multi-consumers or the single-consumer
+ * version, depending on the default behaviour that was specified at
+ * ring creation time (see flags).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects) that will be filled.
+ * @param n
+ * The number of objects to dequeue from the ring to the obj_table.
+ * @return
+ * - 0: Success; objects dequeued.
+ * - -ENOENT: Not enough entries in the ring to dequeue, no object is
+ * dequeued.
+ */
+static inline int __attribute__((always_inline))
+rte_event_ring_dequeue_bulk(struct rte_ring *r, struct rte_event * const obj_table, unsigned n)
+{
+ if (r->cons.sc_dequeue)
+ return rte_event_ring_sc_dequeue_bulk(r, obj_table, n);
+ else
+ return rte_event_ring_mc_dequeue_bulk(r, obj_table, n);
+}
+
+/**
+ * Dequeue one object from a ring (multi-consumers safe).
+ *
+ * This function uses a "compare and set" instruction to move the
+ * consumer index atomically.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_p
+ * A pointer to a void * pointer (object) that will be filled.
+ * @return
+ * - 0: Success; objects dequeued.
+ * - -ENOENT: Not enough entries in the ring to dequeue; no object is
+ * dequeued.
+ */
+static inline int __attribute__((always_inline))
+rte_event_ring_mc_dequeue(struct rte_ring *r, struct rte_event *obj_p)
+{
+ return rte_event_ring_mc_dequeue_bulk(r, obj_p, 1);
+}
+
+/**
+ * Dequeue one object from a ring (NOT multi-consumers safe).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_p
+ * A pointer to a void * pointer (object) that will be filled.
+ * @return
+ * - 0: Success; objects dequeued.
+ * - -ENOENT: Not enough entries in the ring to dequeue, no object is
+ * dequeued.
+ */
+static inline int __attribute__((always_inline))
+rte_event_ring_sc_dequeue(struct rte_ring *r, struct rte_event *obj_p)
+{
+ return rte_event_ring_sc_dequeue_bulk(r, obj_p, 1);
+}
+
+/**
+ * Dequeue one object from a ring.
+ *
+ * This function calls the multi-consumers or the single-consumer
+ * version depending on the default behaviour that was specified at
+ * ring creation time (see flags).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_p
+ * A pointer to a void * pointer (object) that will be filled.
+ * @return
+ * - 0: Success, objects dequeued.
+ * - -ENOENT: Not enough entries in the ring to dequeue, no object is
+ * dequeued.
+ */
+static inline int __attribute__((always_inline))
+rte_event_ring_dequeue(struct rte_ring *r, struct rte_event *obj_p)
+{
+ if (r->cons.sc_dequeue)
+ return rte_event_ring_sc_dequeue(r, obj_p);
+ else
+ return rte_event_ring_mc_dequeue(r, obj_p);
+}
+
+/**
+ * Test if a ring is full.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @return
+ * - 1: The ring is full.
+ * - 0: The ring is not full.
+ */
+static inline int
+rte_event_ring_full(const struct rte_ring *r)
+{
+ return rte_common_ring_full(r);
+}
+
+/**
+ * Test if a ring is empty.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @return
+ * - 1: The ring is empty.
+ * - 0: The ring is not empty.
+ */
+static inline int
+rte_event_ring_empty(const struct rte_ring *r)
+{
+ return rte_common_ring_empty(r);
+}
+
+/**
+ * Return the number of entries in a ring.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @return
+ * The number of entries in the ring.
+ */
+static inline unsigned
+rte_event_ring_count(const struct rte_ring *r)
+{
+ return rte_common_ring_count(r);
+}
+
+/**
+ * Return the number of free entries in a ring.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @return
+ * The number of free entries in the ring.
+ */
+static inline unsigned
+rte_event_ring_free_count(const struct rte_ring *r)
+{
+ return rte_common_ring_free_count(r);
+}
+
+/**
+ * Dump the status of all rings on the console
+ *
+ * @param f
+ * A pointer to a file for output
+ */
+void rte_event_ring_list_dump(FILE *f);
+
+/**
+ * Search a ring from its name
+ *
+ * @param name
+ * The name of the ring.
+ * @return
+ * The pointer to the ring matching the name, or NULL if not found,
+ * with rte_errno set appropriately. Possible rte_errno values include:
+ * - ENOENT - required entry not available to return.
+ */
+struct rte_ring *rte_event_ring_lookup(const char *name);
+
+/**
+ * Enqueue several objects on the ring (multi-producers safe).
+ *
+ * This function uses a "compare and set" instruction to move the
+ * producer index atomically.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects).
+ * @param n
+ * The number of objects to add in the ring from the obj_table.
+ * @return
+ * - n: Actual number of objects enqueued.
+ */
+static inline unsigned __attribute__((always_inline))
+rte_event_ring_mp_enqueue_burst(struct rte_ring *r, const struct rte_event * const obj_table,
+ unsigned n)
+{
+ return __rte_common_ring_mp_do_enqueue(r, obj_table, n, sizeof(struct rte_event), RTE_RING_QUEUE_VARIABLE);
+}
+
+/**
+ * Enqueue several objects on a ring (NOT multi-producers safe).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects).
+ * @param n
+ * The number of objects to add in the ring from the obj_table.
+ * @return
+ * - n: Actual number of objects enqueued.
+ */
+static inline unsigned __attribute__((always_inline))
+rte_event_ring_sp_enqueue_burst(struct rte_ring *r, const struct rte_event * const obj_table,
+ unsigned n)
+{
+ return __rte_common_ring_sp_do_enqueue(r, obj_table, n, sizeof(struct rte_event), RTE_RING_QUEUE_VARIABLE);
+}
+
+/**
+ * Enqueue several objects on a ring.
+ *
+ * This function calls the multi-producer or the single-producer
+ * version depending on the default behavior that was specified at
+ * ring creation time (see flags).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects).
+ * @param n
+ * The number of objects to add in the ring from the obj_table.
+ * @return
+ * - n: Actual number of objects enqueued.
+ */
+static inline unsigned __attribute__((always_inline))
+rte_event_ring_enqueue_burst(struct rte_ring *r, const struct rte_event * const obj_table,
+ unsigned n)
+{
+ if (r->prod.sp_enqueue)
+ return rte_event_ring_sp_enqueue_burst(r, obj_table, n);
+ else
+ return rte_event_ring_mp_enqueue_burst(r, obj_table, n);
+}
+
+/**
+ * Dequeue several objects from a ring (multi-consumers safe). When the request
+ * objects are more than the available objects, only dequeue the actual number
+ * of objects
+ *
+ * This function uses a "compare and set" instruction to move the
+ * consumer index atomically.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects) that will be filled.
+ * @param n
+ * The number of objects to dequeue from the ring to the obj_table.
+ * @return
+ * - n: Actual number of objects dequeued, 0 if ring is empty
+ */
+static inline unsigned __attribute__((always_inline))
+rte_event_ring_mc_dequeue_burst(struct rte_ring *r, struct rte_event * const obj_table, unsigned n)
+{
+ return __rte_common_ring_mc_do_dequeue(r, obj_table, n, sizeof(struct rte_event), RTE_RING_QUEUE_VARIABLE);
+}
+
+/**
+ * Dequeue several objects from a ring (NOT multi-consumers safe).When the
+ * request objects are more than the available objects, only dequeue the
+ * actual number of objects
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects) that will be filled.
+ * @param n
+ * The number of objects to dequeue from the ring to the obj_table.
+ * @return
+ * - n: Actual number of objects dequeued, 0 if ring is empty
+ */
+static inline unsigned __attribute__((always_inline))
+rte_event_ring_sc_dequeue_burst(struct rte_ring *r, struct rte_event * const obj_table, unsigned n)
+{
+ return __rte_common_ring_sc_do_dequeue(r, obj_table, n, sizeof(struct rte_event), RTE_RING_QUEUE_VARIABLE);
+}
+
+/**
+ * Dequeue multiple objects from a ring up to a maximum number.
+ *
+ * This function calls the multi-consumers or the single-consumer
+ * version, depending on the default behaviour that was specified at
+ * ring creation time (see flags).
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @param obj_table
+ * A pointer to a table of void * pointers (objects) that will be filled.
+ * @param n
+ * The number of objects to dequeue from the ring to the obj_table.
+ * @return
+ * - Number of objects dequeued
+ */
+static inline unsigned __attribute__((always_inline))
+rte_event_ring_dequeue_burst(struct rte_ring *r, struct rte_event * const obj_table, unsigned n)
+{
+ if (r->cons.sc_dequeue)
+ return rte_event_ring_sc_dequeue_burst(r, obj_table, n);
+ else
+ return rte_event_ring_mc_dequeue_burst(r, obj_table, n);
+}
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* _RTE_EVENT_RING_H_ */
--
2.9.3
^ permalink raw reply [flat|nested] 5+ messages in thread
end of thread, other threads:[~2017-01-24 10:40 UTC | newest]
Thread overview: 5+ messages (download: mbox.gz / follow: Atom feed)
-- links below jump to the message on this page --
2017-01-24 10:39 [dpdk-dev] [PATCH RFCv2 0/4] generalise rte_ring to allow different datatypes Bruce Richardson
2017-01-24 10:39 ` [dpdk-dev] [PATCH RFCv2 1/4] ring: create common ring files Bruce Richardson
2017-01-24 10:39 ` [dpdk-dev] [PATCH RFCv2 2/4] ring: separate common and rte_ring specific functions Bruce Richardson
2017-01-24 10:39 ` [dpdk-dev] [PATCH RFCv2 3/4] ring: allow common ring to use 8 or 16 byte values Bruce Richardson
2017-01-24 10:39 ` [dpdk-dev] [PATCH RFCv2 4/4] ring: add new event ring class Bruce Richardson
This is a public inbox, see mirroring instructions
for how to clone and mirror all data and code used for this inbox;
as well as URLs for NNTP newsgroup(s).