Patch Detail
get:
Show a patch.
patch:
Update a patch.
put:
Update a patch.
GET /api/patches/67701/?format=api
{ "id": 67701, "url": "http://patches.dpdk.org/api/patches/67701/?format=api", "web_url": "http://patches.dpdk.org/project/dpdk/patch/20200402220959.29885-6-konstantin.ananyev@intel.com/", "project": { "id": 1, "url": "http://patches.dpdk.org/api/projects/1/?format=api", "name": "DPDK", "link_name": "dpdk", "list_id": "dev.dpdk.org", "list_email": "dev@dpdk.org", "web_url": "http://core.dpdk.org", "scm_url": "git://dpdk.org/dpdk", "webscm_url": "http://git.dpdk.org/dpdk", "list_archive_url": "https://inbox.dpdk.org/dev", "list_archive_url_format": "https://inbox.dpdk.org/dev/{}", "commit_url_format": "" }, "msgid": "<20200402220959.29885-6-konstantin.ananyev@intel.com>", "list_archive_url": "https://inbox.dpdk.org/dev/20200402220959.29885-6-konstantin.ananyev@intel.com", "date": "2020-04-02T22:09:55", "name": "[v2,5/9] ring: introduce HTS ring mode", "commit_ref": null, "pull_url": null, "state": "superseded", "archived": true, "hash": "42bd8ac875ec8f6fd5229fa11ef1ca6a23dcb601", "submitter": { "id": 33, "url": "http://patches.dpdk.org/api/people/33/?format=api", "name": "Ananyev, Konstantin", "email": "konstantin.ananyev@intel.com" }, "delegate": { "id": 1, "url": "http://patches.dpdk.org/api/users/1/?format=api", "username": "tmonjalo", "first_name": "Thomas", "last_name": "Monjalon", "email": "thomas@monjalon.net" }, "mbox": "http://patches.dpdk.org/project/dpdk/patch/20200402220959.29885-6-konstantin.ananyev@intel.com/mbox/", "series": [ { "id": 9177, "url": "http://patches.dpdk.org/api/series/9177/?format=api", "web_url": "http://patches.dpdk.org/project/dpdk/list/?series=9177", "date": "2020-04-02T22:09:50", "name": "New sync modes for ring", "version": 2, "mbox": "http://patches.dpdk.org/series/9177/mbox/" } ], "comments": "http://patches.dpdk.org/api/patches/67701/comments/", "check": "fail", "checks": "http://patches.dpdk.org/api/patches/67701/checks/", "tags": {}, "related": [], "headers": { "Return-Path": "<dev-bounces@dpdk.org>", "X-Original-To": "patchwork@inbox.dpdk.org", "Delivered-To": "patchwork@inbox.dpdk.org", "Received": [ "from dpdk.org (dpdk.org [92.243.14.124])\n\tby inbox.dpdk.org (Postfix) with ESMTP id 6BEC2A0562;\n\tFri, 3 Apr 2020 00:11:19 +0200 (CEST)", "from [92.243.14.124] (localhost [127.0.0.1])\n\tby dpdk.org (Postfix) with ESMTP id A83E91C0D4;\n\tFri, 3 Apr 2020 00:10:52 +0200 (CEST)", "from mga01.intel.com (mga01.intel.com [192.55.52.88])\n by dpdk.org (Postfix) with ESMTP id 1CAE01C0C0\n for <dev@dpdk.org>; Fri, 3 Apr 2020 00:10:50 +0200 (CEST)", "from orsmga001.jf.intel.com ([10.7.209.18])\n by fmsmga101.fm.intel.com with ESMTP/TLS/ECDHE-RSA-AES256-GCM-SHA384;\n 02 Apr 2020 15:10:50 -0700", "from sivswdev08.ir.intel.com ([10.237.217.47])\n by orsmga001.jf.intel.com with ESMTP; 02 Apr 2020 15:10:48 -0700" ], "IronPort-SDR": [ "\n NrTFNwf+VEXkc9c51paGbhukTg+mao+J60cSCZAeZoQTPrZq4TYTLEI2j4ErfCe2snhkfwM2jd\n wbYZlkxgLDsw==", "\n NSfpYaJdDhmQwoAQsRD5c3lChJkEjzesI9dI9mAoBIhm1CNoymfqRQLdz1x5u4NmHDbAO2jisV\n XCjruANnVw2w==" ], "X-Amp-Result": "SKIPPED(no attachment in message)", "X-Amp-File-Uploaded": "False", "X-ExtLoop1": "1", "X-IronPort-AV": "E=Sophos;i=\"5.72,337,1580803200\"; d=\"scan'208\";a=\"328975331\"", "From": "Konstantin Ananyev <konstantin.ananyev@intel.com>", "To": "dev@dpdk.org", "Cc": "honnappa.nagarahalli@arm.com, david.marchand@redhat.com,\n jielong.zjl@antfin.com, Konstantin Ananyev <konstantin.ananyev@intel.com>", "Date": "Thu, 2 Apr 2020 23:09:55 +0100", "Message-Id": "<20200402220959.29885-6-konstantin.ananyev@intel.com>", "X-Mailer": "git-send-email 2.18.0", "In-Reply-To": "<20200402220959.29885-1-konstantin.ananyev@intel.com>", "References": "<20200331164330.28854-1-konstantin.ananyev@intel.com>\n <20200402220959.29885-1-konstantin.ananyev@intel.com>", "Subject": "[dpdk-dev] [PATCH v2 5/9] ring: introduce HTS ring mode", "X-BeenThere": "dev@dpdk.org", "X-Mailman-Version": "2.1.15", "Precedence": "list", "List-Id": "DPDK patches and discussions <dev.dpdk.org>", "List-Unsubscribe": "<https://mails.dpdk.org/options/dev>,\n <mailto:dev-request@dpdk.org?subject=unsubscribe>", "List-Archive": "<http://mails.dpdk.org/archives/dev/>", "List-Post": "<mailto:dev@dpdk.org>", "List-Help": "<mailto:dev-request@dpdk.org?subject=help>", "List-Subscribe": "<https://mails.dpdk.org/listinfo/dev>,\n <mailto:dev-request@dpdk.org?subject=subscribe>", "Errors-To": "dev-bounces@dpdk.org", "Sender": "\"dev\" <dev-bounces@dpdk.org>" }, "content": "Introduce head/tail sync mode for MT ring synchronization.\nIn that mode enqueue/dequeue operation is fully serialized:\nonly one thread at a time is allowed to perform given op.\nSuppose to reduce stall times in case when ring is used on\novercommitted cpus (multiple active threads on the same cpu).\n\nSigned-off-by: Konstantin Ananyev <konstantin.ananyev@intel.com>\n---\n lib/librte_ring/Makefile | 3 +\n lib/librte_ring/meson.build | 3 +\n lib/librte_ring/rte_ring.c | 20 ++-\n lib/librte_ring/rte_ring.h | 31 ++++\n lib/librte_ring/rte_ring_elem.h | 13 ++\n lib/librte_ring/rte_ring_hts.h | 210 +++++++++++++++++++++++++\n lib/librte_ring/rte_ring_hts_elem.h | 205 ++++++++++++++++++++++++\n lib/librte_ring/rte_ring_hts_generic.h | 198 +++++++++++++++++++++++\n 8 files changed, 681 insertions(+), 2 deletions(-)\n create mode 100644 lib/librte_ring/rte_ring_hts.h\n create mode 100644 lib/librte_ring/rte_ring_hts_elem.h\n create mode 100644 lib/librte_ring/rte_ring_hts_generic.h", "diff": "diff --git a/lib/librte_ring/Makefile b/lib/librte_ring/Makefile\nindex 8f5c284cc..6fe500f0d 100644\n--- a/lib/librte_ring/Makefile\n+++ b/lib/librte_ring/Makefile\n@@ -19,6 +19,9 @@ SYMLINK-$(CONFIG_RTE_LIBRTE_RING)-include := rte_ring.h \\\n \t\t\t\t\trte_ring_elem.h \\\n \t\t\t\t\trte_ring_generic.h \\\n \t\t\t\t\trte_ring_c11_mem.h \\\n+\t\t\t\t\trte_ring_hts.h \\\n+\t\t\t\t\trte_ring_hts_elem.h \\\n+\t\t\t\t\trte_ring_hts_generic.h \\\n \t\t\t\t\trte_ring_rts.h \\\n \t\t\t\t\trte_ring_rts_elem.h \\\n \t\t\t\t\trte_ring_rts_generic.h\ndiff --git a/lib/librte_ring/meson.build b/lib/librte_ring/meson.build\nindex 612936afb..8e86e037a 100644\n--- a/lib/librte_ring/meson.build\n+++ b/lib/librte_ring/meson.build\n@@ -6,6 +6,9 @@ headers = files('rte_ring.h',\n \t\t'rte_ring_elem.h',\n \t\t'rte_ring_c11_mem.h',\n \t\t'rte_ring_generic.h',\n+\t\t'rte_ring_hts.h',\n+\t\t'rte_ring_hts_elem.h',\n+\t\t'rte_ring_hts_generic.h',\n \t\t'rte_ring_rts.h',\n \t\t'rte_ring_rts_elem.h',\n \t\t'rte_ring_rts_generic.h')\ndiff --git a/lib/librte_ring/rte_ring.c b/lib/librte_ring/rte_ring.c\nindex 222eec0fb..ebe5ccf0d 100644\n--- a/lib/librte_ring/rte_ring.c\n+++ b/lib/librte_ring/rte_ring.c\n@@ -89,9 +89,11 @@ static void\n reset_headtail(void *p)\n {\n \tstruct rte_ring_headtail *ht;\n+\tstruct rte_ring_hts_headtail *ht_hts;\n \tstruct rte_ring_rts_headtail *ht_rts;\n \n \tht = p;\n+\tht_hts = p;\n \tht_rts = p;\n \n \tswitch (ht->sync_type) {\n@@ -104,6 +106,9 @@ reset_headtail(void *p)\n \t\tht_rts->head.raw = 0;\n \t\tht_rts->tail.raw = 0;\n \t\tbreak;\n+\tcase RTE_RING_SYNC_MT_HTS:\n+\t\tht_hts->ht.raw = 0;\n+\t\tbreak;\n \tdefault:\n \t\t/* unknown sync mode */\n \t\tRTE_ASSERT(0);\n@@ -127,9 +132,9 @@ get_sync_type(uint32_t flags, enum rte_ring_sync_type *prod_st,\n \tenum rte_ring_sync_type *cons_st)\n {\n \tstatic const uint32_t prod_st_flags =\n-\t\t(RING_F_SP_ENQ | RING_F_MP_RTS_ENQ);\n+\t\t(RING_F_SP_ENQ | RING_F_MP_RTS_ENQ | RING_F_MP_HTS_ENQ);\n \tstatic const uint32_t cons_st_flags =\n-\t\t(RING_F_SC_DEQ | RING_F_MC_RTS_DEQ);\n+\t\t(RING_F_SC_DEQ | RING_F_MC_RTS_DEQ | RING_F_MC_HTS_DEQ);\n \n \tswitch (flags & prod_st_flags) {\n \tcase 0:\n@@ -141,6 +146,9 @@ get_sync_type(uint32_t flags, enum rte_ring_sync_type *prod_st,\n \tcase RING_F_MP_RTS_ENQ:\n \t\t*prod_st = RTE_RING_SYNC_MT_RTS;\n \t\tbreak;\n+\tcase RING_F_MP_HTS_ENQ:\n+\t\t*prod_st = RTE_RING_SYNC_MT_HTS;\n+\t\tbreak;\n \tdefault:\n \t\treturn -EINVAL;\n \t}\n@@ -155,6 +163,9 @@ get_sync_type(uint32_t flags, enum rte_ring_sync_type *prod_st,\n \tcase RING_F_MC_RTS_DEQ:\n \t\t*cons_st = RTE_RING_SYNC_MT_RTS;\n \t\tbreak;\n+\tcase RING_F_MC_HTS_DEQ:\n+\t\t*cons_st = RTE_RING_SYNC_MT_HTS;\n+\t\tbreak;\n \tdefault:\n \t\treturn -EINVAL;\n \t}\n@@ -176,6 +187,11 @@ rte_ring_init(struct rte_ring *r, const char *name, unsigned count,\n \tRTE_BUILD_BUG_ON((offsetof(struct rte_ring, prod) &\n \t\t\t RTE_CACHE_LINE_MASK) != 0);\n \n+\tRTE_BUILD_BUG_ON(offsetof(struct rte_ring_headtail, sync_type) !=\n+\t\toffsetof(struct rte_ring_hts_headtail, sync_type));\n+\tRTE_BUILD_BUG_ON(offsetof(struct rte_ring_headtail, tail) !=\n+\t\toffsetof(struct rte_ring_hts_headtail, ht.pos.tail));\n+\n \tRTE_BUILD_BUG_ON(offsetof(struct rte_ring_headtail, sync_type) !=\n \t\toffsetof(struct rte_ring_rts_headtail, sync_type));\n \tRTE_BUILD_BUG_ON(offsetof(struct rte_ring_headtail, tail) !=\ndiff --git a/lib/librte_ring/rte_ring.h b/lib/librte_ring/rte_ring.h\nindex f6f084d79..6e4213afa 100644\n--- a/lib/librte_ring/rte_ring.h\n+++ b/lib/librte_ring/rte_ring.h\n@@ -68,6 +68,7 @@ enum rte_ring_sync_type {\n \tRTE_RING_SYNC_ST, /**< single thread only */\n #ifdef ALLOW_EXPERIMENTAL_API\n \tRTE_RING_SYNC_MT_RTS, /**< multi-thread relaxed tail sync */\n+\tRTE_RING_SYNC_MT_HTS, /**< multi-thread head/tail sync */\n #endif\n };\n \n@@ -103,6 +104,19 @@ struct rte_ring_rts_headtail {\n \tvolatile union rte_ring_ht_poscnt head;\n };\n \n+union rte_ring_ht_pos {\n+\tuint64_t raw;\n+\tstruct {\n+\t\tuint32_t head; /**< head position */\n+\t\tuint32_t tail; /**< tail position */\n+\t} pos;\n+};\n+\n+struct rte_ring_hts_headtail {\n+\tvolatile union rte_ring_ht_pos ht;\n+\tenum rte_ring_sync_type sync_type; /**< sync type of prod/cons */\n+};\n+\n /**\n * An RTE ring structure.\n *\n@@ -133,6 +147,7 @@ struct rte_ring {\n \tRTE_STD_C11\n \tunion {\n \t\tstruct rte_ring_headtail prod;\n+\t\tstruct rte_ring_hts_headtail hts_prod;\n \t\tstruct rte_ring_rts_headtail rts_prod;\n \t} __rte_cache_aligned;\n \n@@ -142,6 +157,7 @@ struct rte_ring {\n \tRTE_STD_C11\n \tunion {\n \t\tstruct rte_ring_headtail cons;\n+\t\tstruct rte_ring_hts_headtail hts_cons;\n \t\tstruct rte_ring_rts_headtail rts_cons;\n \t} __rte_cache_aligned;\n \n@@ -164,6 +180,9 @@ struct rte_ring {\n #define RING_F_MP_RTS_ENQ 0x0008 /**< The default enqueue is \"MP RTS\". */\n #define RING_F_MC_RTS_DEQ 0x0010 /**< The default dequeue is \"MC RTS\". */\n \n+#define RING_F_MP_HTS_ENQ 0x0020 /**< The default enqueue is \"MP HTS\". */\n+#define RING_F_MC_HTS_DEQ 0x0040 /**< The default dequeue is \"MC HTS\". */\n+\n #define __IS_SP RTE_RING_SYNC_ST\n #define __IS_MP RTE_RING_SYNC_MT\n #define __IS_SC RTE_RING_SYNC_ST\n@@ -494,6 +513,7 @@ rte_ring_sp_enqueue_bulk(struct rte_ring *r, void * const *obj_table,\n }\n \n #ifdef ALLOW_EXPERIMENTAL_API\n+#include <rte_ring_hts.h>\n #include <rte_ring_rts.h>\n #endif\n \n@@ -529,6 +549,9 @@ rte_ring_enqueue_bulk(struct rte_ring *r, void * const *obj_table,\n \tcase RTE_RING_SYNC_MT_RTS:\n \t\treturn rte_ring_mp_rts_enqueue_bulk(r, obj_table, n,\n \t\t\tfree_space);\n+\tcase RTE_RING_SYNC_MT_HTS:\n+\t\treturn rte_ring_mp_hts_enqueue_bulk(r, obj_table, n,\n+\t\t\tfree_space);\n #endif\n \t}\n \n@@ -676,6 +699,8 @@ rte_ring_dequeue_bulk(struct rte_ring *r, void **obj_table, unsigned int n,\n #ifdef ALLOW_EXPERIMENTAL_API\n \tcase RTE_RING_SYNC_MT_RTS:\n \t\treturn rte_ring_mc_rts_dequeue_bulk(r, obj_table, n, available);\n+\tcase RTE_RING_SYNC_MT_HTS:\n+\t\treturn rte_ring_mc_hts_dequeue_bulk(r, obj_table, n, available);\n #endif\n \t}\n \n@@ -1010,6 +1035,9 @@ rte_ring_enqueue_burst(struct rte_ring *r, void * const *obj_table,\n \tcase RTE_RING_SYNC_MT_RTS:\n \t\treturn rte_ring_mp_rts_enqueue_burst(r, obj_table, n,\n \t\t\tfree_space);\n+\tcase RTE_RING_SYNC_MT_HTS:\n+\t\treturn rte_ring_mp_hts_enqueue_burst(r, obj_table, n,\n+\t\t\tfree_space);\n #endif\n \t}\n \n@@ -1103,6 +1131,9 @@ rte_ring_dequeue_burst(struct rte_ring *r, void **obj_table,\n \tcase RTE_RING_SYNC_MT_RTS:\n \t\treturn rte_ring_mc_rts_dequeue_burst(r, obj_table, n,\n \t\t\tavailable);\n+\tcase RTE_RING_SYNC_MT_HTS:\n+\t\treturn rte_ring_mc_hts_dequeue_burst(r, obj_table, n,\n+\t\t\tavailable);\n #endif\n \t}\n \ndiff --git a/lib/librte_ring/rte_ring_elem.h b/lib/librte_ring/rte_ring_elem.h\nindex 5de0850dc..010a564c1 100644\n--- a/lib/librte_ring/rte_ring_elem.h\n+++ b/lib/librte_ring/rte_ring_elem.h\n@@ -542,6 +542,7 @@ rte_ring_sp_enqueue_bulk_elem(struct rte_ring *r, const void *obj_table,\n \t\t\tRTE_RING_QUEUE_FIXED, __IS_SP, free_space);\n }\n \n+#include <rte_ring_hts_elem.h>\n #include <rte_ring_rts_elem.h>\n \n /**\n@@ -585,6 +586,9 @@ rte_ring_enqueue_bulk_elem(struct rte_ring *r, const void *obj_table,\n \tcase RTE_RING_SYNC_MT_RTS:\n \t\treturn rte_ring_mp_rts_enqueue_bulk_elem(r, obj_table, esize, n,\n \t\t\tfree_space);\n+\tcase RTE_RING_SYNC_MT_HTS:\n+\t\treturn rte_ring_mp_hts_enqueue_bulk_elem(r, obj_table, esize, n,\n+\t\t\tfree_space);\n #endif\n \t}\n \n@@ -766,6 +770,9 @@ rte_ring_dequeue_bulk_elem(struct rte_ring *r, void *obj_table,\n \tcase RTE_RING_SYNC_MT_RTS:\n \t\treturn rte_ring_mc_rts_dequeue_bulk_elem(r, obj_table, esize,\n \t\t\tn, available);\n+\tcase RTE_RING_SYNC_MT_HTS:\n+\t\treturn rte_ring_mc_hts_dequeue_bulk_elem(r, obj_table, esize,\n+\t\t\tn, available);\n #endif\n \t}\n \n@@ -951,6 +958,9 @@ rte_ring_enqueue_burst_elem(struct rte_ring *r, const void *obj_table,\n \tcase RTE_RING_SYNC_MT_RTS:\n \t\treturn rte_ring_mp_rts_enqueue_burst_elem(r, obj_table, esize,\n \t\t\tn, free_space);\n+\tcase RTE_RING_SYNC_MT_HTS:\n+\t\treturn rte_ring_mp_hts_enqueue_burst_elem(r, obj_table, esize,\n+\t\t\tn, free_space);\n #endif\n \t}\n \n@@ -1060,6 +1070,9 @@ rte_ring_dequeue_burst_elem(struct rte_ring *r, void *obj_table,\n \tcase RTE_RING_SYNC_MT_RTS:\n \t\treturn rte_ring_mc_rts_dequeue_burst_elem(r, obj_table, esize,\n \t\t\tn, available);\n+\tcase RTE_RING_SYNC_MT_HTS:\n+\t\treturn rte_ring_mc_hts_dequeue_burst_elem(r, obj_table, esize,\n+\t\t\tn, available);\n #endif\n \t}\n \ndiff --git a/lib/librte_ring/rte_ring_hts.h b/lib/librte_ring/rte_ring_hts.h\nnew file mode 100644\nindex 000000000..062d7be6c\n--- /dev/null\n+++ b/lib/librte_ring/rte_ring_hts.h\n@@ -0,0 +1,210 @@\n+/* SPDX-License-Identifier: BSD-3-Clause\n+ *\n+ * Copyright (c) 2010-2017 Intel Corporation\n+ * Copyright (c) 2007-2009 Kip Macy kmacy@freebsd.org\n+ * All rights reserved.\n+ * Derived from FreeBSD's bufring.h\n+ * Used as BSD-3 Licensed with permission from Kip Macy.\n+ */\n+\n+#ifndef _RTE_RING_HTS_H_\n+#define _RTE_RING_HTS_H_\n+\n+/**\n+ * @file rte_ring_hts.h\n+ * @b EXPERIMENTAL: this API may change without prior notice\n+ * It is not recommended to include this file directly.\n+ * Please include <rte_ring.h> instead.\n+ *\n+ * Contains functions for serialized, aka Head-Tail Sync (HTS) ring mode.\n+ * In that mode enqueue/dequeue operation is fully serialized:\n+ * at any given moement only one enqueue/dequeue operation can proceed.\n+ * This is achieved by thread is allowed to proceed with changing head.value\n+ * only when head.value == tail.value.\n+ * Both head and tail values are updated atomically (as one 64-bit value).\n+ * To achieve that 64-bit CAS is used by head update routine.\n+ */\n+\n+#ifdef __cplusplus\n+extern \"C\" {\n+#endif\n+\n+#include <rte_ring_hts_generic.h>\n+\n+/**\n+ * @internal Enqueue several objects on the HTS ring.\n+ *\n+ * @param r\n+ * A pointer to the ring structure.\n+ * @param obj_table\n+ * A pointer to a table of void * pointers (objects).\n+ * @param n\n+ * The number of objects to add in the ring from the obj_table.\n+ * @param behavior\n+ * RTE_RING_QUEUE_FIXED: Enqueue a fixed number of items from a ring\n+ * RTE_RING_QUEUE_VARIABLE: Enqueue as many items as possible from ring\n+ * @param free_space\n+ * returns the amount of space after the enqueue operation has finished\n+ * @return\n+ * Actual number of objects enqueued.\n+ * If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.\n+ */\n+static __rte_always_inline unsigned int\n+__rte_ring_do_hts_enqueue(struct rte_ring *r, void * const *obj_table,\n+\t\tuint32_t n, enum rte_ring_queue_behavior behavior,\n+\t\tuint32_t *free_space)\n+{\n+\tuint32_t free, head;\n+\n+\tn = __rte_ring_hts_move_prod_head(r, n, behavior, &head, &free);\n+\n+\tif (n != 0) {\n+\t\tENQUEUE_PTRS(r, &r[1], head, obj_table, n, void *);\n+\t\t__rte_ring_hts_update_tail(&r->hts_prod, n, 1);\n+\t}\n+\n+\tif (free_space != NULL)\n+\t\t*free_space = free - n;\n+\treturn n;\n+}\n+\n+/**\n+ * @internal Dequeue several objects from the HTS ring.\n+ *\n+ * @param r\n+ * A pointer to the ring structure.\n+ * @param obj_table\n+ * A pointer to a table of void * pointers (objects).\n+ * @param n\n+ * The number of objects to pull from the ring.\n+ * @param behavior\n+ * RTE_RING_QUEUE_FIXED: Dequeue a fixed number of items from a ring\n+ * RTE_RING_QUEUE_VARIABLE: Dequeue as many items as possible from ring\n+ * @param available\n+ * returns the number of remaining ring entries after the dequeue has finished\n+ * @return\n+ * - Actual number of objects dequeued.\n+ * If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.\n+ */\n+static __rte_always_inline unsigned int\n+__rte_ring_do_hts_dequeue(struct rte_ring *r, void **obj_table,\n+\t\tuint32_t n, enum rte_ring_queue_behavior behavior,\n+\t\tuint32_t *available)\n+{\n+\tuint32_t entries, head;\n+\n+\tn = __rte_ring_hts_move_cons_head(r, n, behavior, &head, &entries);\n+\n+\tif (n != 0) {\n+\t\tDEQUEUE_PTRS(r, &r[1], head, obj_table, n, void *);\n+\t\t__rte_ring_hts_update_tail(&r->hts_cons, n, 0);\n+\t}\n+\n+\tif (available != NULL)\n+\t\t*available = entries - n;\n+\treturn n;\n+}\n+\n+/**\n+ * Enqueue several objects on the HTS ring (multi-producers safe).\n+ *\n+ * @param r\n+ * A pointer to the ring structure.\n+ * @param obj_table\n+ * A pointer to a table of void * pointers (objects).\n+ * @param n\n+ * The number of objects to add in the ring from the obj_table.\n+ * @param free_space\n+ * if non-NULL, returns the amount of space in the ring after the\n+ * enqueue operation has finished.\n+ * @return\n+ * The number of objects enqueued, either 0 or n\n+ */\n+__rte_experimental\n+static __rte_always_inline unsigned int\n+rte_ring_mp_hts_enqueue_bulk(struct rte_ring *r, void * const *obj_table,\n+\t\t\t unsigned int n, unsigned int *free_space)\n+{\n+\treturn __rte_ring_do_hts_enqueue(r, obj_table, n, RTE_RING_QUEUE_FIXED,\n+\t\t\tfree_space);\n+}\n+\n+/**\n+ * Dequeue several objects from an HTS ring (multi-consumers safe).\n+ *\n+ * @param r\n+ * A pointer to the ring structure.\n+ * @param obj_table\n+ * A pointer to a table of void * pointers (objects) that will be filled.\n+ * @param n\n+ * The number of objects to dequeue from the ring to the obj_table.\n+ * @param available\n+ * If non-NULL, returns the number of remaining ring entries after the\n+ * dequeue has finished.\n+ * @return\n+ * The number of objects dequeued, either 0 or n\n+ */\n+__rte_experimental\n+static __rte_always_inline unsigned int\n+rte_ring_mc_hts_dequeue_bulk(struct rte_ring *r, void **obj_table,\n+\t\tunsigned int n, unsigned int *available)\n+{\n+\treturn __rte_ring_do_hts_dequeue(r, obj_table, n, RTE_RING_QUEUE_FIXED,\n+\t\t\tavailable);\n+}\n+\n+/**\n+ * Enqueue several objects on the HTS ring (multi-producers safe).\n+ *\n+ * @param r\n+ * A pointer to the ring structure.\n+ * @param obj_table\n+ * A pointer to a table of void * pointers (objects).\n+ * @param n\n+ * The number of objects to add in the ring from the obj_table.\n+ * @param free_space\n+ * if non-NULL, returns the amount of space in the ring after the\n+ * enqueue operation has finished.\n+ * @return\n+ * - n: Actual number of objects enqueued.\n+ */\n+__rte_experimental\n+static __rte_always_inline unsigned\n+rte_ring_mp_hts_enqueue_burst(struct rte_ring *r, void * const *obj_table,\n+\t\t\t unsigned int n, unsigned int *free_space)\n+{\n+\treturn __rte_ring_do_hts_enqueue(r, obj_table, n,\n+\t\t\tRTE_RING_QUEUE_VARIABLE, free_space);\n+}\n+\n+/**\n+ * Dequeue several objects from an HTS ring (multi-consumers safe).\n+ * When the requested objects are more than the available objects,\n+ * only dequeue the actual number of objects.\n+ *\n+ * @param r\n+ * A pointer to the ring structure.\n+ * @param obj_table\n+ * A pointer to a table of void * pointers (objects) that will be filled.\n+ * @param n\n+ * The number of objects to dequeue from the ring to the obj_table.\n+ * @param available\n+ * If non-NULL, returns the number of remaining ring entries after the\n+ * dequeue has finished.\n+ * @return\n+ * - n: Actual number of objects dequeued, 0 if ring is empty\n+ */\n+__rte_experimental\n+static __rte_always_inline unsigned\n+rte_ring_mc_hts_dequeue_burst(struct rte_ring *r, void **obj_table,\n+\t\tunsigned int n, unsigned int *available)\n+{\n+\treturn __rte_ring_do_hts_dequeue(r, obj_table, n,\n+\t\t\tRTE_RING_QUEUE_VARIABLE, available);\n+}\n+\n+#ifdef __cplusplus\n+}\n+#endif\n+\n+#endif /* _RTE_RING_HTS_H_ */\ndiff --git a/lib/librte_ring/rte_ring_hts_elem.h b/lib/librte_ring/rte_ring_hts_elem.h\nnew file mode 100644\nindex 000000000..34f0d121d\n--- /dev/null\n+++ b/lib/librte_ring/rte_ring_hts_elem.h\n@@ -0,0 +1,205 @@\n+/* SPDX-License-Identifier: BSD-3-Clause\n+ *\n+ * Copyright (c) 2010-2017 Intel Corporation\n+ * Copyright (c) 2007-2009 Kip Macy kmacy@freebsd.org\n+ * All rights reserved.\n+ * Derived from FreeBSD's bufring.h\n+ * Used as BSD-3 Licensed with permission from Kip Macy.\n+ */\n+\n+#ifndef _RTE_RING_HTS_ELEM_H_\n+#define _RTE_RING_HTS_ELEM_H_\n+\n+/**\n+ * @file rte_ring_hts_elem.h\n+ * @b EXPERIMENTAL: this API may change without prior notice\n+ * It is not recommended to include this file directly.\n+ * Please include <rte_ring_elem.h> instead.\n+ *\n+ * Contains *ring_elem* functions for Head-Tail Sync (HTS) ring mode.\n+ * for more details please refer to <rte_ring_hts.h>.\n+ */\n+\n+#ifdef __cplusplus\n+extern \"C\" {\n+#endif\n+\n+#include <rte_ring_hts_generic.h>\n+\n+/**\n+ * @internal Enqueue several objects on the HTS ring.\n+ *\n+ * @param r\n+ * A pointer to the ring structure.\n+ * @param obj_table\n+ * A pointer to a table of void * pointers (objects).\n+ * @param n\n+ * The number of objects to add in the ring from the obj_table.\n+ * @param behavior\n+ * RTE_RING_QUEUE_FIXED: Enqueue a fixed number of items from a ring\n+ * RTE_RING_QUEUE_VARIABLE: Enqueue as many items as possible from ring\n+ * @param free_space\n+ * returns the amount of space after the enqueue operation has finished\n+ * @return\n+ * Actual number of objects enqueued.\n+ * If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.\n+ */\n+static __rte_always_inline unsigned int\n+__rte_ring_do_hts_enqueue_elem(struct rte_ring *r, void * const *obj_table,\n+\tuint32_t esize, uint32_t n, enum rte_ring_queue_behavior behavior,\n+\tuint32_t *free_space)\n+{\n+\tuint32_t free, head;\n+\n+\tn = __rte_ring_hts_move_prod_head(r, n, behavior, &head, &free);\n+\n+\tif (n != 0) {\n+\t\t__rte_ring_enqueue_elems(r, head, obj_table, esize, n);\n+\t\t__rte_ring_hts_update_tail(&r->hts_prod, n, 1);\n+\t}\n+\n+\tif (free_space != NULL)\n+\t\t*free_space = free - n;\n+\treturn n;\n+}\n+\n+/**\n+ * @internal Dequeue several objects from the HTS ring.\n+ *\n+ * @param r\n+ * A pointer to the ring structure.\n+ * @param obj_table\n+ * A pointer to a table of void * pointers (objects).\n+ * @param n\n+ * The number of objects to pull from the ring.\n+ * @param behavior\n+ * RTE_RING_QUEUE_FIXED: Dequeue a fixed number of items from a ring\n+ * RTE_RING_QUEUE_VARIABLE: Dequeue as many items as possible from ring\n+ * @param available\n+ * returns the number of remaining ring entries after the dequeue has finished\n+ * @return\n+ * - Actual number of objects dequeued.\n+ * If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.\n+ */\n+static __rte_always_inline unsigned int\n+__rte_ring_do_hts_dequeue_elem(struct rte_ring *r, void **obj_table,\n+\tuint32_t esize, uint32_t n, enum rte_ring_queue_behavior behavior,\n+\tuint32_t *available)\n+{\n+\tuint32_t entries, head;\n+\n+\tn = __rte_ring_hts_move_cons_head(r, n, behavior, &head, &entries);\n+\n+\tif (n != 0) {\n+\t\t__rte_ring_dequeue_elems(r, head, obj_table, esize, n);\n+\t\t__rte_ring_hts_update_tail(&r->hts_cons, n, 0);\n+\t}\n+\n+\tif (available != NULL)\n+\t\t*available = entries - n;\n+\treturn n;\n+}\n+\n+/**\n+ * Enqueue several objects on the HTS ring (multi-producers safe).\n+ *\n+ * @param r\n+ * A pointer to the ring structure.\n+ * @param obj_table\n+ * A pointer to a table of void * pointers (objects).\n+ * @param n\n+ * The number of objects to add in the ring from the obj_table.\n+ * @param free_space\n+ * if non-NULL, returns the amount of space in the ring after the\n+ * enqueue operation has finished.\n+ * @return\n+ * The number of objects enqueued, either 0 or n\n+ */\n+__rte_experimental\n+static __rte_always_inline unsigned int\n+rte_ring_mp_hts_enqueue_bulk_elem(struct rte_ring *r, void * const *obj_table,\n+\tunsigned int esize, unsigned int n, unsigned int *free_space)\n+{\n+\treturn __rte_ring_do_hts_enqueue_elem(r, obj_table, esize, n,\n+\t\t\tRTE_RING_QUEUE_FIXED, free_space);\n+}\n+\n+/**\n+ * Dequeue several objects from an HTS ring (multi-consumers safe).\n+ *\n+ * @param r\n+ * A pointer to the ring structure.\n+ * @param obj_table\n+ * A pointer to a table of void * pointers (objects) that will be filled.\n+ * @param n\n+ * The number of objects to dequeue from the ring to the obj_table.\n+ * @param available\n+ * If non-NULL, returns the number of remaining ring entries after the\n+ * dequeue has finished.\n+ * @return\n+ * The number of objects dequeued, either 0 or n\n+ */\n+__rte_experimental\n+static __rte_always_inline unsigned int\n+rte_ring_mc_hts_dequeue_bulk_elem(struct rte_ring *r, void **obj_table,\n+\tunsigned int esize, unsigned int n, unsigned int *available)\n+{\n+\treturn __rte_ring_do_hts_dequeue_elem(r, obj_table, esize, n,\n+\t\tRTE_RING_QUEUE_FIXED, available);\n+}\n+\n+/**\n+ * Enqueue several objects on the HTS ring (multi-producers safe).\n+ *\n+ * @param r\n+ * A pointer to the ring structure.\n+ * @param obj_table\n+ * A pointer to a table of void * pointers (objects).\n+ * @param n\n+ * The number of objects to add in the ring from the obj_table.\n+ * @param free_space\n+ * if non-NULL, returns the amount of space in the ring after the\n+ * enqueue operation has finished.\n+ * @return\n+ * - n: Actual number of objects enqueued.\n+ */\n+__rte_experimental\n+static __rte_always_inline unsigned\n+rte_ring_mp_hts_enqueue_burst_elem(struct rte_ring *r, void * const *obj_table,\n+\tunsigned int esize, unsigned int n, unsigned int *free_space)\n+{\n+\treturn __rte_ring_do_hts_enqueue_elem(r, obj_table, esize, n,\n+\t\t\tRTE_RING_QUEUE_VARIABLE, free_space);\n+}\n+\n+/**\n+ * Dequeue several objects from an HTS ring (multi-consumers safe).\n+ * When the requested objects are more than the available objects,\n+ * only dequeue the actual number of objects.\n+ *\n+ * @param r\n+ * A pointer to the ring structure.\n+ * @param obj_table\n+ * A pointer to a table of void * pointers (objects) that will be filled.\n+ * @param n\n+ * The number of objects to dequeue from the ring to the obj_table.\n+ * @param available\n+ * If non-NULL, returns the number of remaining ring entries after the\n+ * dequeue has finished.\n+ * @return\n+ * - n: Actual number of objects dequeued, 0 if ring is empty\n+ */\n+__rte_experimental\n+static __rte_always_inline unsigned\n+rte_ring_mc_hts_dequeue_burst_elem(struct rte_ring *r, void **obj_table,\n+\tunsigned int esize, unsigned int n, unsigned int *available)\n+{\n+\treturn __rte_ring_do_hts_dequeue_elem(r, obj_table, esize, n,\n+\t\t\tRTE_RING_QUEUE_VARIABLE, available);\n+}\n+\n+#ifdef __cplusplus\n+}\n+#endif\n+\n+#endif /* _RTE_RING_HTS_ELEM_H_ */\ndiff --git a/lib/librte_ring/rte_ring_hts_generic.h b/lib/librte_ring/rte_ring_hts_generic.h\nnew file mode 100644\nindex 000000000..0b3931ffa\n--- /dev/null\n+++ b/lib/librte_ring/rte_ring_hts_generic.h\n@@ -0,0 +1,198 @@\n+/* SPDX-License-Identifier: BSD-3-Clause\n+ *\n+ * Copyright (c) 2010-2020 Intel Corporation\n+ * Copyright (c) 2007-2009 Kip Macy kmacy@freebsd.org\n+ * All rights reserved.\n+ * Derived from FreeBSD's bufring.h\n+ * Used as BSD-3 Licensed with permission from Kip Macy.\n+ */\n+\n+#ifndef _RTE_RING_HTS_GENERIC_H_\n+#define _RTE_RING_HTS_GENERIC_H_\n+\n+/**\n+ * @file rte_ring_hts_generic.h\n+ * It is not recommended to include this file directly,\n+ * include <rte_ring.h> instead.\n+ * Contains internal helper functions for head/tail sync (HTS) ring mode.\n+ * For more information please refer to <rte_ring_hts.h>.\n+ */\n+\n+static __rte_always_inline void\n+__rte_ring_hts_update_tail(struct rte_ring_hts_headtail *ht, uint32_t num,\n+\tuint32_t enqueue)\n+{\n+\tunion rte_ring_ht_pos p;\n+\n+\tif (enqueue)\n+\t\trte_smp_wmb();\n+\telse\n+\t\trte_smp_rmb();\n+\n+\tp.raw = rte_atomic64_read((rte_atomic64_t *)(uintptr_t)&ht->ht.raw);\n+\n+\tp.pos.head = p.pos.tail + num;\n+\tp.pos.tail = p.pos.head;\n+\n+\trte_atomic64_set((rte_atomic64_t *)(uintptr_t)&ht->ht.raw, p.raw);\n+}\n+\n+/**\n+ * @internal waits till tail will become equal to head.\n+ * Means no writer/reader is active for that ring.\n+ * Suppose to work as serialization point.\n+ */\n+static __rte_always_inline void\n+__rte_ring_hts_head_wait(const struct rte_ring_hts_headtail *ht,\n+\t\tunion rte_ring_ht_pos *p)\n+{\n+\tp->raw = rte_atomic64_read((rte_atomic64_t *)\n+\t\t\t(uintptr_t)&ht->ht.raw);\n+\n+\twhile (p->pos.head != p->pos.tail) {\n+\t\trte_pause();\n+\t\tp->raw = rte_atomic64_read((rte_atomic64_t *)\n+\t\t\t\t(uintptr_t)&ht->ht.raw);\n+\t}\n+}\n+\n+/**\n+ * @internal This function updates the producer head for enqueue\n+ *\n+ * @param r\n+ * A pointer to the ring structure\n+ * @param is_sp\n+ * Indicates whether multi-producer path is needed or not\n+ * @param n\n+ * The number of elements we will want to enqueue, i.e. how far should the\n+ * head be moved\n+ * @param behavior\n+ * RTE_RING_QUEUE_FIXED: Enqueue a fixed number of items from a ring\n+ * RTE_RING_QUEUE_VARIABLE: Enqueue as many items as possible from ring\n+ * @param old_head\n+ * Returns head value as it was before the move, i.e. where enqueue starts\n+ * @param new_head\n+ * Returns the current/new head value i.e. where enqueue finishes\n+ * @param free_entries\n+ * Returns the amount of free space in the ring BEFORE head was moved\n+ * @return\n+ * Actual number of objects enqueued.\n+ * If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.\n+ */\n+static __rte_always_inline unsigned int\n+__rte_ring_hts_move_prod_head(struct rte_ring *r, unsigned int num,\n+\tenum rte_ring_queue_behavior behavior, uint32_t *old_head,\n+\tuint32_t *free_entries)\n+{\n+\tuint32_t n;\n+\tunion rte_ring_ht_pos np, op;\n+\n+\tconst uint32_t capacity = r->capacity;\n+\n+\tdo {\n+\t\t/* Reset n to the initial burst count */\n+\t\tn = num;\n+\n+\t\t/* wait for tail to be equal to head */\n+\t\t__rte_ring_hts_head_wait(&r->hts_prod, &op);\n+\n+\t\t/* add rmb barrier to avoid load/load reorder in weak\n+\t\t * memory model. It is noop on x86\n+\t\t */\n+\t\trte_smp_rmb();\n+\n+\t\t/*\n+\t\t * The subtraction is done between two unsigned 32bits value\n+\t\t * (the result is always modulo 32 bits even if we have\n+\t\t * *old_head > cons_tail). So 'free_entries' is always between 0\n+\t\t * and capacity (which is < size).\n+\t\t */\n+\t\t*free_entries = capacity + r->cons.tail - op.pos.head;\n+\n+\t\t/* check that we have enough room in ring */\n+\t\tif (unlikely(n > *free_entries))\n+\t\t\tn = (behavior == RTE_RING_QUEUE_FIXED) ?\n+\t\t\t\t\t0 : *free_entries;\n+\n+\t\tif (n == 0)\n+\t\t\treturn 0;\n+\n+\t\tnp.pos.tail = op.pos.tail;\n+\t\tnp.pos.head = op.pos.head + n;\n+\n+\t} while (rte_atomic64_cmpset(&r->hts_prod.ht.raw,\n+\t\t\top.raw, np.raw) == 0);\n+\n+\t*old_head = op.pos.head;\n+\treturn n;\n+}\n+\n+/**\n+ * @internal This function updates the consumer head for dequeue\n+ *\n+ * @param r\n+ * A pointer to the ring structure\n+ * @param is_sc\n+ * Indicates whether multi-consumer path is needed or not\n+ * @param n\n+ * The number of elements we will want to enqueue, i.e. how far should the\n+ * head be moved\n+ * @param behavior\n+ * RTE_RING_QUEUE_FIXED: Dequeue a fixed number of items from a ring\n+ * RTE_RING_QUEUE_VARIABLE: Dequeue as many items as possible from ring\n+ * @param old_head\n+ * Returns head value as it was before the move, i.e. where dequeue starts\n+ * @param new_head\n+ * Returns the current/new head value i.e. where dequeue finishes\n+ * @param entries\n+ * Returns the number of entries in the ring BEFORE head was moved\n+ * @return\n+ * - Actual number of objects dequeued.\n+ * If behavior == RTE_RING_QUEUE_FIXED, this will be 0 or n only.\n+ */\n+static __rte_always_inline unsigned int\n+__rte_ring_hts_move_cons_head(struct rte_ring *r, unsigned int num,\n+\tenum rte_ring_queue_behavior behavior, uint32_t *old_head,\n+\tuint32_t *entries)\n+{\n+\tuint32_t n;\n+\tunion rte_ring_ht_pos np, op;\n+\n+\t/* move cons.head atomically */\n+\tdo {\n+\t\t/* Restore n as it may change every loop */\n+\t\tn = num;\n+\n+\t\t/* wait for tail to be equal to head */\n+\t\t__rte_ring_hts_head_wait(&r->hts_cons, &op);\n+\n+\t\t/* add rmb barrier to avoid load/load reorder in weak\n+\t\t * memory model. It is noop on x86\n+\t\t */\n+\t\trte_smp_rmb();\n+\n+\t\t/* The subtraction is done between two unsigned 32bits value\n+\t\t * (the result is always modulo 32 bits even if we have\n+\t\t * cons_head > prod_tail). So 'entries' is always between 0\n+\t\t * and size(ring)-1.\n+\t\t */\n+\t\t*entries = r->prod.tail - op.pos.head;\n+\n+\t\t/* Set the actual entries for dequeue */\n+\t\tif (n > *entries)\n+\t\t\tn = (behavior == RTE_RING_QUEUE_FIXED) ? 0 : *entries;\n+\n+\t\tif (unlikely(n == 0))\n+\t\t\treturn 0;\n+\n+\t\tnp.pos.tail = op.pos.tail;\n+\t\tnp.pos.head = op.pos.head + n;\n+\n+\t} while (rte_atomic64_cmpset(&r->hts_cons.ht.raw,\n+\t\t\top.raw, np.raw) == 0);\n+\n+\t*old_head = op.pos.head;\n+\treturn n;\n+}\n+\n+#endif /* _RTE_RING_HTS_GENERIC_H_ */\n", "prefixes": [ "v2", "5/9" ] }