-/*-
- * BSD LICENSE
- *
- * Copyright(c) 2010-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.
- */
-
-/*
- * Derived from FreeBSD's bufring.h
- *
- **************************************************************************
+/* SPDX-License-Identifier: BSD-3-Clause
*
+ * Copyright (c) 2010-2020 Intel Corporation
* 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.
- *
- ***************************************************************************/
+ * Derived from FreeBSD's bufring.h
+ * Used as BSD-3 Licensed with permission from Kip Macy.
+ */
#ifndef _RTE_RING_H_
#define _RTE_RING_H_
* - 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.
+ * Note: the ring implementation is not preemptible. Refer to Programmer's
+ * guide/Environment Abstraction Layer/Multiple pthread/Known Issues/rte_ring
+ * for more information.
*
*/
extern "C" {
#endif
-#include <stdio.h>
-#include <stdint.h>
-#include <sys/queue.h>
-#include <errno.h>
-#include <rte_common.h>
-#include <rte_config.h>
-#include <rte_memory.h>
-#include <rte_lcore.h>
-#include <rte_atomic.h>
-#include <rte_branch_prediction.h>
-#include <rte_memzone.h>
-#include <rte_pause.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 */
-};
-
-#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)
-
-struct rte_memzone; /* forward declaration, so as not to require memzone.h */
-
-#if RTE_CACHE_LINE_SIZE < 128
-#define PROD_ALIGN (RTE_CACHE_LINE_SIZE * 2)
-#define CONS_ALIGN (RTE_CACHE_LINE_SIZE * 2)
-#else
-#define PROD_ALIGN RTE_CACHE_LINE_SIZE
-#define CONS_ALIGN RTE_CACHE_LINE_SIZE
-#endif
-
-/* structure to hold a pair of head/tail values and other metadata */
-struct rte_ring_headtail {
- volatile uint32_t head; /**< Prod/consumer head. */
- volatile uint32_t tail; /**< Prod/consumer tail. */
- uint32_t single; /**< True if single prod/cons */
-};
-
-/**
- * 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] __rte_cache_aligned; /**< Name of the ring. */
- int flags; /**< Flags supplied at creation. */
- const struct rte_memzone *memzone;
- /**< Memzone, if any, containing the rte_ring */
- uint32_t size; /**< Size of ring. */
- uint32_t mask; /**< Mask (size-1) of ring. */
- uint32_t capacity; /**< Usable size of ring */
-
- /** Ring producer status. */
- struct rte_ring_headtail prod __rte_aligned(PROD_ALIGN);
-
- /** Ring consumer status. */
- struct rte_ring_headtail cons __rte_aligned(CONS_ALIGN);
-};
-
-#define RING_F_SP_ENQ 0x0001 /**< The default enqueue is "single-producer". */
-#define RING_F_SC_DEQ 0x0002 /**< The default dequeue is "single-consumer". */
-/**
- * Ring is to hold exactly requested number of entries.
- * Without this flag set, the ring size requested must be a power of 2, and the
- * usable space will be that size - 1. With the flag, the requested size will
- * be rounded up to the next power of two, but the usable space will be exactly
- * that requested. Worst case, if a power-of-2 size is requested, half the
- * ring space will be wasted.
- */
-#define RING_F_EXACT_SZ 0x0004
-#define RTE_RING_SZ_MASK (0x7fffffffU) /**< Ring size mask */
-
-/* @internal defines for passing to the enqueue dequeue worker functions */
-#define __IS_SP 1
-#define __IS_MP 0
-#define __IS_SC 1
-#define __IS_MC 0
+#include <rte_ring_core.h>
/**
* Calculate the memory size needed for a ring
* 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".
+ * - One of mutually exclusive flags that define producer behavior:
+ * - 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".
+ * - RING_F_MP_RTS_ENQ: If this flag is set, the default behavior when
+ * using ``rte_ring_enqueue()`` or ``rte_ring_enqueue_bulk()``
+ * is "multi-producer RTS mode".
+ * If none of these flags is set, then default "multi-producer"
+ * behavior is selected.
+ * - One of mutually exclusive flags that define consumer behavior:
+ * - 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".
+ * - RING_F_MC_RTS_DEQ: If this flag is set, the default behavior when
+ * using ``rte_ring_dequeue()`` or ``rte_ring_dequeue_bulk()``
+ * is "multi-consumer RTS mode".
+ * If none of these flags is set, then default "multi-consumer"
+ * behavior is selected.
* @return
* 0 on success, or a negative value on error.
*/
* 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".
+ * - One of mutually exclusive flags that define producer behavior:
+ * - 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".
+ * - RING_F_MP_RTS_ENQ: If this flag is set, the default behavior when
+ * using ``rte_ring_enqueue()`` or ``rte_ring_enqueue_bulk()``
+ * is "multi-producer RTS mode".
+ * If none of these flags is set, then default "multi-producer"
+ * behavior is selected.
+ * - One of mutually exclusive flags that define consumer behavior:
+ * - 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".
+ * - RING_F_MC_RTS_DEQ: If this flag is set, the default behavior when
+ * using ``rte_ring_dequeue()`` or ``rte_ring_dequeue_bulk()``
+ * is "multi-consumer RTS mode".
+ * If none of these flags is set, then default "multi-consumer"
+ * behavior is selected.
* @return
* On success, the pointer to the new allocated ring. NULL on error with
* rte_errno set appropriately. Possible errno values include:
*/
struct rte_ring *rte_ring_create(const char *name, unsigned count,
int socket_id, unsigned flags);
+
/**
* De-allocate all memory used by the ring.
*
} \
} while (0)
-/* Move common functions to generic file */
+/* Between load and load. there might be cpu reorder in weak model
+ * (powerpc/arm).
+ * There are 2 choices for the users
+ * 1.use rmb() memory barrier
+ * 2.use one-direction load_acquire/store_release barrier,defined by
+ * CONFIG_RTE_USE_C11_MEM_MODEL=y
+ * It depends on performance test results.
+ * By default, move common functions to rte_ring_generic.h
+ */
+#ifdef RTE_USE_C11_MEM_MODEL
+#include "rte_ring_c11_mem.h"
+#else
#include "rte_ring_generic.h"
+#endif
/**
* @internal Enqueue several objects on the ring
static __rte_always_inline unsigned int
__rte_ring_do_enqueue(struct rte_ring *r, void * const *obj_table,
unsigned int n, enum rte_ring_queue_behavior behavior,
- int is_sp, unsigned int *free_space)
+ unsigned int is_sp, unsigned int *free_space)
{
uint32_t prod_head, prod_next;
uint32_t free_entries;
static __rte_always_inline unsigned int
__rte_ring_do_dequeue(struct rte_ring *r, void **obj_table,
unsigned int n, enum rte_ring_queue_behavior behavior,
- int is_sc, unsigned int *available)
+ unsigned int is_sc, unsigned int *available)
{
uint32_t cons_head, cons_next;
uint32_t entries;
- n = __rte_ring_move_cons_head(r, is_sc, n, behavior,
+ n = __rte_ring_move_cons_head(r, (int)is_sc, n, behavior,
&cons_head, &cons_next, &entries);
if (n == 0)
goto end;
unsigned int n, unsigned int *free_space)
{
return __rte_ring_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_FIXED,
- __IS_MP, free_space);
+ RTE_RING_SYNC_MT, free_space);
}
/**
unsigned int n, unsigned int *free_space)
{
return __rte_ring_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_FIXED,
- __IS_SP, free_space);
+ RTE_RING_SYNC_ST, free_space);
}
+#ifdef ALLOW_EXPERIMENTAL_API
+#include <rte_ring_elem.h>
+#endif
+
/**
* Enqueue several objects on a ring.
*
rte_ring_enqueue_bulk(struct rte_ring *r, void * const *obj_table,
unsigned int n, unsigned int *free_space)
{
- return __rte_ring_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_FIXED,
- r->prod.single, free_space);
+ switch (r->prod.sync_type) {
+ case RTE_RING_SYNC_MT:
+ return rte_ring_mp_enqueue_bulk(r, obj_table, n, free_space);
+ case RTE_RING_SYNC_ST:
+ return rte_ring_sp_enqueue_bulk(r, obj_table, n, free_space);
+#ifdef ALLOW_EXPERIMENTAL_API
+ case RTE_RING_SYNC_MT_RTS:
+ return rte_ring_mp_rts_enqueue_bulk(r, obj_table, n,
+ free_space);
+#endif
+ }
+
+ /* valid ring should never reach this point */
+ RTE_ASSERT(0);
+ return 0;
}
/**
unsigned int n, unsigned int *available)
{
return __rte_ring_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_FIXED,
- __IS_MC, available);
+ RTE_RING_SYNC_MT, available);
}
/**
unsigned int n, unsigned int *available)
{
return __rte_ring_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_FIXED,
- __IS_SC, available);
+ RTE_RING_SYNC_ST, available);
}
/**
rte_ring_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned int n,
unsigned int *available)
{
- return __rte_ring_do_dequeue(r, obj_table, n, RTE_RING_QUEUE_FIXED,
- r->cons.single, available);
+ switch (r->cons.sync_type) {
+ case RTE_RING_SYNC_MT:
+ return rte_ring_mc_dequeue_bulk(r, obj_table, n, available);
+ case RTE_RING_SYNC_ST:
+ return rte_ring_sc_dequeue_bulk(r, obj_table, n, available);
+#ifdef ALLOW_EXPERIMENTAL_API
+ case RTE_RING_SYNC_MT_RTS:
+ return rte_ring_mc_rts_dequeue_bulk(r, obj_table, n, available);
+#endif
+ }
+
+ /* valid ring should never reach this point */
+ RTE_ASSERT(0);
+ return 0;
}
/**
return rte_ring_dequeue_bulk(r, obj_p, 1, NULL) ? 0 : -ENOENT;
}
+/**
+ * Flush a ring.
+ *
+ * This function flush all the elements in a ring
+ *
+ * @b EXPERIMENTAL: this API may change without prior notice
+ *
+ * @warning
+ * Make sure the ring is not in use while calling this function.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ */
+__rte_experimental
+void
+rte_ring_reset(struct rte_ring *r);
+
/**
* Return the number of entries in a ring.
*
return r->capacity;
}
+/**
+ * Return sync type used by producer in the ring.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @return
+ * Producer sync type value.
+ */
+static inline enum rte_ring_sync_type
+rte_ring_get_prod_sync_type(const struct rte_ring *r)
+{
+ return r->prod.sync_type;
+}
+
+/**
+ * Check is the ring for single producer.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @return
+ * true if ring is SP, zero otherwise.
+ */
+static inline int
+rte_ring_is_prod_single(const struct rte_ring *r)
+{
+ return (rte_ring_get_prod_sync_type(r) == RTE_RING_SYNC_ST);
+}
+
+/**
+ * Return sync type used by consumer in the ring.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @return
+ * Consumer sync type value.
+ */
+static inline enum rte_ring_sync_type
+rte_ring_get_cons_sync_type(const struct rte_ring *r)
+{
+ return r->cons.sync_type;
+}
+
+/**
+ * Check is the ring for single consumer.
+ *
+ * @param r
+ * A pointer to the ring structure.
+ * @return
+ * true if ring is SC, zero otherwise.
+ */
+static inline int
+rte_ring_is_cons_single(const struct rte_ring *r)
+{
+ return (rte_ring_get_cons_sync_type(r) == RTE_RING_SYNC_ST);
+}
+
/**
* Dump the status of all rings on the console
*
unsigned int n, unsigned int *free_space)
{
return __rte_ring_do_enqueue(r, obj_table, n,
- RTE_RING_QUEUE_VARIABLE, __IS_MP, free_space);
+ RTE_RING_QUEUE_VARIABLE, RTE_RING_SYNC_MT, free_space);
}
/**
unsigned int n, unsigned int *free_space)
{
return __rte_ring_do_enqueue(r, obj_table, n,
- RTE_RING_QUEUE_VARIABLE, __IS_SP, free_space);
+ RTE_RING_QUEUE_VARIABLE, RTE_RING_SYNC_ST, free_space);
}
/**
rte_ring_enqueue_burst(struct rte_ring *r, void * const *obj_table,
unsigned int n, unsigned int *free_space)
{
- return __rte_ring_do_enqueue(r, obj_table, n, RTE_RING_QUEUE_VARIABLE,
- r->prod.single, free_space);
+ switch (r->prod.sync_type) {
+ case RTE_RING_SYNC_MT:
+ return rte_ring_mp_enqueue_burst(r, obj_table, n, free_space);
+ case RTE_RING_SYNC_ST:
+ return rte_ring_sp_enqueue_burst(r, obj_table, n, free_space);
+#ifdef ALLOW_EXPERIMENTAL_API
+ case RTE_RING_SYNC_MT_RTS:
+ return rte_ring_mp_rts_enqueue_burst(r, obj_table, n,
+ free_space);
+#endif
+ }
+
+ /* valid ring should never reach this point */
+ RTE_ASSERT(0);
+ return 0;
}
/**
unsigned int n, unsigned int *available)
{
return __rte_ring_do_dequeue(r, obj_table, n,
- RTE_RING_QUEUE_VARIABLE, __IS_MC, available);
+ RTE_RING_QUEUE_VARIABLE, RTE_RING_SYNC_MT, available);
}
/**
unsigned int n, unsigned int *available)
{
return __rte_ring_do_dequeue(r, obj_table, n,
- RTE_RING_QUEUE_VARIABLE, __IS_SC, available);
+ RTE_RING_QUEUE_VARIABLE, RTE_RING_SYNC_ST, available);
}
/**
rte_ring_dequeue_burst(struct rte_ring *r, void **obj_table,
unsigned int n, unsigned int *available)
{
- return __rte_ring_do_dequeue(r, obj_table, n,
- RTE_RING_QUEUE_VARIABLE,
- r->cons.single, available);
+ switch (r->cons.sync_type) {
+ case RTE_RING_SYNC_MT:
+ return rte_ring_mc_dequeue_burst(r, obj_table, n, available);
+ case RTE_RING_SYNC_ST:
+ return rte_ring_sc_dequeue_burst(r, obj_table, n, available);
+#ifdef ALLOW_EXPERIMENTAL_API
+ case RTE_RING_SYNC_MT_RTS:
+ return rte_ring_mc_rts_dequeue_burst(r, obj_table, n,
+ available);
+#endif
+ }
+
+ /* valid ring should never reach this point */
+ RTE_ASSERT(0);
+ return 0;
}
#ifdef __cplusplus
git.droids-corp.org / dpdk.git / blobdiff