ring: introduce RTS ring mode

[dpdk.git] / lib / librte_ring / rte_ring_elem.h

diff --git a/lib/librte_ring/rte_ring_elem.h b/lib/librte_ring/rte_ring_elem.h
index 7406c0b..4030753 100644 (file)
--- a/lib/librte_ring/rte_ring_elem.h
+++ b/lib/librte_ring/rte_ring_elem.h
@@ -74,12 +74,24 @@ ssize_t rte_ring_get_memsize_elem(unsigned int esize, unsigned int count);
  *   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:
@@ -528,6 +540,10 @@ rte_ring_sp_enqueue_bulk_elem(struct rte_ring *r, const void *obj_table,
                        RTE_RING_QUEUE_FIXED, RTE_RING_SYNC_ST, free_space);
 }
 
+#ifdef ALLOW_EXPERIMENTAL_API
+#include <rte_ring_rts.h>
+#endif
+
 /**
  * Enqueue several objects on a ring.
  *
@@ -557,6 +573,26 @@ rte_ring_enqueue_bulk_elem(struct rte_ring *r, const void *obj_table,
 {
        return __rte_ring_do_enqueue_elem(r, obj_table, esize, n,
                        RTE_RING_QUEUE_FIXED, r->prod.sync_type, free_space);
+
+       switch (r->prod.sync_type) {
+       case RTE_RING_SYNC_MT:
+               return rte_ring_mp_enqueue_bulk_elem(r, obj_table, esize, n,
+                       free_space);
+       case RTE_RING_SYNC_ST:
+               return rte_ring_sp_enqueue_bulk_elem(r, obj_table, esize, n,
+                       free_space);
+#ifdef ALLOW_EXPERIMENTAL_API
+       case RTE_RING_SYNC_MT_RTS:
+               return rte_ring_mp_rts_enqueue_bulk_elem(r, obj_table, esize, n,
+                       free_space);
+#endif
+       }
+
+       /* valid ring should never reach this point */
+       RTE_ASSERT(0);
+       if (free_space != NULL)
+               *free_space = 0;
+       return 0;
 }
 
 /**
@@ -661,7 +697,7 @@ rte_ring_mc_dequeue_bulk_elem(struct rte_ring *r, void *obj_table,
                unsigned int esize, unsigned int n, unsigned int *available)
 {
        return __rte_ring_do_dequeue_elem(r, obj_table, esize, n,
-                               RTE_RING_QUEUE_FIXED, RTE_RING_SYNC_MT, available);
+                       RTE_RING_QUEUE_FIXED, RTE_RING_SYNC_MT, available);
 }
 
 /**
@@ -719,8 +755,25 @@ static __rte_always_inline unsigned int
 rte_ring_dequeue_bulk_elem(struct rte_ring *r, void *obj_table,
                unsigned int esize, unsigned int n, unsigned int *available)
 {
-       return __rte_ring_do_dequeue_elem(r, obj_table, esize, n,
-                       RTE_RING_QUEUE_FIXED, r->cons.sync_type, available);
+       switch (r->cons.sync_type) {
+       case RTE_RING_SYNC_MT:
+               return rte_ring_mc_dequeue_bulk_elem(r, obj_table, esize, n,
+                       available);
+       case RTE_RING_SYNC_ST:
+               return rte_ring_sc_dequeue_bulk_elem(r, obj_table, esize, n,
+                       available);
+#ifdef ALLOW_EXPERIMENTAL_API
+       case RTE_RING_SYNC_MT_RTS:
+               return rte_ring_mc_rts_dequeue_bulk_elem(r, obj_table, esize,
+                       n, available);
+#endif
+       }
+
+       /* valid ring should never reach this point */
+       RTE_ASSERT(0);
+       if (available != NULL)
+               *available = 0;
+       return 0;
 }
 
 /**
@@ -887,8 +940,25 @@ static __rte_always_inline unsigned
 rte_ring_enqueue_burst_elem(struct rte_ring *r, const void *obj_table,
                unsigned int esize, unsigned int n, unsigned int *free_space)
 {
-       return __rte_ring_do_enqueue_elem(r, obj_table, esize, n,
-                       RTE_RING_QUEUE_VARIABLE, r->prod.sync_type, free_space);
+       switch (r->prod.sync_type) {
+       case RTE_RING_SYNC_MT:
+               return rte_ring_mp_enqueue_burst_elem(r, obj_table, esize, n,
+                       free_space);
+       case RTE_RING_SYNC_ST:
+               return rte_ring_sp_enqueue_burst_elem(r, obj_table, esize, n,
+                       free_space);
+#ifdef ALLOW_EXPERIMENTAL_API
+       case RTE_RING_SYNC_MT_RTS:
+               return rte_ring_mp_rts_enqueue_burst_elem(r, obj_table, esize,
+                       n, free_space);
+#endif
+       }
+
+       /* valid ring should never reach this point */
+       RTE_ASSERT(0);
+       if (free_space != NULL)
+               *free_space = 0;
+       return 0;
 }
 
 /**
@@ -979,9 +1049,25 @@ static __rte_always_inline unsigned int
 rte_ring_dequeue_burst_elem(struct rte_ring *r, void *obj_table,
                unsigned int esize, unsigned int n, unsigned int *available)
 {
-       return __rte_ring_do_dequeue_elem(r, obj_table, esize, n,
-                               RTE_RING_QUEUE_VARIABLE,
-                               r->cons.sync_type, available);
+       switch (r->cons.sync_type) {
+       case RTE_RING_SYNC_MT:
+               return rte_ring_mc_dequeue_burst_elem(r, obj_table, esize, n,
+                       available);
+       case RTE_RING_SYNC_ST:
+               return rte_ring_sc_dequeue_burst_elem(r, obj_table, esize, n,
+                       available);
+#ifdef ALLOW_EXPERIMENTAL_API
+       case RTE_RING_SYNC_MT_RTS:
+               return rte_ring_mc_rts_dequeue_burst_elem(r, obj_table, esize,
+                       n, available);
+#endif
+       }
+
+       /* valid ring should never reach this point */
+       RTE_ASSERT(0);
+       if (available != NULL)
+               *available = 0;
+       return 0;
 }
 
 #include <rte_ring.h>