Adding upstream version 14.2.21.upstream/14.2.21upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

9 files changed, 3037 insertions, 0 deletions

diff --git a/src/seastar/dpdk/drivers/event/sw/Makefile b/src/seastar/dpdk/drivers/event/sw/Makefile
new file mode 100644
index 00000000..857a87cc
--- /dev/null
+++ b/src/seastar/dpdk/drivers/event/sw/Makefile
@@ -0,0 +1,62 @@
+#   BSD LICENSE
+#
+#   Copyright(c) 2016-2017 Intel Corporation. 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_SDK)/mk/rte.vars.mk
+
+# library name
+LIB = librte_pmd_sw_event.a
+
+# build flags
+CFLAGS += -O3
+CFLAGS += $(WERROR_FLAGS)
+# for older GCC versions, allow us to initialize an event using
+# designated initializers.
+ifeq ($(CONFIG_RTE_TOOLCHAIN_GCC),y)
+ifeq ($(shell test $(GCC_VERSION) -le 50 && echo 1), 1)
+CFLAGS += -Wno-missing-field-initializers
+endif
+endif
+
+# library version
+LIBABIVER := 1
+
+# versioning export map
+EXPORT_MAP := rte_pmd_evdev_sw_version.map
+
+# library source files
+SRCS-$(CONFIG_RTE_LIBRTE_PMD_SW_EVENTDEV) += sw_evdev.c
+SRCS-$(CONFIG_RTE_LIBRTE_PMD_SW_EVENTDEV) += sw_evdev_worker.c
+SRCS-$(CONFIG_RTE_LIBRTE_PMD_SW_EVENTDEV) += sw_evdev_scheduler.c
+SRCS-$(CONFIG_RTE_LIBRTE_PMD_SW_EVENTDEV) += sw_evdev_xstats.c
+
+# export include files
+SYMLINK-y-include +=
+
+include $(RTE_SDK)/mk/rte.lib.mk
diff --git a/src/seastar/dpdk/drivers/event/sw/event_ring.h b/src/seastar/dpdk/drivers/event/sw/event_ring.h
new file mode 100644
index 00000000..cdaee95d
--- /dev/null
+++ b/src/seastar/dpdk/drivers/event/sw/event_ring.h
@@ -0,0 +1,185 @@
+/*-
+ *   BSD LICENSE
+ *
+ *   Copyright(c) 2016-2017 Intel Corporation. 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.
+ */
+
+/*
+ * Generic ring structure for passing events from one core to another.
+ *
+ * Used by the software scheduler for the producer and consumer rings for
+ * each port, i.e. for passing events from worker cores to scheduler and
+ * vice-versa. Designed for single-producer, single-consumer use with two
+ * cores working on each ring.
+ */
+
+#ifndef _EVENT_RING_
+#define _EVENT_RING_
+
+#include <stdint.h>
+
+#include <rte_common.h>
+#include <rte_memory.h>
+#include <rte_malloc.h>
+
+#define QE_RING_NAMESIZE 32
+
+struct qe_ring {
+	char name[QE_RING_NAMESIZE] __rte_cache_aligned;
+	uint32_t ring_size; /* size of memory block allocated to the ring */
+	uint32_t mask;      /* mask for read/write values == ring_size -1 */
+	uint32_t size;      /* actual usable space in the ring */
+	volatile uint32_t write_idx __rte_cache_aligned;
+	volatile uint32_t read_idx __rte_cache_aligned;
+
+	struct rte_event ring[0] __rte_cache_aligned;
+};
+
+#ifndef force_inline
+#define force_inline inline __attribute__((always_inline))
+#endif
+
+static inline struct qe_ring *
+qe_ring_create(const char *name, unsigned int size, unsigned int socket_id)
+{
+	struct qe_ring *retval;
+	const uint32_t ring_size = rte_align32pow2(size + 1);
+	size_t memsize = sizeof(*retval) +
+			(ring_size * sizeof(retval->ring[0]));
+
+	retval = rte_zmalloc_socket(NULL, memsize, 0, socket_id);
+	if (retval == NULL)
+		goto end;
+
+	snprintf(retval->name, sizeof(retval->name), "EVDEV_RG_%s", name);
+	retval->ring_size = ring_size;
+	retval->mask = ring_size - 1;
+	retval->size = size;
+end:
+	return retval;
+}
+
+static inline void
+qe_ring_destroy(struct qe_ring *r)
+{
+	rte_free(r);
+}
+
+static force_inline unsigned int
+qe_ring_count(const struct qe_ring *r)
+{
+	return r->write_idx - r->read_idx;
+}
+
+static force_inline unsigned int
+qe_ring_free_count(const struct qe_ring *r)
+{
+	return r->size - qe_ring_count(r);
+}
+
+static force_inline unsigned int
+qe_ring_enqueue_burst(struct qe_ring *r, const struct rte_event *qes,
+		unsigned int nb_qes, uint16_t *free_count)
+{
+	const uint32_t size = r->size;
+	const uint32_t mask = r->mask;
+	const uint32_t read = r->read_idx;
+	uint32_t write = r->write_idx;
+	const uint32_t space = read + size - write;
+	uint32_t i;
+
+	if (space < nb_qes)
+		nb_qes = space;
+
+	for (i = 0; i < nb_qes; i++, write++)
+		r->ring[write & mask] = qes[i];
+
+	rte_smp_wmb();
+
+	if (nb_qes != 0)
+		r->write_idx = write;
+
+	*free_count = space - nb_qes;
+
+	return nb_qes;
+}
+
+static force_inline unsigned int
+qe_ring_enqueue_burst_with_ops(struct qe_ring *r, const struct rte_event *qes,
+		unsigned int nb_qes, uint8_t *ops)
+{
+	const uint32_t size = r->size;
+	const uint32_t mask = r->mask;
+	const uint32_t read = r->read_idx;
+	uint32_t write = r->write_idx;
+	const uint32_t space = read + size - write;
+	uint32_t i;
+
+	if (space < nb_qes)
+		nb_qes = space;
+
+	for (i = 0; i < nb_qes; i++, write++) {
+		r->ring[write & mask] = qes[i];
+		r->ring[write & mask].op = ops[i];
+	}
+
+	rte_smp_wmb();
+
+	if (nb_qes != 0)
+		r->write_idx = write;
+
+	return nb_qes;
+}
+
+static force_inline unsigned int
+qe_ring_dequeue_burst(struct qe_ring *r, struct rte_event *qes,
+		unsigned int nb_qes)
+{
+	const uint32_t mask = r->mask;
+	uint32_t read = r->read_idx;
+	const uint32_t write = r->write_idx;
+	const uint32_t items = write - read;
+	uint32_t i;
+
+	if (items < nb_qes)
+		nb_qes = items;
+
+
+	for (i = 0; i < nb_qes; i++, read++)
+		qes[i] = r->ring[read & mask];
+
+	rte_smp_rmb();
+
+	if (nb_qes != 0)
+		r->read_idx += nb_qes;
+
+	return nb_qes;
+}
+
+#endif
diff --git a/src/seastar/dpdk/drivers/event/sw/iq_ring.h b/src/seastar/dpdk/drivers/event/sw/iq_ring.h
new file mode 100644
index 00000000..d480d156
--- /dev/null
+++ b/src/seastar/dpdk/drivers/event/sw/iq_ring.h
@@ -0,0 +1,176 @@
+/*-
+ *   BSD LICENSE
+ *
+ *   Copyright(c) 2016-2017 Intel Corporation. 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.
+ */
+
+/*
+ * Ring structure definitions used for the internal ring buffers of the
+ * SW eventdev implementation. These are designed for single-core use only.
+ */
+#ifndef _IQ_RING_
+#define _IQ_RING_
+
+#include <stdint.h>
+
+#include <rte_common.h>
+#include <rte_memory.h>
+#include <rte_malloc.h>
+#include <rte_eventdev.h>
+
+#define IQ_RING_NAMESIZE 12
+#define QID_IQ_DEPTH 512
+#define QID_IQ_MASK (uint16_t)(QID_IQ_DEPTH - 1)
+
+struct iq_ring {
+	char name[IQ_RING_NAMESIZE] __rte_cache_aligned;
+	uint16_t write_idx;
+	uint16_t read_idx;
+
+	struct rte_event ring[QID_IQ_DEPTH];
+};
+
+#ifndef force_inline
+#define force_inline inline __attribute__((always_inline))
+#endif
+
+static inline struct iq_ring *
+iq_ring_create(const char *name, unsigned int socket_id)
+{
+	struct iq_ring *retval;
+
+	retval = rte_malloc_socket(NULL, sizeof(*retval), 0, socket_id);
+	if (retval == NULL)
+		goto end;
+
+	snprintf(retval->name, sizeof(retval->name), "%s", name);
+	retval->write_idx = retval->read_idx = 0;
+end:
+	return retval;
+}
+
+static inline void
+iq_ring_destroy(struct iq_ring *r)
+{
+	rte_free(r);
+}
+
+static force_inline uint16_t
+iq_ring_count(const struct iq_ring *r)
+{
+	return r->write_idx - r->read_idx;
+}
+
+static force_inline uint16_t
+iq_ring_free_count(const struct iq_ring *r)
+{
+	return QID_IQ_MASK - iq_ring_count(r);
+}
+
+static force_inline uint16_t
+iq_ring_enqueue_burst(struct iq_ring *r, struct rte_event *qes, uint16_t nb_qes)
+{
+	const uint16_t read = r->read_idx;
+	uint16_t write = r->write_idx;
+	const uint16_t space = read + QID_IQ_MASK - write;
+	uint16_t i;
+
+	if (space < nb_qes)
+		nb_qes = space;
+
+	for (i = 0; i < nb_qes; i++, write++)
+		r->ring[write & QID_IQ_MASK] = qes[i];
+
+	r->write_idx = write;
+
+	return nb_qes;
+}
+
+static force_inline uint16_t
+iq_ring_dequeue_burst(struct iq_ring *r, struct rte_event *qes, uint16_t nb_qes)
+{
+	uint16_t read = r->read_idx;
+	const uint16_t write = r->write_idx;
+	const uint16_t items = write - read;
+	uint16_t i;
+
+	for (i = 0; i < nb_qes; i++, read++)
+		qes[i] = r->ring[read & QID_IQ_MASK];
+
+	if (items < nb_qes)
+		nb_qes = items;
+
+	r->read_idx += nb_qes;
+
+	return nb_qes;
+}
+
+/* assumes there is space, from a previous dequeue_burst */
+static force_inline uint16_t
+iq_ring_put_back(struct iq_ring *r, struct rte_event *qes, uint16_t nb_qes)
+{
+	uint16_t i, read = r->read_idx;
+
+	for (i = nb_qes; i-- > 0; )
+		r->ring[--read & QID_IQ_MASK] = qes[i];
+
+	r->read_idx = read;
+	return nb_qes;
+}
+
+static force_inline const struct rte_event *
+iq_ring_peek(const struct iq_ring *r)
+{
+	return &r->ring[r->read_idx & QID_IQ_MASK];
+}
+
+static force_inline void
+iq_ring_pop(struct iq_ring *r)
+{
+	r->read_idx++;
+}
+
+static force_inline int
+iq_ring_enqueue(struct iq_ring *r, const struct rte_event *qe)
+{
+	const uint16_t read = r->read_idx;
+	const uint16_t write = r->write_idx;
+	const uint16_t space = read + QID_IQ_MASK - write;
+
+	if (space == 0)
+		return -1;
+
+	r->ring[write & QID_IQ_MASK] = *qe;
+
+	r->write_idx = write + 1;
+
+	return 0;
+}
+
+#endif
diff --git a/src/seastar/dpdk/drivers/event/sw/rte_pmd_evdev_sw_version.map b/src/seastar/dpdk/drivers/event/sw/rte_pmd_evdev_sw_version.map
new file mode 100644
index 00000000..5352e7e3
--- /dev/null
+++ b/src/seastar/dpdk/drivers/event/sw/rte_pmd_evdev_sw_version.map
@@ -0,0 +1,3 @@
+DPDK_17.05 {
+	local: *;
+};
diff --git a/src/seastar/dpdk/drivers/event/sw/sw_evdev.c b/src/seastar/dpdk/drivers/event/sw/sw_evdev.c
new file mode 100644
index 00000000..a31aaa66
--- /dev/null
+++ b/src/seastar/dpdk/drivers/event/sw/sw_evdev.c
@@ -0,0 +1,833 @@
+/*-
+ *   BSD LICENSE
+ *
+ *   Copyright(c) 2016-2017 Intel Corporation. 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 <string.h>
+
+#include <rte_vdev.h>
+#include <rte_memzone.h>
+#include <rte_kvargs.h>
+#include <rte_ring.h>
+#include <rte_errno.h>
+
+#include "sw_evdev.h"
+#include "iq_ring.h"
+#include "event_ring.h"
+
+#define EVENTDEV_NAME_SW_PMD event_sw
+#define NUMA_NODE_ARG "numa_node"
+#define SCHED_QUANTA_ARG "sched_quanta"
+#define CREDIT_QUANTA_ARG "credit_quanta"
+
+static void
+sw_info_get(struct rte_eventdev *dev, struct rte_event_dev_info *info);
+
+static int
+sw_port_link(struct rte_eventdev *dev, void *port, const uint8_t queues[],
+		const uint8_t priorities[], uint16_t num)
+{
+	struct sw_port *p = port;
+	struct sw_evdev *sw = sw_pmd_priv(dev);
+	int i;
+
+	RTE_SET_USED(priorities);
+	for (i = 0; i < num; i++) {
+		struct sw_qid *q = &sw->qids[queues[i]];
+
+		/* check for qid map overflow */
+		if (q->cq_num_mapped_cqs >= RTE_DIM(q->cq_map)) {
+			rte_errno = -EDQUOT;
+			break;
+		}
+
+		if (p->is_directed && p->num_qids_mapped > 0) {
+			rte_errno = -EDQUOT;
+			break;
+		}
+
+		if (q->type == SW_SCHED_TYPE_DIRECT) {
+			/* check directed qids only map to one port */
+			if (p->num_qids_mapped > 0) {
+				rte_errno = -EDQUOT;
+				break;
+			}
+			/* check port only takes a directed flow */
+			if (num > 1) {
+				rte_errno = -EDQUOT;
+				break;
+			}
+
+			p->is_directed = 1;
+			p->num_qids_mapped = 1;
+		} else if (q->type == RTE_SCHED_TYPE_ORDERED) {
+			p->num_ordered_qids++;
+			p->num_qids_mapped++;
+		} else if (q->type == RTE_SCHED_TYPE_ATOMIC) {
+			p->num_qids_mapped++;
+		}
+
+		q->cq_map[q->cq_num_mapped_cqs] = p->id;
+		rte_smp_wmb();
+		q->cq_num_mapped_cqs++;
+	}
+	return i;
+}
+
+static int
+sw_port_unlink(struct rte_eventdev *dev, void *port, uint8_t queues[],
+		uint16_t nb_unlinks)
+{
+	struct sw_port *p = port;
+	struct sw_evdev *sw = sw_pmd_priv(dev);
+	unsigned int i, j;
+
+	int unlinked = 0;
+	for (i = 0; i < nb_unlinks; i++) {
+		struct sw_qid *q = &sw->qids[queues[i]];
+		for (j = 0; j < q->cq_num_mapped_cqs; j++) {
+			if (q->cq_map[j] == p->id) {
+				q->cq_map[j] =
+					q->cq_map[q->cq_num_mapped_cqs - 1];
+				rte_smp_wmb();
+				q->cq_num_mapped_cqs--;
+				unlinked++;
+
+				p->num_qids_mapped--;
+
+				if (q->type == RTE_SCHED_TYPE_ORDERED)
+					p->num_ordered_qids--;
+
+				continue;
+			}
+		}
+	}
+	return unlinked;
+}
+
+static int
+sw_port_setup(struct rte_eventdev *dev, uint8_t port_id,
+		const struct rte_event_port_conf *conf)
+{
+	struct sw_evdev *sw = sw_pmd_priv(dev);
+	struct sw_port *p = &sw->ports[port_id];
+	char buf[QE_RING_NAMESIZE];
+	unsigned int i;
+
+	struct rte_event_dev_info info;
+	sw_info_get(dev, &info);
+
+	/* detect re-configuring and return credits to instance if needed */
+	if (p->initialized) {
+		/* taking credits from pool is done one quanta at a time, and
+		 * credits may be spend (counted in p->inflights) or still
+		 * available in the port (p->inflight_credits). We must return
+		 * the sum to no leak credits
+		 */
+		int possible_inflights = p->inflight_credits + p->inflights;
+		rte_atomic32_sub(&sw->inflights, possible_inflights);
+	}
+
+	*p = (struct sw_port){0}; /* zero entire structure */
+	p->id = port_id;
+	p->sw = sw;
+
+	snprintf(buf, sizeof(buf), "sw%d_%s", dev->data->dev_id,
+			"rx_worker_ring");
+	p->rx_worker_ring = qe_ring_create(buf, MAX_SW_PROD_Q_DEPTH,
+			dev->data->socket_id);
+	if (p->rx_worker_ring == NULL) {
+		SW_LOG_ERR("Error creating RX worker ring for port %d\n",
+				port_id);
+		return -1;
+	}
+
+	p->inflight_max = conf->new_event_threshold;
+
+	snprintf(buf, sizeof(buf), "sw%d_%s", dev->data->dev_id,
+			"cq_worker_ring");
+	p->cq_worker_ring = qe_ring_create(buf, conf->dequeue_depth,
+			dev->data->socket_id);
+	if (p->cq_worker_ring == NULL) {
+		qe_ring_destroy(p->rx_worker_ring);
+		SW_LOG_ERR("Error creating CQ worker ring for port %d\n",
+				port_id);
+		return -1;
+	}
+	sw->cq_ring_space[port_id] = conf->dequeue_depth;
+
+	/* set hist list contents to empty */
+	for (i = 0; i < SW_PORT_HIST_LIST; i++) {
+		p->hist_list[i].fid = -1;
+		p->hist_list[i].qid = -1;
+	}
+	dev->data->ports[port_id] = p;
+
+	rte_smp_wmb();
+	p->initialized = 1;
+	return 0;
+}
+
+static void
+sw_port_release(void *port)
+{
+	struct sw_port *p = (void *)port;
+	if (p == NULL)
+		return;
+
+	qe_ring_destroy(p->rx_worker_ring);
+	qe_ring_destroy(p->cq_worker_ring);
+	memset(p, 0, sizeof(*p));
+}
+
+static int32_t
+qid_init(struct sw_evdev *sw, unsigned int idx, int type,
+		const struct rte_event_queue_conf *queue_conf)
+{
+	unsigned int i;
+	int dev_id = sw->data->dev_id;
+	int socket_id = sw->data->socket_id;
+	char buf[IQ_RING_NAMESIZE];
+	struct sw_qid *qid = &sw->qids[idx];
+
+	for (i = 0; i < SW_IQS_MAX; i++) {
+		snprintf(buf, sizeof(buf), "q_%u_iq_%d", idx, i);
+		qid->iq[i] = iq_ring_create(buf, socket_id);
+		if (!qid->iq[i]) {
+			SW_LOG_DBG("ring create failed");
+			goto cleanup;
+		}
+	}
+
+	/* Initialize the FID structures to no pinning (-1), and zero packets */
+	const struct sw_fid_t fid = {.cq = -1, .pcount = 0};
+	for (i = 0; i < RTE_DIM(qid->fids); i++)
+		qid->fids[i] = fid;
+
+	qid->id = idx;
+	qid->type = type;
+	qid->priority = queue_conf->priority;
+
+	if (qid->type == RTE_SCHED_TYPE_ORDERED) {
+		char ring_name[RTE_RING_NAMESIZE];
+		uint32_t window_size;
+
+		/* rte_ring and window_size_mask require require window_size to
+		 * be a power-of-2.
+		 */
+		window_size = rte_align32pow2(
+				queue_conf->nb_atomic_order_sequences);
+
+		qid->window_size = window_size - 1;
+
+		if (!window_size) {
+			SW_LOG_DBG(
+				"invalid reorder_window_size for ordered queue\n"
+				);
+			goto cleanup;
+		}
+
+		snprintf(buf, sizeof(buf), "sw%d_iq_%d_rob", dev_id, i);
+		qid->reorder_buffer = rte_zmalloc_socket(buf,
+				window_size * sizeof(qid->reorder_buffer[0]),
+				0, socket_id);
+		if (!qid->reorder_buffer) {
+			SW_LOG_DBG("reorder_buffer malloc failed\n");
+			goto cleanup;
+		}
+
+		memset(&qid->reorder_buffer[0],
+		       0,
+		       window_size * sizeof(qid->reorder_buffer[0]));
+
+		snprintf(ring_name, sizeof(ring_name), "sw%d_q%d_freelist",
+				dev_id, idx);
+
+		/* lookup the ring, and if it already exists, free it */
+		struct rte_ring *cleanup = rte_ring_lookup(ring_name);
+		if (cleanup)
+			rte_ring_free(cleanup);
+
+		qid->reorder_buffer_freelist = rte_ring_create(ring_name,
+				window_size,
+				socket_id,
+				RING_F_SP_ENQ | RING_F_SC_DEQ);
+		if (!qid->reorder_buffer_freelist) {
+			SW_LOG_DBG("freelist ring create failed");
+			goto cleanup;
+		}
+
+		/* Populate the freelist with reorder buffer entries. Enqueue
+		 * 'window_size - 1' entries because the rte_ring holds only
+		 * that many.
+		 */
+		for (i = 0; i < window_size - 1; i++) {
+			if (rte_ring_sp_enqueue(qid->reorder_buffer_freelist,
+						&qid->reorder_buffer[i]) < 0)
+				goto cleanup;
+		}
+
+		qid->reorder_buffer_index = 0;
+		qid->cq_next_tx = 0;
+	}
+
+	qid->initialized = 1;
+
+	return 0;
+
+cleanup:
+	for (i = 0; i < SW_IQS_MAX; i++) {
+		if (qid->iq[i])
+			iq_ring_destroy(qid->iq[i]);
+	}
+
+	if (qid->reorder_buffer) {
+		rte_free(qid->reorder_buffer);
+		qid->reorder_buffer = NULL;
+	}
+
+	if (qid->reorder_buffer_freelist) {
+		rte_ring_free(qid->reorder_buffer_freelist);
+		qid->reorder_buffer_freelist = NULL;
+	}
+
+	return -EINVAL;
+}
+
+static int
+sw_queue_setup(struct rte_eventdev *dev, uint8_t queue_id,
+		const struct rte_event_queue_conf *conf)
+{
+	int type;
+
+	/* SINGLE_LINK can be OR-ed with other types, so handle first */
+	if (RTE_EVENT_QUEUE_CFG_SINGLE_LINK & conf->event_queue_cfg) {
+		type = SW_SCHED_TYPE_DIRECT;
+	} else {
+		switch (conf->event_queue_cfg) {
+		case RTE_EVENT_QUEUE_CFG_ATOMIC_ONLY:
+			type = RTE_SCHED_TYPE_ATOMIC;
+			break;
+		case RTE_EVENT_QUEUE_CFG_ORDERED_ONLY:
+			type = RTE_SCHED_TYPE_ORDERED;
+			break;
+		case RTE_EVENT_QUEUE_CFG_PARALLEL_ONLY:
+			type = RTE_SCHED_TYPE_PARALLEL;
+			break;
+		case RTE_EVENT_QUEUE_CFG_ALL_TYPES:
+			SW_LOG_ERR("QUEUE_CFG_ALL_TYPES not supported\n");
+			return -ENOTSUP;
+		default:
+			SW_LOG_ERR("Unknown queue type %d requested\n",
+				   conf->event_queue_cfg);
+			return -EINVAL;
+		}
+	}
+
+	struct sw_evdev *sw = sw_pmd_priv(dev);
+	return qid_init(sw, queue_id, type, conf);
+}
+
+static void
+sw_queue_release(struct rte_eventdev *dev, uint8_t id)
+{
+	struct sw_evdev *sw = sw_pmd_priv(dev);
+	struct sw_qid *qid = &sw->qids[id];
+	uint32_t i;
+
+	for (i = 0; i < SW_IQS_MAX; i++)
+		iq_ring_destroy(qid->iq[i]);
+
+	if (qid->type == RTE_SCHED_TYPE_ORDERED) {
+		rte_free(qid->reorder_buffer);
+		rte_ring_free(qid->reorder_buffer_freelist);
+	}
+	memset(qid, 0, sizeof(*qid));
+}
+
+static void
+sw_queue_def_conf(struct rte_eventdev *dev, uint8_t queue_id,
+				 struct rte_event_queue_conf *conf)
+{
+	RTE_SET_USED(dev);
+	RTE_SET_USED(queue_id);
+
+	static const struct rte_event_queue_conf default_conf = {
+		.nb_atomic_flows = 4096,
+		.nb_atomic_order_sequences = 1,
+		.event_queue_cfg = RTE_EVENT_QUEUE_CFG_ATOMIC_ONLY,
+		.priority = RTE_EVENT_DEV_PRIORITY_NORMAL,
+	};
+
+	*conf = default_conf;
+}
+
+static void
+sw_port_def_conf(struct rte_eventdev *dev, uint8_t port_id,
+		 struct rte_event_port_conf *port_conf)
+{
+	RTE_SET_USED(dev);
+	RTE_SET_USED(port_id);
+
+	port_conf->new_event_threshold = 1024;
+	port_conf->dequeue_depth = 16;
+	port_conf->enqueue_depth = 16;
+}
+
+static int
+sw_dev_configure(const struct rte_eventdev *dev)
+{
+	struct sw_evdev *sw = sw_pmd_priv(dev);
+	const struct rte_eventdev_data *data = dev->data;
+	const struct rte_event_dev_config *conf = &data->dev_conf;
+
+	sw->qid_count = conf->nb_event_queues;
+	sw->port_count = conf->nb_event_ports;
+	sw->nb_events_limit = conf->nb_events_limit;
+	rte_atomic32_set(&sw->inflights, 0);
+
+	if (conf->event_dev_cfg & RTE_EVENT_DEV_CFG_PER_DEQUEUE_TIMEOUT)
+		return -ENOTSUP;
+
+	return 0;
+}
+
+static void
+sw_info_get(struct rte_eventdev *dev, struct rte_event_dev_info *info)
+{
+	RTE_SET_USED(dev);
+
+	static const struct rte_event_dev_info evdev_sw_info = {
+			.driver_name = SW_PMD_NAME,
+			.max_event_queues = RTE_EVENT_MAX_QUEUES_PER_DEV,
+			.max_event_queue_flows = SW_QID_NUM_FIDS,
+			.max_event_queue_priority_levels = SW_Q_PRIORITY_MAX,
+			.max_event_priority_levels = SW_IQS_MAX,
+			.max_event_ports = SW_PORTS_MAX,
+			.max_event_port_dequeue_depth = MAX_SW_CONS_Q_DEPTH,
+			.max_event_port_enqueue_depth = MAX_SW_PROD_Q_DEPTH,
+			.max_num_events = SW_INFLIGHT_EVENTS_TOTAL,
+			.event_dev_cap = (RTE_EVENT_DEV_CAP_QUEUE_QOS |
+					RTE_EVENT_DEV_CAP_EVENT_QOS),
+	};
+
+	*info = evdev_sw_info;
+}
+
+static void
+sw_dump(struct rte_eventdev *dev, FILE *f)
+{
+	const struct sw_evdev *sw = sw_pmd_priv(dev);
+
+	static const char * const q_type_strings[] = {
+			"Ordered", "Atomic", "Parallel", "Directed"
+	};
+	uint32_t i;
+	fprintf(f, "EventDev %s: ports %d, qids %d\n", "todo-fix-name",
+			sw->port_count, sw->qid_count);
+
+	fprintf(f, "\trx   %"PRIu64"\n\tdrop %"PRIu64"\n\ttx   %"PRIu64"\n",
+		sw->stats.rx_pkts, sw->stats.rx_dropped, sw->stats.tx_pkts);
+	fprintf(f, "\tsched calls: %"PRIu64"\n", sw->sched_called);
+	fprintf(f, "\tsched cq/qid call: %"PRIu64"\n", sw->sched_cq_qid_called);
+	fprintf(f, "\tsched no IQ enq: %"PRIu64"\n", sw->sched_no_iq_enqueues);
+	fprintf(f, "\tsched no CQ enq: %"PRIu64"\n", sw->sched_no_cq_enqueues);
+	uint32_t inflights = rte_atomic32_read(&sw->inflights);
+	uint32_t credits = sw->nb_events_limit - inflights;
+	fprintf(f, "\tinflight %d, credits: %d\n", inflights, credits);
+
+#define COL_RED "\x1b[31m"
+#define COL_RESET "\x1b[0m"
+
+	for (i = 0; i < sw->port_count; i++) {
+		int max, j;
+		const struct sw_port *p = &sw->ports[i];
+		if (!p->initialized) {
+			fprintf(f, "  %sPort %d not initialized.%s\n",
+				COL_RED, i, COL_RESET);
+			continue;
+		}
+		fprintf(f, "  Port %d %s\n", i,
+			p->is_directed ? " (SingleCons)" : "");
+		fprintf(f, "\trx   %"PRIu64"\tdrop %"PRIu64"\ttx   %"PRIu64
+			"\t%sinflight %d%s\n", sw->ports[i].stats.rx_pkts,
+			sw->ports[i].stats.rx_dropped,
+			sw->ports[i].stats.tx_pkts,
+			(p->inflights == p->inflight_max) ?
+				COL_RED : COL_RESET,
+			sw->ports[i].inflights, COL_RESET);
+
+		fprintf(f, "\tMax New: %u"
+			"\tAvg cycles PP: %"PRIu64"\tCredits: %u\n",
+			sw->ports[i].inflight_max,
+			sw->ports[i].avg_pkt_ticks,
+			sw->ports[i].inflight_credits);
+		fprintf(f, "\tReceive burst distribution:\n");
+		float zp_percent = p->zero_polls * 100.0 / p->total_polls;
+		fprintf(f, zp_percent < 10 ? "\t\t0:%.02f%% " : "\t\t0:%.0f%% ",
+				zp_percent);
+		for (max = (int)RTE_DIM(p->poll_buckets); max-- > 0;)
+			if (p->poll_buckets[max] != 0)
+				break;
+		for (j = 0; j <= max; j++) {
+			if (p->poll_buckets[j] != 0) {
+				float poll_pc = p->poll_buckets[j] * 100.0 /
+					p->total_polls;
+				fprintf(f, "%u-%u:%.02f%% ",
+					((j << SW_DEQ_STAT_BUCKET_SHIFT) + 1),
+					((j+1) << SW_DEQ_STAT_BUCKET_SHIFT),
+					poll_pc);
+			}
+		}
+		fprintf(f, "\n");
+
+		if (p->rx_worker_ring) {
+			uint64_t used = qe_ring_count(p->rx_worker_ring);
+			uint64_t space = qe_ring_free_count(p->rx_worker_ring);
+			const char *col = (space == 0) ? COL_RED : COL_RESET;
+			fprintf(f, "\t%srx ring used: %4"PRIu64"\tfree: %4"
+					PRIu64 COL_RESET"\n", col, used, space);
+		} else
+			fprintf(f, "\trx ring not initialized.\n");
+
+		if (p->cq_worker_ring) {
+			uint64_t used = qe_ring_count(p->cq_worker_ring);
+			uint64_t space = qe_ring_free_count(p->cq_worker_ring);
+			const char *col = (space == 0) ? COL_RED : COL_RESET;
+			fprintf(f, "\t%scq ring used: %4"PRIu64"\tfree: %4"
+					PRIu64 COL_RESET"\n", col, used, space);
+		} else
+			fprintf(f, "\tcq ring not initialized.\n");
+	}
+
+	for (i = 0; i < sw->qid_count; i++) {
+		const struct sw_qid *qid = &sw->qids[i];
+		if (!qid->initialized) {
+			fprintf(f, "  %sQueue %d not initialized.%s\n",
+				COL_RED, i, COL_RESET);
+			continue;
+		}
+		int affinities_per_port[SW_PORTS_MAX] = {0};
+		uint32_t inflights = 0;
+
+		fprintf(f, "  Queue %d (%s)\n", i, q_type_strings[qid->type]);
+		fprintf(f, "\trx   %"PRIu64"\tdrop %"PRIu64"\ttx   %"PRIu64"\n",
+			qid->stats.rx_pkts, qid->stats.rx_dropped,
+			qid->stats.tx_pkts);
+		if (qid->type == RTE_SCHED_TYPE_ORDERED) {
+			struct rte_ring *rob_buf_free =
+				qid->reorder_buffer_freelist;
+			if (rob_buf_free)
+				fprintf(f, "\tReorder entries in use: %u\n",
+					rte_ring_free_count(rob_buf_free));
+			else
+				fprintf(f,
+					"\tReorder buffer not initialized\n");
+		}
+
+		uint32_t flow;
+		for (flow = 0; flow < RTE_DIM(qid->fids); flow++)
+			if (qid->fids[flow].cq != -1) {
+				affinities_per_port[qid->fids[flow].cq]++;
+				inflights += qid->fids[flow].pcount;
+			}
+
+		uint32_t cq;
+		fprintf(f, "\tInflights: %u\tFlows pinned per port: ",
+				inflights);
+		for (cq = 0; cq < sw->port_count; cq++)
+			fprintf(f, "%d ", affinities_per_port[cq]);
+		fprintf(f, "\n");
+
+		uint32_t iq;
+		uint32_t iq_printed = 0;
+		for (iq = 0; iq < SW_IQS_MAX; iq++) {
+			if (!qid->iq[iq]) {
+				fprintf(f, "\tiq %d is not initialized.\n", iq);
+				iq_printed = 1;
+				continue;
+			}
+			uint32_t used = iq_ring_count(qid->iq[iq]);
+			uint32_t free = iq_ring_free_count(qid->iq[iq]);
+			const char *col = (free == 0) ? COL_RED : COL_RESET;
+			if (used > 0) {
+				fprintf(f, "\t%siq %d: Used %d\tFree %d"
+					COL_RESET"\n", col, iq, used, free);
+				iq_printed = 1;
+			}
+		}
+		if (iq_printed == 0)
+			fprintf(f, "\t-- iqs empty --\n");
+	}
+}
+
+static int
+sw_start(struct rte_eventdev *dev)
+{
+	unsigned int i, j;
+	struct sw_evdev *sw = sw_pmd_priv(dev);
+	/* check all ports are set up */
+	for (i = 0; i < sw->port_count; i++)
+		if (sw->ports[i].rx_worker_ring == NULL) {
+			SW_LOG_ERR("Port %d not configured\n", i);
+			return -ESTALE;
+		}
+
+	/* check all queues are configured and mapped to ports*/
+	for (i = 0; i < sw->qid_count; i++)
+		if (sw->qids[i].iq[0] == NULL ||
+				sw->qids[i].cq_num_mapped_cqs == 0) {
+			SW_LOG_ERR("Queue %d not configured\n", i);
+			return -ENOLINK;
+		}
+
+	/* build up our prioritized array of qids */
+	/* We don't use qsort here, as if all/multiple entries have the same
+	 * priority, the result is non-deterministic. From "man 3 qsort":
+	 * "If two members compare as equal, their order in the sorted
+	 * array is undefined."
+	 */
+	uint32_t qidx = 0;
+	for (j = 0; j <= RTE_EVENT_DEV_PRIORITY_LOWEST; j++) {
+		for (i = 0; i < sw->qid_count; i++) {
+			if (sw->qids[i].priority == j) {
+				sw->qids_prioritized[qidx] = &sw->qids[i];
+				qidx++;
+			}
+		}
+	}
+
+	if (sw_xstats_init(sw) < 0)
+		return -EINVAL;
+
+	rte_smp_wmb();
+	sw->started = 1;
+
+	return 0;
+}
+
+static void
+sw_stop(struct rte_eventdev *dev)
+{
+	struct sw_evdev *sw = sw_pmd_priv(dev);
+	sw_xstats_uninit(sw);
+	sw->started = 0;
+	rte_smp_wmb();
+}
+
+static int
+sw_close(struct rte_eventdev *dev)
+{
+	struct sw_evdev *sw = sw_pmd_priv(dev);
+	uint32_t i;
+
+	for (i = 0; i < sw->qid_count; i++)
+		sw_queue_release(dev, i);
+	sw->qid_count = 0;
+
+	for (i = 0; i < sw->port_count; i++)
+		sw_port_release(&sw->ports[i]);
+	sw->port_count = 0;
+
+	memset(&sw->stats, 0, sizeof(sw->stats));
+	sw->sched_called = 0;
+	sw->sched_no_iq_enqueues = 0;
+	sw->sched_no_cq_enqueues = 0;
+	sw->sched_cq_qid_called = 0;
+
+	return 0;
+}
+
+static int
+assign_numa_node(const char *key __rte_unused, const char *value, void *opaque)
+{
+	int *socket_id = opaque;
+	*socket_id = atoi(value);
+	if (*socket_id >= RTE_MAX_NUMA_NODES)
+		return -1;
+	return 0;
+}
+
+static int
+set_sched_quanta(const char *key __rte_unused, const char *value, void *opaque)
+{
+	int *quanta = opaque;
+	*quanta = atoi(value);
+	if (*quanta < 0 || *quanta >= 4096)
+		return -1;
+	return 0;
+}
+
+static int
+set_credit_quanta(const char *key __rte_unused, const char *value, void *opaque)
+{
+	int *credit = opaque;
+	*credit = atoi(value);
+	if (*credit < 0 || *credit >= 128)
+		return -1;
+	return 0;
+}
+
+static int
+sw_probe(struct rte_vdev_device *vdev)
+{
+	static const struct rte_eventdev_ops evdev_sw_ops = {
+			.dev_configure = sw_dev_configure,
+			.dev_infos_get = sw_info_get,
+			.dev_close = sw_close,
+			.dev_start = sw_start,
+			.dev_stop = sw_stop,
+			.dump = sw_dump,
+
+			.queue_def_conf = sw_queue_def_conf,
+			.queue_setup = sw_queue_setup,
+			.queue_release = sw_queue_release,
+			.port_def_conf = sw_port_def_conf,
+			.port_setup = sw_port_setup,
+			.port_release = sw_port_release,
+			.port_link = sw_port_link,
+			.port_unlink = sw_port_unlink,
+
+			.xstats_get = sw_xstats_get,
+			.xstats_get_names = sw_xstats_get_names,
+			.xstats_get_by_name = sw_xstats_get_by_name,
+			.xstats_reset = sw_xstats_reset,
+	};
+
+	static const char *const args[] = {
+		NUMA_NODE_ARG,
+		SCHED_QUANTA_ARG,
+		CREDIT_QUANTA_ARG,
+		NULL
+	};
+	const char *name;
+	const char *params;
+	struct rte_eventdev *dev;
+	struct sw_evdev *sw;
+	int socket_id = rte_socket_id();
+	int sched_quanta  = SW_DEFAULT_SCHED_QUANTA;
+	int credit_quanta = SW_DEFAULT_CREDIT_QUANTA;
+
+	name = rte_vdev_device_name(vdev);
+	params = rte_vdev_device_args(vdev);
+	if (params != NULL && params[0] != '\0') {
+		struct rte_kvargs *kvlist = rte_kvargs_parse(params, args);
+
+		if (!kvlist) {
+			SW_LOG_INFO(
+				"Ignoring unsupported parameters when creating device '%s'\n",
+				name);
+		} else {
+			int ret = rte_kvargs_process(kvlist, NUMA_NODE_ARG,
+					assign_numa_node, &socket_id);
+			if (ret != 0) {
+				SW_LOG_ERR(
+					"%s: Error parsing numa node parameter",
+					name);
+				rte_kvargs_free(kvlist);
+				return ret;
+			}
+
+			ret = rte_kvargs_process(kvlist, SCHED_QUANTA_ARG,
+					set_sched_quanta, &sched_quanta);
+			if (ret != 0) {
+				SW_LOG_ERR(
+					"%s: Error parsing sched quanta parameter",
+					name);
+				rte_kvargs_free(kvlist);
+				return ret;
+			}
+
+			ret = rte_kvargs_process(kvlist, CREDIT_QUANTA_ARG,
+					set_credit_quanta, &credit_quanta);
+			if (ret != 0) {
+				SW_LOG_ERR(
+					"%s: Error parsing credit quanta parameter",
+					name);
+				rte_kvargs_free(kvlist);
+				return ret;
+			}
+
+			rte_kvargs_free(kvlist);
+		}
+	}
+
+	SW_LOG_INFO(
+			"Creating eventdev sw device %s, numa_node=%d, sched_quanta=%d, credit_quanta=%d\n",
+			name, socket_id, sched_quanta, credit_quanta);
+
+	dev = rte_event_pmd_vdev_init(name,
+			sizeof(struct sw_evdev), socket_id);
+	if (dev == NULL) {
+		SW_LOG_ERR("eventdev vdev init() failed");
+		return -EFAULT;
+	}
+	dev->dev_ops = &evdev_sw_ops;
+	dev->enqueue = sw_event_enqueue;
+	dev->enqueue_burst = sw_event_enqueue_burst;
+	dev->dequeue = sw_event_dequeue;
+	dev->dequeue_burst = sw_event_dequeue_burst;
+	dev->schedule = sw_event_schedule;
+
+	if (rte_eal_process_type() != RTE_PROC_PRIMARY)
+		return 0;
+
+	sw = dev->data->dev_private;
+	sw->data = dev->data;
+
+	/* copy values passed from vdev command line to instance */
+	sw->credit_update_quanta = credit_quanta;
+	sw->sched_quanta = sched_quanta;
+
+	return 0;
+}
+
+static int
+sw_remove(struct rte_vdev_device *vdev)
+{
+	const char *name;
+
+	name = rte_vdev_device_name(vdev);
+	if (name == NULL)
+		return -EINVAL;
+
+	SW_LOG_INFO("Closing eventdev sw device %s\n", name);
+
+	return rte_event_pmd_vdev_uninit(name);
+}
+
+static struct rte_vdev_driver evdev_sw_pmd_drv = {
+	.probe = sw_probe,
+	.remove = sw_remove
+};
+
+RTE_PMD_REGISTER_VDEV(EVENTDEV_NAME_SW_PMD, evdev_sw_pmd_drv);
+RTE_PMD_REGISTER_PARAM_STRING(event_sw, NUMA_NODE_ARG "=<int> "
+		SCHED_QUANTA_ARG "=<int>" CREDIT_QUANTA_ARG "=<int>");
diff --git a/src/seastar/dpdk/drivers/event/sw/sw_evdev.h b/src/seastar/dpdk/drivers/event/sw/sw_evdev.h
new file mode 100644
index 00000000..61c671d6
--- /dev/null
+++ b/src/seastar/dpdk/drivers/event/sw/sw_evdev.h
@@ -0,0 +1,318 @@
+/*-
+ *   BSD LICENSE
+ *
+ *   Copyright(c) 2016-2017 Intel Corporation. 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 _SW_EVDEV_H_
+#define _SW_EVDEV_H_
+
+#include <rte_eventdev.h>
+#include <rte_eventdev_pmd.h>
+#include <rte_atomic.h>
+
+#define SW_DEFAULT_CREDIT_QUANTA 32
+#define SW_DEFAULT_SCHED_QUANTA 128
+#define SW_QID_NUM_FIDS 16384
+#define SW_IQS_MAX 4
+#define SW_Q_PRIORITY_MAX 255
+#define SW_PORTS_MAX 64
+#define MAX_SW_CONS_Q_DEPTH 128
+#define SW_INFLIGHT_EVENTS_TOTAL 4096
+/* allow for lots of over-provisioning */
+#define MAX_SW_PROD_Q_DEPTH 4096
+#define SW_FRAGMENTS_MAX 16
+
+/* report dequeue burst sizes in buckets */
+#define SW_DEQ_STAT_BUCKET_SHIFT 2
+/* how many packets pulled from port by sched */
+#define SCHED_DEQUEUE_BURST_SIZE 32
+
+#define SW_PORT_HIST_LIST (MAX_SW_PROD_Q_DEPTH) /* size of our history list */
+#define NUM_SAMPLES 64 /* how many data points use for average stats */
+
+#define EVENTDEV_NAME_SW_PMD event_sw
+#define SW_PMD_NAME RTE_STR(event_sw)
+
+#define SW_SCHED_TYPE_DIRECT (RTE_SCHED_TYPE_PARALLEL + 1)
+
+#define SW_NUM_POLL_BUCKETS (MAX_SW_CONS_Q_DEPTH >> SW_DEQ_STAT_BUCKET_SHIFT)
+
+enum {
+	QE_FLAG_VALID_SHIFT = 0,
+	QE_FLAG_COMPLETE_SHIFT,
+	QE_FLAG_NOT_EOP_SHIFT,
+	_QE_FLAG_COUNT
+};
+
+#define QE_FLAG_VALID    (1 << QE_FLAG_VALID_SHIFT)    /* for NEW FWD, FRAG */
+#define QE_FLAG_COMPLETE (1 << QE_FLAG_COMPLETE_SHIFT) /* set for FWD, DROP  */
+#define QE_FLAG_NOT_EOP  (1 << QE_FLAG_NOT_EOP_SHIFT)  /* set for FRAG only  */
+
+static const uint8_t sw_qe_flag_map[] = {
+		QE_FLAG_VALID /* NEW Event */,
+		QE_FLAG_VALID | QE_FLAG_COMPLETE /* FWD Event */,
+		QE_FLAG_COMPLETE /* RELEASE Event */,
+
+		/* Values which can be used for future support for partial
+		 * events, i.e. where one event comes back to the scheduler
+		 * as multiple which need to be tracked together
+		 */
+		QE_FLAG_VALID | QE_FLAG_COMPLETE | QE_FLAG_NOT_EOP,
+};
+
+#ifdef RTE_LIBRTE_PMD_EVDEV_SW_DEBUG
+#define SW_LOG_INFO(fmt, args...) \
+	RTE_LOG(INFO, EVENTDEV, "[%s] %s() line %u: " fmt "\n", \
+			SW_PMD_NAME, \
+			__func__, __LINE__, ## args)
+
+#define SW_LOG_DBG(fmt, args...) \
+	RTE_LOG(DEBUG, EVENTDEV, "[%s] %s() line %u: " fmt "\n", \
+			SW_PMD_NAME, \
+			__func__, __LINE__, ## args)
+#else
+#define SW_LOG_INFO(fmt, args...)
+#define SW_LOG_DBG(fmt, args...)
+#endif
+
+#define SW_LOG_ERR(fmt, args...) \
+	RTE_LOG(ERR, EVENTDEV, "[%s] %s() line %u: " fmt "\n", \
+			SW_PMD_NAME, \
+			__func__, __LINE__, ## args)
+
+/* Records basic event stats at a given point. Used in port and qid structs */
+struct sw_point_stats {
+	uint64_t rx_pkts;
+	uint64_t rx_dropped;
+	uint64_t tx_pkts;
+};
+
+/* structure used to track what port a flow (FID) is pinned to */
+struct sw_fid_t {
+	/* which CQ this FID is currently pinned to */
+	int32_t cq;
+	/* number of packets gone to the CQ with this FID */
+	uint32_t pcount;
+};
+
+struct reorder_buffer_entry {
+	uint16_t num_fragments;		/**< Number of packet fragments */
+	uint16_t fragment_index;	/**< Points to the oldest valid frag */
+	uint8_t ready;			/**< Entry is ready to be reordered */
+	struct rte_event fragments[SW_FRAGMENTS_MAX];
+};
+
+struct sw_qid {
+	/* set when the QID has been initialized */
+	uint8_t initialized;
+	/* The type of this QID */
+	int8_t type;
+	/* Integer ID representing the queue. This is used in history lists,
+	 * to identify the stage of processing.
+	 */
+	uint32_t id;
+	struct sw_point_stats stats;
+
+	/* Internal priority rings for packets */
+	struct iq_ring *iq[SW_IQS_MAX];
+	uint32_t iq_pkt_mask; /* A mask to indicate packets in an IQ */
+	uint64_t iq_pkt_count[SW_IQS_MAX];
+
+	/* Information on what CQs are polling this IQ */
+	uint32_t cq_num_mapped_cqs;
+	uint32_t cq_next_tx; /* cq to write next (non-atomic) packet */
+	uint32_t cq_map[SW_PORTS_MAX];
+
+	/* Track flow ids for atomic load balancing */
+	struct sw_fid_t fids[SW_QID_NUM_FIDS];
+
+	/* Track packet order for reordering when needed */
+	struct reorder_buffer_entry *reorder_buffer; /*< pkts await reorder */
+	struct rte_ring *reorder_buffer_freelist; /* available reorder slots */
+	uint32_t reorder_buffer_index; /* oldest valid reorder buffer entry */
+	uint32_t window_size;          /* Used to wrap reorder_buffer_index */
+
+	uint8_t priority;
+};
+
+struct sw_hist_list_entry {
+	int32_t qid;
+	int32_t fid;
+	struct reorder_buffer_entry *rob_entry;
+};
+
+struct sw_evdev;
+
+struct sw_port {
+	/* new enqueue / dequeue API doesn't have an instance pointer, only the
+	 * pointer to the port being enqueue/dequeued from
+	 */
+	struct sw_evdev *sw;
+
+	/* set when the port is initialized */
+	uint8_t initialized;
+	/* A numeric ID for the port */
+	uint8_t id;
+
+	int16_t is_directed; /** Takes from a single directed QID */
+	/**
+	 * For loadbalanced we can optimise pulling packets from
+	 * producers if there is no reordering involved
+	 */
+	int16_t num_ordered_qids;
+
+	/** Ring and buffer for pulling events from workers for scheduling */
+	struct qe_ring *rx_worker_ring __rte_cache_aligned;
+	/** Ring and buffer for pushing packets to workers after scheduling */
+	struct qe_ring *cq_worker_ring;
+
+	/* hole */
+
+	/* num releases yet to be completed on this port */
+	uint16_t outstanding_releases __rte_cache_aligned;
+	uint16_t inflight_max; /* app requested max inflights for this port */
+	uint16_t inflight_credits; /* num credits this port has right now */
+
+	uint16_t last_dequeue_burst_sz; /* how big the burst was */
+	uint64_t last_dequeue_ticks; /* used to track burst processing time */
+	uint64_t avg_pkt_ticks;      /* tracks average over NUM_SAMPLES burst */
+	uint64_t total_polls;        /* how many polls were counted in stats */
+	uint64_t zero_polls;         /* tracks polls returning nothing */
+	uint32_t poll_buckets[SW_NUM_POLL_BUCKETS];
+		/* bucket values in 4s for shorter reporting */
+
+	/* History list structs, containing info on pkts egressed to worker */
+	uint16_t hist_head __rte_cache_aligned;
+	uint16_t hist_tail;
+	uint16_t inflights;
+	struct sw_hist_list_entry hist_list[SW_PORT_HIST_LIST];
+
+	/* track packets in and out of this port */
+	struct sw_point_stats stats;
+
+
+	uint32_t pp_buf_start;
+	uint32_t pp_buf_count;
+	uint16_t cq_buf_count;
+	struct rte_event pp_buf[SCHED_DEQUEUE_BURST_SIZE];
+	struct rte_event cq_buf[MAX_SW_CONS_Q_DEPTH];
+
+	uint8_t num_qids_mapped;
+};
+
+struct sw_evdev {
+	struct rte_eventdev_data *data;
+
+	uint32_t port_count;
+	uint32_t qid_count;
+	uint32_t xstats_count;
+	struct sw_xstats_entry *xstats;
+	uint32_t xstats_count_mode_dev;
+	uint32_t xstats_count_mode_port;
+	uint32_t xstats_count_mode_queue;
+
+	/* Contains all ports - load balanced and directed */
+	struct sw_port ports[SW_PORTS_MAX] __rte_cache_aligned;
+
+	rte_atomic32_t inflights __rte_cache_aligned;
+
+	/*
+	 * max events in this instance. Cached here for performance.
+	 * (also available in data->conf.nb_events_limit)
+	 */
+	uint32_t nb_events_limit;
+
+	/* Internal queues - one per logical queue */
+	struct sw_qid qids[RTE_EVENT_MAX_QUEUES_PER_DEV] __rte_cache_aligned;
+
+	/* Cache how many packets are in each cq */
+	uint16_t cq_ring_space[SW_PORTS_MAX] __rte_cache_aligned;
+
+	/* Array of pointers to load-balanced QIDs sorted by priority level */
+	struct sw_qid *qids_prioritized[RTE_EVENT_MAX_QUEUES_PER_DEV];
+
+	/* Stats */
+	struct sw_point_stats stats __rte_cache_aligned;
+	uint64_t sched_called;
+	int32_t sched_quanta;
+	uint64_t sched_no_iq_enqueues;
+	uint64_t sched_no_cq_enqueues;
+	uint64_t sched_cq_qid_called;
+
+	uint8_t started;
+	uint32_t credit_update_quanta;
+
+	/* store num stats and offset of the stats for each port */
+	uint16_t xstats_count_per_port[SW_PORTS_MAX];
+	uint16_t xstats_offset_for_port[SW_PORTS_MAX];
+	/* store num stats and offset of the stats for each queue */
+	uint16_t xstats_count_per_qid[RTE_EVENT_MAX_QUEUES_PER_DEV];
+	uint16_t xstats_offset_for_qid[RTE_EVENT_MAX_QUEUES_PER_DEV];
+};
+
+static inline struct sw_evdev *
+sw_pmd_priv(const struct rte_eventdev *eventdev)
+{
+	return eventdev->data->dev_private;
+}
+
+static inline const struct sw_evdev *
+sw_pmd_priv_const(const struct rte_eventdev *eventdev)
+{
+	return eventdev->data->dev_private;
+}
+
+uint16_t sw_event_enqueue(void *port, const struct rte_event *ev);
+uint16_t sw_event_enqueue_burst(void *port, const struct rte_event ev[],
+		uint16_t num);
+
+uint16_t sw_event_dequeue(void *port, struct rte_event *ev, uint64_t wait);
+uint16_t sw_event_dequeue_burst(void *port, struct rte_event *ev, uint16_t num,
+			uint64_t wait);
+void sw_event_schedule(struct rte_eventdev *dev);
+int sw_xstats_init(struct sw_evdev *dev);
+int sw_xstats_uninit(struct sw_evdev *dev);
+int sw_xstats_get_names(const struct rte_eventdev *dev,
+	enum rte_event_dev_xstats_mode mode, uint8_t queue_port_id,
+	struct rte_event_dev_xstats_name *xstats_names,
+	unsigned int *ids, unsigned int size);
+int sw_xstats_get(const struct rte_eventdev *dev,
+		enum rte_event_dev_xstats_mode mode, uint8_t queue_port_id,
+		const unsigned int ids[], uint64_t values[], unsigned int n);
+uint64_t sw_xstats_get_by_name(const struct rte_eventdev *dev,
+		const char *name, unsigned int *id);
+int sw_xstats_reset(struct rte_eventdev *dev,
+		enum rte_event_dev_xstats_mode mode,
+		int16_t queue_port_id,
+		const uint32_t ids[],
+		uint32_t nb_ids);
+
+
+#endif /* _SW_EVDEV_H_ */
diff --git a/src/seastar/dpdk/drivers/event/sw/sw_evdev_scheduler.c b/src/seastar/dpdk/drivers/event/sw/sw_evdev_scheduler.c
new file mode 100644
index 00000000..a333a6f0
--- /dev/null
+++ b/src/seastar/dpdk/drivers/event/sw/sw_evdev_scheduler.c
@@ -0,0 +1,601 @@
+/*-
+ *   BSD LICENSE
+ *
+ *   Copyright(c) 2016-2017 Intel Corporation. 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>
+#include <rte_hash_crc.h>
+#include "sw_evdev.h"
+#include "iq_ring.h"
+#include "event_ring.h"
+
+#define SW_IQS_MASK (SW_IQS_MAX-1)
+
+/* Retrieve the highest priority IQ or -1 if no pkts available. Doing the
+ * CLZ twice is faster than caching the value due to data dependencies
+ */
+#define PKT_MASK_TO_IQ(pkts) \
+	(__builtin_ctz(pkts | (1 << SW_IQS_MAX)))
+
+#if SW_IQS_MAX != 4
+#error Misconfigured PRIO_TO_IQ caused by SW_IQS_MAX value change
+#endif
+#define PRIO_TO_IQ(prio) (prio >> 6)
+
+#define MAX_PER_IQ_DEQUEUE 48
+#define FLOWID_MASK (SW_QID_NUM_FIDS-1)
+/* use cheap bit mixing, we only need to lose a few bits */
+#define SW_HASH_FLOWID(f) (((f) ^ (f >> 10)) & FLOWID_MASK)
+
+static inline uint32_t
+sw_schedule_atomic_to_cq(struct sw_evdev *sw, struct sw_qid * const qid,
+		uint32_t iq_num, unsigned int count)
+{
+	struct rte_event qes[MAX_PER_IQ_DEQUEUE]; /* count <= MAX */
+	struct rte_event blocked_qes[MAX_PER_IQ_DEQUEUE];
+	uint32_t nb_blocked = 0;
+	uint32_t i;
+
+	if (count > MAX_PER_IQ_DEQUEUE)
+		count = MAX_PER_IQ_DEQUEUE;
+
+	/* This is the QID ID. The QID ID is static, hence it can be
+	 * used to identify the stage of processing in history lists etc
+	 */
+	uint32_t qid_id = qid->id;
+
+	iq_ring_dequeue_burst(qid->iq[iq_num], qes, count);
+	for (i = 0; i < count; i++) {
+		const struct rte_event *qe = &qes[i];
+		const uint16_t flow_id = SW_HASH_FLOWID(qes[i].flow_id);
+		struct sw_fid_t *fid = &qid->fids[flow_id];
+		int cq = fid->cq;
+
+		if (cq < 0) {
+			uint32_t cq_idx = qid->cq_next_tx++;
+			if (qid->cq_next_tx == qid->cq_num_mapped_cqs)
+				qid->cq_next_tx = 0;
+			cq = qid->cq_map[cq_idx];
+
+			/* find least used */
+			int cq_free_cnt = sw->cq_ring_space[cq];
+			for (cq_idx = 0; cq_idx < qid->cq_num_mapped_cqs;
+					cq_idx++) {
+				int test_cq = qid->cq_map[cq_idx];
+				int test_cq_free = sw->cq_ring_space[test_cq];
+				if (test_cq_free > cq_free_cnt) {
+					cq = test_cq;
+					cq_free_cnt = test_cq_free;
+				}
+			}
+
+			fid->cq = cq; /* this pins early */
+		}
+
+		if (sw->cq_ring_space[cq] == 0 ||
+				sw->ports[cq].inflights == SW_PORT_HIST_LIST) {
+			blocked_qes[nb_blocked++] = *qe;
+			continue;
+		}
+
+		struct sw_port *p = &sw->ports[cq];
+
+		/* at this point we can queue up the packet on the cq_buf */
+		fid->pcount++;
+		p->cq_buf[p->cq_buf_count++] = *qe;
+		p->inflights++;
+		sw->cq_ring_space[cq]--;
+
+		int head = (p->hist_head++ & (SW_PORT_HIST_LIST-1));
+		p->hist_list[head].fid = flow_id;
+		p->hist_list[head].qid = qid_id;
+
+		p->stats.tx_pkts++;
+		qid->stats.tx_pkts++;
+
+		/* if we just filled in the last slot, flush the buffer */
+		if (sw->cq_ring_space[cq] == 0) {
+			struct qe_ring *worker = p->cq_worker_ring;
+			qe_ring_enqueue_burst(worker, p->cq_buf,
+					p->cq_buf_count,
+					&sw->cq_ring_space[cq]);
+			p->cq_buf_count = 0;
+		}
+	}
+	iq_ring_put_back(qid->iq[iq_num], blocked_qes, nb_blocked);
+
+	return count - nb_blocked;
+}
+
+static inline uint32_t
+sw_schedule_parallel_to_cq(struct sw_evdev *sw, struct sw_qid * const qid,
+		uint32_t iq_num, unsigned int count, int keep_order)
+{
+	uint32_t i;
+	uint32_t cq_idx = qid->cq_next_tx;
+
+	/* This is the QID ID. The QID ID is static, hence it can be
+	 * used to identify the stage of processing in history lists etc
+	 */
+	uint32_t qid_id = qid->id;
+
+	if (count > MAX_PER_IQ_DEQUEUE)
+		count = MAX_PER_IQ_DEQUEUE;
+
+	if (keep_order)
+		/* only schedule as many as we have reorder buffer entries */
+		count = RTE_MIN(count,
+				rte_ring_count(qid->reorder_buffer_freelist));
+
+	for (i = 0; i < count; i++) {
+		const struct rte_event *qe = iq_ring_peek(qid->iq[iq_num]);
+		uint32_t cq_check_count = 0;
+		uint32_t cq;
+
+		/*
+		 *  for parallel, just send to next available CQ in round-robin
+		 * fashion. So scan for an available CQ. If all CQs are full
+		 * just return and move on to next QID
+		 */
+		do {
+			if (++cq_check_count > qid->cq_num_mapped_cqs)
+				goto exit;
+			cq = qid->cq_map[cq_idx];
+			if (++cq_idx == qid->cq_num_mapped_cqs)
+				cq_idx = 0;
+		} while (qe_ring_free_count(sw->ports[cq].cq_worker_ring) == 0 ||
+				sw->ports[cq].inflights == SW_PORT_HIST_LIST);
+
+		struct sw_port *p = &sw->ports[cq];
+		if (sw->cq_ring_space[cq] == 0 ||
+				p->inflights == SW_PORT_HIST_LIST)
+			break;
+
+		sw->cq_ring_space[cq]--;
+
+		qid->stats.tx_pkts++;
+
+		const int head = (p->hist_head & (SW_PORT_HIST_LIST-1));
+		p->hist_list[head].fid = SW_HASH_FLOWID(qe->flow_id);
+		p->hist_list[head].qid = qid_id;
+
+		if (keep_order)
+			rte_ring_sc_dequeue(qid->reorder_buffer_freelist,
+					(void *)&p->hist_list[head].rob_entry);
+
+		sw->ports[cq].cq_buf[sw->ports[cq].cq_buf_count++] = *qe;
+		iq_ring_pop(qid->iq[iq_num]);
+
+		rte_compiler_barrier();
+		p->inflights++;
+		p->stats.tx_pkts++;
+		p->hist_head++;
+	}
+exit:
+	qid->cq_next_tx = cq_idx;
+	return i;
+}
+
+static uint32_t
+sw_schedule_dir_to_cq(struct sw_evdev *sw, struct sw_qid * const qid,
+		uint32_t iq_num, unsigned int count __rte_unused)
+{
+	uint32_t cq_id = qid->cq_map[0];
+	struct sw_port *port = &sw->ports[cq_id];
+
+	/* get max burst enq size for cq_ring */
+	uint32_t count_free = sw->cq_ring_space[cq_id];
+	if (count_free == 0)
+		return 0;
+
+	/* burst dequeue from the QID IQ ring */
+	struct iq_ring *ring = qid->iq[iq_num];
+	uint32_t ret = iq_ring_dequeue_burst(ring,
+			&port->cq_buf[port->cq_buf_count], count_free);
+	port->cq_buf_count += ret;
+
+	/* Update QID, Port and Total TX stats */
+	qid->stats.tx_pkts += ret;
+	port->stats.tx_pkts += ret;
+
+	/* Subtract credits from cached value */
+	sw->cq_ring_space[cq_id] -= ret;
+
+	return ret;
+}
+
+static uint32_t
+sw_schedule_qid_to_cq(struct sw_evdev *sw)
+{
+	uint32_t pkts = 0;
+	uint32_t qid_idx;
+
+	sw->sched_cq_qid_called++;
+
+	for (qid_idx = 0; qid_idx < sw->qid_count; qid_idx++) {
+		struct sw_qid *qid = sw->qids_prioritized[qid_idx];
+
+		int type = qid->type;
+		int iq_num = PKT_MASK_TO_IQ(qid->iq_pkt_mask);
+
+		/* zero mapped CQs indicates directed */
+		if (iq_num >= SW_IQS_MAX)
+			continue;
+
+		uint32_t pkts_done = 0;
+		uint32_t count = iq_ring_count(qid->iq[iq_num]);
+
+		if (count > 0) {
+			if (type == SW_SCHED_TYPE_DIRECT)
+				pkts_done += sw_schedule_dir_to_cq(sw, qid,
+						iq_num, count);
+			else if (type == RTE_SCHED_TYPE_ATOMIC)
+				pkts_done += sw_schedule_atomic_to_cq(sw, qid,
+						iq_num, count);
+			else
+				pkts_done += sw_schedule_parallel_to_cq(sw, qid,
+						iq_num, count,
+						type == RTE_SCHED_TYPE_ORDERED);
+		}
+
+		/* Check if the IQ that was polled is now empty, and unset it
+		 * in the IQ mask if its empty.
+		 */
+		int all_done = (pkts_done == count);
+
+		qid->iq_pkt_mask &= ~(all_done << (iq_num));
+		pkts += pkts_done;
+	}
+
+	return pkts;
+}
+
+/* This function will perform re-ordering of packets, and injecting into
+ * the appropriate QID IQ. As LB and DIR QIDs are in the same array, but *NOT*
+ * contiguous in that array, this function accepts a "range" of QIDs to scan.
+ */
+static uint16_t
+sw_schedule_reorder(struct sw_evdev *sw, int qid_start, int qid_end)
+{
+	/* Perform egress reordering */
+	struct rte_event *qe;
+	uint32_t pkts_iter = 0;
+
+	for (; qid_start < qid_end; qid_start++) {
+		struct sw_qid *qid = &sw->qids[qid_start];
+		int i, num_entries_in_use;
+
+		if (qid->type != RTE_SCHED_TYPE_ORDERED)
+			continue;
+
+		num_entries_in_use = rte_ring_free_count(
+					qid->reorder_buffer_freelist);
+
+		for (i = 0; i < num_entries_in_use; i++) {
+			struct reorder_buffer_entry *entry;
+			int j;
+
+			entry = &qid->reorder_buffer[qid->reorder_buffer_index];
+
+			if (!entry->ready)
+				break;
+
+			for (j = 0; j < entry->num_fragments; j++) {
+				uint16_t dest_qid;
+				uint16_t dest_iq;
+
+				int idx = entry->fragment_index + j;
+				qe = &entry->fragments[idx];
+
+				dest_qid = qe->queue_id;
+				dest_iq  = PRIO_TO_IQ(qe->priority);
+
+				if (dest_qid >= sw->qid_count) {
+					sw->stats.rx_dropped++;
+					continue;
+				}
+
+				struct sw_qid *dest_qid_ptr =
+					&sw->qids[dest_qid];
+				const struct iq_ring *dest_iq_ptr =
+					dest_qid_ptr->iq[dest_iq];
+				if (iq_ring_free_count(dest_iq_ptr) == 0)
+					break;
+
+				pkts_iter++;
+
+				struct sw_qid *q = &sw->qids[dest_qid];
+				struct iq_ring *r = q->iq[dest_iq];
+
+				/* we checked for space above, so enqueue must
+				 * succeed
+				 */
+				iq_ring_enqueue(r, qe);
+				q->iq_pkt_mask |= (1 << (dest_iq));
+				q->iq_pkt_count[dest_iq]++;
+				q->stats.rx_pkts++;
+			}
+
+			entry->ready = (j != entry->num_fragments);
+			entry->num_fragments -= j;
+			entry->fragment_index += j;
+
+			if (!entry->ready) {
+				entry->fragment_index = 0;
+
+				rte_ring_sp_enqueue(
+						qid->reorder_buffer_freelist,
+						entry);
+
+				qid->reorder_buffer_index++;
+				qid->reorder_buffer_index %= qid->window_size;
+			}
+		}
+	}
+	return pkts_iter;
+}
+
+static inline void __attribute__((always_inline))
+sw_refill_pp_buf(struct sw_evdev *sw, struct sw_port *port)
+{
+	RTE_SET_USED(sw);
+	struct qe_ring *worker = port->rx_worker_ring;
+	port->pp_buf_start = 0;
+	port->pp_buf_count = qe_ring_dequeue_burst(worker, port->pp_buf,
+			RTE_DIM(port->pp_buf));
+}
+
+static inline uint32_t __attribute__((always_inline))
+__pull_port_lb(struct sw_evdev *sw, uint32_t port_id, int allow_reorder)
+{
+	static struct reorder_buffer_entry dummy_rob;
+	uint32_t pkts_iter = 0;
+	struct sw_port *port = &sw->ports[port_id];
+
+	/* If shadow ring has 0 pkts, pull from worker ring */
+	if (port->pp_buf_count == 0)
+		sw_refill_pp_buf(sw, port);
+
+	while (port->pp_buf_count) {
+		const struct rte_event *qe = &port->pp_buf[port->pp_buf_start];
+		struct sw_hist_list_entry *hist_entry = NULL;
+		uint8_t flags = qe->op;
+		const uint16_t eop = !(flags & QE_FLAG_NOT_EOP);
+		int needs_reorder = 0;
+		/* if no-reordering, having PARTIAL == NEW */
+		if (!allow_reorder && !eop)
+			flags = QE_FLAG_VALID;
+
+		/*
+		 * if we don't have space for this packet in an IQ,
+		 * then move on to next queue. Technically, for a
+		 * packet that needs reordering, we don't need to check
+		 * here, but it simplifies things not to special-case
+		 */
+		uint32_t iq_num = PRIO_TO_IQ(qe->priority);
+		struct sw_qid *qid = &sw->qids[qe->queue_id];
+
+		if ((flags & QE_FLAG_VALID) &&
+				iq_ring_free_count(qid->iq[iq_num]) == 0)
+			break;
+
+		/* now process based on flags. Note that for directed
+		 * queues, the enqueue_flush masks off all but the
+		 * valid flag. This makes FWD and PARTIAL enqueues just
+		 * NEW type, and makes DROPS no-op calls.
+		 */
+		if ((flags & QE_FLAG_COMPLETE) && port->inflights > 0) {
+			const uint32_t hist_tail = port->hist_tail &
+					(SW_PORT_HIST_LIST - 1);
+
+			hist_entry = &port->hist_list[hist_tail];
+			const uint32_t hist_qid = hist_entry->qid;
+			const uint32_t hist_fid = hist_entry->fid;
+
+			struct sw_fid_t *fid =
+				&sw->qids[hist_qid].fids[hist_fid];
+			fid->pcount -= eop;
+			if (fid->pcount == 0)
+				fid->cq = -1;
+
+			if (allow_reorder) {
+				/* set reorder ready if an ordered QID */
+				uintptr_t rob_ptr =
+					(uintptr_t)hist_entry->rob_entry;
+				const uintptr_t valid = (rob_ptr != 0);
+				needs_reorder = valid;
+				rob_ptr |=
+					((valid - 1) & (uintptr_t)&dummy_rob);
+				struct reorder_buffer_entry *tmp_rob_ptr =
+					(struct reorder_buffer_entry *)rob_ptr;
+				tmp_rob_ptr->ready = eop * needs_reorder;
+			}
+
+			port->inflights -= eop;
+			port->hist_tail += eop;
+		}
+		if (flags & QE_FLAG_VALID) {
+			port->stats.rx_pkts++;
+
+			if (allow_reorder && needs_reorder) {
+				struct reorder_buffer_entry *rob_entry =
+						hist_entry->rob_entry;
+
+				hist_entry->rob_entry = NULL;
+				/* Although fragmentation not currently
+				 * supported by eventdev API, we support it
+				 * here. Open: How do we alert the user that
+				 * they've exceeded max frags?
+				 */
+				int num_frag = rob_entry->num_fragments;
+				if (num_frag == SW_FRAGMENTS_MAX)
+					sw->stats.rx_dropped++;
+				else {
+					int idx = rob_entry->num_fragments++;
+					rob_entry->fragments[idx] = *qe;
+				}
+				goto end_qe;
+			}
+
+			/* Use the iq_num from above to push the QE
+			 * into the qid at the right priority
+			 */
+
+			qid->iq_pkt_mask |= (1 << (iq_num));
+			iq_ring_enqueue(qid->iq[iq_num], qe);
+			qid->iq_pkt_count[iq_num]++;
+			qid->stats.rx_pkts++;
+			pkts_iter++;
+		}
+
+end_qe:
+		port->pp_buf_start++;
+		port->pp_buf_count--;
+	} /* while (avail_qes) */
+
+	return pkts_iter;
+}
+
+static uint32_t
+sw_schedule_pull_port_lb(struct sw_evdev *sw, uint32_t port_id)
+{
+	return __pull_port_lb(sw, port_id, 1);
+}
+
+static uint32_t
+sw_schedule_pull_port_no_reorder(struct sw_evdev *sw, uint32_t port_id)
+{
+	return __pull_port_lb(sw, port_id, 0);
+}
+
+static uint32_t
+sw_schedule_pull_port_dir(struct sw_evdev *sw, uint32_t port_id)
+{
+	uint32_t pkts_iter = 0;
+	struct sw_port *port = &sw->ports[port_id];
+
+	/* If shadow ring has 0 pkts, pull from worker ring */
+	if (port->pp_buf_count == 0)
+		sw_refill_pp_buf(sw, port);
+
+	while (port->pp_buf_count) {
+		const struct rte_event *qe = &port->pp_buf[port->pp_buf_start];
+		uint8_t flags = qe->op;
+
+		if ((flags & QE_FLAG_VALID) == 0)
+			goto end_qe;
+
+		uint32_t iq_num = PRIO_TO_IQ(qe->priority);
+		struct sw_qid *qid = &sw->qids[qe->queue_id];
+		struct iq_ring *iq_ring = qid->iq[iq_num];
+
+		if (iq_ring_free_count(iq_ring) == 0)
+			break; /* move to next port */
+
+		port->stats.rx_pkts++;
+
+		/* Use the iq_num from above to push the QE
+		 * into the qid at the right priority
+		 */
+		qid->iq_pkt_mask |= (1 << (iq_num));
+		iq_ring_enqueue(iq_ring, qe);
+		qid->iq_pkt_count[iq_num]++;
+		qid->stats.rx_pkts++;
+		pkts_iter++;
+
+end_qe:
+		port->pp_buf_start++;
+		port->pp_buf_count--;
+	} /* while port->pp_buf_count */
+
+	return pkts_iter;
+}
+
+void
+sw_event_schedule(struct rte_eventdev *dev)
+{
+	struct sw_evdev *sw = sw_pmd_priv(dev);
+	uint32_t in_pkts, out_pkts;
+	uint32_t out_pkts_total = 0, in_pkts_total = 0;
+	int32_t sched_quanta = sw->sched_quanta;
+	uint32_t i;
+
+	sw->sched_called++;
+	if (!sw->started)
+		return;
+
+	do {
+		uint32_t in_pkts_this_iteration = 0;
+
+		/* Pull from rx_ring for ports */
+		do {
+			in_pkts = 0;
+			for (i = 0; i < sw->port_count; i++)
+				if (sw->ports[i].is_directed)
+					in_pkts += sw_schedule_pull_port_dir(sw, i);
+				else if (sw->ports[i].num_ordered_qids > 0)
+					in_pkts += sw_schedule_pull_port_lb(sw, i);
+				else
+					in_pkts += sw_schedule_pull_port_no_reorder(sw, i);
+
+			/* QID scan for re-ordered */
+			in_pkts += sw_schedule_reorder(sw, 0,
+					sw->qid_count);
+			in_pkts_this_iteration += in_pkts;
+		} while (in_pkts > 4 &&
+				(int)in_pkts_this_iteration < sched_quanta);
+
+		out_pkts = 0;
+		out_pkts += sw_schedule_qid_to_cq(sw);
+		out_pkts_total += out_pkts;
+		in_pkts_total += in_pkts_this_iteration;
+
+		if (in_pkts == 0 && out_pkts == 0)
+			break;
+	} while ((int)out_pkts_total < sched_quanta);
+
+	/* push all the internal buffered QEs in port->cq_ring to the
+	 * worker cores: aka, do the ring transfers batched.
+	 */
+	for (i = 0; i < sw->port_count; i++) {
+		struct qe_ring *worker = sw->ports[i].cq_worker_ring;
+		qe_ring_enqueue_burst(worker, sw->ports[i].cq_buf,
+				sw->ports[i].cq_buf_count,
+				&sw->cq_ring_space[i]);
+		sw->ports[i].cq_buf_count = 0;
+	}
+
+	sw->stats.tx_pkts += out_pkts_total;
+	sw->stats.rx_pkts += in_pkts_total;
+
+	sw->sched_no_iq_enqueues += (in_pkts_total == 0);
+	sw->sched_no_cq_enqueues += (out_pkts_total == 0);
+
+}
diff --git a/src/seastar/dpdk/drivers/event/sw/sw_evdev_worker.c b/src/seastar/dpdk/drivers/event/sw/sw_evdev_worker.c
new file mode 100644
index 00000000..9cb6bef5
--- /dev/null
+++ b/src/seastar/dpdk/drivers/event/sw/sw_evdev_worker.c
@@ -0,0 +1,185 @@
+/*-
+ *   BSD LICENSE
+ *
+ *   Copyright(c) 2016-2017 Intel Corporation. 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_atomic.h>
+#include <rte_cycles.h>
+
+#include "sw_evdev.h"
+#include "event_ring.h"
+
+#define PORT_ENQUEUE_MAX_BURST_SIZE 64
+
+static inline void
+sw_event_release(struct sw_port *p, uint8_t index)
+{
+	/*
+	 * Drops the next outstanding event in our history. Used on dequeue
+	 * to clear any history before dequeuing more events.
+	 */
+	RTE_SET_USED(index);
+
+	/* create drop message */
+	struct rte_event ev;
+	ev.op = sw_qe_flag_map[RTE_EVENT_OP_RELEASE];
+
+	uint16_t free_count;
+	qe_ring_enqueue_burst(p->rx_worker_ring, &ev, 1, &free_count);
+
+	/* each release returns one credit */
+	p->outstanding_releases--;
+	p->inflight_credits++;
+}
+
+uint16_t
+sw_event_enqueue_burst(void *port, const struct rte_event ev[], uint16_t num)
+{
+	int32_t i;
+	uint8_t new_ops[PORT_ENQUEUE_MAX_BURST_SIZE];
+	struct sw_port *p = port;
+	struct sw_evdev *sw = (void *)p->sw;
+	uint32_t sw_inflights = rte_atomic32_read(&sw->inflights);
+
+	if (unlikely(p->inflight_max < sw_inflights))
+		return 0;
+
+	if (num > PORT_ENQUEUE_MAX_BURST_SIZE)
+		num = PORT_ENQUEUE_MAX_BURST_SIZE;
+
+	if (p->inflight_credits < num) {
+		/* check if event enqueue brings port over max threshold */
+		uint32_t credit_update_quanta = sw->credit_update_quanta;
+		if (sw_inflights + credit_update_quanta > sw->nb_events_limit)
+			return 0;
+
+		rte_atomic32_add(&sw->inflights, credit_update_quanta);
+		p->inflight_credits += (credit_update_quanta);
+
+		if (p->inflight_credits < num)
+			return 0;
+	}
+
+	for (i = 0; i < num; i++) {
+		int op = ev[i].op;
+		int outstanding = p->outstanding_releases > 0;
+		const uint8_t invalid_qid = (ev[i].queue_id >= sw->qid_count);
+
+		p->inflight_credits -= (op == RTE_EVENT_OP_NEW);
+		p->inflight_credits += (op == RTE_EVENT_OP_RELEASE) *
+					outstanding;
+
+		new_ops[i] = sw_qe_flag_map[op];
+		new_ops[i] &= ~(invalid_qid << QE_FLAG_VALID_SHIFT);
+
+		/* FWD and RELEASE packets will both resolve to taken (assuming
+		 * correct usage of the API), providing very high correct
+		 * prediction rate.
+		 */
+		if ((new_ops[i] & QE_FLAG_COMPLETE) && outstanding)
+			p->outstanding_releases--;
+
+		/* error case: branch to avoid touching p->stats */
+		if (unlikely(invalid_qid)) {
+			p->stats.rx_dropped++;
+			p->inflight_credits++;
+		}
+	}
+
+	/* returns number of events actually enqueued */
+	uint32_t enq = qe_ring_enqueue_burst_with_ops(p->rx_worker_ring, ev, i,
+					     new_ops);
+	if (p->outstanding_releases == 0 && p->last_dequeue_burst_sz != 0) {
+		uint64_t burst_ticks = rte_get_timer_cycles() -
+				p->last_dequeue_ticks;
+		uint64_t burst_pkt_ticks =
+			burst_ticks / p->last_dequeue_burst_sz;
+		p->avg_pkt_ticks -= p->avg_pkt_ticks / NUM_SAMPLES;
+		p->avg_pkt_ticks += burst_pkt_ticks / NUM_SAMPLES;
+		p->last_dequeue_ticks = 0;
+	}
+	return enq;
+}
+
+uint16_t
+sw_event_enqueue(void *port, const struct rte_event *ev)
+{
+	return sw_event_enqueue_burst(port, ev, 1);
+}
+
+uint16_t
+sw_event_dequeue_burst(void *port, struct rte_event *ev, uint16_t num,
+		uint64_t wait)
+{
+	RTE_SET_USED(wait);
+	struct sw_port *p = (void *)port;
+	struct sw_evdev *sw = (void *)p->sw;
+	struct qe_ring *ring = p->cq_worker_ring;
+	uint32_t credit_update_quanta = sw->credit_update_quanta;
+
+	/* check that all previous dequeues have been released */
+	if (!p->is_directed) {
+		uint16_t out_rels = p->outstanding_releases;
+		uint16_t i;
+		for (i = 0; i < out_rels; i++)
+			sw_event_release(p, i);
+	}
+
+	/* returns number of events actually dequeued */
+	uint16_t ndeq = qe_ring_dequeue_burst(ring, ev, num);
+	if (unlikely(ndeq == 0)) {
+		p->outstanding_releases = 0;
+		p->zero_polls++;
+		p->total_polls++;
+		goto end;
+	}
+
+	/* only add credits for directed ports - LB ports send RELEASEs */
+	p->inflight_credits += ndeq * p->is_directed;
+	p->outstanding_releases = ndeq;
+	p->last_dequeue_burst_sz = ndeq;
+	p->last_dequeue_ticks = rte_get_timer_cycles();
+	p->poll_buckets[(ndeq - 1) >> SW_DEQ_STAT_BUCKET_SHIFT]++;
+	p->total_polls++;
+
+end:
+	if (p->inflight_credits >= credit_update_quanta * 2 &&
+			p->inflight_credits > credit_update_quanta + ndeq) {
+		rte_atomic32_sub(&sw->inflights, credit_update_quanta);
+		p->inflight_credits -= credit_update_quanta;
+	}
+	return ndeq;
+}
+
+uint16_t
+sw_event_dequeue(void *port, struct rte_event *ev, uint64_t wait)
+{
+	return sw_event_dequeue_burst(port, ev, 1, wait);
+}
diff --git a/src/seastar/dpdk/drivers/event/sw/sw_evdev_xstats.c b/src/seastar/dpdk/drivers/event/sw/sw_evdev_xstats.c
new file mode 100644
index 00000000..c7b1abe8
--- /dev/null
+++ b/src/seastar/dpdk/drivers/event/sw/sw_evdev_xstats.c
@@ -0,0 +1,674 @@
+/*-
+ *   BSD LICENSE
+ *
+ *   Copyright(c) 2016-2017 Intel Corporation. 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 "sw_evdev.h"
+#include "iq_ring.h"
+#include "event_ring.h"
+
+enum xstats_type {
+	/* common stats */
+	rx,
+	tx,
+	dropped,
+	inflight,
+	calls,
+	credits,
+	/* device instance specific */
+	no_iq_enq,
+	no_cq_enq,
+	/* port_specific */
+	rx_used,
+	rx_free,
+	tx_used,
+	tx_free,
+	pkt_cycles,
+	poll_return, /* for zero-count and used also for port bucket loop */
+	/* qid_specific */
+	iq_size,
+	iq_used,
+	/* qid port mapping specific */
+	pinned,
+};
+
+typedef uint64_t (*xstats_fn)(const struct sw_evdev *dev,
+		uint16_t obj_idx, /* port or queue id */
+		enum xstats_type stat, int extra_arg);
+
+struct sw_xstats_entry {
+	struct rte_event_dev_xstats_name name;
+	xstats_fn fn;
+	uint16_t obj_idx;
+	enum xstats_type stat;
+	enum rte_event_dev_xstats_mode mode;
+	int extra_arg;
+	uint8_t reset_allowed; /* when set, this value can be reset */
+	uint64_t reset_value; /* an offset to be taken away to emulate resets */
+};
+
+static uint64_t
+get_dev_stat(const struct sw_evdev *sw, uint16_t obj_idx __rte_unused,
+		enum xstats_type type, int extra_arg __rte_unused)
+{
+	switch (type) {
+	case rx: return sw->stats.rx_pkts;
+	case tx: return sw->stats.tx_pkts;
+	case dropped: return sw->stats.rx_dropped;
+	case calls: return sw->sched_called;
+	case no_iq_enq: return sw->sched_no_iq_enqueues;
+	case no_cq_enq: return sw->sched_no_cq_enqueues;
+	default: return -1;
+	}
+}
+
+static uint64_t
+get_port_stat(const struct sw_evdev *sw, uint16_t obj_idx,
+		enum xstats_type type, int extra_arg __rte_unused)
+{
+	const struct sw_port *p = &sw->ports[obj_idx];
+
+	switch (type) {
+	case rx: return p->stats.rx_pkts;
+	case tx: return p->stats.tx_pkts;
+	case dropped: return p->stats.rx_dropped;
+	case inflight: return p->inflights;
+	case pkt_cycles: return p->avg_pkt_ticks;
+	case calls: return p->total_polls;
+	case credits: return p->inflight_credits;
+	case poll_return: return p->zero_polls;
+	case rx_used: return qe_ring_count(p->rx_worker_ring);
+	case rx_free: return qe_ring_free_count(p->rx_worker_ring);
+	case tx_used: return qe_ring_count(p->cq_worker_ring);
+	case tx_free: return qe_ring_free_count(p->cq_worker_ring);
+	default: return -1;
+	}
+}
+
+static uint64_t
+get_port_bucket_stat(const struct sw_evdev *sw, uint16_t obj_idx,
+		enum xstats_type type, int extra_arg)
+{
+	const struct sw_port *p = &sw->ports[obj_idx];
+
+	switch (type) {
+	case poll_return: return p->poll_buckets[extra_arg];
+	default: return -1;
+	}
+}
+
+static uint64_t
+get_qid_stat(const struct sw_evdev *sw, uint16_t obj_idx,
+		enum xstats_type type, int extra_arg __rte_unused)
+{
+	const struct sw_qid *qid = &sw->qids[obj_idx];
+
+	switch (type) {
+	case rx: return qid->stats.rx_pkts;
+	case tx: return qid->stats.tx_pkts;
+	case dropped: return qid->stats.rx_dropped;
+	case inflight:
+		do {
+			uint64_t infl = 0;
+			unsigned int i;
+			for (i = 0; i < RTE_DIM(qid->fids); i++)
+				infl += qid->fids[i].pcount;
+			return infl;
+		} while (0);
+		break;
+	case iq_size: return RTE_DIM(qid->iq[0]->ring);
+	default: return -1;
+	}
+}
+
+static uint64_t
+get_qid_iq_stat(const struct sw_evdev *sw, uint16_t obj_idx,
+		enum xstats_type type, int extra_arg)
+{
+	const struct sw_qid *qid = &sw->qids[obj_idx];
+	const int iq_idx = extra_arg;
+
+	switch (type) {
+	case iq_used: return iq_ring_count(qid->iq[iq_idx]);
+	default: return -1;
+	}
+}
+
+static uint64_t
+get_qid_port_stat(const struct sw_evdev *sw, uint16_t obj_idx,
+		enum xstats_type type, int extra_arg)
+{
+	const struct sw_qid *qid = &sw->qids[obj_idx];
+	uint16_t port = extra_arg;
+
+	switch (type) {
+	case pinned:
+		do {
+			uint64_t pin = 0;
+			unsigned int i;
+			for (i = 0; i < RTE_DIM(qid->fids); i++)
+				if (qid->fids[i].cq == port)
+					pin++;
+			return pin;
+		} while (0);
+		break;
+	default: return -1;
+	}
+}
+
+int
+sw_xstats_init(struct sw_evdev *sw)
+{
+	/*
+	 * define the stats names and types. Used to build up the device
+	 * xstats array
+	 * There are multiple set of stats:
+	 *   - device-level,
+	 *   - per-port,
+	 *   - per-port-dequeue-burst-sizes
+	 *   - per-qid,
+	 *   - per-iq
+	 *   - per-port-per-qid
+	 *
+	 * For each of these sets, we have three parallel arrays, one for the
+	 * names, the other for the stat type parameter to be passed in the fn
+	 * call to get that stat. The third array allows resetting or not.
+	 * All these arrays must be kept in sync
+	 */
+	static const char * const dev_stats[] = { "rx", "tx", "drop",
+			"sched_calls", "sched_no_iq_enq", "sched_no_cq_enq",
+	};
+	static const enum xstats_type dev_types[] = { rx, tx, dropped,
+			calls, no_iq_enq, no_cq_enq,
+	};
+	/* all device stats are allowed to be reset */
+
+	static const char * const port_stats[] = {"rx", "tx", "drop",
+			"inflight", "avg_pkt_cycles", "credits",
+			"rx_ring_used", "rx_ring_free",
+			"cq_ring_used", "cq_ring_free",
+			"dequeue_calls", "dequeues_returning_0",
+	};
+	static const enum xstats_type port_types[] = { rx, tx, dropped,
+			inflight, pkt_cycles, credits,
+			rx_used, rx_free, tx_used, tx_free,
+			calls, poll_return,
+	};
+	static const uint8_t port_reset_allowed[] = {1, 1, 1,
+			0, 1, 0,
+			0, 0, 0, 0,
+			1, 1,
+	};
+
+	static const char * const port_bucket_stats[] = {
+			"dequeues_returning" };
+	static const enum xstats_type port_bucket_types[] = { poll_return };
+	/* all bucket dequeues are allowed to be reset, handled in loop below */
+
+	static const char * const qid_stats[] = {"rx", "tx", "drop",
+			"inflight", "iq_size"
+	};
+	static const enum xstats_type qid_types[] = { rx, tx, dropped,
+			inflight, iq_size
+	};
+	static const uint8_t qid_reset_allowed[] = {1, 1, 1,
+			0, 0
+	};
+
+	static const char * const qid_iq_stats[] = { "used" };
+	static const enum xstats_type qid_iq_types[] = { iq_used };
+	/* reset allowed */
+
+	static const char * const qid_port_stats[] = { "pinned_flows" };
+	static const enum xstats_type qid_port_types[] = { pinned };
+	/* reset allowed */
+	/* ---- end of stat definitions ---- */
+
+	/* check sizes, since a missed comma can lead to strings being
+	 * joined by the compiler.
+	 */
+	RTE_BUILD_BUG_ON(RTE_DIM(dev_stats) != RTE_DIM(dev_types));
+	RTE_BUILD_BUG_ON(RTE_DIM(port_stats) != RTE_DIM(port_types));
+	RTE_BUILD_BUG_ON(RTE_DIM(qid_stats) != RTE_DIM(qid_types));
+	RTE_BUILD_BUG_ON(RTE_DIM(qid_iq_stats) != RTE_DIM(qid_iq_types));
+	RTE_BUILD_BUG_ON(RTE_DIM(qid_port_stats) != RTE_DIM(qid_port_types));
+	RTE_BUILD_BUG_ON(RTE_DIM(port_bucket_stats) !=
+			RTE_DIM(port_bucket_types));
+
+	RTE_BUILD_BUG_ON(RTE_DIM(port_stats) != RTE_DIM(port_reset_allowed));
+	RTE_BUILD_BUG_ON(RTE_DIM(qid_stats) != RTE_DIM(qid_reset_allowed));
+
+	/* other vars */
+	const uint32_t cons_bkt_shift =
+		(MAX_SW_CONS_Q_DEPTH >> SW_DEQ_STAT_BUCKET_SHIFT);
+	const unsigned int count = RTE_DIM(dev_stats) +
+			sw->port_count * RTE_DIM(port_stats) +
+			sw->port_count * RTE_DIM(port_bucket_stats) *
+				(cons_bkt_shift + 1) +
+			sw->qid_count * RTE_DIM(qid_stats) +
+			sw->qid_count * SW_IQS_MAX * RTE_DIM(qid_iq_stats) +
+			sw->qid_count * sw->port_count *
+				RTE_DIM(qid_port_stats);
+	unsigned int i, port, qid, iq, bkt, stat = 0;
+
+	sw->xstats = rte_zmalloc_socket(NULL, sizeof(sw->xstats[0]) * count, 0,
+			sw->data->socket_id);
+	if (sw->xstats == NULL)
+		return -ENOMEM;
+
+#define sname sw->xstats[stat].name.name
+	for (i = 0; i < RTE_DIM(dev_stats); i++, stat++) {
+		sw->xstats[stat] = (struct sw_xstats_entry){
+			.fn = get_dev_stat,
+			.stat = dev_types[i],
+			.mode = RTE_EVENT_DEV_XSTATS_DEVICE,
+			.reset_allowed = 1,
+		};
+		snprintf(sname, sizeof(sname), "dev_%s", dev_stats[i]);
+	}
+	sw->xstats_count_mode_dev = stat;
+
+	for (port = 0; port < sw->port_count; port++) {
+		sw->xstats_offset_for_port[port] = stat;
+
+		uint32_t count_offset = stat;
+
+		for (i = 0; i < RTE_DIM(port_stats); i++, stat++) {
+			sw->xstats[stat] = (struct sw_xstats_entry){
+				.fn = get_port_stat,
+				.obj_idx = port,
+				.stat = port_types[i],
+				.mode = RTE_EVENT_DEV_XSTATS_PORT,
+				.reset_allowed = port_reset_allowed[i],
+			};
+			snprintf(sname, sizeof(sname), "port_%u_%s",
+					port, port_stats[i]);
+		}
+
+		for (bkt = 0; bkt < (sw->ports[port].cq_worker_ring->size >>
+				SW_DEQ_STAT_BUCKET_SHIFT) + 1; bkt++) {
+			for (i = 0; i < RTE_DIM(port_bucket_stats); i++) {
+				sw->xstats[stat] = (struct sw_xstats_entry){
+					.fn = get_port_bucket_stat,
+					.obj_idx = port,
+					.stat = port_bucket_types[i],
+					.mode = RTE_EVENT_DEV_XSTATS_PORT,
+					.extra_arg = bkt,
+					.reset_allowed = 1,
+				};
+				snprintf(sname, sizeof(sname),
+					"port_%u_%s_%u-%u",
+					port, port_bucket_stats[i],
+					(bkt << SW_DEQ_STAT_BUCKET_SHIFT) + 1,
+					(bkt + 1) << SW_DEQ_STAT_BUCKET_SHIFT);
+				stat++;
+			}
+		}
+
+		sw->xstats_count_per_port[port] = stat - count_offset;
+	}
+
+	sw->xstats_count_mode_port = stat - sw->xstats_count_mode_dev;
+
+	for (qid = 0; qid < sw->qid_count; qid++) {
+		uint32_t count_offset = stat;
+		sw->xstats_offset_for_qid[qid] = stat;
+
+		for (i = 0; i < RTE_DIM(qid_stats); i++, stat++) {
+			sw->xstats[stat] = (struct sw_xstats_entry){
+				.fn = get_qid_stat,
+				.obj_idx = qid,
+				.stat = qid_types[i],
+				.mode = RTE_EVENT_DEV_XSTATS_QUEUE,
+				.reset_allowed = qid_reset_allowed[i],
+			};
+			snprintf(sname, sizeof(sname), "qid_%u_%s",
+					qid, qid_stats[i]);
+		}
+		for (iq = 0; iq < SW_IQS_MAX; iq++)
+			for (i = 0; i < RTE_DIM(qid_iq_stats); i++, stat++) {
+				sw->xstats[stat] = (struct sw_xstats_entry){
+					.fn = get_qid_iq_stat,
+					.obj_idx = qid,
+					.stat = qid_iq_types[i],
+					.mode = RTE_EVENT_DEV_XSTATS_QUEUE,
+					.extra_arg = iq,
+					.reset_allowed = 0,
+				};
+				snprintf(sname, sizeof(sname),
+						"qid_%u_iq_%u_%s",
+						qid, iq,
+						qid_iq_stats[i]);
+			}
+
+		for (port = 0; port < sw->port_count; port++)
+			for (i = 0; i < RTE_DIM(qid_port_stats); i++, stat++) {
+				sw->xstats[stat] = (struct sw_xstats_entry){
+					.fn = get_qid_port_stat,
+					.obj_idx = qid,
+					.stat = qid_port_types[i],
+					.mode = RTE_EVENT_DEV_XSTATS_QUEUE,
+					.extra_arg = port,
+					.reset_allowed = 0,
+				};
+				snprintf(sname, sizeof(sname),
+						"qid_%u_port_%u_%s",
+						qid, port,
+						qid_port_stats[i]);
+			}
+
+		sw->xstats_count_per_qid[qid] = stat - count_offset;
+	}
+
+	sw->xstats_count_mode_queue = stat -
+		(sw->xstats_count_mode_dev + sw->xstats_count_mode_port);
+#undef sname
+
+	sw->xstats_count = stat;
+
+	return stat;
+}
+
+int
+sw_xstats_uninit(struct sw_evdev *sw)
+{
+	rte_free(sw->xstats);
+	sw->xstats_count = 0;
+	return 0;
+}
+
+int
+sw_xstats_get_names(const struct rte_eventdev *dev,
+		enum rte_event_dev_xstats_mode mode, uint8_t queue_port_id,
+		struct rte_event_dev_xstats_name *xstats_names,
+		unsigned int *ids, unsigned int size)
+{
+	const struct sw_evdev *sw = sw_pmd_priv_const(dev);
+	unsigned int i;
+	unsigned int xidx = 0;
+	RTE_SET_USED(mode);
+	RTE_SET_USED(queue_port_id);
+
+	uint32_t xstats_mode_count = 0;
+	uint32_t start_offset = 0;
+
+	switch (mode) {
+	case RTE_EVENT_DEV_XSTATS_DEVICE:
+		xstats_mode_count = sw->xstats_count_mode_dev;
+		break;
+	case RTE_EVENT_DEV_XSTATS_PORT:
+		if (queue_port_id >= (signed int)sw->port_count)
+			break;
+		xstats_mode_count = sw->xstats_count_per_port[queue_port_id];
+		start_offset = sw->xstats_offset_for_port[queue_port_id];
+		break;
+	case RTE_EVENT_DEV_XSTATS_QUEUE:
+		if (queue_port_id >= (signed int)sw->qid_count)
+			break;
+		xstats_mode_count = sw->xstats_count_per_qid[queue_port_id];
+		start_offset = sw->xstats_offset_for_qid[queue_port_id];
+		break;
+	default:
+		SW_LOG_ERR("Invalid mode received in sw_xstats_get_names()\n");
+		return -EINVAL;
+	};
+
+	if (xstats_mode_count > size || !ids || !xstats_names)
+		return xstats_mode_count;
+
+	for (i = 0; i < sw->xstats_count && xidx < size; i++) {
+		if (sw->xstats[i].mode != mode)
+			continue;
+
+		if (mode != RTE_EVENT_DEV_XSTATS_DEVICE &&
+				queue_port_id != sw->xstats[i].obj_idx)
+			continue;
+
+		xstats_names[xidx] = sw->xstats[i].name;
+		if (ids)
+			ids[xidx] = start_offset + xidx;
+		xidx++;
+	}
+	return xidx;
+}
+
+static int
+sw_xstats_update(struct sw_evdev *sw, enum rte_event_dev_xstats_mode mode,
+		uint8_t queue_port_id, const unsigned int ids[],
+		uint64_t values[], unsigned int n, const uint32_t reset,
+		const uint32_t ret_if_n_lt_nstats)
+{
+	unsigned int i;
+	unsigned int xidx = 0;
+	RTE_SET_USED(mode);
+	RTE_SET_USED(queue_port_id);
+
+	uint32_t xstats_mode_count = 0;
+
+	switch (mode) {
+	case RTE_EVENT_DEV_XSTATS_DEVICE:
+		xstats_mode_count = sw->xstats_count_mode_dev;
+		break;
+	case RTE_EVENT_DEV_XSTATS_PORT:
+		if (queue_port_id >= (signed int)sw->port_count)
+			goto invalid_value;
+		xstats_mode_count = sw->xstats_count_per_port[queue_port_id];
+		break;
+	case RTE_EVENT_DEV_XSTATS_QUEUE:
+		if (queue_port_id >= (signed int)sw->qid_count)
+			goto invalid_value;
+		xstats_mode_count = sw->xstats_count_per_qid[queue_port_id];
+		break;
+	default:
+		SW_LOG_ERR("Invalid mode received in sw_xstats_get()\n");
+		goto invalid_value;
+	};
+
+	/* this function can check num stats and return them (xstats_get() style
+	 * behaviour) or ignore n for reset() of a single stat style behaviour.
+	 */
+	if (ret_if_n_lt_nstats && xstats_mode_count > n)
+		return xstats_mode_count;
+
+	for (i = 0; i < n && xidx < xstats_mode_count; i++) {
+		struct sw_xstats_entry *xs = &sw->xstats[ids[i]];
+		if (ids[i] > sw->xstats_count || xs->mode != mode)
+			continue;
+
+		if (mode != RTE_EVENT_DEV_XSTATS_DEVICE &&
+				queue_port_id != xs->obj_idx)
+			continue;
+
+		uint64_t val = xs->fn(sw, xs->obj_idx, xs->stat, xs->extra_arg)
+					- xs->reset_value;
+
+		if (values)
+			values[xidx] = val;
+
+		if (xs->reset_allowed && reset)
+			xs->reset_value = val;
+
+		xidx++;
+	}
+
+	return xidx;
+invalid_value:
+	return -EINVAL;
+}
+
+int
+sw_xstats_get(const struct rte_eventdev *dev,
+		enum rte_event_dev_xstats_mode mode, uint8_t queue_port_id,
+		const unsigned int ids[], uint64_t values[], unsigned int n)
+{
+	struct sw_evdev *sw = sw_pmd_priv(dev);
+	const uint32_t reset = 0;
+	const uint32_t ret_n_lt_stats = 1;
+	return sw_xstats_update(sw, mode, queue_port_id, ids, values, n,
+				reset, ret_n_lt_stats);
+}
+
+uint64_t
+sw_xstats_get_by_name(const struct rte_eventdev *dev,
+		const char *name, unsigned int *id)
+{
+	const struct sw_evdev *sw = sw_pmd_priv_const(dev);
+	unsigned int i;
+
+	for (i = 0; i < sw->xstats_count; i++) {
+		struct sw_xstats_entry *xs = &sw->xstats[i];
+		if (strncmp(xs->name.name, name,
+				RTE_EVENT_DEV_XSTATS_NAME_SIZE) == 0){
+			if (id != NULL)
+				*id = i;
+			return xs->fn(sw, xs->obj_idx, xs->stat, xs->extra_arg)
+					- xs->reset_value;
+		}
+	}
+	if (id != NULL)
+		*id = (uint32_t)-1;
+	return (uint64_t)-1;
+}
+
+static void
+sw_xstats_reset_range(struct sw_evdev *sw, uint32_t start, uint32_t num)
+{
+	uint32_t i;
+	for (i = start; i < start + num; i++) {
+		struct sw_xstats_entry *xs = &sw->xstats[i];
+		if (!xs->reset_allowed)
+			continue;
+
+		uint64_t val = xs->fn(sw, xs->obj_idx, xs->stat, xs->extra_arg)
+					- xs->reset_value;
+		xs->reset_value = val;
+	}
+}
+
+static int
+sw_xstats_reset_queue(struct sw_evdev *sw, uint8_t queue_id,
+		const uint32_t ids[], uint32_t nb_ids)
+{
+	const uint32_t reset = 1;
+	const uint32_t ret_n_lt_stats = 0;
+	if (ids) {
+		uint32_t nb_reset = sw_xstats_update(sw,
+					RTE_EVENT_DEV_XSTATS_QUEUE,
+					queue_id, ids, NULL, nb_ids,
+					reset, ret_n_lt_stats);
+		return nb_reset == nb_ids ? 0 : -EINVAL;
+	}
+
+	if (ids == NULL)
+		sw_xstats_reset_range(sw, sw->xstats_offset_for_qid[queue_id],
+				      sw->xstats_count_per_qid[queue_id]);
+
+	return 0;
+}
+
+static int
+sw_xstats_reset_port(struct sw_evdev *sw, uint8_t port_id,
+		const uint32_t ids[], uint32_t nb_ids)
+{
+	const uint32_t reset = 1;
+	const uint32_t ret_n_lt_stats = 0;
+	int offset = sw->xstats_offset_for_port[port_id];
+	int nb_stat = sw->xstats_count_per_port[port_id];
+
+	if (ids) {
+		uint32_t nb_reset = sw_xstats_update(sw,
+					RTE_EVENT_DEV_XSTATS_PORT, port_id,
+					ids, NULL, nb_ids,
+					reset, ret_n_lt_stats);
+		return nb_reset == nb_ids ? 0 : -EINVAL;
+	}
+
+	sw_xstats_reset_range(sw, offset, nb_stat);
+	return 0;
+}
+
+static int
+sw_xstats_reset_dev(struct sw_evdev *sw, const uint32_t ids[], uint32_t nb_ids)
+{
+	uint32_t i;
+	if (ids) {
+		for (i = 0; i < nb_ids; i++) {
+			uint32_t id = ids[i];
+			if (id >= sw->xstats_count_mode_dev)
+				return -EINVAL;
+			sw_xstats_reset_range(sw, id, 1);
+		}
+	} else {
+		for (i = 0; i < sw->xstats_count_mode_dev; i++)
+			sw_xstats_reset_range(sw, i, 1);
+	}
+
+	return 0;
+}
+
+int
+sw_xstats_reset(struct rte_eventdev *dev,
+		enum rte_event_dev_xstats_mode mode,
+		int16_t queue_port_id,
+		const uint32_t ids[],
+		uint32_t nb_ids)
+{
+	struct sw_evdev *sw = sw_pmd_priv(dev);
+	uint32_t i, err;
+
+	/* handle -1 for queue_port_id here, looping over all ports/queues */
+	switch (mode) {
+	case RTE_EVENT_DEV_XSTATS_DEVICE:
+		sw_xstats_reset_dev(sw, ids, nb_ids);
+		break;
+	case RTE_EVENT_DEV_XSTATS_PORT:
+		if (queue_port_id == -1) {
+			for (i = 0; i < sw->port_count; i++) {
+				err = sw_xstats_reset_port(sw, i, ids, nb_ids);
+				if (err)
+					return -EINVAL;
+			}
+		} else if (queue_port_id < (int16_t)sw->port_count)
+			sw_xstats_reset_port(sw, queue_port_id, ids, nb_ids);
+		break;
+	case RTE_EVENT_DEV_XSTATS_QUEUE:
+		if (queue_port_id == -1) {
+			for (i = 0; i < sw->qid_count; i++) {
+				err = sw_xstats_reset_queue(sw, i, ids, nb_ids);
+				if (err)
+					return -EINVAL;
+			}
+		} else if (queue_port_id < (int16_t)sw->qid_count)
+			sw_xstats_reset_queue(sw, queue_port_id, ids, nb_ids);
+		break;
+	};
+
+	return 0;
+}