Patch Detail
get:
Show a patch.
patch:
Update a patch.
put:
Update a patch.
GET /api/patches/20110/?format=api
{ "id": 20110, "url": "http://patches.dpdk.org/api/patches/20110/?format=api", "web_url": "http://patches.dpdk.org/project/dpdk/patch/1485879273-86228-12-git-send-email-harry.van.haaren@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": "<1485879273-86228-12-git-send-email-harry.van.haaren@intel.com>", "list_archive_url": "https://inbox.dpdk.org/dev/1485879273-86228-12-git-send-email-harry.van.haaren@intel.com", "date": "2017-01-31T16:14:29", "name": "[dpdk-dev,v2,11/15] event/sw: add scheduling logic", "commit_ref": null, "pull_url": null, "state": "superseded", "archived": true, "hash": "07ab78b0f6b9013f00849ff4a18a38107f65e98e", "submitter": { "id": 317, "url": "http://patches.dpdk.org/api/people/317/?format=api", "name": "Van Haaren, Harry", "email": "harry.van.haaren@intel.com" }, "delegate": { "id": 10, "url": "http://patches.dpdk.org/api/users/10/?format=api", "username": "bruce", "first_name": "Bruce", "last_name": "Richardson", "email": "bruce.richardson@intel.com" }, "mbox": "http://patches.dpdk.org/project/dpdk/patch/1485879273-86228-12-git-send-email-harry.van.haaren@intel.com/mbox/", "series": [], "comments": "http://patches.dpdk.org/api/patches/20110/comments/", "check": "warning", "checks": "http://patches.dpdk.org/api/patches/20110/checks/", "tags": {}, "related": [], "headers": { "Return-Path": "<dev-bounces@dpdk.org>", "X-Original-To": "patchwork@dpdk.org", "Delivered-To": "patchwork@dpdk.org", "Received": [ "from [92.243.14.124] (localhost [IPv6:::1])\n\tby dpdk.org (Postfix) with ESMTP id 0507BF962;\n\tTue, 31 Jan 2017 17:15:39 +0100 (CET)", "from mga09.intel.com (mga09.intel.com [134.134.136.24])\n\tby dpdk.org (Postfix) with ESMTP id 797072BE9\n\tfor <dev@dpdk.org>; Tue, 31 Jan 2017 17:15:01 +0100 (CET)", "from fmsmga003.fm.intel.com ([10.253.24.29])\n\tby orsmga102.jf.intel.com with ESMTP; 31 Jan 2017 08:15:00 -0800", "from silpixa00398672.ir.intel.com ([10.237.223.128])\n\tby FMSMGA003.fm.intel.com with ESMTP; 31 Jan 2017 08:14:59 -0800" ], "X-ExtLoop1": "1", "X-IronPort-AV": "E=Sophos;i=\"5.33,315,1477983600\"; d=\"scan'208\";a=\"815468201\"", "From": "Harry van Haaren <harry.van.haaren@intel.com>", "To": "dev@dpdk.org", "Cc": "jerin.jacob@caviumnetworks.com,\n\tBruce Richardson <bruce.richardson@intel.com>,\n\tGage Eads <gage.eads@intel.com>,\n\tHarry van Haaren <harry.van.haaren@intel.com>", "Date": "Tue, 31 Jan 2017 16:14:29 +0000", "Message-Id": "<1485879273-86228-12-git-send-email-harry.van.haaren@intel.com>", "X-Mailer": "git-send-email 2.7.4", "In-Reply-To": "<1485879273-86228-1-git-send-email-harry.van.haaren@intel.com>", "References": "<1484580885-148524-1-git-send-email-harry.van.haaren@intel.com>\n\t<1485879273-86228-1-git-send-email-harry.van.haaren@intel.com>", "Subject": "[dpdk-dev] [PATCH v2 11/15] event/sw: add scheduling logic", "X-BeenThere": "dev@dpdk.org", "X-Mailman-Version": "2.1.15", "Precedence": "list", "List-Id": "DPDK patches and discussions <dev.dpdk.org>", "List-Unsubscribe": "<http://dpdk.org/ml/options/dev>,\n\t<mailto:dev-request@dpdk.org?subject=unsubscribe>", "List-Archive": "<http://dpdk.org/ml/archives/dev/>", "List-Post": "<mailto:dev@dpdk.org>", "List-Help": "<mailto:dev-request@dpdk.org?subject=help>", "List-Subscribe": "<http://dpdk.org/ml/listinfo/dev>,\n\t<mailto:dev-request@dpdk.org?subject=subscribe>", "Errors-To": "dev-bounces@dpdk.org", "Sender": "\"dev\" <dev-bounces@dpdk.org>" }, "content": "From: Bruce Richardson <bruce.richardson@intel.com>\n\nAdd in the scheduling function which takes the events from the\nproducer queues and buffers them before scheduling them to consumer\nqueues. The scheduling logic includes support for atomic, reordered,\nand parallel scheduling of flows.\n\nSigned-off-by: Bruce Richardson <bruce.richardson@intel.com>\nSigned-off-by: Gage Eads <gage.eads@intel.com>\nSigned-off-by: Harry van Haaren <harry.van.haaren@intel.com>\n---\n drivers/event/sw/Makefile | 1 +\n drivers/event/sw/sw_evdev.c | 1 +\n drivers/event/sw/sw_evdev.h | 11 +\n drivers/event/sw/sw_evdev_scheduler.c | 602 ++++++++++++++++++++++++++++++++++\n 4 files changed, 615 insertions(+)\n create mode 100644 drivers/event/sw/sw_evdev_scheduler.c", "diff": "diff --git a/drivers/event/sw/Makefile b/drivers/event/sw/Makefile\nindex b6ecd91..a7f5b3d 100644\n--- a/drivers/event/sw/Makefile\n+++ b/drivers/event/sw/Makefile\n@@ -54,6 +54,7 @@ EXPORT_MAP := rte_pmd_evdev_sw_version.map\n # library source files\n SRCS-$(CONFIG_RTE_LIBRTE_PMD_SW_EVENTDEV) += sw_evdev.c\n SRCS-$(CONFIG_RTE_LIBRTE_PMD_SW_EVENTDEV) += sw_evdev_worker.c\n+SRCS-$(CONFIG_RTE_LIBRTE_PMD_SW_EVENTDEV) += sw_evdev_scheduler.c\n \n # export include files\n SYMLINK-y-include +=\ndiff --git a/drivers/event/sw/sw_evdev.c b/drivers/event/sw/sw_evdev.c\nindex b719f65..17b5e49 100644\n--- a/drivers/event/sw/sw_evdev.c\n+++ b/drivers/event/sw/sw_evdev.c\n@@ -521,6 +521,7 @@ sw_probe(const char *name, const char *params)\n \tdev->enqueue_burst = sw_event_enqueue_burst;\n \tdev->dequeue = sw_event_dequeue;\n \tdev->dequeue_burst = sw_event_dequeue_burst;\n+\tdev->schedule = sw_event_schedule;\n \n \tsw = dev->data->dev_private;\n \tsw->data = dev->data;\ndiff --git a/drivers/event/sw/sw_evdev.h b/drivers/event/sw/sw_evdev.h\nindex ea39bb2..63b0979 100644\n--- a/drivers/event/sw/sw_evdev.h\n+++ b/drivers/event/sw/sw_evdev.h\n@@ -240,8 +240,18 @@ struct sw_evdev {\n \t/* Cache how many packets are in each cq */\n \tuint16_t cq_ring_space[SW_PORTS_MAX] __rte_cache_aligned;\n \n+\t/* Array of pointers to load-balanced QIDs sorted by priority level */\n+\tstruct sw_qid *qids_prioritized[RTE_EVENT_MAX_QUEUES_PER_DEV];\n+\n+\t/* Stats */\n+\tstruct sw_point_stats stats __rte_cache_aligned;\n+\tuint64_t sched_called;\n \tint32_t sched_quanta;\n+\tuint64_t sched_no_iq_enqueues;\n+\tuint64_t sched_no_cq_enqueues;\n+\tuint64_t sched_cq_qid_called;\n \n+\tuint8_t started;\n \tuint32_t credit_update_quanta;\n };\n \n@@ -266,5 +276,6 @@ uint16_t sw_event_enqueue_burst(void *port, const struct rte_event ev[],\n uint16_t sw_event_dequeue(void *port, struct rte_event *ev, uint64_t wait);\n uint16_t sw_event_dequeue_burst(void *port, struct rte_event *ev, uint16_t num,\n \t\t\tuint64_t wait);\n+void sw_event_schedule(struct rte_eventdev *dev);\n \n #endif /* _SW_EVDEV_H_ */\ndiff --git a/drivers/event/sw/sw_evdev_scheduler.c b/drivers/event/sw/sw_evdev_scheduler.c\nnew file mode 100644\nindex 0000000..3b30efe\n--- /dev/null\n+++ b/drivers/event/sw/sw_evdev_scheduler.c\n@@ -0,0 +1,602 @@\n+/*-\n+ * BSD LICENSE\n+ *\n+ * Copyright(c) 2016-2017 Intel Corporation. All rights reserved.\n+ *\n+ * Redistribution and use in source and binary forms, with or without\n+ * modification, are permitted provided that the following conditions\n+ * are met:\n+ *\n+ * * Redistributions of source code must retain the above copyright\n+ * notice, this list of conditions and the following disclaimer.\n+ * * Redistributions in binary form must reproduce the above copyright\n+ * notice, this list of conditions and the following disclaimer in\n+ * the documentation and/or other materials provided with the\n+ * distribution.\n+ * * Neither the name of Intel Corporation nor the names of its\n+ * contributors may be used to endorse or promote products derived\n+ * from this software without specific prior written permission.\n+ *\n+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS\n+ * \"AS IS\" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT\n+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR\n+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT\n+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,\n+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT\n+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,\n+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY\n+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT\n+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE\n+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.\n+ */\n+\n+#include <rte_ring.h>\n+#include <rte_hash_crc.h>\n+#include \"sw_evdev.h\"\n+#include \"iq_ring.h\"\n+#include \"event_ring.h\"\n+\n+#define SW_IQS_MASK (SW_IQS_MAX-1)\n+\n+/* Retrieve the highest priority IQ or -1 if no pkts available. Doing the\n+ * CLZ twice is faster than caching the value due to data dependencies\n+ */\n+#define PKT_MASK_TO_IQ(pkts) \\\n+\t(__builtin_ctz(pkts | (1 << SW_IQS_MAX)))\n+\n+/* Clamp the highest priorities to the max value as allowed by\n+ * the mask. Assums MASK is (powerOfTwo - 1). Priority 0 (highest) are shifted\n+ * into leftmost IQ so that clz() reads it first on dequeue\n+ */\n+#define PRIO_TO_IQ(prio) (prio > SW_IQS_MASK ? SW_IQS_MASK : prio)\n+\n+#define MAX_PER_IQ_DEQUEUE 48\n+#define FLOWID_MASK (SW_QID_NUM_FIDS-1)\n+\n+static inline uint32_t\n+sw_schedule_atomic_to_cq(struct sw_evdev *sw, struct sw_qid * const qid,\n+\t\tuint32_t iq_num, unsigned int count)\n+{\n+\tstruct rte_event qes[MAX_PER_IQ_DEQUEUE]; /* count <= MAX */\n+\tstruct rte_event blocked_qes[MAX_PER_IQ_DEQUEUE];\n+\tuint32_t nb_blocked = 0;\n+\tuint32_t i;\n+\n+\tif (count > MAX_PER_IQ_DEQUEUE)\n+\t\tcount = MAX_PER_IQ_DEQUEUE;\n+\n+\t/* This is the QID ID. The QID ID is static, hence it can be\n+\t * used to identify the stage of processing in history lists etc\n+\t */\n+\tuint32_t qid_id = qid->id;\n+\n+\tiq_ring_dequeue_burst(qid->iq[iq_num], qes, count);\n+\tfor (i = 0; i < count; i++) {\n+\t\tconst struct rte_event *qe = &qes[i];\n+\t\t/* use cheap bit mixing, we only need to lose a few bits */\n+\t\tuint32_t flow_id32 = (qes[i].flow_id) ^ (qes[i].flow_id >> 10);\n+\t\tconst uint16_t flow_id = FLOWID_MASK & flow_id32;\n+\t\tstruct sw_fid_t *fid = &qid->fids[flow_id];\n+\t\tint cq = fid->cq;\n+\n+\t\tif (cq < 0) {\n+\t\t\tuint32_t cq_idx = qid->cq_next_tx++;\n+\t\t\tif (qid->cq_next_tx == qid->cq_num_mapped_cqs)\n+\t\t\t\tqid->cq_next_tx = 0;\n+\t\t\tcq = qid->cq_map[cq_idx];\n+\n+\t\t\t/* find least used */\n+\t\t\tint cq_free_cnt = sw->cq_ring_space[cq];\n+\t\t\tfor (cq_idx = 0; cq_idx < qid->cq_num_mapped_cqs;\n+\t\t\t\t\tcq_idx++) {\n+\t\t\t\tint test_cq = qid->cq_map[cq_idx];\n+\t\t\t\tint test_cq_free = sw->cq_ring_space[test_cq];\n+\t\t\t\tif (test_cq_free > cq_free_cnt) {\n+\t\t\t\t\tcq = test_cq;\n+\t\t\t\t\tcq_free_cnt = test_cq_free;\n+\t\t\t\t}\n+\t\t\t}\n+\n+\t\t\tfid->cq = cq; /* this pins early */\n+\t\t}\n+\n+\t\tif (sw->cq_ring_space[cq] == 0 ||\n+\t\t\t\tsw->ports[cq].inflights == SW_PORT_HIST_LIST) {\n+\t\t\tblocked_qes[nb_blocked++] = *qe;\n+\t\t\tcontinue;\n+\t\t}\n+\n+\t\tstruct sw_port *p = &sw->ports[cq];\n+\n+\t\t/* at this point we can queue up the packet on the cq_buf */\n+\t\tfid->pcount++;\n+\t\tp->cq_buf[p->cq_buf_count++] = *qe;\n+\t\tp->inflights++;\n+\t\tsw->cq_ring_space[cq]--;\n+\n+\t\tint head = (p->hist_head++ & (SW_PORT_HIST_LIST-1));\n+\t\tp->hist_list[head].fid = flow_id;\n+\t\tp->hist_list[head].qid = qid_id;\n+\n+\t\tp->stats.tx_pkts++;\n+\t\tqid->stats.tx_pkts++;\n+\n+\t\t/* if we just filled in the last slot, flush the buffer */\n+\t\tif (sw->cq_ring_space[cq] == 0) {\n+\t\t\tstruct qe_ring *worker = p->cq_worker_ring;\n+\t\t\tqe_ring_enqueue_burst(worker, p->cq_buf,\n+\t\t\t\t\tp->cq_buf_count,\n+\t\t\t\t\t&sw->cq_ring_space[cq]);\n+\t\t\tp->cq_buf_count = 0;\n+\t\t}\n+\t}\n+\tiq_ring_put_back(qid->iq[iq_num], blocked_qes, nb_blocked);\n+\n+\treturn count - nb_blocked;\n+}\n+\n+static inline uint32_t\n+sw_schedule_parallel_to_cq(struct sw_evdev *sw, struct sw_qid * const qid,\n+\t\tuint32_t iq_num, unsigned int count, int keep_order)\n+{\n+\tuint32_t i;\n+\tuint32_t cq_idx = qid->cq_next_tx;\n+\n+\t/* This is the QID ID. The QID ID is static, hence it can be\n+\t * used to identify the stage of processing in history lists etc\n+\t */\n+\tuint32_t qid_id = qid->id;\n+\n+\tif (count > MAX_PER_IQ_DEQUEUE)\n+\t\tcount = MAX_PER_IQ_DEQUEUE;\n+\n+\tif (keep_order)\n+\t\t/* only schedule as many as we have reorder buffer entries */\n+\t\tcount = RTE_MIN(count,\n+\t\t\t\trte_ring_count(qid->reorder_buffer_freelist));\n+\n+\tfor (i = 0; i < count; i++) {\n+\t\tconst struct rte_event *qe = iq_ring_peek(qid->iq[iq_num]);\n+\t\tuint32_t cq_check_count = 0;\n+\t\tuint32_t cq;\n+\n+\t\t/*\n+\t\t * for parallel, just send to next available CQ in round-robin\n+\t\t * fashion. So scan for an available CQ. If all CQs are full\n+\t\t * just return and move on to next QID\n+\t\t */\n+\t\tdo {\n+\t\t\tif (++cq_check_count > qid->cq_num_mapped_cqs)\n+\t\t\t\tgoto exit;\n+\t\t\tcq = qid->cq_map[cq_idx];\n+\t\t\tif (++cq_idx == qid->cq_num_mapped_cqs)\n+\t\t\t\tcq_idx = 0;\n+\t\t} while (qe_ring_free_count(sw->ports[cq].cq_worker_ring) == 0 ||\n+\t\t\t\tsw->ports[cq].inflights == SW_PORT_HIST_LIST);\n+\n+\t\tstruct sw_port *p = &sw->ports[cq];\n+\t\tif (sw->cq_ring_space[cq] == 0 ||\n+\t\t\t\tp->inflights == SW_PORT_HIST_LIST)\n+\t\t\tbreak;\n+\n+\t\tsw->cq_ring_space[cq]--;\n+\n+\t\tqid->stats.tx_pkts++;\n+\n+\t\tconst int head = (p->hist_head & (SW_PORT_HIST_LIST-1));\n+\n+\t\tp->hist_list[head].fid = qe->flow_id;\n+\t\tp->hist_list[head].qid = qid_id;\n+\n+\t\tif (keep_order)\n+\t\t\trte_ring_sc_dequeue(qid->reorder_buffer_freelist,\n+\t\t\t\t\t(void *)&p->hist_list[head].rob_entry);\n+\n+\t\tsw->ports[cq].cq_buf[sw->ports[cq].cq_buf_count++] = *qe;\n+\t\tiq_ring_pop(qid->iq[iq_num]);\n+\n+\t\trte_compiler_barrier();\n+\t\tp->inflights++;\n+\t\tp->stats.tx_pkts++;\n+\t\tp->hist_head++;\n+\t}\n+exit:\n+\tqid->cq_next_tx = cq_idx;\n+\treturn i;\n+}\n+\n+static uint32_t\n+sw_schedule_dir_to_cq(struct sw_evdev *sw, struct sw_qid * const qid,\n+\t\tuint32_t iq_num, unsigned int count __rte_unused)\n+{\n+\tuint32_t cq_id = qid->cq_map[0];\n+\tstruct sw_port *port = &sw->ports[cq_id];\n+\n+\t/* get max burst enq size for cq_ring */\n+\tuint32_t count_free = sw->cq_ring_space[cq_id];\n+\tif (count_free == 0)\n+\t\treturn 0;\n+\n+\t/* burst dequeue from the QID IQ ring */\n+\tstruct iq_ring *ring = qid->iq[iq_num];\n+\tuint32_t ret = iq_ring_dequeue_burst(ring,\n+\t\t\t&port->cq_buf[port->cq_buf_count], count_free);\n+\tport->cq_buf_count += ret;\n+\n+\t/* Update QID, Port and Total TX stats */\n+\tqid->stats.tx_pkts += ret;\n+\tport->stats.tx_pkts += ret;\n+\n+\t/* Subtract credits from cached value */\n+\tsw->cq_ring_space[cq_id] -= ret;\n+\n+\treturn ret;\n+}\n+\n+static uint32_t\n+sw_schedule_qid_to_cq(struct sw_evdev *sw)\n+{\n+\tuint32_t pkts = 0;\n+\tuint32_t qid_idx;\n+\n+\tsw->sched_cq_qid_called++;\n+\n+\tfor (qid_idx = 0; qid_idx < sw->qid_count; qid_idx++) {\n+\t\tstruct sw_qid *qid = sw->qids_prioritized[qid_idx];\n+\n+\t\tint type = qid->type;\n+\t\tint iq_num = PKT_MASK_TO_IQ(qid->iq_pkt_mask);\n+\n+\t\t/* zero mapped CQs indicates directed */\n+\t\tif (iq_num >= SW_IQS_MAX)\n+\t\t\tcontinue;\n+\n+\t\tuint32_t pkts_done = 0;\n+\t\tuint32_t count = iq_ring_count(qid->iq[iq_num]);\n+\n+\t\tif (count > 0) {\n+\t\t\tif (type == RTE_SCHED_TYPE_DIRECT)\n+\t\t\t\tpkts_done += sw_schedule_dir_to_cq(sw, qid,\n+\t\t\t\t\t\tiq_num, count);\n+\t\t\telse if (type == RTE_SCHED_TYPE_ATOMIC)\n+\t\t\t\tpkts_done += sw_schedule_atomic_to_cq(sw, qid,\n+\t\t\t\t\t\tiq_num, count);\n+\t\t\telse\n+\t\t\t\tpkts_done += sw_schedule_parallel_to_cq(sw, qid,\n+\t\t\t\t\t\tiq_num, count,\n+\t\t\t\t\t\ttype == RTE_SCHED_TYPE_ORDERED);\n+\t\t}\n+\n+\t\t/* Check if the IQ that was polled is now empty, and unset it\n+\t\t * in the IQ mask if its empty.\n+\t\t */\n+\t\tint all_done = (pkts_done == count);\n+\n+\t\tqid->iq_pkt_mask &= ~(all_done << (iq_num));\n+\t\tpkts += pkts_done;\n+\t}\n+\n+\treturn pkts;\n+}\n+\n+/* This function will perform re-ordering of packets, and injecting into\n+ * the appropriate QID IQ. As LB and DIR QIDs are in the same array, but *NOT*\n+ * contiguous in that array, this function accepts a \"range\" of QIDs to scan.\n+ */\n+static uint16_t\n+sw_schedule_reorder(struct sw_evdev *sw, int qid_start, int qid_end)\n+{\n+\t/* Perform egress reordering */\n+\tstruct rte_event *qe;\n+\tuint32_t pkts_iter = 0;\n+\n+\tfor (; qid_start < qid_end; qid_start++) {\n+\t\tstruct sw_qid *qid = &sw->qids[qid_start];\n+\t\tint i, num_entries_in_use;\n+\n+\t\tif (qid->type != RTE_SCHED_TYPE_ORDERED)\n+\t\t\tcontinue;\n+\n+\t\tnum_entries_in_use = rte_ring_free_count(\n+\t\t\t\t\tqid->reorder_buffer_freelist);\n+\n+\t\tfor (i = 0; i < num_entries_in_use; i++) {\n+\t\t\tstruct reorder_buffer_entry *entry;\n+\t\t\tint j;\n+\n+\t\t\tentry = &qid->reorder_buffer[qid->reorder_buffer_index];\n+\n+\t\t\tif (!entry->ready)\n+\t\t\t\tbreak;\n+\n+\t\t\tfor (j = 0; j < entry->num_fragments; j++) {\n+\t\t\t\tuint16_t dest_qid;\n+\t\t\t\tuint16_t dest_iq;\n+\n+\t\t\t\tint idx = entry->fragment_index + j;\n+\t\t\t\tqe = &entry->fragments[idx];\n+\n+\t\t\t\tdest_qid = qe->queue_id;\n+\t\t\t\tdest_iq = PRIO_TO_IQ(qe->priority);\n+\n+\t\t\t\tif (dest_qid >= sw->qid_count) {\n+\t\t\t\t\tsw->stats.rx_dropped++;\n+\t\t\t\t\tcontinue;\n+\t\t\t\t}\n+\n+\t\t\t\tstruct sw_qid *dest_qid_ptr =\n+\t\t\t\t\t&sw->qids[dest_qid];\n+\t\t\t\tconst struct iq_ring *dest_iq_ptr =\n+\t\t\t\t\tdest_qid_ptr->iq[dest_iq];\n+\t\t\t\tif (iq_ring_free_count(dest_iq_ptr) == 0)\n+\t\t\t\t\tbreak;\n+\n+\t\t\t\tpkts_iter++;\n+\n+\t\t\t\tstruct sw_qid *q = &sw->qids[dest_qid];\n+\t\t\t\tstruct iq_ring *r = q->iq[dest_iq];\n+\n+\t\t\t\t/* we checked for space above, so enqueue must\n+\t\t\t\t * succeed\n+\t\t\t\t */\n+\t\t\t\tiq_ring_enqueue(r, qe);\n+\t\t\t\tq->iq_pkt_mask |= (1 << (dest_iq));\n+\t\t\t\tq->iq_pkt_count[dest_iq]++;\n+\t\t\t\tq->stats.rx_pkts++;\n+\t\t\t}\n+\n+\t\t\tentry->ready = (j != entry->num_fragments);\n+\t\t\tentry->num_fragments -= j;\n+\t\t\tentry->fragment_index += j;\n+\n+\t\t\tif (!entry->ready) {\n+\t\t\t\tentry->fragment_index = 0;\n+\n+\t\t\t\trte_ring_sp_enqueue(qid->reorder_buffer_freelist,\n+\t\t\t\t\t\t entry);\n+\n+\t\t\t\tqid->reorder_buffer_index++;\n+\t\t\t\tqid->reorder_buffer_index %= qid->window_size;\n+\t\t\t}\n+\t\t}\n+\t}\n+\treturn pkts_iter;\n+}\n+\n+static inline void __attribute__((always_inline))\n+sw_refill_pp_buf(struct sw_evdev *sw, struct sw_port *port)\n+{\n+\tRTE_SET_USED(sw);\n+\tstruct qe_ring *worker = port->rx_worker_ring;\n+\tport->pp_buf_start = 0;\n+\tport->pp_buf_count = qe_ring_dequeue_burst(worker, port->pp_buf,\n+\t\t\tRTE_DIM(port->pp_buf));\n+}\n+\n+static inline uint32_t __attribute__((always_inline))\n+__pull_port_lb(struct sw_evdev *sw, uint32_t port_id, int allow_reorder)\n+{\n+\tstatic const struct reorder_buffer_entry dummy_rob;\n+\tuint32_t pkts_iter = 0;\n+\tstruct sw_port *port = &sw->ports[port_id];\n+\n+\t/* If shadow ring has 0 pkts, pull from worker ring */\n+\tif (port->pp_buf_count == 0)\n+\t\tsw_refill_pp_buf(sw, port);\n+\n+\twhile (port->pp_buf_count) {\n+\t\tconst struct rte_event *qe = &port->pp_buf[port->pp_buf_start];\n+\t\tstruct sw_hist_list_entry *hist_entry = NULL;\n+\t\tuint8_t flags = qe->op;\n+\t\tconst uint16_t eop = !(flags & QE_FLAG_NOT_EOP);\n+\t\tint needs_reorder = 0;\n+\t\t/* if no-reordering, having PARTIAL == NEW */\n+\t\tif (!allow_reorder && !eop)\n+\t\t\tflags = QE_FLAG_VALID;\n+\n+\t\t/*\n+\t\t * if we don't have space for this packet in an IQ,\n+\t\t * then move on to next queue. Technically, for a\n+\t\t * packet that needs reordering, we don't need to check\n+\t\t * here, but it simplifies things not to special-case\n+\t\t */\n+\t\tuint32_t iq_num = PRIO_TO_IQ(qe->priority);\n+\t\tstruct sw_qid *qid = &sw->qids[qe->queue_id];\n+\n+\t\tif ((flags & QE_FLAG_VALID) &&\n+\t\t\t\tiq_ring_free_count(qid->iq[iq_num]) == 0)\n+\t\t\tbreak;\n+\n+\t\t/* now process based on flags. Note that for directed\n+\t\t * queues, the enqueue_flush masks off all but the\n+\t\t * valid flag. This makes FWD and PARTIAL enqueues just\n+\t\t * NEW type, and makes DROPS no-op calls.\n+\t\t */\n+\t\tif ((flags & QE_FLAG_COMPLETE) && port->inflights > 0) {\n+\t\t\tconst uint32_t hist_tail = port->hist_tail &\n+\t\t\t\t\t(SW_PORT_HIST_LIST - 1);\n+\n+\t\t\thist_entry = &port->hist_list[hist_tail];\n+\t\t\tconst uint32_t hist_qid = hist_entry->qid;\n+\t\t\tconst uint32_t hist_fid = hist_entry->fid;\n+\n+\t\t\tstruct sw_fid_t *fid =\n+\t\t\t\t&sw->qids[hist_qid].fids[hist_fid];\n+\t\t\tfid->pcount -= eop;\n+\t\t\tif (fid->pcount == 0)\n+\t\t\t\tfid->cq = -1;\n+\n+\t\t\tif (allow_reorder) {\n+\t\t\t\t/* set reorder ready if an ordered QID */\n+\t\t\t\tuintptr_t rob_ptr =\n+\t\t\t\t\t(uintptr_t)hist_entry->rob_entry;\n+\t\t\t\tconst uintptr_t valid = (rob_ptr != 0);\n+\t\t\t\tneeds_reorder = valid;\n+\t\t\t\trob_ptr |=\n+\t\t\t\t\t((valid - 1) & (uintptr_t)&dummy_rob);\n+\t\t\t\tstruct reorder_buffer_entry *tmp_rob_ptr =\n+\t\t\t\t\t(struct reorder_buffer_entry *)rob_ptr;\n+\t\t\t\ttmp_rob_ptr->ready = eop * needs_reorder;\n+\t\t\t}\n+\n+\t\t\tport->inflights -= eop;\n+\t\t\tport->hist_tail += eop;\n+\t\t}\n+\t\tif (flags & QE_FLAG_VALID) {\n+\t\t\tport->stats.rx_pkts++;\n+\n+\t\t\tif (allow_reorder && needs_reorder) {\n+\t\t\t\tstruct reorder_buffer_entry *rob_entry =\n+\t\t\t\t\t\thist_entry->rob_entry;\n+\n+\t\t\t\t/* Although fragmentation not currently\n+\t\t\t\t * supported by eventdev API, we support it\n+\t\t\t\t * here. Open: How do we alert the user that\n+\t\t\t\t * they've exceeded max frags?\n+\t\t\t\t */\n+\t\t\t\tint num_frag = rob_entry->num_fragments;\n+\t\t\t\tif (num_frag == SW_FRAGMENTS_MAX)\n+\t\t\t\t\tsw->stats.rx_dropped++;\n+\t\t\t\telse {\n+\t\t\t\t\tint idx = rob_entry->num_fragments++;\n+\t\t\t\t\trob_entry->fragments[idx] = *qe;\n+\t\t\t\t}\n+\t\t\t\tgoto end_qe;\n+\t\t\t}\n+\n+\t\t\t/* Use the iq_num from above to push the QE\n+\t\t\t * into the qid at the right priority\n+\t\t\t */\n+\n+\t\t\tqid->iq_pkt_mask |= (1 << (iq_num));\n+\t\t\tiq_ring_enqueue(qid->iq[iq_num], qe);\n+\t\t\tqid->iq_pkt_count[iq_num]++;\n+\t\t\tqid->stats.rx_pkts++;\n+\t\t\tpkts_iter++;\n+\t\t}\n+\n+end_qe:\n+\t\tport->pp_buf_start++;\n+\t\tport->pp_buf_count--;\n+\t} /* while (avail_qes) */\n+\n+\treturn pkts_iter;\n+}\n+\n+static uint32_t\n+sw_schedule_pull_port_lb(struct sw_evdev *sw, uint32_t port_id)\n+{\n+\treturn __pull_port_lb(sw, port_id, 1);\n+}\n+\n+static uint32_t\n+sw_schedule_pull_port_no_reorder(struct sw_evdev *sw, uint32_t port_id)\n+{\n+\treturn __pull_port_lb(sw, port_id, 0);\n+}\n+\n+static uint32_t\n+sw_schedule_pull_port_dir(struct sw_evdev *sw, uint32_t port_id)\n+{\n+\tuint32_t pkts_iter = 0;\n+\tstruct sw_port *port = &sw->ports[port_id];\n+\n+\t/* If shadow ring has 0 pkts, pull from worker ring */\n+\tif (port->pp_buf_count == 0)\n+\t\tsw_refill_pp_buf(sw, port);\n+\n+\twhile (port->pp_buf_count) {\n+\t\tconst struct rte_event *qe = &port->pp_buf[port->pp_buf_start];\n+\t\tuint8_t flags = qe->op;\n+\n+\t\tif ((flags & QE_FLAG_VALID) == 0)\n+\t\t\tgoto end_qe;\n+\n+\t\tuint32_t iq_num = PRIO_TO_IQ(qe->priority);\n+\t\tstruct sw_qid *qid = &sw->qids[qe->queue_id];\n+\t\tstruct iq_ring *iq_ring = qid->iq[iq_num];\n+\n+\t\tif (iq_ring_free_count(iq_ring) == 0)\n+\t\t\tbreak; /* move to next port */\n+\n+\t\tport->stats.rx_pkts++;\n+\n+\t\t/* Use the iq_num from above to push the QE\n+\t\t * into the qid at the right priority\n+\t\t */\n+\t\tqid->iq_pkt_mask |= (1 << (iq_num));\n+\t\tiq_ring_enqueue(iq_ring, qe);\n+\t\tqid->iq_pkt_count[iq_num]++;\n+\t\tqid->stats.rx_pkts++;\n+\t\tpkts_iter++;\n+\n+end_qe:\n+\t\tport->pp_buf_start++;\n+\t\tport->pp_buf_count--;\n+\t} /* while port->pp_buf_count */\n+\n+\treturn pkts_iter;\n+}\n+\n+void\n+sw_event_schedule(struct rte_eventdev *dev)\n+{\n+\tstruct sw_evdev *sw = sw_pmd_priv(dev);\n+\tuint32_t in_pkts, out_pkts;\n+\tuint32_t out_pkts_total = 0, in_pkts_total = 0;\n+\tint32_t sched_quanta = sw->sched_quanta;\n+\tuint32_t i;\n+\n+\tsw->sched_called++;\n+\tif (!sw->started)\n+\t\treturn;\n+\n+\tdo {\n+\t\tuint32_t in_pkts_this_iteration = 0;\n+\n+\t\t/* Pull from rx_ring for ports */\n+\t\tdo {\n+\t\t\tin_pkts = 0;\n+\t\t\tfor (i = 0; i < sw->port_count; i++)\n+\t\t\t\t/* TODO: use a function pointer in the port */\n+\t\t\t\tif (sw->ports[i].is_directed)\n+\t\t\t\t\tin_pkts += sw_schedule_pull_port_dir(sw, i);\n+\t\t\t\telse if (sw->ports[i].num_ordered_qids > 0)\n+\t\t\t\t\tin_pkts += sw_schedule_pull_port_lb(sw, i);\n+\t\t\t\telse\n+\t\t\t\t\tin_pkts += sw_schedule_pull_port_no_reorder(sw, i);\n+\n+\t\t\t/* QID scan for re-ordered */\n+\t\t\tin_pkts += sw_schedule_reorder(sw, 0,\n+\t\t\t\t\tsw->qid_count);\n+\t\t\tin_pkts_this_iteration += in_pkts;\n+\t\t} while (in_pkts > 4 &&\n+\t\t\t\t(int)in_pkts_this_iteration < sched_quanta);\n+\n+\t\tout_pkts = 0;\n+\t\tout_pkts += sw_schedule_qid_to_cq(sw);\n+\t\tout_pkts_total += out_pkts;\n+\t\tin_pkts_total += in_pkts_this_iteration;\n+\n+\t\tif (in_pkts == 0 && out_pkts == 0)\n+\t\t\tbreak;\n+\t} while ((int)out_pkts_total < sched_quanta);\n+\n+\t/* push all the internal buffered QEs in port->cq_ring to the\n+\t * worker cores: aka, do the ring transfers batched.\n+\t */\n+\tfor (i = 0; i < sw->port_count; i++) {\n+\t\tstruct qe_ring *worker = sw->ports[i].cq_worker_ring;\n+\t\tqe_ring_enqueue_burst(worker, sw->ports[i].cq_buf,\n+\t\t\t\tsw->ports[i].cq_buf_count,\n+\t\t\t\t&sw->cq_ring_space[i]);\n+\t\tsw->ports[i].cq_buf_count = 0;\n+\t}\n+\n+\tsw->stats.tx_pkts += out_pkts_total;\n+\tsw->stats.rx_pkts += in_pkts_total;\n+\n+\tsw->sched_no_iq_enqueues += (in_pkts_total == 0);\n+\tsw->sched_no_cq_enqueues += (out_pkts_total == 0);\n+\n+}\n", "prefixes": [ "dpdk-dev", "v2", "11/15" ] }