[dpdk-dev] [PATCH 00/15] next-eventdev: event/sw Software Eventdev

[dpdk-dev] [PATCH 00/15] next-eventdev: event/sw Software Eventdev

[Harry van Haaren]

The following patchset adds software eventdev implementation
to the next-eventdev tree, and applies to current git HEAD.

This implementation is based on the previous software eventdev
RFC patchset[1], updated to integrate with the latest rte_eventdev.h
API.

The first two patches make changes to the eventdev API,
then the software implementation is added, and finally
tests are added for the sw eventdev implementation.

This patchset contains the work of multiple developers,
please see signoffs on each patch.

Signed-off-by: Harry van Haaren <harry.van.haaren@intel.com>

[1] http://dpdk.org/ml/archives/dev/2016-November/050285.html

Bruce Richardson (15):
  eventdev: remove unneeded dependencies
  eventdev: add APIs for extended stats
  event/sw: add new software-only eventdev driver
  event/sw: add function to return device capabilities
  event/sw: add configure function
  event/sw: add fns to return default port/queue config
  event/sw: add support for event queues
  event/sw: add support for event ports
  event/sw: add support for linking queues to ports
  event/sw: add worker core functions
  event/sw: add scheduling logic
  event/sw: add start, stop and close functions
  event/sw: add dump function for easier debugging
  event/sw: add xstats support
  app/test: add unit tests for SW eventdev driver

 app/test/Makefile                             |    5 +-
 app/test/test_sw_eventdev.c                   | 2031 +++++++++++++++++++++++++
 config/common_base                            |    5 +
 drivers/event/Makefile                        |    1 +
 drivers/event/sw/Makefile                     |   70 +
 drivers/event/sw/event_ring.h                 |  179 +++
 drivers/event/sw/iq_ring.h                    |  176 +++
 drivers/event/sw/rte_pmd_evdev_sw_version.map |    3 +
 drivers/event/sw/sw_evdev.c                   |  709 +++++++++
 drivers/event/sw/sw_evdev.h                   |  287 ++++
 drivers/event/sw/sw_evdev_scheduler.c         |  586 +++++++
 drivers/event/sw/sw_evdev_worker.c            |  169 ++
 drivers/event/sw/sw_evdev_xstats.c            |  401 +++++
 lib/librte_eventdev/Makefile                  |    1 -
 lib/librte_eventdev/rte_eventdev.c            |   64 +
 lib/librte_eventdev/rte_eventdev.h            |   80 +-
 lib/librte_eventdev/rte_eventdev_pmd.h        |   58 +
 lib/librte_eventdev/rte_eventdev_version.map  |    3 +
 mk/rte.app.mk                                 |    1 +
 19 files changed, 4825 insertions(+), 4 deletions(-)
 create mode 100644 app/test/test_sw_eventdev.c
 create mode 100644 drivers/event/sw/Makefile
 create mode 100644 drivers/event/sw/event_ring.h
 create mode 100644 drivers/event/sw/iq_ring.h
 create mode 100644 drivers/event/sw/rte_pmd_evdev_sw_version.map
 create mode 100644 drivers/event/sw/sw_evdev.c
 create mode 100644 drivers/event/sw/sw_evdev.h
 create mode 100644 drivers/event/sw/sw_evdev_scheduler.c
 create mode 100644 drivers/event/sw/sw_evdev_worker.c
 create mode 100644 drivers/event/sw/sw_evdev_xstats.c

-- 
2.7.4

[dpdk-dev] [PATCH 01/15] eventdev: remove unneeded dependencies

[Harry van Haaren]

From: Bruce Richardson <bruce.richardson@intel.com>

Since eventdev uses event structures rather than working directly on
mbufs, there is no actual dependencies on the mbuf library. The
inclusion of an mbuf pointer element inside the event itself does not
require the inclusion of the mbuf header file. Similarly the pci
header is not needed, but following their removal, rte_memory.h is
needed for the definition of the __rte_cache_aligned macro.

Signed-off-by: Bruce Richardson <bruce.richardson@intel.com>
Signed-off-by: Harry van Haaren <harry.van.haaren@intel.com>
---
 lib/librte_eventdev/Makefile       | 1 -
 lib/librte_eventdev/rte_eventdev.h | 5 +++--
 2 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/lib/librte_eventdev/Makefile b/lib/librte_eventdev/Makefile
index dac0663..396e5ec 100644
--- a/lib/librte_eventdev/Makefile
+++ b/lib/librte_eventdev/Makefile
@@ -52,6 +52,5 @@ EXPORT_MAP := rte_eventdev_version.map
 
 # library dependencies
 DEPDIRS-y += lib/librte_eal
-DEPDIRS-y += lib/librte_mbuf
 
 include $(RTE_SDK)/mk/rte.lib.mk
diff --git a/lib/librte_eventdev/rte_eventdev.h b/lib/librte_eventdev/rte_eventdev.h
index e1bd05f..c2f9310 100644
--- a/lib/librte_eventdev/rte_eventdev.h
+++ b/lib/librte_eventdev/rte_eventdev.h
@@ -244,8 +244,9 @@ extern "C" {
 #endif
 
 #include <rte_common.h>
-#include <rte_pci.h>
-#include <rte_mbuf.h>
+#include <rte_memory.h>
+
+struct rte_mbuf; /* we just use mbuf pointers; no need to include rte_mbuf.h */
 
 /* Event device capability bitmap flags */
 #define RTE_EVENT_DEV_CAP_QUEUE_QOS           (1ULL << 0)
-- 
2.7.4

Re: [dpdk-dev] [PATCH 01/15] eventdev: remove unneeded dependencies

[Jerin Jacob]

On Mon, Jan 16, 2017 at 03:40:41PM +0000, Harry van Haaren wrote:
> From: Bruce Richardson <bruce.richardson@intel.com>
> 
> Since eventdev uses event structures rather than working directly on
> mbufs, there is no actual dependencies on the mbuf library. The
> inclusion of an mbuf pointer element inside the event itself does not
> require the inclusion of the mbuf header file. Similarly the pci
> header is not needed, but following their removal, rte_memory.h is
> needed for the definition of the __rte_cache_aligned macro.
> 
> Signed-off-by: Bruce Richardson <bruce.richardson@intel.com>
> Signed-off-by: Harry van Haaren <harry.van.haaren@intel.com>
> ---
>  lib/librte_eventdev/Makefile       | 1 -
>  lib/librte_eventdev/rte_eventdev.h | 5 +++--
>  2 files changed, 3 insertions(+), 3 deletions(-)
> 
> diff --git a/lib/librte_eventdev/Makefile b/lib/librte_eventdev/Makefile
> index dac0663..396e5ec 100644
> --- a/lib/librte_eventdev/Makefile
> +++ b/lib/librte_eventdev/Makefile
> @@ -52,6 +52,5 @@ EXPORT_MAP := rte_eventdev_version.map
>  
>  # library dependencies
>  DEPDIRS-y += lib/librte_eal
> -DEPDIRS-y += lib/librte_mbuf
>  
>  include $(RTE_SDK)/mk/rte.lib.mk
> diff --git a/lib/librte_eventdev/rte_eventdev.h b/lib/librte_eventdev/rte_eventdev.h
> index e1bd05f..c2f9310 100644
> --- a/lib/librte_eventdev/rte_eventdev.h
> +++ b/lib/librte_eventdev/rte_eventdev.h
> @@ -244,8 +244,9 @@ extern "C" {
>  #endif
>  
>  #include <rte_common.h>
> -#include <rte_pci.h>
> -#include <rte_mbuf.h>
> +#include <rte_memory.h>
> +
> +struct rte_mbuf; /* we just use mbuf pointers; no need to include rte_mbuf.h */

This "struct rte_mbuf" reference is not present in dpdk-next-eventdev tree.
Are you planning to rebase to dpdk-next-eventdev?

>  
>  /* Event device capability bitmap flags */
>  #define RTE_EVENT_DEV_CAP_QUEUE_QOS           (1ULL << 0)
> -- 
> 2.7.4
> 

Re: [dpdk-dev] [PATCH 01/15] eventdev: remove unneeded dependencies

[Van Haaren, Harry]

> From: Jerin Jacob [mailto:jerin.jacob@caviumnetworks.com]
> >
> >  #include <rte_common.h>
> > -#include <rte_pci.h>
> > -#include <rte_mbuf.h>
> > +#include <rte_memory.h>
> > +
> > +struct rte_mbuf; /* we just use mbuf pointers; no need to include rte_mbuf.h */
> 
> This "struct rte_mbuf" reference is not present in dpdk-next-eventdev tree.
> Are you planning to rebase to dpdk-next-eventdev?


The idea was to remove the include of the header file, as we never dereference the mbuf pointer, and hence we shouldn't include a header we don't require.

The struct rte_mbuf here is just a forward declaration for the actual rte_mbuf. This allows the rte_event to contain a struct rte_mbuf* without the compiler complaining that it doesn't understand the type.


The current patches apply to dpdk-next-eventdev HEAD, I don't think I understand what you're asking about rebasing.

Re: [dpdk-dev] [PATCH 01/15] eventdev: remove unneeded dependencies

[Jerin Jacob]

On Tue, Jan 17, 2017 at 09:59:59AM +0000, Van Haaren, Harry wrote:
> > From: Jerin Jacob [mailto:jerin.jacob@caviumnetworks.com]
> > >
> > >  #include <rte_common.h>
> > > -#include <rte_pci.h>
> > > -#include <rte_mbuf.h>
> > > +#include <rte_memory.h>
> > > +
> > > +struct rte_mbuf; /* we just use mbuf pointers; no need to include rte_mbuf.h */
> > 
> > This "struct rte_mbuf" reference is not present in dpdk-next-eventdev tree.
> > Are you planning to rebase to dpdk-next-eventdev?
> 
> 
> The idea was to remove the include of the header file, as we never dereference the mbuf pointer, and hence we shouldn't include a header we don't require.
> 
> The struct rte_mbuf here is just a forward declaration for the actual rte_mbuf. This allows the rte_event to contain a struct rte_mbuf* without the compiler complaining that it doesn't understand the type.
> 
> 
> The current patches apply to dpdk-next-eventdev HEAD, I don't think I understand what you're asking about rebasing.

Thanks for the clarification. It is clear now.
I got confused with following comment in the cover-letter.

This implementation is based on the previous software eventdev
RFC patchset[1], updated to integrate with the latest rte_eventdev.h
API.
[1] http://dpdk.org/ml/archives/dev/2016-November/050285.html

Re: [dpdk-dev] [PATCH 01/15] eventdev: remove unneeded dependencies

[Jerin Jacob]

On Mon, Jan 16, 2017 at 03:40:41PM +0000, Harry van Haaren wrote:
> From: Bruce Richardson <bruce.richardson@intel.com>
> 
> Since eventdev uses event structures rather than working directly on
> mbufs, there is no actual dependencies on the mbuf library. The
> inclusion of an mbuf pointer element inside the event itself does not
> require the inclusion of the mbuf header file. Similarly the pci
> header is not needed, but following their removal, rte_memory.h is
> needed for the definition of the __rte_cache_aligned macro.
> 
> Signed-off-by: Bruce Richardson <bruce.richardson@intel.com>
> Signed-off-by: Harry van Haaren <harry.van.haaren@intel.com>

Acked-by: Jerin Jacob <jerin.jacob@caviumnetworks.com>

> ---
>  lib/librte_eventdev/Makefile       | 1 -
>  lib/librte_eventdev/rte_eventdev.h | 5 +++--
>  2 files changed, 3 insertions(+), 3 deletions(-)
> 
> diff --git a/lib/librte_eventdev/Makefile b/lib/librte_eventdev/Makefile
> index dac0663..396e5ec 100644
> --- a/lib/librte_eventdev/Makefile
> +++ b/lib/librte_eventdev/Makefile
> @@ -52,6 +52,5 @@ EXPORT_MAP := rte_eventdev_version.map
>  
>  # library dependencies
>  DEPDIRS-y += lib/librte_eal
> -DEPDIRS-y += lib/librte_mbuf
>  
>  include $(RTE_SDK)/mk/rte.lib.mk
> diff --git a/lib/librte_eventdev/rte_eventdev.h b/lib/librte_eventdev/rte_eventdev.h
> index e1bd05f..c2f9310 100644
> --- a/lib/librte_eventdev/rte_eventdev.h
> +++ b/lib/librte_eventdev/rte_eventdev.h
> @@ -244,8 +244,9 @@ extern "C" {
>  #endif
>  
>  #include <rte_common.h>
> -#include <rte_pci.h>
> -#include <rte_mbuf.h>
> +#include <rte_memory.h>
> +
> +struct rte_mbuf; /* we just use mbuf pointers; no need to include rte_mbuf.h */
>  
>  /* Event device capability bitmap flags */
>  #define RTE_EVENT_DEV_CAP_QUEUE_QOS           (1ULL << 0)
> -- 
> 2.7.4
> 

[dpdk-dev] [PATCH 02/15] eventdev: add APIs for extended stats

[Harry van Haaren]

From: Bruce Richardson <bruce.richardson@intel.com>

Add in APIs for extended stats so that eventdev implementations can report
out information on their internal state. The APIs are based on, but not
identical to, the equivalent ethdev functions.

Signed-off-by: Bruce Richardson <bruce.richardson@intel.com>
Signed-off-by: Harry van Haaren <harry.van.haaren@intel.com>
---
 lib/librte_eventdev/rte_eventdev.c           | 64 ++++++++++++++++++++++++
 lib/librte_eventdev/rte_eventdev.h           | 75 ++++++++++++++++++++++++++++
 lib/librte_eventdev/rte_eventdev_pmd.h       | 58 +++++++++++++++++++++
 lib/librte_eventdev/rte_eventdev_version.map |  3 ++
 4 files changed, 200 insertions(+)

diff --git a/lib/librte_eventdev/rte_eventdev.c b/lib/librte_eventdev/rte_eventdev.c
index c8f3e94..ca11d54 100644
--- a/lib/librte_eventdev/rte_eventdev.c
+++ b/lib/librte_eventdev/rte_eventdev.c
@@ -920,6 +920,70 @@ rte_event_dev_dump(uint8_t dev_id, FILE *f)
 
 }
 
+static int
+get_xstats_count(uint8_t dev_id)
+{
+	struct rte_eventdev *dev = &rte_eventdevs[dev_id];
+	if (dev->dev_ops->get_xstat_names != NULL)
+		return (*dev->dev_ops->get_xstat_names)(dev, NULL, 0);
+	return 0;
+}
+
+int
+rte_event_dev_get_xstat_names(uint8_t dev_id,
+	struct rte_event_dev_xstat_name *xstats_names,
+	unsigned int size)
+{
+	RTE_EVENTDEV_VALID_DEVID_OR_ERR_RET(dev_id, -EINVAL);
+	const int cnt_expected_entries = get_xstats_count(dev_id);
+	if (xstats_names == NULL || cnt_expected_entries < 0 ||
+			(int)size < cnt_expected_entries)
+		return cnt_expected_entries;
+
+	/* dev_id checked in get_xstats_count() */
+	const struct rte_eventdev *dev = &rte_eventdevs[dev_id];
+
+	if (dev->dev_ops->get_xstat_names != NULL)
+		return (*dev->dev_ops->get_xstat_names)(dev,
+				xstats_names, size);
+
+	return 0;
+}
+
+/* retrieve eventdev extended statistics */
+int
+rte_event_dev_get_xstats(uint8_t dev_id, const unsigned int ids[],
+	uint64_t values[], unsigned int n)
+{
+	RTE_EVENTDEV_VALID_DEVID_OR_ERR_RET(dev_id, -EINVAL);
+	const struct rte_eventdev *dev = &rte_eventdevs[dev_id];
+
+	/* implemented by the driver */
+	if (dev->dev_ops->get_xstats != NULL)
+		return (*dev->dev_ops->get_xstats)(dev, ids, values, n);
+	return 0;
+}
+
+uint64_t
+rte_event_dev_get_xstat_by_name(uint8_t dev_id, const char *name,
+		unsigned int *id)
+{
+	const struct rte_eventdev *dev = &rte_eventdevs[dev_id];
+	unsigned int temp = -1;
+
+	if (id != NULL)
+		*id = (unsigned int)-1;
+	else
+		id = &temp; /* ensure driver never gets a NULL value */
+
+	RTE_EVENTDEV_VALID_DEVID_OR_ERR_RET(dev_id, 0);
+
+	/* implemented by driver */
+	if (dev->dev_ops->get_xstat_by_name != NULL)
+		return (*dev->dev_ops->get_xstat_by_name)(dev, name, id);
+	return 0;
+}
+
 int
 rte_event_dev_start(uint8_t dev_id)
 {
diff --git a/lib/librte_eventdev/rte_eventdev.h b/lib/librte_eventdev/rte_eventdev.h
index c2f9310..681cbfa 100644
--- a/lib/librte_eventdev/rte_eventdev.h
+++ b/lib/librte_eventdev/rte_eventdev.h
@@ -1401,6 +1401,81 @@ rte_event_port_links_get(uint8_t dev_id, uint8_t port_id,
 int
 rte_event_dev_dump(uint8_t dev_id, FILE *f);
 
+/** Maximum name length for extended statistics counters */
+#define RTE_EVENT_DEV_XSTAT_NAME_SIZE 64
+
+/**
+ * A name-key lookup element for extended statistics.
+ *
+ * This structure is used to map between names and ID numbers
+ * for extended ethdev statistics.
+ */
+struct rte_event_dev_xstat_name {
+	char name[RTE_EVENT_DEV_XSTAT_NAME_SIZE];
+};
+
+/**
+ * Retrieve names of extended statistics of an event device.
+ *
+ * @param dev_id
+ *   The identifier of the event device.
+ * @param xstat_names
+ *  Block of memory to insert names into. Must be at least size in capacity.
+ *  If set to NULL, function returns required capacity.
+ * @param size
+ *  Capacity of xstat_names (number of names).
+ * @return
+ *   - positive value lower or equal to size: success. The return value
+ *     is the number of entries filled in the stats table.
+ *   - positive value higher than size: error, the given statistics table
+ *     is too small. The return value corresponds to the size that should
+ *     be given to succeed. The entries in the table are not valid and
+ *     shall not be used by the caller.
+ *   - negative value on error (invalid port id)
+ */
+int
+rte_event_dev_get_xstat_names(uint8_t dev_id,
+		struct rte_event_dev_xstat_name *xstat_names,
+		unsigned int size);
+
+/**
+ * Retrieve extended statistics of an event device.
+ *
+ * @param dev_id
+ *   The identifier of the device.
+ * @param ids
+ *   The id numbers of the stats to get. The ids can be got from the stat
+ *   position in the stat list from rte_event_dev_get_xstat_names(), or
+ *   by using rte_eventdev_get_xstat_by_name()
+ * @param values
+ *   The values for each stats request by ID.
+ * @param n
+ *   The number of stats requested
+ * @return
+ *   Number of stat entries filled into the values array
+ */
+int
+rte_event_dev_get_xstats(uint8_t dev_id, const unsigned int ids[],
+		uint64_t values[], unsigned int n);
+
+/**
+ * Retrieve the value of a single stat by requesting it by name.
+ *
+ * @param dev_id
+ *   The identifier of the device
+ * @param name
+ *   The stat name to retrieve
+ * @param id
+ *   If non-NULL, the numerical id of the stat will be returned, so that further
+ *   requests for the stat can be got using rte_eventdev_xstats_get, which will
+ *   be faster as it doesn't need to scan a list of names for the stat.
+ *   If the stat cannot be found, the id returned will be (unsigned)-1.
+ * @return
+ *   The stat value, or -1 if not found.
+ */
+uint64_t
+rte_event_dev_get_xstat_by_name(uint8_t dev_id, const char *name, unsigned int *id);
+
 #ifdef __cplusplus
 }
 #endif
diff --git a/lib/librte_eventdev/rte_eventdev_pmd.h b/lib/librte_eventdev/rte_eventdev_pmd.h
index c84c9a2..7a6229d 100644
--- a/lib/librte_eventdev/rte_eventdev_pmd.h
+++ b/lib/librte_eventdev/rte_eventdev_pmd.h
@@ -418,6 +418,59 @@ typedef void (*eventdev_dequeue_timeout_ticks_t)(struct rte_eventdev *dev,
  */
 typedef void (*eventdev_dump_t)(struct rte_eventdev *dev, FILE *f);
 
+/**
+ * Retrieve a set of statistics from device
+ *
+ * @param dev
+ *   Event device pointer
+ * @param ids
+ *   The stat ids to retrieve
+ * @param values
+ *   The returned stat values
+ * @param n
+ *   The number of id values and entries in the values array
+ * @return
+ *   The number of stat values successfully filled into the values array
+ */
+typedef int (*eventdev_get_xstats_t)(const struct rte_eventdev *dev,
+	const unsigned int ids[], uint64_t values[], unsigned int n);
+
+/**
+ * Get names of extended stats of an event device
+ *
+ * @param dev
+ *   Event device pointer
+ * @param xstat_names
+ *   Array of name values to be filled in
+ * @param size
+ *   Number of values in the xstat_names array
+ * @return
+ *   When size >= the number of stats, return the number of stat values filled
+ *   into the array.
+ *   When size < the number of available stats, return the number of stats
+ *   values, and do not fill in any data into xstat_names.
+ */
+typedef int (*eventdev_get_xstat_names_t)(const struct rte_eventdev *dev,
+	struct rte_event_dev_xstat_name *xstat_names, unsigned int size);
+
+/**
+ * Get value of one stats and optionally return its id
+ *
+ * @param dev
+ *   Event device pointer
+ * @param name
+ *   The name of the stat to retrieve
+ * @param id
+ *   Pointer to an unsigned int where we store the stat-id for future reference.
+ *   This pointer may be null if the id is not required.
+ * @return
+ *   The value of the stat, or (uint64_t)-1 if the stat is not found.
+ *   If the stat is not found, the id value will be returned as (unsigned)-1,
+ *   if id pointer is non-NULL
+ */
+typedef uint64_t (*eventdev_get_xstat_by_name)(const struct rte_eventdev *dev,
+		const char *name, unsigned int *id);
+
 /** Event device operations function pointer table */
 struct rte_eventdev_ops {
 	eventdev_info_get_t dev_infos_get;	/**< Get device info. */
@@ -448,6 +501,11 @@ struct rte_eventdev_ops {
 	/**< Converts ns to *timeout_ticks* value for rte_event_dequeue() */
 	eventdev_dump_t dump;
 	/* Dump internal information */
+
+	eventdev_get_xstats_t get_xstats;    /**< Get extended device statistics. */
+	eventdev_get_xstat_names_t get_xstat_names; /**< Get names of extended stats. */
+	eventdev_get_xstat_by_name get_xstat_by_name; /**< Get one value by name */
+
 };
 
 /**
diff --git a/lib/librte_eventdev/rte_eventdev_version.map b/lib/librte_eventdev/rte_eventdev_version.map
index 68b8c81..b138eb3 100644
--- a/lib/librte_eventdev/rte_eventdev_version.map
+++ b/lib/librte_eventdev/rte_eventdev_version.map
@@ -12,6 +12,9 @@ DPDK_17.02 {
 	rte_event_dev_stop;
 	rte_event_dev_close;
 	rte_event_dev_dump;
+	rte_event_dev_get_xstats;
+	rte_event_dev_get_xstat_names;
+	rte_event_dev_get_xstat_by_name;
 
 	rte_event_port_default_conf_get;
 	rte_event_port_setup;
-- 
2.7.4

Re: [dpdk-dev] [PATCH 02/15] eventdev: add APIs for extended stats

[Jerin Jacob]

On Mon, Jan 16, 2017 at 03:40:42PM +0000, Harry van Haaren wrote:
> From: Bruce Richardson <bruce.richardson@intel.com>
> 
> Add in APIs for extended stats so that eventdev implementations can report
> out information on their internal state. The APIs are based on, but not
> identical to, the equivalent ethdev functions.

Overall the xstat API looks good. I think, we need to extend the API to
support event port and event queue specific xstats too.

> 
> Signed-off-by: Bruce Richardson <bruce.richardson@intel.com>
> Signed-off-by: Harry van Haaren <harry.van.haaren@intel.com>
> ---
>  lib/librte_eventdev/rte_eventdev.c           | 64 ++++++++++++++++++++++++
>  lib/librte_eventdev/rte_eventdev.h           | 75 ++++++++++++++++++++++++++++
>  lib/librte_eventdev/rte_eventdev_pmd.h       | 58 +++++++++++++++++++++
>  lib/librte_eventdev/rte_eventdev_version.map |  3 ++
>  4 files changed, 200 insertions(+)
> 
> +
> +uint64_t
> +rte_event_dev_get_xstat_by_name(uint8_t dev_id, const char *name,
> +		unsigned int *id)
> +{
> +	const struct rte_eventdev *dev = &rte_eventdevs[dev_id];
> +	unsigned int temp = -1;
> +
> +	if (id != NULL)
> +		*id = (unsigned int)-1;
> +	else
> +		id = &temp; /* ensure driver never gets a NULL value */
> +
> +	RTE_EVENTDEV_VALID_DEVID_OR_ERR_RET(dev_id, 0);
> +
> +	/* implemented by driver */
> +	if (dev->dev_ops->get_xstat_by_name != NULL)
> +		return (*dev->dev_ops->get_xstat_by_name)(dev, name, id);
> +	return 0;

Shouldn't we return -1 as per API specification?

> +}
> +
>  int
>  rte_event_dev_start(uint8_t dev_id)
>  {
> diff --git a/lib/librte_eventdev/rte_eventdev.h b/lib/librte_eventdev/rte_eventdev.h
> index c2f9310..681cbfa 100644
> --- a/lib/librte_eventdev/rte_eventdev.h
> +++ b/lib/librte_eventdev/rte_eventdev.h
> @@ -1401,6 +1401,81 @@ rte_event_port_links_get(uint8_t dev_id, uint8_t port_id,
>  int
>  rte_event_dev_dump(uint8_t dev_id, FILE *f);
>  
> +/** Maximum name length for extended statistics counters */
> +#define RTE_EVENT_DEV_XSTAT_NAME_SIZE 64
> +
> +/**
> + * A name-key lookup element for extended statistics.
> + *
> + * This structure is used to map between names and ID numbers
> + * for extended ethdev statistics.
> + */
> +struct rte_event_dev_xstat_name {
> +	char name[RTE_EVENT_DEV_XSTAT_NAME_SIZE];
> +};
> +
> +/**
> + * Retrieve names of extended statistics of an event device.
> + *
> + * @param dev_id
> + *   The identifier of the event device.
> + * @param xstat_names
> + *  Block of memory to insert names into. Must be at least size in capacity.
> + *  If set to NULL, function returns required capacity.
> + * @param size
> + *  Capacity of xstat_names (number of names).
> + * @return
> + *   - positive value lower or equal to size: success. The return value
> + *     is the number of entries filled in the stats table.
> + *   - positive value higher than size: error, the given statistics table
> + *     is too small. The return value corresponds to the size that should
> + *     be given to succeed. The entries in the table are not valid and
> + *     shall not be used by the caller.
> + *   - negative value on error (invalid port id)

How about updating the rte_errno also in the failure case? It may useful
for application developer to have single check for error condition

> + */
> +int
> +rte_event_dev_get_xstat_names(uint8_t dev_id,
> +		struct rte_event_dev_xstat_name *xstat_names,
> +		unsigned int size);
> +
> +/**
> + * Retrieve extended statistics of an event device.
> + *
> + * @param dev_id
> + *   The identifier of the device.
> + * @param ids
> + *   The id numbers of the stats to get. The ids can be got from the stat
> + *   position in the stat list from rte_event_dev_get_xstat_names(), or
> + *   by using rte_eventdev_get_xstat_by_name()
> + * @param values

Please add [out] to indicate it is a output parameter
example: @param[out] values

> + *   The values for each stats request by ID.
> + * @param n
> + *   The number of stats requested
> + * @return
> + *   Number of stat entries filled into the values array
> + */
> +int
> +rte_event_dev_get_xstats(uint8_t dev_id, const unsigned int ids[],
> +		uint64_t values[], unsigned int n);
> +
> +/**
> + * Retrieve the value of a single stat by requesting it by name.
> + *
> + * @param dev_id
> + *   The identifier of the device
> + * @param name
> + *   The stat name to retrieve
> + * @param id

Please add [out] to indicate it is output parameter

> + *   If non-NULL, the numerical id of the stat will be returned, so that further
> + *   requests for the stat can be got using rte_eventdev_xstats_get, which will

The function prototype is rte_event_dev_get_xstats. Change the
rte_eventdev_xstats_get to rte_event_dev_get_xstats in the above description

The rest of the file is following rte_eventdev_xxx_xxx_get syntax for
get functions. How about changing rte_eventdev_xxx_xxx_get syntax to
maintain the consistency?


> + *   be faster as it doesn't need to scan a list of names for the stat.
> + *   If the stat cannot be found, the id returned will be (unsigned)-1.
> + * @return
> + *   The stat value, or -1 if not found.
> + */
> +uint64_t
> +rte_event_dev_get_xstat_by_name(uint8_t dev_id, const char *name, unsigned int *id);

[dpdk-dev] [PATCH 03/15] event/sw: add new software-only eventdev driver

[Harry van Haaren]

From: Bruce Richardson <bruce.richardson@intel.com>

This adds the minimal changes to allow a SW eventdev implementation to
be compiled, linked and created at run time. The eventdev does nothing,
but can be created via vdev on commandline, e.g.

  sudo ./x86_64-native-linuxapp-gcc/app/test --vdev=event_sw0
  ...
  PMD: Creating eventdev sw device event_sw0, numa_node=0, sched_quanta=128
  RTE>>

Signed-off-by: Bruce Richardson <bruce.richardson@intel.com>
Signed-off-by: Harry van Haaren <harry.van.haaren@intel.com>
---
 config/common_base                            |   5 +
 drivers/event/Makefile                        |   1 +
 drivers/event/sw/Makefile                     |  67 ++++++++++++
 drivers/event/sw/rte_pmd_evdev_sw_version.map |   3 +
 drivers/event/sw/sw_evdev.c                   | 148 ++++++++++++++++++++++++++
 drivers/event/sw/sw_evdev.h                   |  57 ++++++++++
 mk/rte.app.mk                                 |   1 +
 7 files changed, 282 insertions(+)
 create mode 100644 drivers/event/sw/Makefile
 create mode 100644 drivers/event/sw/rte_pmd_evdev_sw_version.map
 create mode 100644 drivers/event/sw/sw_evdev.c
 create mode 100644 drivers/event/sw/sw_evdev.h

diff --git a/config/common_base b/config/common_base
index 00af811..6f91172 100644
--- a/config/common_base
+++ b/config/common_base
@@ -434,6 +434,11 @@ CONFIG_RTE_LIBRTE_PMD_SKELETON_EVENTDEV=y
 CONFIG_RTE_LIBRTE_PMD_SKELETON_EVENTDEV_DEBUG=n
 
 #
+# Compile PMD for software event device
+#
+CONFIG_RTE_LIBRTE_PMD_SW_EVENTDEV=y
+
+#
 # Compile librte_ring
 #
 CONFIG_RTE_LIBRTE_RING=y
diff --git a/drivers/event/Makefile b/drivers/event/Makefile
index 678279f..353441c 100644
--- a/drivers/event/Makefile
+++ b/drivers/event/Makefile
@@ -32,5 +32,6 @@
 include $(RTE_SDK)/mk/rte.vars.mk
 
 DIRS-$(CONFIG_RTE_LIBRTE_PMD_SKELETON_EVENTDEV) += skeleton
+DIRS-$(CONFIG_RTE_LIBRTE_PMD_SW_EVENTDEV) += sw
 
 include $(RTE_SDK)/mk/rte.subdir.mk
diff --git a/drivers/event/sw/Makefile b/drivers/event/sw/Makefile
new file mode 100644
index 0000000..c891eb5
--- /dev/null
+++ b/drivers/event/sw/Makefile
@@ -0,0 +1,67 @@
+#   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
+
+# export include files
+SYMLINK-y-include +=
+
+# library dependencies
+DEPDIRS-$(CONFIG_RTE_LIBRTE_PMD_SW_EVENTDEV) += lib/librte_eal
+DEPDIRS-$(CONFIG_RTE_LIBRTE_PMD_SW_EVENTDEV) += lib/librte_eventdev
+DEPDIRS-$(CONFIG_RTE_LIBRTE_PMD_SW_EVENTDEV) += lib/librte_kvargs
+DEPDIRS-$(CONFIG_RTE_LIBRTE_PMD_SW_EVENTDEV) += lib/librte_ring
+
+include $(RTE_SDK)/mk/rte.lib.mk
+
diff --git a/drivers/event/sw/rte_pmd_evdev_sw_version.map b/drivers/event/sw/rte_pmd_evdev_sw_version.map
new file mode 100644
index 0000000..1f84b68
--- /dev/null
+++ b/drivers/event/sw/rte_pmd_evdev_sw_version.map
@@ -0,0 +1,3 @@
+DPDK_17.02 {
+	local: *;
+};
diff --git a/drivers/event/sw/sw_evdev.c b/drivers/event/sw/sw_evdev.c
new file mode 100644
index 0000000..29762b8
--- /dev/null
+++ b/drivers/event/sw/sw_evdev.c
@@ -0,0 +1,148 @@
+/*-
+ *   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 "sw_evdev.h"
+
+#define EVENTDEV_NAME_SW_PMD event_sw
+#define NUMA_NODE_ARG "numa_node"
+#define SCHED_QUANTA_ARG "sched_quanta"
+
+int sched_quanta = 128;
+
+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
+sw_probe(const char *name, const char *params)
+{
+	static const struct rte_eventdev_ops evdev_sw_ops = {
+	};
+
+	static const char *const args[] = { NUMA_NODE_ARG, SCHED_QUANTA_ARG, NULL };
+	struct rte_eventdev *dev;
+	struct sw_evdev *sw;
+	int socket_id = rte_socket_id();
+
+	if (params != NULL && params[0] != '\0') {
+		struct rte_kvargs *kvlist = rte_kvargs_parse(params, args);
+
+		if (!kvlist) {
+			RTE_LOG(INFO, PMD,
+				"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) {
+				RTE_LOG(ERR, PMD,
+					"%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) {
+				RTE_LOG(ERR, PMD,
+					"%s: Error parsing sched quanta parameter",
+					name);
+				rte_kvargs_free(kvlist);
+				return ret;
+			}
+
+			rte_kvargs_free(kvlist);
+		}
+	}
+
+	RTE_LOG(INFO, PMD,
+			"Creating eventdev sw device %s, numa_node=%d, sched_quanta=%d\n",
+			name, socket_id, sched_quanta);
+
+	dev = rte_event_pmd_vdev_init(name,
+			sizeof(struct sw_evdev), socket_id);
+	if (dev == NULL) {
+		printf("eventdev vdev init() failed");
+		return -EFAULT;
+	}
+	dev->dev_ops = &evdev_sw_ops;
+
+	sw = dev->data->dev_private;
+	sw->data = dev->data;
+
+	return 0;
+}
+
+static int
+sw_remove(const char *name)
+{
+	if (name == NULL)
+		return -EINVAL;
+
+	RTE_LOG(INFO, PMD, "Closing eventdev sw device %s\n", name);
+	/* TODO unregister eventdev and release memzone */
+
+	return 0;
+}
+
+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>");
diff --git a/drivers/event/sw/sw_evdev.h b/drivers/event/sw/sw_evdev.h
new file mode 100644
index 0000000..1400417
--- /dev/null
+++ b/drivers/event/sw/sw_evdev.h
@@ -0,0 +1,57 @@
+/*-
+ *   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>
+
+struct sw_evdev {
+	struct rte_eventdev_data *data;
+};
+
+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;
+}
+
+extern int sched_quanta;
+
+#endif /* _SW_EVDEV_H_ */
diff --git a/mk/rte.app.mk b/mk/rte.app.mk
index 8341c13..8dc0f9f 100644
--- a/mk/rte.app.mk
+++ b/mk/rte.app.mk
@@ -150,6 +150,7 @@ endif # CONFIG_RTE_LIBRTE_CRYPTODEV
 
 ifeq ($(CONFIG_RTE_LIBRTE_EVENTDEV),y)
 _LDLIBS-$(CONFIG_RTE_LIBRTE_PMD_SKELETON_EVENTDEV) += -lrte_pmd_skeleton_event
+_LDLIBS-$(CONFIG_RTE_LIBRTE_PMD_SW_EVENTDEV) += -lrte_pmd_sw_event
 endif # CONFIG_RTE_LIBRTE_EVENTDEV
 
 endif # !CONFIG_RTE_BUILD_SHARED_LIBS
-- 
2.7.4

[dpdk-dev] [PATCH 04/15] event/sw: add function to return device capabilities

[Harry van Haaren]

From: Bruce Richardson <bruce.richardson@intel.com>

Add in the info_get function to return details on the queues, flow,
prioritization capabilities, etc. that this device has.

Signed-off-by: Bruce Richardson <bruce.richardson@intel.com>
Signed-off-by: Harry van Haaren <harry.van.haaren@intel.com>
---
 drivers/event/sw/sw_evdev.c | 23 +++++++++++++++++++++++
 drivers/event/sw/sw_evdev.h | 11 +++++++++++
 2 files changed, 34 insertions(+)

diff --git a/drivers/event/sw/sw_evdev.c b/drivers/event/sw/sw_evdev.c
index 29762b8..d75bf50 100644
--- a/drivers/event/sw/sw_evdev.c
+++ b/drivers/event/sw/sw_evdev.c
@@ -45,6 +45,28 @@
 
 int sched_quanta = 128;
 
+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 = 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 int
 assign_numa_node(const char *key __rte_unused, const char *value, void *opaque)
 {
@@ -69,6 +91,7 @@ static int
 sw_probe(const char *name, const char *params)
 {
 	static const struct rte_eventdev_ops evdev_sw_ops = {
+			.dev_infos_get = sw_info_get,
 	};
 
 	static const char *const args[] = { NUMA_NODE_ARG, SCHED_QUANTA_ARG, NULL };
diff --git a/drivers/event/sw/sw_evdev.h b/drivers/event/sw/sw_evdev.h
index 1400417..b6d99fd 100644
--- a/drivers/event/sw/sw_evdev.h
+++ b/drivers/event/sw/sw_evdev.h
@@ -36,6 +36,17 @@
 #include <rte_eventdev.h>
 #include <rte_eventdev_pmd.h>
 
+#define PMD_NAME "event_sw"
+
+#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
+
 struct sw_evdev {
 	struct rte_eventdev_data *data;
 };
-- 
2.7.4

[dpdk-dev] [PATCH 05/15] event/sw: add configure function

[Harry van Haaren]

From: Bruce Richardson <bruce.richardson@intel.com>

Signed-off-by: Bruce Richardson <bruce.richardson@intel.com>
Signed-off-by: Harry van Haaren <harry.van.haaren@intel.com>
---
 drivers/event/sw/sw_evdev.c | 17 +++++++++++++++++
 drivers/event/sw/sw_evdev.h | 13 +++++++++++++
 2 files changed, 30 insertions(+)

diff --git a/drivers/event/sw/sw_evdev.c b/drivers/event/sw/sw_evdev.c
index d75bf50..1bdcc05 100644
--- a/drivers/event/sw/sw_evdev.c
+++ b/drivers/event/sw/sw_evdev.c
@@ -45,6 +45,22 @@
 
 int sched_quanta = 128;
 
+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;
+	uint32_t quanta_pool = conf->nb_events_limit / SW_INFLIGHT_QUANTA_SIZE;
+	rte_atomic32_set(&sw->inflight_quanta, quanta_pool);
+
+	return 0;
+}
+
 static void
 sw_info_get(struct rte_eventdev *dev, struct rte_event_dev_info *info)
 {
@@ -91,6 +107,7 @@ static int
 sw_probe(const char *name, const char *params)
 {
 	static const struct rte_eventdev_ops evdev_sw_ops = {
+			.dev_configure = sw_dev_configure,
 			.dev_infos_get = sw_info_get,
 	};
 
diff --git a/drivers/event/sw/sw_evdev.h b/drivers/event/sw/sw_evdev.h
index b6d99fd..514b0b1 100644
--- a/drivers/event/sw/sw_evdev.h
+++ b/drivers/event/sw/sw_evdev.h
@@ -35,6 +35,7 @@
 
 #include <rte_eventdev.h>
 #include <rte_eventdev_pmd.h>
+#include <rte_atomic.h>
 
 #define PMD_NAME "event_sw"
 
@@ -46,9 +47,21 @@
 #define SW_INFLIGHT_EVENTS_TOTAL 4096
 /* allow for lots of over-provisioning */
 #define MAX_SW_PROD_Q_DEPTH 4096
+#define SW_INFLIGHT_QUANTA_SIZE 32
 
 struct sw_evdev {
 	struct rte_eventdev_data *data;
+
+	uint32_t port_count;
+	uint32_t qid_count;
+
+	rte_atomic32_t inflight_quanta __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;
+
 };
 
 static inline struct sw_evdev *
-- 
2.7.4

[dpdk-dev] [PATCH 06/15] event/sw: add fns to return default port/queue config

[Harry van Haaren]

From: Bruce Richardson <bruce.richardson@intel.com>

Signed-off-by: Bruce Richardson <bruce.richardson@intel.com>
Signed-off-by: Harry van Haaren <harry.van.haaren@intel.com>
---
 drivers/event/sw/sw_evdev.c | 32 ++++++++++++++++++++++++++++++++
 1 file changed, 32 insertions(+)

diff --git a/drivers/event/sw/sw_evdev.c b/drivers/event/sw/sw_evdev.c
index 1bdcc05..8cc8e06 100644
--- a/drivers/event/sw/sw_evdev.c
+++ b/drivers/event/sw/sw_evdev.c
@@ -45,6 +45,35 @@
 
 int sched_quanta = 128;
 
+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)
 {
@@ -109,6 +138,9 @@ sw_probe(const char *name, const char *params)
 	static const struct rte_eventdev_ops evdev_sw_ops = {
 			.dev_configure = sw_dev_configure,
 			.dev_infos_get = sw_info_get,
+
+			.queue_def_conf = sw_queue_def_conf,
+			.port_def_conf = sw_port_def_conf,
 	};
 
 	static const char *const args[] = { NUMA_NODE_ARG, SCHED_QUANTA_ARG, NULL };
-- 
2.7.4

[dpdk-dev] [PATCH 07/15] event/sw: add support for event queues

[Harry van Haaren]

From: Bruce Richardson <bruce.richardson@intel.com>

Add in the data structures for the event queues, and the eventdev
functions to create and destroy those queues.

Signed-off-by: Bruce Richardson <bruce.richardson@intel.com>
Signed-off-by: Harry van Haaren <harry.van.haaren@intel.com>
---
 drivers/event/sw/iq_ring.h  | 176 ++++++++++++++++++++++++++++++++++++++++++++
 drivers/event/sw/sw_evdev.c | 156 +++++++++++++++++++++++++++++++++++++++
 drivers/event/sw/sw_evdev.h |  75 +++++++++++++++++++
 3 files changed, 407 insertions(+)
 create mode 100644 drivers/event/sw/iq_ring.h

diff --git a/drivers/event/sw/iq_ring.h b/drivers/event/sw/iq_ring.h
new file mode 100644
index 0000000..5994e70
--- /dev/null
+++ b/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/drivers/event/sw/sw_evdev.c b/drivers/event/sw/sw_evdev.c
index 8cc8e06..f420d2b 100644
--- a/drivers/event/sw/sw_evdev.c
+++ b/drivers/event/sw/sw_evdev.c
@@ -38,6 +38,7 @@
 #include <rte_ring.h>
 
 #include "sw_evdev.h"
+#include "iq_ring.h"
 
 #define EVENTDEV_NAME_SW_PMD event_sw
 #define NUMA_NODE_ARG "numa_node"
@@ -45,6 +46,159 @@
 
 int sched_quanta = 128;
 
+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);
+		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;
+	}
+
+	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;
+
+	/* TODO: Error check queue types and appropriate values */
+	switch (conf->event_queue_cfg) {
+	case RTE_EVENT_QUEUE_CFG_SINGLE_LINK:
+		type = RTE_SCHED_TYPE_DIRECT;
+		break;
+	case RTE_EVENT_QUEUE_CFG_DEFAULT:
+		/* fallthrough */
+	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;
+	default:
+		printf("%s : unknown queue type %d requested\n", __func__,
+				conf->event_queue_cfg);
+		return -1;
+	}
+
+	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)
@@ -140,6 +294,8 @@ sw_probe(const char *name, const char *params)
 			.dev_infos_get = sw_info_get,
 
 			.queue_def_conf = sw_queue_def_conf,
+			.queue_setup = sw_queue_setup,
+			.queue_release = sw_queue_release,
 			.port_def_conf = sw_port_def_conf,
 	};
 
diff --git a/drivers/event/sw/sw_evdev.h b/drivers/event/sw/sw_evdev.h
index 514b0b1..e225eb7 100644
--- a/drivers/event/sw/sw_evdev.h
+++ b/drivers/event/sw/sw_evdev.h
@@ -48,6 +48,78 @@
 /* allow for lots of over-provisioning */
 #define MAX_SW_PROD_Q_DEPTH 4096
 #define SW_INFLIGHT_QUANTA_SIZE 32
+#define SW_FRAGMENTS_MAX 16
+
+/* have a new scheduling type for 1:1 queue to port links */
+#define RTE_SCHED_TYPE_DIRECT (RTE_SCHED_TYPE_PARALLEL + 1)
+
+#ifdef RTE_LIBRTE_PMD_EVDEV_SW_DEBUG
+#define SW_LOG_INFO(fmt, args...) \
+	RTE_LOG(INFO, PMD, "[%s] %s() line %u: " fmt "\n", \
+			PMD_NAME, \
+			__func__, __LINE__, ## args)
+
+#define SW_LOG_DBG(fmt, args...) \
+	RTE_LOG(DEBUG, PMD, "[%s] %s() line %u: " fmt "\n", \
+			PMD_NAME, \
+			__func__, __LINE__, ## args)
+#else
+#define SW_LOG_INFO(fmt, args...)
+#define SW_LOG_DBG(fmt, args...)
+#endif
+
+/* 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 {
+	/* The type of this QID */
+	int 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; /* packets awaiting reordering */
+	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_evdev {
 	struct rte_eventdev_data *data;
@@ -62,6 +134,9 @@ struct sw_evdev {
 	 */
 	uint32_t nb_events_limit;
 
+	/* Internal queues - one per logical queue */
+	struct sw_qid qids[RTE_EVENT_MAX_QUEUES_PER_DEV] __rte_cache_aligned;
+
 };
 
 static inline struct sw_evdev *
-- 
2.7.4

[dpdk-dev] [PATCH 08/15] event/sw: add support for event ports

[Harry van Haaren]

From: Bruce Richardson <bruce.richardson@intel.com>

Add in the data-structures for the ports used by workers to send
packets to/from the scheduler. Also add in the functions to
create/destroy those ports.

Signed-off-by: Bruce Richardson <bruce.richardson@intel.com>
Signed-off-by: Harry van Haaren <harry.van.haaren@intel.com>
---
 drivers/event/sw/event_ring.h | 179 ++++++++++++++++++++++++++++++++++++++++++
 drivers/event/sw/sw_evdev.c   |  71 +++++++++++++++++
 drivers/event/sw/sw_evdev.h   |  76 ++++++++++++++++++
 3 files changed, 326 insertions(+)
 create mode 100644 drivers/event/sw/event_ring.h

diff --git a/drivers/event/sw/event_ring.h b/drivers/event/sw/event_ring.h
new file mode 100644
index 0000000..67aa72e
--- /dev/null
+++ b/drivers/event/sw/event_ring.h
@@ -0,0 +1,179 @@
+/*-
+ *   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];
+
+	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];
+	}
+
+	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];
+
+	if (nb_qes != 0)
+		r->read_idx += nb_qes;
+
+	return nb_qes;
+}
+
+#endif
diff --git a/drivers/event/sw/sw_evdev.c b/drivers/event/sw/sw_evdev.c
index f420d2b..51f4ffd 100644
--- a/drivers/event/sw/sw_evdev.c
+++ b/drivers/event/sw/sw_evdev.c
@@ -39,6 +39,7 @@
 
 #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"
@@ -46,6 +47,74 @@
 
 int sched_quanta = 128;
 
+static void
+sw_info_get(struct rte_eventdev *dev, struct rte_event_dev_info *info);
+
+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);
+
+	uint8_t enq_oversize = conf->enqueue_depth > info.max_event_port_enqueue_depth;
+	uint8_t deq_oversize = conf->dequeue_depth > info.max_event_port_dequeue_depth;
+	if (enq_oversize || deq_oversize)
+		return -EINVAL;
+
+	*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) {
+		printf("%s %d: error creating RX worker ring\n",
+				__func__, __LINE__);
+		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);
+		printf("%s %d: error creating CQ worker ring\n",
+				__func__, __LINE__);
+		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;
+
+	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)
@@ -297,6 +366,8 @@ sw_probe(const char *name, const char *params)
 			.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,
 	};
 
 	static const char *const args[] = { NUMA_NODE_ARG, SCHED_QUANTA_ARG, NULL };
diff --git a/drivers/event/sw/sw_evdev.h b/drivers/event/sw/sw_evdev.h
index e225eb7..383ea9c 100644
--- a/drivers/event/sw/sw_evdev.h
+++ b/drivers/event/sw/sw_evdev.h
@@ -49,6 +49,9 @@
 #define MAX_SW_PROD_Q_DEPTH 4096
 #define SW_INFLIGHT_QUANTA_SIZE 32
 #define SW_FRAGMENTS_MAX 16
+#define SW_DEQ_STAT_BUCKET_SHIFT 2 /* report dequeue burst sizes in buckets */
+#define SCHED_DEQUEUE_BURST_SIZE 32 /* how many packets pulled from port by sched */
+#define SW_PORT_HIST_LIST (MAX_SW_PROD_Q_DEPTH) /* size of our history list */
 
 /* have a new scheduling type for 1:1 queue to port links */
 #define RTE_SCHED_TYPE_DIRECT (RTE_SCHED_TYPE_PARALLEL + 1)
@@ -121,12 +124,82 @@ struct sw_qid {
 	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;
+
+	/* A numeric ID for the port. This should be used to access the
+	 * statistics as returned by *rte_qm_stats_get*, and in other places
+	 * where the API requires accessing a port by integer. It is not valid
+	 * to assume that ports will be allocated in a linear sequence.
+	 */
+	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 */
+	int16_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 bursts */
+	uint64_t total_polls;        /* how many polls were counted in stats */
+	uint64_t zero_polls;         /* tracks polls returning nothing */
+	uint32_t poll_buckets[MAX_SW_CONS_Q_DEPTH >> SW_DEQ_STAT_BUCKET_SHIFT];
+		/* 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;
 
+	/* Contains all ports - load balanced and directed */
+	struct sw_port ports[SW_PORTS_MAX] __rte_cache_aligned;
+
 	rte_atomic32_t inflight_quanta __rte_cache_aligned;
 	/*
 	 * max events in this instance. Cached here for performance.
@@ -137,6 +210,9 @@ struct sw_evdev {
 	/* 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;
+
 };
 
 static inline struct sw_evdev *
-- 
2.7.4

[dpdk-dev] [PATCH 09/15] event/sw: add support for linking queues to ports

[Harry van Haaren]

From: Bruce Richardson <bruce.richardson@intel.com>

Signed-off-by: Bruce Richardson <bruce.richardson@intel.com>
Signed-off-by: Harry van Haaren <harry.van.haaren@intel.com>
---
 drivers/event/sw/sw_evdev.c | 67 +++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 67 insertions(+)

diff --git a/drivers/event/sw/sw_evdev.c b/drivers/event/sw/sw_evdev.c
index 51f4ffd..6c2593e 100644
--- a/drivers/event/sw/sw_evdev.c
+++ b/drivers/event/sw/sw_evdev.c
@@ -51,6 +51,71 @@ static void
 sw_info_get(struct rte_eventdev *dev, struct rte_event_dev_info *info);
 
 static int
+sw_port_link(void *port, const uint8_t queues[], const uint8_t priorities[],
+		uint16_t num)
+{
+	struct sw_port *p = (void *)port;
+	struct sw_evdev *sw = p->sw;
+	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))
+			break;
+
+		if (p->is_directed && p->num_qids_mapped > 0)
+			break;
+
+		if (q->type == RTE_SCHED_TYPE_DIRECT) {
+			/* check directed qids only map to one port */
+			if (p->num_qids_mapped > 0)
+				break;
+			/* check port only takes a directed flow */
+			if (num > 1)
+				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(void *port, uint8_t queues[], uint16_t nb_unlinks)
+{
+	struct sw_port *p = (void *)port;
+	struct sw_evdev *sw = p->sw;
+	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++;
+				continue;
+			}
+	}
+	return unlinked;
+}
+
+static int
 sw_port_setup(struct rte_eventdev *dev, uint8_t port_id,
 		const struct rte_event_port_conf *conf)
 {
@@ -368,6 +433,8 @@ sw_probe(const char *name, const char *params)
 			.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,
 	};
 
 	static const char *const args[] = { NUMA_NODE_ARG, SCHED_QUANTA_ARG, NULL };
-- 
2.7.4

[dpdk-dev] [PATCH 10/15] event/sw: add worker core functions

[Harry van Haaren]

From: Bruce Richardson <bruce.richardson@intel.com>

add the event enqueue, dequeue and release functions to the eventdev.
These also include tracking of stats for observability in the load of
the scheduler.
Internally in the enqueue function, the various types of enqueue
operations, to forward an existing event, to send a new event, to
drop a previous event, are converted to a series of flags which will
be used by the scheduler code to perform the needed actions for that
event.

Signed-off-by: Bruce Richardson <bruce.richardson@intel.com>
Signed-off-by: Gage Eads <gage.eads@intel.com>
Signed-off-by: Harry van Haaren <harry.van.haaren@intel.com>
---
 drivers/event/sw/Makefile          |   1 +
 drivers/event/sw/sw_evdev.c        |   4 +
 drivers/event/sw/sw_evdev.h        |  33 ++++++++
 drivers/event/sw/sw_evdev_worker.c | 169 +++++++++++++++++++++++++++++++++++++
 4 files changed, 207 insertions(+)
 create mode 100644 drivers/event/sw/sw_evdev_worker.c

diff --git a/drivers/event/sw/Makefile b/drivers/event/sw/Makefile
index c891eb5..7c23b73 100644
--- a/drivers/event/sw/Makefile
+++ b/drivers/event/sw/Makefile
@@ -53,6 +53,7 @@ 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
 
 # export include files
 SYMLINK-y-include +=
diff --git a/drivers/event/sw/sw_evdev.c b/drivers/event/sw/sw_evdev.c
index 6c2593e..fbf7598 100644
--- a/drivers/event/sw/sw_evdev.c
+++ b/drivers/event/sw/sw_evdev.c
@@ -485,6 +485,10 @@ sw_probe(const char *name, const char *params)
 		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;
 
 	sw = dev->data->dev_private;
 	sw->data = dev->data;
diff --git a/drivers/event/sw/sw_evdev.h b/drivers/event/sw/sw_evdev.h
index 383ea9c..97bad17 100644
--- a/drivers/event/sw/sw_evdev.h
+++ b/drivers/event/sw/sw_evdev.h
@@ -52,10 +52,35 @@
 #define SW_DEQ_STAT_BUCKET_SHIFT 2 /* report dequeue burst sizes in buckets */
 #define SCHED_DEQUEUE_BURST_SIZE 32 /* how many packets pulled from port by sched */
 #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 */
 
 /* have a new scheduling type for 1:1 queue to port links */
 #define RTE_SCHED_TYPE_DIRECT (RTE_SCHED_TYPE_PARALLEL + 1)
 
+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)  /* set 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, PMD, "[%s] %s() line %u: " fmt "\n", \
@@ -229,4 +254,12 @@ sw_pmd_priv_const(const struct rte_eventdev *eventdev)
 
 extern int sched_quanta;
 
+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);
+
 #endif /* _SW_EVDEV_H_ */
diff --git a/drivers/event/sw/sw_evdev_worker.c b/drivers/event/sw/sw_evdev_worker.c
new file mode 100644
index 0000000..3cfaa6f
--- /dev/null
+++ b/drivers/event/sw/sw_evdev_worker.c
@@ -0,0 +1,169 @@
+/*-
+ *   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 = {
+		.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);
+
+	p->outstanding_releases--;
+}
+
+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 quantas_avail = *(uint32_t *)(&sw->inflight_quanta);
+	uint32_t credits_avail = quantas_avail * SW_INFLIGHT_QUANTA_SIZE;
+	uint32_t sw_inflights = (sw->nb_events_limit - credits_avail);
+
+	if (p->inflight_max < sw_inflights)
+		return 0;
+	if (num > PORT_ENQUEUE_MAX_BURST_SIZE)
+		num = PORT_ENQUEUE_MAX_BURST_SIZE;
+
+	/* request SW_INFLIGHT_QUANTA_SIZE more credits if needed */
+	while (p->inflight_credits < num) {
+		int failed_to_get = rte_atomic32_dec_and_test(&sw->inflight_quanta);
+		if (failed_to_get)
+			num = p->inflight_credits;
+		else
+			p->inflight_credits += (SW_INFLIGHT_QUANTA_SIZE);
+	}
+
+	for (i = 0; i < num; i++) {
+		int op = ev[i].op;
+		p->inflight_credits -= (op != RTE_EVENT_OP_RELEASE);
+		new_ops[i] = sw_qe_flag_map[op];
+		const uint8_t invalid_qid = (ev[i].queue_id >= sw->qid_count);
+		new_ops[i] &= ~(invalid_qid << QE_FLAG_VALID_SHIFT);
+		if ((new_ops[i] & QE_FLAG_COMPLETE) && p->outstanding_releases)
+			p->outstanding_releases--;
+
+		if (invalid_qid)
+			p->stats.rx_dropped++;
+	}
+
+	/* 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 qe_ring *ring = p->cq_worker_ring;
+
+	/* 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);
+	}
+
+	/* Intel modification: may not be in final API */
+	if (ev == 0)
+		return 0;
+
+	/* returns number of events actually dequeued */
+	uint16_t ndeq = qe_ring_dequeue_burst(ring, ev, num);
+	if (ndeq == 0) {
+		p->outstanding_releases = 0;
+		p->zero_polls++;
+		p->total_polls++;
+		goto end;
+	}
+
+	p->inflight_credits += ndeq;
+	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 >= SW_INFLIGHT_QUANTA_SIZE * 2 &&
+			p->inflight_credits > SW_INFLIGHT_QUANTA_SIZE + ndeq) {
+		rte_atomic32_inc(&p->sw->inflight_quanta);
+		p->inflight_credits -= SW_INFLIGHT_QUANTA_SIZE;
+	}
+	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);
+}
-- 
2.7.4

[dpdk-dev] [PATCH 11/15] event/sw: add scheduling logic

[Harry van Haaren]

From: Bruce Richardson <bruce.richardson@intel.com>

Add in the scheduling function which takes the events from the
producer queues and buffers them before scheduling them to consumer
queues. The scheduling logic includes support for atomic, reordered,
and parallel scheduling of flows.

Signed-off-by: Bruce Richardson <bruce.richardson@intel.com>
Signed-off-by: Gage Eads <gage.eads@intel.com>
Signed-off-by: Harry van Haaren <harry.van.haaren@intel.com>
---
 drivers/event/sw/Makefile             |   1 +
 drivers/event/sw/sw_evdev.c           |   1 +
 drivers/event/sw/sw_evdev.h           |  12 +
 drivers/event/sw/sw_evdev_scheduler.c | 586 ++++++++++++++++++++++++++++++++++
 4 files changed, 600 insertions(+)
 create mode 100644 drivers/event/sw/sw_evdev_scheduler.c

diff --git a/drivers/event/sw/Makefile b/drivers/event/sw/Makefile
index 7c23b73..e96c457 100644
--- a/drivers/event/sw/Makefile
+++ b/drivers/event/sw/Makefile
@@ -54,6 +54,7 @@ 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
 
 # export include files
 SYMLINK-y-include +=
diff --git a/drivers/event/sw/sw_evdev.c b/drivers/event/sw/sw_evdev.c
index fbf7598..021f3ab 100644
--- a/drivers/event/sw/sw_evdev.c
+++ b/drivers/event/sw/sw_evdev.c
@@ -489,6 +489,7 @@ sw_probe(const char *name, const char *params)
 	dev->enqueue_burst = sw_event_enqueue_burst;
 	dev->dequeue = sw_event_dequeue;
 	dev->dequeue_burst = sw_event_dequeue_burst;
+	dev->schedule = sw_event_schedule;
 
 	sw = dev->data->dev_private;
 	sw->data = dev->data;
diff --git a/drivers/event/sw/sw_evdev.h b/drivers/event/sw/sw_evdev.h
index 97bad17..a0f3668 100644
--- a/drivers/event/sw/sw_evdev.h
+++ b/drivers/event/sw/sw_evdev.h
@@ -238,6 +238,17 @@ struct sw_evdev {
 	/* 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;
+	uint64_t sched_no_iq_enqueues;
+	uint64_t sched_no_cq_enqueues;
+	uint64_t sched_cq_qid_called;
+
+	uint8_t started;
 };
 
 static inline struct sw_evdev *
@@ -261,5 +272,6 @@ uint16_t sw_event_enqueue_burst(void *port, const struct rte_event ev[],
 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);
 
 #endif /* _SW_EVDEV_H_ */
diff --git a/drivers/event/sw/sw_evdev_scheduler.c b/drivers/event/sw/sw_evdev_scheduler.c
new file mode 100644
index 0000000..76829f3
--- /dev/null
+++ b/drivers/event/sw/sw_evdev_scheduler.c
@@ -0,0 +1,586 @@
+/*-
+ *   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)))
+
+/* Clamp the highest priorities to the max value as allowed by
+ * the mask. Assums MASK is (powerOfTwo - 1). Priority 0 (highest) are shifted
+ * into leftmost IQ so that clz() reads it first on dequeue
+ */
+#define PRIO_TO_IQ(prio) (prio > SW_IQS_MASK ? SW_IQS_MASK : prio)
+
+#define MAX_PER_IQ_DEQUEUE 48
+#define FLOWID_MASK (SW_QID_NUM_FIDS-1)
+
+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];
+		/* use cheap bit mixing, since we only need to lose a few bits */
+		uint32_t flow_id32 = (qes[i].flow_id) ^ (qes[i].flow_id >> 10);
+		const uint16_t flow_id = FLOWID_MASK & flow_id32;
+		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 = 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 == RTE_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;
+
+				qe = &entry->fragments[entry->fragment_index + j];
+
+				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 const 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 *)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;
+
+				/* 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?
+				 */
+				if (rob_entry->num_fragments == SW_FRAGMENTS_MAX)
+					sw->stats.rx_dropped++;
+				else
+					rob_entry->fragments[rob_entry->num_fragments++] = *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;
+	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++)
+				/* TODO: use a function pointer in the port itself */
+				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);
+
+}
-- 
2.7.4

[dpdk-dev] [PATCH 12/15] event/sw: add start, stop and close functions

[Harry van Haaren]

From: Bruce Richardson <bruce.richardson@intel.com>

Signed-off-by: Bruce Richardson <bruce.richardson@intel.com>
Signed-off-by: Harry van Haaren <harry.van.haaren@intel.com>
---
 drivers/event/sw/sw_evdev.c | 74 +++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 74 insertions(+)

diff --git a/drivers/event/sw/sw_evdev.c b/drivers/event/sw/sw_evdev.c
index 021f3ab..318f1d7 100644
--- a/drivers/event/sw/sw_evdev.c
+++ b/drivers/event/sw/sw_evdev.c
@@ -401,6 +401,77 @@ sw_info_get(struct rte_eventdev *dev, struct rte_event_dev_info *info)
 }
 
 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) {
+			printf("%s %d: port %d not configured\n",
+			       __func__, __LINE__, i);
+			return -1;
+		}
+
+	/* 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) {
+			printf("%s %d: queue %d not configured\n",
+			       __func__, __LINE__, i);
+			return -1;
+		}
+
+	/* 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++;
+			}
+		}
+	}
+	sw->started = 1;
+	return 0;
+}
+
+static void
+sw_stop(struct rte_eventdev *dev)
+{
+	struct sw_evdev *sw = sw_pmd_priv(dev);
+	sw->started = 0;
+}
+
+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;
@@ -426,6 +497,9 @@ sw_probe(const char *name, const char *params)
 	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,
 
 			.queue_def_conf = sw_queue_def_conf,
 			.queue_setup = sw_queue_setup,
-- 
2.7.4

[dpdk-dev] [PATCH 13/15] event/sw: add dump function for easier debugging

[Harry van Haaren]

From: Bruce Richardson <bruce.richardson@intel.com>

Signed-off-by: Bruce Richardson <bruce.richardson@intel.com>
Signed-off-by: Harry van Haaren <harry.van.haaren@intel.com>
---
 drivers/event/sw/sw_evdev.c | 109 ++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 109 insertions(+)

diff --git a/drivers/event/sw/sw_evdev.c b/drivers/event/sw/sw_evdev.c
index 318f1d7..ea34b99 100644
--- a/drivers/event/sw/sw_evdev.c
+++ b/drivers/event/sw/sw_evdev.c
@@ -400,6 +400,114 @@ sw_info_get(struct rte_eventdev *dev, struct rte_event_dev_info *info)
 	*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 quanta = *((const uint32_t *)(&sw->inflight_quanta));
+	uint32_t credits = sw->nb_events_limit - (quanta * SW_INFLIGHT_QUANTA_SIZE);
+	fprintf(f, "\tinflight sw credits: %d\n", 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];
+		fprintf(f, "  Port %d %s\n", i,
+				p->is_directed ? " (SingleCons)" : "");
+		fprintf(f, "\trx   %"PRIu64"\tdrop %"PRIu64"\ttx   %"PRIu64"\tinflight %d\n",
+				sw->ports[i].stats.rx_pkts,
+			sw->ports[i].stats.rx_dropped,
+			sw->ports[i].stats.tx_pkts, sw->ports[i].inflights);
+
+		fprintf(f, "\tAvg cycles PP: %"PRIu64"\tCredits: %u\n",
+			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)
+				printf("%u-%u:%.02f%% ",
+						((j << SW_DEQ_STAT_BUCKET_SHIFT) + 1),
+						((j+1) << SW_DEQ_STAT_BUCKET_SHIFT),
+						p->poll_buckets[j] * 100.0 / p->total_polls);
+		printf("\n");
+
+		uint64_t rx_used = qe_ring_count(p->rx_worker_ring);
+		uint64_t rx_free = qe_ring_free_count(p->rx_worker_ring);
+		const char *rxcol = (rx_free == 0) ? COL_RED : COL_RESET;
+		fprintf(f, "\t%srx ring used: %4"PRIu64"\tfree: %4"PRIu64 COL_RESET"\n",
+				rxcol, rx_used, rx_free);
+
+		uint64_t tx_used = qe_ring_count(p->cq_worker_ring);
+		uint64_t tx_free = qe_ring_free_count(p->cq_worker_ring);
+		const char *txcol = (tx_free == 0) ? COL_RED : COL_RESET;
+		fprintf(f, "\t%scq ring used: %4"PRIu64"\tfree: %4"PRIu64 COL_RESET"\n",
+				txcol, tx_used, tx_free);
+	}
+
+	for (i = 0; i < sw->qid_count; i++) {
+		const struct sw_qid *qid = &sw->qids[i];
+		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)
+			fprintf(f, "\tReorder entries in use: %u\n",
+				rte_ring_free_count(qid->reorder_buffer_freelist));
+
+		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++) {
+			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)
 {
@@ -500,6 +608,7 @@ sw_probe(const char *name, const char *params)
 			.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,
-- 
2.7.4

[dpdk-dev] [PATCH 14/15] event/sw: add xstats support

[Harry van Haaren]

From: Bruce Richardson <bruce.richardson@intel.com>

Add support for xstats to report out on the state of the eventdev.
Useful for debugging and for unit tests, as well as observability
at runtime and performance tuning of apps to work well with the
scheduler.

Signed-off-by: Bruce Richardson <bruce.richardson@intel.com>
Signed-off-by: Harry van Haaren <harry.van.haaren@intel.com>
---
 drivers/event/sw/Makefile          |   1 +
 drivers/event/sw/sw_evdev.c        |   7 +
 drivers/event/sw/sw_evdev.h        |  10 +
 drivers/event/sw/sw_evdev_xstats.c | 401 +++++++++++++++++++++++++++++++++++++
 4 files changed, 419 insertions(+)
 create mode 100644 drivers/event/sw/sw_evdev_xstats.c

diff --git a/drivers/event/sw/Makefile b/drivers/event/sw/Makefile
index e96c457..890d2af 100644
--- a/drivers/event/sw/Makefile
+++ b/drivers/event/sw/Makefile
@@ -55,6 +55,7 @@ EXPORT_MAP := rte_pmd_evdev_sw_version.map
 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 +=
diff --git a/drivers/event/sw/sw_evdev.c b/drivers/event/sw/sw_evdev.c
index ea34b99..958e798 100644
--- a/drivers/event/sw/sw_evdev.c
+++ b/drivers/event/sw/sw_evdev.c
@@ -545,6 +545,8 @@ sw_start(struct rte_eventdev *dev)
 			}
 		}
 	}
+	if (sw_xstats_init(sw) < 0)
+		return -1;
 	sw->started = 1;
 	return 0;
 }
@@ -553,6 +555,7 @@ static void
 sw_stop(struct rte_eventdev *dev)
 {
 	struct sw_evdev *sw = sw_pmd_priv(dev);
+	sw_xstats_uninit(sw);
 	sw->started = 0;
 }
 
@@ -618,6 +621,10 @@ sw_probe(const char *name, const char *params)
 			.port_release = sw_port_release,
 			.port_link = sw_port_link,
 			.port_unlink = sw_port_unlink,
+
+			.get_xstat_names = sw_get_xstat_names,
+			.get_xstats = sw_get_xstats,
+			.get_xstat_by_name = sw_get_xstat_by_name,
 	};
 
 	static const char *const args[] = { NUMA_NODE_ARG, SCHED_QUANTA_ARG, NULL };
diff --git a/drivers/event/sw/sw_evdev.h b/drivers/event/sw/sw_evdev.h
index a0f3668..d82e61a 100644
--- a/drivers/event/sw/sw_evdev.h
+++ b/drivers/event/sw/sw_evdev.h
@@ -221,6 +221,8 @@ struct sw_evdev {
 
 	uint32_t port_count;
 	uint32_t qid_count;
+	uint32_t xstats_count;
+	struct sw_xstats_entry *xstats;
 
 	/* Contains all ports - load balanced and directed */
 	struct sw_port ports[SW_PORTS_MAX] __rte_cache_aligned;
@@ -273,5 +275,13 @@ 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_get_xstat_names(const struct rte_eventdev *dev,
+	struct rte_event_dev_xstat_name *xstats_names, unsigned int size);
+int sw_get_xstats(const struct rte_eventdev *dev, const unsigned int ids[],
+	uint64_t values[], unsigned int n);
+uint64_t sw_get_xstat_by_name(const struct rte_eventdev *dev,
+		const char *name, unsigned int *id);
 
 #endif /* _SW_EVDEV_H_ */
diff --git a/drivers/event/sw/sw_evdev_xstats.c b/drivers/event/sw/sw_evdev_xstats.c
new file mode 100644
index 0000000..f879303
--- /dev/null
+++ b/drivers/event/sw/sw_evdev_xstats.c
@@ -0,0 +1,401 @@
+/*-
+ *   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_xstat_name name;
+	xstats_fn fn;
+	uint16_t obj_idx;
+	enum xstats_type stat;
+	int extra_arg;
+};
+
+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 two parallel arrays, one for the
+	 * names, the other for the stat type parameter to be passed in the fn
+	 * call to get that stat. These two 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,
+	};
+
+	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 char * const port_bucket_stats[] = {
+			"dequeues_returning" };
+	static const enum xstats_type port_bucket_types[] = { poll_return };
+
+	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 char * const qid_iq_stats[] = { "used" };
+	static const enum xstats_type qid_iq_types[] = { iq_used };
+
+	static const char * const qid_port_stats[] = { "pinned_flows" };
+	static const enum xstats_type qid_port_types[] = { pinned };
+	/* ---- 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(port_bucket_stats) != RTE_DIM(port_bucket_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));
+
+	/* other vars */
+	const unsigned int count = RTE_DIM(dev_stats) +
+			sw->port_count * RTE_DIM(port_stats) +
+			sw->port_count * RTE_DIM(port_bucket_stats) *
+				((MAX_SW_CONS_Q_DEPTH >> SW_DEQ_STAT_BUCKET_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],
+		};
+		snprintf(sname, sizeof(sname), "dev_%s", dev_stats[i]);
+	}
+
+	for (port = 0; port < sw->port_count; port++) {
+		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],
+			};
+			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++, stat++) {
+				sw->xstats[stat] = (struct sw_xstats_entry){
+					.fn = get_port_bucket_stat,
+					.obj_idx = port,
+					.stat = port_bucket_types[i],
+					.extra_arg = bkt,
+				};
+				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);
+			}
+		}
+	}
+
+	for (qid = 0; qid < sw->qid_count; qid++) {
+		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],
+			};
+			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],
+					.extra_arg = iq
+				};
+				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],
+					.extra_arg = port
+				};
+				snprintf(sname, sizeof(sname),
+						"qid_%u_port_%u_%s",
+						qid, port,
+						qid_port_stats[i]);
+			}
+	}
+#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_get_xstat_names(const struct rte_eventdev *dev,
+	struct rte_event_dev_xstat_name *xstats_names, unsigned int size)
+{
+	const struct sw_evdev *sw = sw_pmd_priv_const(dev);
+	unsigned int i;
+
+	for (i = 0; i < sw->xstats_count && i < size; i++)
+		xstats_names[i] = sw->xstats[i].name;
+	return sw->xstats_count;
+}
+
+int
+sw_get_xstats(const struct rte_eventdev *dev, const unsigned int ids[],
+	uint64_t values[], unsigned int n)
+{
+	const struct sw_evdev *sw = sw_pmd_priv_const(dev);
+	unsigned int i;
+
+	for (i = 0; i < n; i++) {
+		struct sw_xstats_entry *xs = &sw->xstats[ids[i]];
+		if (ids[i] > sw->xstats_count)
+			continue;
+		values[i] = xs->fn(sw, xs->obj_idx, xs->stat, xs->extra_arg);
+	}
+
+	return 0;
+}
+
+uint64_t
+sw_get_xstat_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_XSTAT_NAME_SIZE) == 0){
+			if (id != NULL)
+				*id = i;
+			return xs->fn(sw, xs->obj_idx, xs->stat, xs->extra_arg);
+		}
+	}
+	if (id != NULL)
+		*id = (uint32_t)-1;
+	return (uint64_t)-1;
+}
-- 
2.7.4

[dpdk-dev] [PATCH 15/15] app/test: add unit tests for SW eventdev driver

[Harry van Haaren]

From: Bruce Richardson <bruce.richardson@intel.com>

Since the sw driver is a standalone lookaside device that has no HW
requirements, we can provide a set of unit tests that test its
functionality across the different queue types and with different input
scenarios.

Signed-off-by: Bruce Richardson <bruce.richardson@intel.com>
Signed-off-by: David Hunt <david.hunt@intel.com>
Signed-off-by: Harry van Haaren <harry.van.haaren@intel.com>
---
 app/test/Makefile           |    5 +-
 app/test/test_sw_eventdev.c | 2031 +++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 2035 insertions(+), 1 deletion(-)
 create mode 100644 app/test/test_sw_eventdev.c

diff --git a/app/test/Makefile b/app/test/Makefile
index e28c079..1770c09 100644
--- a/app/test/Makefile
+++ b/app/test/Makefile
@@ -197,7 +197,10 @@ SRCS-$(CONFIG_RTE_LIBRTE_CRYPTODEV) += test_cryptodev_blockcipher.c
 SRCS-$(CONFIG_RTE_LIBRTE_CRYPTODEV) += test_cryptodev_perf.c
 SRCS-$(CONFIG_RTE_LIBRTE_CRYPTODEV) += test_cryptodev.c
 
-SRCS-$(CONFIG_RTE_LIBRTE_EVENTDEV) += test_eventdev.c
+ifeq ($(CONFIG_RTE_LIBRTE_EVENTDEV),y)
+SRCS-y += test_eventdev.c
+SRCS-$(CONFIG_RTE_LIBRTE_PMD_SW_EVENTDEV) += test_sw_eventdev.c
+endif
 
 SRCS-$(CONFIG_RTE_LIBRTE_KVARGS) += test_kvargs.c
 
diff --git a/app/test/test_sw_eventdev.c b/app/test/test_sw_eventdev.c
new file mode 100644
index 0000000..13a8218
--- /dev/null
+++ b/app/test/test_sw_eventdev.c
@@ -0,0 +1,2031 @@
+/*-
+ *   BSD LICENSE
+ *
+ *   Copyright(c) 2016-2017 Intel Corporation. All rights reserved.
+ *   All rights reserved.
+ *
+ *   Redistribution and use in source and binary forms, with or without
+ *   modification, are permitted provided that the following conditions
+ *   are met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ *       notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above copyright
+ *       notice, this list of conditions and the following disclaimer in
+ *       the documentation and/or other materials provided with the
+ *       distribution.
+ *     * Neither the name of Intel Corporation nor the names of its
+ *       contributors may be used to endorse or promote products derived
+ *       from this software without specific prior written permission.
+ *
+ *   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ *   "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ *   LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ *   A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ *   OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ *   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ *   LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ *   DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ *   THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ *   (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ *   OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#include <stdio.h>
+#include <string.h>
+#include <stdint.h>
+#include <errno.h>
+#include <unistd.h>
+#include <sys/queue.h>
+
+#include <rte_memory.h>
+#include <rte_memzone.h>
+#include <rte_launch.h>
+#include <rte_eal.h>
+#include <rte_per_lcore.h>
+#include <rte_lcore.h>
+#include <rte_debug.h>
+#include <rte_ethdev.h>
+#include <rte_cycles.h>
+
+#include <rte_eventdev.h>
+#include "test.h"
+
+#define MAX_PORTS 16
+#define MAX_QIDS 16
+#define NUM_PACKETS (1<<18)
+
+static int evdev;
+
+struct test {
+	struct rte_mempool *mbuf_pool;
+	uint8_t port[MAX_PORTS];
+	uint8_t qid[MAX_QIDS];
+	int nb_qids;
+};
+
+static struct rte_event release_ev = {.op = RTE_EVENT_OP_RELEASE };
+
+static inline struct rte_mbuf *
+rte_gen_arp(int portid, struct rte_mempool *mp)
+{
+	/*
+	* len = 14 + 46
+	* ARP, Request who-has 10.0.0.1 tell 10.0.0.2, length 46
+	*/
+	static const uint8_t arp_request[] = {
+		/*0x0000:*/ 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xec, 0xa8,
+		0x6b, 0xfd, 0x02, 0x29, 0x08, 0x06, 0x00, 0x01,
+		/*0x0010:*/ 0x08, 0x00, 0x06, 0x04, 0x00, 0x01, 0xec, 0xa8,
+		0x6b, 0xfd, 0x02, 0x29, 0x0a, 0x00, 0x00, 0x01,
+		/*0x0020:*/ 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x0a, 0x00,
+		0x00, 0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+		/*0x0030:*/ 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+		0x00, 0x00, 0x00, 0x00
+	};
+	struct rte_mbuf *m;
+	int pkt_len = sizeof(arp_request) - 1;
+
+	m = rte_pktmbuf_alloc(mp);
+	if (!m)
+		return 0;
+
+	memcpy((void *)((uintptr_t)m->buf_addr + m->data_off),
+		arp_request, pkt_len);
+	rte_pktmbuf_pkt_len(m) = pkt_len;
+	rte_pktmbuf_data_len(m) = pkt_len;
+
+	RTE_SET_USED(portid);
+
+	return m;
+}
+
+/* initialization and qm config */
+static inline int
+init(struct test *t, int nb_queues, int nb_ports)
+{
+	struct rte_event_dev_config config = {
+			.nb_event_queues = nb_queues,
+			.nb_event_ports = nb_ports,
+			.nb_event_queue_flows = 1024,
+			.nb_events_limit = 4096,
+			.nb_event_port_dequeue_depth = 128,
+			.nb_event_port_enqueue_depth = 128,
+	};
+	int ret;
+
+	void *temp = t->mbuf_pool; /* save and restore mbuf pool */
+
+	memset(t, 0, sizeof(*t));
+	t->mbuf_pool = temp;
+
+	ret = rte_event_dev_configure(evdev, &config);
+	if (ret < 0)
+		printf("%d: Error configuring device\n", __LINE__);
+	return ret;
+};
+
+static inline int
+create_ports(struct test *t, int num_ports)
+{
+	int i;
+	static const struct rte_event_port_conf conf = {
+			.new_event_threshold = 1024,
+			.dequeue_depth = 32,
+			.enqueue_depth = 64,
+	};
+	if (num_ports > MAX_PORTS)
+		return -1;
+
+	for (i = 0; i < num_ports; i++) {
+		if (rte_event_port_setup(evdev, i, &conf) < 0) {
+			printf("Error setting up port %d\n", i);
+			return -1;
+		}
+		t->port[i] = i;
+	}
+
+	return 0;
+}
+
+static inline int
+create_lb_qids(struct test *t, int num_qids, uint32_t flags)
+{
+	int i;
+
+	/* Q creation */
+	const struct rte_event_queue_conf conf = {
+			.event_queue_cfg = flags,
+			.priority = RTE_EVENT_DEV_PRIORITY_NORMAL,
+			.nb_atomic_flows = 1024,
+			.nb_atomic_order_sequences = 1024,
+	};
+
+	for (i = t->nb_qids; i < t->nb_qids + num_qids; i++) {
+		if (rte_event_queue_setup(evdev, i, &conf) < 0) {
+			printf("%d: error creating qid %d\n", __LINE__, i);
+			return -1;
+		}
+		t->qid[i] = i;
+	}
+	t->nb_qids += num_qids;
+	if (t->nb_qids > MAX_QIDS)
+		return -1;
+
+	return 0;
+}
+
+static inline int
+create_atomic_qids(struct test *t, int num_qids)
+{
+	return create_lb_qids(t, num_qids, RTE_EVENT_QUEUE_CFG_ATOMIC_ONLY);
+}
+
+static inline int
+create_ordered_qids(struct test *t, int num_qids)
+{
+	return create_lb_qids(t, num_qids, RTE_EVENT_QUEUE_CFG_ORDERED_ONLY);
+}
+
+
+static inline int
+create_unordered_qids(struct test *t, int num_qids)
+{
+	return create_lb_qids(t, num_qids, RTE_EVENT_QUEUE_CFG_PARALLEL_ONLY);
+}
+
+static inline int
+create_directed_qids(struct test *t, int num_qids, const uint8_t ports[])
+{
+	int i;
+
+	/* Q creation */
+	static const struct rte_event_queue_conf conf = {
+			.priority = RTE_EVENT_DEV_PRIORITY_NORMAL,
+			.event_queue_cfg = RTE_EVENT_QUEUE_CFG_SINGLE_LINK,
+			.nb_atomic_flows = 1024,
+			.nb_atomic_order_sequences = 1024,
+	};
+
+	for (i = t->nb_qids; i < t->nb_qids + num_qids; i++) {
+		if (rte_event_queue_setup(evdev, i, &conf) < 0) {
+			printf("%d: error creating qid %d\n", __LINE__, i);
+			return -1;
+		}
+		t->qid[i] = i;
+
+		if (rte_event_port_link(evdev, ports[i - t->nb_qids],
+				&t->qid[i], NULL, 1) != 1) {
+			printf("%d: error creating link for qid %d\n",
+					__LINE__, i);
+			return -1;
+		}
+	}
+	t->nb_qids += num_qids;
+	if (t->nb_qids > MAX_QIDS)
+		return -1;
+
+	return 0;
+}
+
+/* destruction */
+static inline int
+cleanup(struct test *t __rte_unused)
+{
+	rte_event_dev_stop(evdev);
+	rte_event_dev_close(evdev);
+	return 0;
+};
+
+struct test_event_dev_stats {
+	uint64_t rx_pkts;       /**< Total packets received */
+	uint64_t rx_dropped;    /**< Total packets dropped (Eg Invalid QID) */
+	uint64_t tx_pkts;       /**< Total packets transmitted */
+
+	/** Packets received on this port */
+	uint64_t port_rx_pkts[MAX_PORTS];
+	/** Packets dropped on this port */
+	uint64_t port_rx_dropped[MAX_PORTS];
+	/** Packets inflight on this port */
+	uint64_t port_inflight[MAX_PORTS];
+	/** Packets transmitted on this port */
+	uint64_t port_tx_pkts[MAX_PORTS];
+	/** Packets received on this qid */
+	uint64_t qid_rx_pkts[MAX_QIDS];
+	/** Packets dropped on this qid */
+	uint64_t qid_rx_dropped[MAX_QIDS];
+	/** Packets transmitted on this qid */
+	uint64_t qid_tx_pkts[MAX_QIDS];
+};
+
+static inline int
+test_event_dev_stats_get(int dev_id, struct test_event_dev_stats *stats)
+{
+	static uint32_t i;
+	static uint32_t total_ids[3]; /* rx, tx and drop */
+	static uint32_t port_rx_pkts_ids[MAX_PORTS];
+	static uint32_t port_rx_dropped_ids[MAX_PORTS];
+	static uint32_t port_inflight_ids[MAX_PORTS];
+	static uint32_t port_tx_pkts_ids[MAX_PORTS];
+	static uint32_t qid_rx_pkts_ids[MAX_QIDS];
+	static uint32_t qid_rx_dropped_ids[MAX_QIDS];
+	static uint32_t qid_tx_pkts_ids[MAX_QIDS];
+
+
+	stats->rx_pkts = rte_event_dev_get_xstat_by_name(dev_id,
+			"dev_rx", &total_ids[0]);
+	stats->rx_dropped = rte_event_dev_get_xstat_by_name(dev_id,
+			"dev_drop", &total_ids[1]);
+	stats->tx_pkts = rte_event_dev_get_xstat_by_name(dev_id,
+			"dev_tx", &total_ids[2]);
+	for (i = 0; i < MAX_PORTS; i++) {
+		char name[32];
+		snprintf(name, sizeof(name), "port_%u_rx", i);
+		stats->port_rx_pkts[i] = rte_event_dev_get_xstat_by_name(
+				dev_id, name, &port_rx_pkts_ids[i]);
+		snprintf(name, sizeof(name), "port_%u_drop", i);
+		stats->port_rx_dropped[i] = rte_event_dev_get_xstat_by_name(
+				dev_id, name, &port_rx_dropped_ids[i]);
+		snprintf(name, sizeof(name), "port_%u_inflight", i);
+		stats->port_inflight[i] = rte_event_dev_get_xstat_by_name(
+				dev_id, name, &port_inflight_ids[i]);
+		snprintf(name, sizeof(name), "port_%u_tx", i);
+		stats->port_tx_pkts[i] = rte_event_dev_get_xstat_by_name(
+				dev_id, name, &port_tx_pkts_ids[i]);
+	}
+	for (i = 0; i < MAX_QIDS; i++) {
+		char name[32];
+		snprintf(name, sizeof(name), "qid_%u_rx", i);
+		stats->qid_rx_pkts[i] = rte_event_dev_get_xstat_by_name(
+				dev_id, name, &qid_rx_pkts_ids[i]);
+		snprintf(name, sizeof(name), "qid_%u_drop", i);
+		stats->qid_rx_dropped[i] = rte_event_dev_get_xstat_by_name(
+				dev_id, name, &qid_rx_dropped_ids[i]);
+		snprintf(name, sizeof(name), "qid_%u_tx", i);
+		stats->qid_tx_pkts[i] = rte_event_dev_get_xstat_by_name(
+				dev_id, name, &qid_tx_pkts_ids[i]);
+
+	}
+
+	return 0;
+}
+
+/* run_prio_packet_test
+ * This performs a basic packet priority check on the test instance passed in.
+ * It is factored out of the main priority tests as the same tests must be
+ * performed to ensure prioritization of each type of QID.
+ *
+ * Requirements:
+ *  - An initialized test structure, including mempool
+ *  - t->port[0] is initialized for both Enq / Deq of packets to the QID
+ *  - t->qid[0] is the QID to be tested
+ *  - if LB QID, the CQ must be mapped to the QID.
+ */
+static int
+run_prio_packet_test(struct test *t)
+{
+	int err;
+	const uint32_t MAGIC_SEQN[] = {4711, 1234};
+	const uint32_t PRIORITY[] = {3, 0};
+	unsigned int i;
+	for (i = 0; i < RTE_DIM(MAGIC_SEQN); i++) {
+		/* generate pkt and enqueue */
+		struct rte_event ev;
+		struct rte_mbuf *arp = rte_gen_arp(0, t->mbuf_pool);
+		if (!arp) {
+			printf("%d: gen of pkt failed\n", __LINE__);
+			return -1;
+		}
+		arp->seqn = MAGIC_SEQN[i];
+
+		ev = (struct rte_event){
+			.priority = PRIORITY[i],
+			.op = RTE_EVENT_OP_NEW,
+			.queue_id = t->qid[0],
+			.mbuf = arp
+		};
+		err = rte_event_enqueue_burst(evdev, t->port[0], &ev, 1);
+		if (err < 0) {
+			printf("%d: error failed to enqueue\n", __LINE__);
+			return -1;
+		}
+	}
+
+	rte_event_schedule(evdev);
+
+	struct test_event_dev_stats stats;
+	err = test_event_dev_stats_get(evdev, &stats);
+	if (err) {
+		printf("%d: error failed to get stats\n", __LINE__);
+		return -1;
+	}
+
+	if (stats.port_rx_pkts[t->port[0]] != 2) {
+		printf("%d: error stats incorrect for directed port\n", __LINE__);
+		rte_event_dev_dump(evdev, stdout);
+		return -1;
+	}
+
+	struct rte_event ev, ev2;
+	uint32_t deq_pkts;
+	deq_pkts = rte_event_dequeue_burst(evdev, t->port[0], &ev, 1, 0);
+	if (deq_pkts != 1) {
+		printf("%d: error failed to deq\n", __LINE__);
+		rte_event_dev_dump(evdev, stdout);
+		return -1;
+	}
+	if (ev.mbuf->seqn != MAGIC_SEQN[1]) {
+		printf("%d: first packet out not highest priority\n", __LINE__);
+		rte_event_dev_dump(evdev, stdout);
+		return -1;
+	}
+	rte_pktmbuf_free(ev.mbuf);
+
+	deq_pkts = rte_event_dequeue_burst(evdev, t->port[0], &ev2, 1, 0);
+	if (deq_pkts != 1) {
+		printf("%d: error failed to deq\n", __LINE__);
+		rte_event_dev_dump(evdev, stdout);
+		return -1;
+	}
+	if (ev2.mbuf->seqn != MAGIC_SEQN[0]) {
+		printf("%d: second packet out not lower priority\n", __LINE__);
+		rte_event_dev_dump(evdev, stdout);
+		return -1;
+	}
+	rte_pktmbuf_free(ev2.mbuf);
+
+	cleanup(t);
+	return 0;
+}
+
+static int
+test_single_directed_packet(struct test *t)
+{
+	const int rx_enq = 0;
+	const int wrk_enq = 2;
+	int err;
+
+	/* Create qm instance with 3 directed QIDs going to 3 ports */
+	if (init(t, 3, 3) < 0 ||
+			create_ports(t, 3) < 0 ||
+			create_directed_qids(t, 3, t->port) < 0)
+		return -1;
+
+	if (rte_event_dev_start(evdev) < 0) {
+		printf("%d: Error with start call\n", __LINE__);
+		return -1;
+	}
+
+	/************** FORWARD ****************/
+	struct rte_mbuf *arp = rte_gen_arp(0, t->mbuf_pool);
+	struct rte_event ev = {
+			.op = RTE_EVENT_OP_NEW,
+			.queue_id = wrk_enq,
+			.mbuf = arp,
+	};
+
+	if (!arp) {
+		printf("%d: gen of pkt failed\n", __LINE__);
+		return -1;
+	}
+
+	const uint32_t MAGIC_SEQN = 4711;
+	arp->seqn = MAGIC_SEQN;
+
+	/* generate pkt and enqueue */
+	err = rte_event_enqueue_burst(evdev, rx_enq, &ev, 1);
+	if (err < 0) {
+		printf("%d: error failed to enqueue\n", __LINE__);
+		return -1;
+	}
+
+	/* Run schedule() as dir packets may need to be re-ordered */
+	rte_event_schedule(evdev);
+
+	struct test_event_dev_stats stats;
+	err = test_event_dev_stats_get(evdev, &stats);
+	if (err) {
+		printf("%d: error failed to get stats\n", __LINE__);
+		return -1;
+	}
+
+	if (stats.port_rx_pkts[rx_enq] != 1) {
+		printf("%d: error stats incorrect for directed port\n", __LINE__);
+		return -1;
+	}
+
+	uint32_t deq_pkts;
+	deq_pkts = rte_event_dequeue_burst(evdev, wrk_enq, &ev, 1, 0);
+	if (deq_pkts != 1) {
+		printf("%d: error failed to deq\n", __LINE__);
+		return -1;
+	}
+
+	err = test_event_dev_stats_get(evdev, &stats);
+	if (stats.port_rx_pkts[wrk_enq] != 0 &&
+			stats.port_rx_pkts[wrk_enq] != 1) {
+		printf("%d: error directed stats post-dequeue\n", __LINE__);
+		return -1;
+	}
+
+	if (ev.mbuf->seqn != MAGIC_SEQN) {
+		printf("%d: error magic sequence number not dequeued\n", __LINE__);
+		return -1;
+	}
+
+	rte_pktmbuf_free(ev.mbuf);
+	cleanup(t);
+	return 0;
+}
+
+
+static int
+test_priority_directed(struct test *t)
+{
+	if (init(t, 1, 1) < 0 ||
+			create_ports(t, 1) < 0 ||
+			create_directed_qids(t, 1, t->port) < 0) {
+		printf("%d: Error initializing device\n", __LINE__);
+		return -1;
+	}
+
+	if (rte_event_dev_start(evdev) < 0) {
+		printf("%d: Error with start call\n", __LINE__);
+		return -1;
+	}
+
+	return run_prio_packet_test(t);
+}
+
+static int
+test_priority_atomic(struct test *t)
+{
+	if (init(t, 1, 1) < 0 ||
+			create_ports(t, 1) < 0 ||
+			create_atomic_qids(t, 1) < 0) {
+		printf("%d: Error initializing device\n", __LINE__);
+		return -1;
+	}
+
+	/* map the QID */
+	if (rte_event_port_link(evdev, t->port[0], &t->qid[0], NULL, 1) != 1) {
+		printf("%d: error mapping qid to port\n", __LINE__);
+		return -1;
+	}
+	if (rte_event_dev_start(evdev) < 0) {
+		printf("%d: Error with start call\n", __LINE__);
+		return -1;
+	}
+
+	return run_prio_packet_test(t);
+}
+
+static int
+test_priority_ordered(struct test *t)
+{
+	if (init(t, 1, 1) < 0 ||
+			create_ports(t, 1) < 0 ||
+			create_ordered_qids(t, 1) < 0) {
+		printf("%d: Error initializing device\n", __LINE__);
+		return -1;
+	}
+
+	/* map the QID */
+	if (rte_event_port_link(evdev, t->port[0], &t->qid[0], NULL, 1) != 1) {
+		printf("%d: error mapping qid to port\n", __LINE__);
+		return -1;
+	}
+	if (rte_event_dev_start(evdev) < 0) {
+		printf("%d: Error with start call\n", __LINE__);
+		return -1;
+	}
+
+	return run_prio_packet_test(t);
+}
+
+static int
+test_priority_unordered(struct test *t)
+{
+	if (init(t, 1, 1) < 0 ||
+			create_ports(t, 1) < 0 ||
+			create_unordered_qids(t, 1) < 0) {
+		printf("%d: Error initializing device\n", __LINE__);
+		return -1;
+	}
+
+	/* map the QID */
+	if (rte_event_port_link(evdev, t->port[0], &t->qid[0], NULL, 1) != 1) {
+		printf("%d: error mapping qid to port\n", __LINE__);
+		return -1;
+	}
+	if (rte_event_dev_start(evdev) < 0) {
+		printf("%d: Error with start call\n", __LINE__);
+		return -1;
+	}
+
+	return run_prio_packet_test(t);
+}
+
+static int
+burst_packets(struct test *t)
+{
+	/************** CONFIG ****************/
+	uint32_t i;
+	int err;
+	int ret;
+
+	/* Create qm instance with 2 ports and 2 queues */
+	if (init(t, 2, 2) < 0 ||
+			create_ports(t, 2) < 0 ||
+			create_atomic_qids(t, 2) < 0) {
+		printf("%d: Error initializing device\n", __LINE__);
+		return -1;
+	}
+
+	/* CQ mapping to QID */
+	ret = rte_event_port_link(evdev, t->port[0], &t->qid[0], NULL, 1);
+	if (ret != 1) {
+		printf("%d: error mapping lb qid0\n", __LINE__);
+		return -1;
+	}
+	ret = rte_event_port_link(evdev, t->port[1], &t->qid[1], NULL, 1);
+	if (ret != 1) {
+		printf("%d: error mapping lb qid1\n", __LINE__);
+		return -1;
+	}
+
+	if (rte_event_dev_start(evdev) < 0) {
+		printf("%d: Error with start call\n", __LINE__);
+		return -1;
+	}
+
+	/************** FORWARD ****************/
+	const uint32_t rx_port = 0;
+	const uint32_t NUM_PKTS = 2;
+
+	for (i = 0; i < NUM_PKTS; i++) {
+		struct rte_mbuf *arp = rte_gen_arp(0, t->mbuf_pool);
+		if (!arp) {
+			printf("%d: error generating pkt\n", __LINE__);
+			return -1;
+		}
+
+		struct rte_event ev = {
+				.op = RTE_EVENT_OP_NEW,
+				.queue_id = i % 2,
+				.flow_id = i % 3,
+				.mbuf = arp,
+		};
+		/* generate pkt and enqueue */
+		err = rte_event_enqueue_burst(evdev, t->port[rx_port], &ev, 1);
+		if (err < 0) {
+			printf("%d: Failed to enqueue\n", __LINE__);
+			return -1;
+		}
+	}
+	rte_event_schedule(evdev);
+
+	/* Check stats for all NUM_PKTS arrived to sched core */
+	struct test_event_dev_stats stats;
+
+	err = test_event_dev_stats_get(evdev, &stats);
+	if (err) {
+		printf("%d: failed to get stats\n", __LINE__);
+		return -1;
+	}
+	if (stats.rx_pkts != NUM_PKTS || stats.tx_pkts != NUM_PKTS) {
+		printf("%d: Sched core didn't receive all %d pkts\n",
+				__LINE__, NUM_PKTS);
+		rte_event_dev_dump(evdev, stdout);
+		return -1;
+	}
+
+	uint32_t deq_pkts;
+	int p;
+
+	deq_pkts = 0;
+	/******** DEQ QID 1 *******/
+	do {
+		struct rte_event ev;
+		p = rte_event_dequeue_burst(evdev, t->port[0], &ev, 1, 0);
+		deq_pkts += p;
+		rte_pktmbuf_free(ev.mbuf);
+	} while (p);
+
+	if (deq_pkts != NUM_PKTS/2) {
+		printf("%d: Half of NUM_PKTS didn't arrive at port 1\n", __LINE__);
+		return -1;
+	}
+
+	/******** DEQ QID 2 *******/
+	deq_pkts = 0;
+	do {
+		struct rte_event ev;
+		p = rte_event_dequeue_burst(evdev, t->port[1], &ev, 1, 0);
+		deq_pkts += p;
+		rte_pktmbuf_free(ev.mbuf);
+	} while (p);
+	if (deq_pkts != NUM_PKTS/2) {
+		printf("%d: Half of NUM_PKTS didn't arrive at port 2\n", __LINE__);
+		return -1;
+	}
+
+	cleanup(t);
+	return 0;
+}
+
+static int
+abuse_inflights(struct test *t)
+{
+	const int rx_enq = 0;
+	const int wrk_enq = 2;
+	int err;
+
+	/* Create qm instance with 4 ports */
+	if (init(t, 1, 4) < 0 ||
+			create_ports(t, 4) < 0 ||
+			create_atomic_qids(t, 1) < 0) {
+		printf("%d: Error initializing device\n", __LINE__);
+		return -1;
+	}
+
+	/* CQ mapping to QID */
+	err = rte_event_port_link(evdev, t->port[wrk_enq], NULL, NULL, 0);
+	if (err != 1) {
+		printf("%d: error mapping lb qid\n", __LINE__);
+		cleanup(t);
+		return -1;
+	}
+
+	if (rte_event_dev_start(evdev) < 0) {
+		printf("%d: Error with start call\n", __LINE__);
+		return -1;
+	}
+
+	/* Enqueue op only */
+	err = rte_event_enqueue_burst(evdev, t->port[rx_enq], &release_ev, 1);
+	if (err < 0) {
+		printf("%d: Failed to enqueue\n", __LINE__);
+		return -1;
+	}
+
+	/* schedule */
+	rte_event_schedule(evdev);
+
+	struct test_event_dev_stats stats;
+
+	err = test_event_dev_stats_get(evdev, &stats);
+	if (err) {
+		printf("%d: failed to get stats\n", __LINE__);
+		return -1;
+	}
+
+	if (stats.rx_pkts != 0 ||
+			stats.tx_pkts != 0 ||
+			stats.port_inflight[wrk_enq] != 0) {
+		printf("%d: Sched core didn't handle pkt as expected\n", __LINE__);
+		return -1;
+	}
+
+	cleanup(t);
+	return 0;
+}
+
+static int
+qid_priorities(struct test *t)
+{
+	/* Test works by having a CQ with enough empty space for all packets,
+	 * and enqueueing 3 packets to 3 QIDs. They must return based on the
+	 * priority of the QID, not the ingress order, to pass the test
+	 */
+	unsigned int i;
+	/* Create qm instance with 1 ports, and 3 qids */
+	if (init(t, 3, 1) < 0 ||
+			create_ports(t, 1) < 0) {
+		printf("%d: Error initializing device\n", __LINE__);
+		return -1;
+	}
+
+	for (i = 0; i < 3; i++) {
+		/* Create QID */
+		const struct rte_event_queue_conf conf = {
+				.event_queue_cfg = RTE_EVENT_QUEUE_CFG_ATOMIC_ONLY,
+				/* increase priority (0 == highest), as we go */
+				.priority = RTE_EVENT_DEV_PRIORITY_NORMAL - i,
+				.nb_atomic_flows = 1024,
+				.nb_atomic_order_sequences = 1024,
+		};
+
+		if (rte_event_queue_setup(evdev, i, &conf) < 0) {
+			printf("%d: error creating qid %d\n", __LINE__, i);
+			return -1;
+		}
+		t->qid[i] = i;
+	}
+	t->nb_qids = i;
+	/* map all QIDs to port */
+	rte_event_port_link(evdev, t->port[0], NULL, NULL, 0);
+
+	if (rte_event_dev_start(evdev) < 0) {
+		printf("%d: Error with start call\n", __LINE__);
+		return -1;
+	}
+
+	/* enqueue 3 packets, setting seqn and QID as needed to check priority */
+	for (i = 0; i < 3; i++) {
+		struct rte_event ev;
+		struct rte_mbuf *arp = rte_gen_arp(0, t->mbuf_pool);
+		if (!arp) {
+			printf("%d: gen of pkt failed\n", __LINE__);
+			return -1;
+		}
+		ev.queue_id = t->qid[i];
+		ev.op = RTE_EVENT_OP_NEW;
+		ev.mbuf = arp;
+		arp->seqn = i;
+
+		int err = rte_event_enqueue_burst(evdev, t->port[0], &ev, 1);
+		if (err != 1) {
+			printf("%d: Failed to enqueue\n", __LINE__);
+			return -1;
+		}
+	}
+
+	rte_event_schedule(evdev);
+
+	/* dequeue packets, verify priority was upheld */
+	struct rte_event ev[32];
+	uint32_t deq_pkts = rte_event_dequeue_burst(evdev, t->port[0], ev, 32, 0);
+	if (deq_pkts != 3) {
+		printf("%d: failed to deq packets\n", __LINE__);
+		rte_event_dev_dump(evdev, stdout);
+		return -1;
+	}
+	for (i = 0; i < 3; i++) {
+		if (ev[i].mbuf->seqn != 2-i) {
+			printf("%d: qid priority test: seqn %d incorrectly prioritized\n",
+					__LINE__, i);
+		}
+	}
+
+	cleanup(t);
+	return 0;
+}
+
+static int
+load_balancing(struct test *t)
+{
+	const int rx_enq = 0;
+	int err;
+	uint32_t i;
+
+	if (init(t, 1, 4) < 0 ||
+			create_ports(t, 4) < 0 ||
+			create_atomic_qids(t, 1) < 0) {
+		printf("%d: Error initializing device\n", __LINE__);
+		return -1;
+	}
+
+	for (i = 0; i < 3; i++) {
+		/* map port 1 - 3 inclusive */
+		if (rte_event_port_link(evdev, t->port[i+1], &t->qid[0],
+				NULL, 1) != 1) {
+			printf("%d: error mapping qid to port %d\n", __LINE__, i);
+			return -1;
+		}
+	}
+
+	if (rte_event_dev_start(evdev) < 0) {
+		printf("%d: Error with start call\n", __LINE__);
+		return -1;
+	}
+
+	/************** FORWARD ****************/
+	/*
+	 * Create a set of flows that test the load-balancing operation of the
+	 * sqm implementation. Fill CQ 0 and 1 with flows 0 and 1, and test
+	 * with a new flow, which should be sent to the 3rd mapped CQ
+	 */
+	static uint32_t flows[] = {0, 1, 1, 0, 0, 2, 2, 0, 2};
+
+	for (i = 0; i < RTE_DIM(flows); i++) {
+		struct rte_mbuf *arp = rte_gen_arp(0, t->mbuf_pool);
+		if (!arp) {
+			printf("%d: gen of pkt failed\n", __LINE__);
+			return -1;
+		}
+
+		struct rte_event ev = {
+				.op = RTE_EVENT_OP_NEW,
+				.queue_id = t->qid[0],
+				.flow_id = flows[i],
+				.mbuf = arp,
+		};
+		/* generate pkt and enqueue */
+		err = rte_event_enqueue_burst(evdev, t->port[rx_enq], &ev, 1);
+		if (err < 0) {
+			printf("%d: Failed to enqueue\n", __LINE__);
+			return -1;
+		}
+	}
+
+	rte_event_schedule(evdev);
+
+	struct test_event_dev_stats stats;
+	err = test_event_dev_stats_get(evdev, &stats);
+	if (err) {
+		printf("%d: failed to get stats\n", __LINE__);
+		return -1;
+	}
+
+	if (stats.port_inflight[1] != 4) {
+		printf("%d:%s: port 1 inflight not correct\n", __LINE__, __func__);
+		return -1;
+	}
+	if (stats.port_inflight[2] != 2) {
+		printf("%d:%s: port 2 inflight not correct\n", __LINE__, __func__);
+		return -1;
+	}
+	if (stats.port_inflight[3] != 3) {
+		printf("%d:%s: port 3 inflight not correct\n", __LINE__, __func__);
+		return -1;
+	}
+
+	cleanup(t);
+	return 0;
+}
+
+static int
+load_balancing_history(struct test *t)
+{
+	struct test_event_dev_stats stats = {0};
+	const int rx_enq = 0;
+	int err;
+	uint32_t i;
+
+	/* Create qm instance with 1 atomic QID going to 3 ports + 1 prod port */
+	if (init(t, 1, 4) < 0 ||
+			create_ports(t, 4) < 0 ||
+			create_atomic_qids(t, 1) < 0)
+		return -1;
+
+	/* CQ mapping to QID */
+	if (rte_event_port_link(evdev, t->port[1], &t->qid[0], NULL, 1) != 1) {
+		printf("%d: error mapping port 1 qid\n", __LINE__);
+		return -1;
+	}
+	if (rte_event_port_link(evdev, t->port[2], &t->qid[0], NULL, 1) != 1) {
+		printf("%d: error mapping port 2 qid\n", __LINE__);
+		return -1;
+	}
+	if (rte_event_port_link(evdev, t->port[3], &t->qid[0], NULL, 1) != 1) {
+		printf("%d: error mapping port 3 qid\n", __LINE__);
+		return -1;
+	}
+	if (rte_event_dev_start(evdev) < 0) {
+		printf("%d: Error with start call\n", __LINE__);
+		return -1;
+	}
+
+	/*
+	 * Create a set of flows that test the load-balancing operation of the
+	 * sqm implementation. Fill CQ 0, 1 and 2 with flows 0, 1 and 2, drop
+	 * the packet from CQ 0, send in a new set of flows. Ensure that:
+	 *  1. The new flow 3 gets into the empty CQ0
+	 *  2. packets for existing flow gets added into CQ1
+	 *  3. Next flow 0 pkt is now onto CQ2, since CQ0 and CQ1 now contain
+	 *     more outstanding pkts
+	 *
+	 *  This test makes sure that when a flow ends (i.e. all packets
+	 *  have been completed for that flow), that the flow can be moved
+	 *  to a different CQ when new packets come in for that flow.
+	 */
+	static uint32_t flows1[] = {0, 1, 1, 2};
+
+	for (i = 0; i < RTE_DIM(flows1); i++) {
+		struct rte_mbuf *arp = rte_gen_arp(0, t->mbuf_pool);
+		struct rte_event ev = {
+				.flow_id = flows1[i],
+				.op = RTE_EVENT_OP_NEW,
+				.queue_id = t->qid[0],
+				.event_type = RTE_EVENT_TYPE_CPU,
+				.priority = RTE_EVENT_DEV_PRIORITY_NORMAL,
+				.mbuf = arp
+		};
+
+		if (!arp) {
+			printf("%d: gen of pkt failed\n", __LINE__);
+			return -1;
+		}
+		arp->hash.rss = flows1[i];
+		err = rte_event_enqueue_burst(evdev, t->port[rx_enq], &ev, 1);
+		if (err < 0) {
+			printf("%d: Failed to enqueue\n", __LINE__);
+			return -1;
+		}
+	}
+
+	/* call the scheduler */
+	rte_event_schedule(evdev);
+
+	/* Dequeue the flow 0 packet from port 1, so that we can then drop */
+	struct rte_event ev;
+	if (!rte_event_dequeue_burst(evdev, t->port[1], &ev, 1, 0)) {
+		printf("%d: failed to dequeue\n", __LINE__);
+		return -1;
+	}
+	if (ev.mbuf->hash.rss != flows1[0]) {
+		printf("%d: unexpected flow received\n", __LINE__);
+		return -1;
+	}
+
+	/* drop the flow 0 packet from port 1 */
+	rte_event_enqueue_burst(evdev, t->port[1], &release_ev, 1);
+
+	/* call the scheduler */
+	rte_event_schedule(evdev);
+
+	/*
+	 * Set up the next set of flows, first a new flow to fill up
+	 * CQ 0, so that the next flow 0 packet should go to CQ2
+	 */
+	static uint32_t flows2[] = { 3, 3, 3, 1, 1, 0 };
+
+	for (i = 0; i < RTE_DIM(flows2); i++) {
+		struct rte_mbuf *arp = rte_gen_arp(0, t->mbuf_pool);
+		struct rte_event ev = {
+				.flow_id = flows2[i],
+				.op = RTE_EVENT_OP_NEW,
+				.queue_id = t->qid[0],
+				.event_type = RTE_EVENT_TYPE_CPU,
+				.priority = RTE_EVENT_DEV_PRIORITY_NORMAL,
+				.mbuf = arp
+		};
+
+		if (!arp) {
+			printf("%d: gen of pkt failed\n", __LINE__);
+			return -1;
+		}
+		arp->hash.rss = flows2[i];
+
+		err = rte_event_enqueue_burst(evdev, t->port[rx_enq], &ev, 1);
+		if (err < 0) {
+			printf("%d: Failed to enqueue\n", __LINE__);
+			return -1;
+		}
+	}
+
+	/* schedule */
+	rte_event_schedule(evdev);
+
+	/* rte_event_dev_dump(stdout, t->qm); */
+	err = test_event_dev_stats_get(evdev, &stats);
+	if (err) {
+		printf("%d:failed to get stats\n", __LINE__);
+		return -1;
+	}
+
+	/*
+	 * Now check the resulting inflights on each port.
+	 */
+	if (stats.port_inflight[1] != 3) {
+		printf("%d:%s: port 1 inflight not correct\n", __LINE__, __func__);
+		printf("Inflights, ports 1, 2, 3: %u, %u, %u\n",
+				(unsigned int)stats.port_inflight[1],
+				(unsigned int)stats.port_inflight[2],
+				(unsigned int)stats.port_inflight[3]);
+		return -1;
+	}
+	if (stats.port_inflight[2] != 4) {
+		printf("%d:%s: port 2 inflight not correct\n", __LINE__, __func__);
+		printf("Inflights, ports 1, 2, 3: %u, %u, %u\n",
+				(unsigned int)stats.port_inflight[1],
+				(unsigned int)stats.port_inflight[2],
+				(unsigned int)stats.port_inflight[3]);
+		return -1;
+	}
+	if (stats.port_inflight[3] != 2) {
+		printf("%d:%s: port 3 inflight not correct\n", __LINE__, __func__);
+		printf("Inflights, ports 1, 2, 3: %u, %u, %u\n",
+				(unsigned int)stats.port_inflight[1],
+				(unsigned int)stats.port_inflight[2],
+				(unsigned int)stats.port_inflight[3]);
+		return -1;
+	}
+
+	for (i = 1; i <= 3; i++) {
+		struct rte_event ev;
+		while (rte_event_dequeue_burst(evdev, i, &ev, 1, 0))
+			rte_event_enqueue_burst(evdev, i, &release_ev, 1);
+	}
+	rte_event_schedule(evdev);
+
+	cleanup(t);
+	return 0;
+}
+
+static int
+invalid_qid(struct test *t)
+{
+	struct test_event_dev_stats stats;
+	const int rx_enq = 0;
+	int err;
+	uint32_t i;
+
+	if (init(t, 1, 4) < 0 ||
+			create_ports(t, 4) < 0 ||
+			create_atomic_qids(t, 1) < 0) {
+		printf("%d: Error initializing device\n", __LINE__);
+		return -1;
+	}
+
+	/* CQ mapping to QID */
+	for (i = 0; i < 4; i++) {
+		err = rte_event_port_link(evdev, t->port[i], &t->qid[0], NULL, 1);
+		if (err != 1) {
+			printf("%d: error mapping port 1 qid\n", __LINE__);
+			return -1;
+		}
+	}
+
+	if (rte_event_dev_start(evdev) < 0) {
+		printf("%d: Error with start call\n", __LINE__);
+		return -1;
+	}
+
+	/*
+	 * Send in a packet with an invalid qid to the scheduler.
+	 * We should see the packed enqueued OK, but the inflights for
+	 * that packet should not be incremented, and the rx_dropped
+	 * should be incremented.
+	 */
+	static uint32_t flows1[] = {20};
+
+	for (i = 0; i < RTE_DIM(flows1); i++) {
+		struct rte_mbuf *arp = rte_gen_arp(0, t->mbuf_pool);
+		if (!arp) {
+			printf("%d: gen of pkt failed\n", __LINE__);
+			return -1;
+		}
+
+		struct rte_event ev = {
+				.op = RTE_EVENT_OP_NEW,
+				.queue_id = t->qid[0] + flows1[i],
+				.flow_id = i,
+				.mbuf = arp,
+		};
+		/* generate pkt and enqueue */
+		err = rte_event_enqueue_burst(evdev, t->port[rx_enq], &ev, 1);
+		if (err < 0) {
+			printf("%d: Failed to enqueue\n", __LINE__);
+			return -1;
+		}
+	}
+
+	/* call the scheduler */
+	rte_event_schedule(evdev);
+
+	err = test_event_dev_stats_get(evdev, &stats);
+	if (err) {
+		printf("%d: failed to get stats\n", __LINE__);
+		return -1;
+	}
+
+	/*
+	 * Now check the resulting inflights on the port, and the rx_dropped.
+	 */
+	if (stats.port_inflight[0] != 0) {
+		printf("%d:%s: port 1 inflight count not correct\n", __LINE__, __func__);
+		rte_event_dev_dump(evdev, stdout);
+		return -1;
+	}
+	if (stats.port_rx_dropped[0] != 1) {
+		printf("%d:%s: port 1 drops\n", __LINE__, __func__);
+		rte_event_dev_dump(evdev, stdout);
+		return -1;
+	}
+	/* each packet drop should only be counted in one place - port or dev */
+	if (stats.rx_dropped != 0) {
+		printf("%d:%s: port 1 dropped count not correct\n", __LINE__, __func__);
+		rte_event_dev_dump(evdev, stdout);
+		return -1;
+	}
+
+	cleanup(t);
+	return 0;
+}
+
+static int
+single_packet(struct test *t)
+{
+	const uint32_t MAGIC_SEQN = 7321;
+	struct rte_event ev;
+	struct test_event_dev_stats stats;
+	const int rx_enq = 0;
+	const int wrk_enq = 2;
+	int err;
+
+	/* Create qm instance with 4 ports */
+	if (init(t, 1, 4) < 0 ||
+			create_ports(t, 4) < 0 ||
+			create_atomic_qids(t, 1) < 0) {
+		printf("%d: Error initializing device\n", __LINE__);
+		return -1;
+	}
+
+	/* CQ mapping to QID */
+	err = rte_event_port_link(evdev, t->port[wrk_enq], NULL, NULL, 0);
+	if (err != 1) {
+		printf("%d: error mapping lb qid\n", __LINE__);
+		cleanup(t);
+		return -1;
+	}
+
+	if (rte_event_dev_start(evdev) < 0) {
+		printf("%d: Error with start call\n", __LINE__);
+		return -1;
+	}
+
+	/************** Gen pkt and enqueue ****************/
+	struct rte_mbuf *arp = rte_gen_arp(0, t->mbuf_pool);
+	if (!arp) {
+		printf("%d: gen of pkt failed\n", __LINE__);
+		return -1;
+	}
+
+	ev.op = RTE_EVENT_OP_NEW;
+	ev.priority = RTE_EVENT_DEV_PRIORITY_NORMAL;
+	ev.mbuf = arp;
+	ev.queue_id = 0;
+	ev.flow_id = 3;
+	arp->seqn = MAGIC_SEQN;
+
+	err = rte_event_enqueue_burst(evdev, t->port[rx_enq], &ev, 1);
+	if (err < 0) {
+		printf("%d: Failed to enqueue\n", __LINE__);
+		return -1;
+	}
+
+	rte_event_schedule(evdev);
+
+	err = test_event_dev_stats_get(evdev, &stats);
+	if (err) {
+		printf("%d: failed to get stats\n", __LINE__);
+		return -1;
+	}
+
+	if (stats.rx_pkts != 1 ||
+			stats.tx_pkts != 1 ||
+			stats.port_inflight[wrk_enq] != 1) {
+		printf("%d: Sched core didn't handle pkt as expected\n", __LINE__);
+		rte_event_dev_dump(evdev, stdout);
+		return -1;
+	}
+
+	uint32_t deq_pkts;
+
+	deq_pkts = rte_event_dequeue_burst(evdev, t->port[wrk_enq], &ev, 1, 0);
+	if (deq_pkts < 1) {
+		printf("%d: Failed to deq\n", __LINE__);
+		return -1;
+	}
+
+	err = test_event_dev_stats_get(evdev, &stats);
+	if (err) {
+		printf("%d: failed to get stats\n", __LINE__);
+		return -1;
+	}
+
+	err = test_event_dev_stats_get(evdev, &stats);
+	if (ev.mbuf->seqn != MAGIC_SEQN) {
+		printf("%d: magic sequence number not dequeued\n", __LINE__);
+		return -1;
+	}
+
+	rte_pktmbuf_free(ev.mbuf);
+	err = rte_event_enqueue_burst(evdev, t->port[wrk_enq], &release_ev, 1);
+	if (err < 0) {
+		printf("%d: Failed to enqueue\n", __LINE__);
+		return -1;
+	}
+	rte_event_schedule(evdev);
+
+	err = test_event_dev_stats_get(evdev, &stats);
+	if (stats.port_inflight[wrk_enq] != 0) {
+		printf("%d: port inflight not correct\n", __LINE__);
+		return -1;
+	}
+
+	cleanup(t);
+	return 0;
+}
+
+static int
+inflight_counts(struct test *t)
+{
+	struct rte_event ev;
+	struct test_event_dev_stats stats;
+	const int rx_enq = 0;
+	const int p1 = 1;
+	const int p2 = 2;
+	int err;
+	int i;
+
+	/* Create qm instance with 4 ports */
+	if (init(t, 2, 3) < 0 ||
+			create_ports(t, 3) < 0 ||
+			create_atomic_qids(t, 2) < 0) {
+		printf("%d: Error initializing device\n", __LINE__);
+		return -1;
+	}
+
+	/* CQ mapping to QID */
+	err = rte_event_port_link(evdev, t->port[p1], &t->qid[0], NULL, 1);
+	if (err != 1) {
+		printf("%d: error mapping lb qid\n", __LINE__);
+		cleanup(t);
+		return -1;
+	}
+	err = rte_event_port_link(evdev, t->port[p2], &t->qid[1], NULL, 1);
+	if (err != 1) {
+		printf("%d: error mapping lb qid\n", __LINE__);
+		cleanup(t);
+		return -1;
+	}
+
+	if (rte_event_dev_start(evdev) < 0) {
+		printf("%d: Error with start call\n", __LINE__);
+		return -1;
+	}
+
+	/************** FORWARD ****************/
+#define QID1_NUM 5
+	for (i = 0; i < QID1_NUM; i++) {
+		struct rte_mbuf *arp = rte_gen_arp(0, t->mbuf_pool);
+
+		if (!arp) {
+			printf("%d: gen of pkt failed\n", __LINE__);
+			goto err;
+		}
+
+		ev.queue_id =  t->qid[0];
+		ev.op = RTE_EVENT_OP_NEW;
+		ev.mbuf = arp;
+		err = rte_event_enqueue_burst(evdev, t->port[rx_enq], &ev, 1);
+		if (err != 1) {
+			printf("%d: Failed to enqueue\n", __LINE__);
+			goto err;
+		}
+	}
+#define QID2_NUM 3
+	for (i = 0; i < QID2_NUM; i++) {
+		struct rte_mbuf *arp = rte_gen_arp(0, t->mbuf_pool);
+
+		if (!arp) {
+			printf("%d: gen of pkt failed\n", __LINE__);
+			goto err;
+		}
+		ev.queue_id =  t->qid[1];
+		ev.op = RTE_EVENT_OP_NEW;
+		ev.mbuf = arp;
+		err = rte_event_enqueue_burst(evdev, t->port[rx_enq], &ev, 1);
+		if (err != 1) {
+			printf("%d: Failed to enqueue\n", __LINE__);
+			goto err;
+		}
+	}
+
+	/* schedule */
+	rte_event_schedule(evdev);
+
+	err = test_event_dev_stats_get(evdev, &stats);
+	if (err) {
+		printf("%d: failed to get stats\n", __LINE__);
+		goto err;
+	}
+
+	if (stats.rx_pkts != QID1_NUM + QID2_NUM ||
+			stats.tx_pkts != QID1_NUM + QID2_NUM) {
+		printf("%d: Sched core didn't handle pkt as expected\n", __LINE__);
+		goto err;
+	}
+
+	if (stats.port_inflight[p1] != QID1_NUM) {
+		printf("%d: %s port 1 inflight not correct\n", __LINE__, __func__);
+		goto err;
+	}
+	if (stats.port_inflight[p2] != QID2_NUM) {
+		printf("%d: %s port 2 inflight not correct\n", __LINE__, __func__);
+		goto err;
+	}
+
+	/************** DEQUEUE INFLIGHT COUNT CHECKS  ****************/
+	/* port 1 */
+	struct rte_event events[QID1_NUM + QID2_NUM];
+	uint32_t deq_pkts = rte_event_dequeue_burst(evdev, t->port[p1], events,
+			RTE_DIM(events), 0);
+
+	if (deq_pkts != QID1_NUM) {
+		printf("%d: Port 1: DEQUEUE inflight failed\n", __LINE__);
+		goto err;
+	}
+	err = test_event_dev_stats_get(evdev, &stats);
+	if (stats.port_inflight[p1] != QID1_NUM) {
+		printf("%d: port 1 inflight decrement after DEQ != 0\n", __LINE__);
+		goto err;
+	}
+	for (i = 0; i < QID1_NUM; i++) {
+		err = rte_event_enqueue_burst(evdev, t->port[p1], &release_ev, 1);
+
+		if (err != 1) {
+			printf("%d: %s rte enqueue of inf release failed\n",
+				__LINE__, __func__);
+			goto err;
+		}
+	}
+
+	/*
+	 * As the scheduler core decrements inflights, it needs to run to
+	 * process packets to act on the drop messages
+	 */
+	rte_event_schedule(evdev);
+
+	err = test_event_dev_stats_get(evdev, &stats);
+	if (stats.port_inflight[p1] != 0) {
+		printf("%d: port 1 inflight NON NULL after DROP\n", __LINE__);
+		goto err;
+	}
+
+	/* port2 */
+	deq_pkts = rte_event_dequeue_burst(evdev, t->port[p2], events,
+			RTE_DIM(events), 0);
+	if (deq_pkts != QID2_NUM) {
+		printf("%d: Port 2: DEQUEUE inflight failed\n", __LINE__);
+		goto err;
+	}
+	err = test_event_dev_stats_get(evdev, &stats);
+	if (stats.port_inflight[p2] != QID2_NUM) {
+		printf("%d: port 1 inflight decrement after DEQ != 0\n", __LINE__);
+		goto err;
+	}
+	for (i = 0; i < QID2_NUM; i++) {
+		err = rte_event_enqueue_burst(evdev, t->port[p2], &release_ev, 1);
+
+		if (err != 1) {
+			printf("%d: %s rte enqueue of inf release failed\n",
+				__LINE__, __func__);
+			goto err;
+		}
+	}
+
+	/*
+	 * As the scheduler core decrements inflights, it needs to run to
+	 * process packets to act on the drop messages
+	 */
+	rte_event_schedule(evdev);
+
+	err = test_event_dev_stats_get(evdev, &stats);
+	if (stats.port_inflight[p2] != 0) {
+		printf("%d: port 2 inflight NON NULL after DROP\n", __LINE__);
+		goto err;
+	}
+	cleanup(t);
+	return 0;
+
+err:
+	rte_event_dev_dump(evdev, stdout);
+	cleanup(t);
+	return -1;
+}
+
+static int
+parallel_basic(struct test *t, int check_order)
+{
+	const uint8_t rx_port = 0;
+	const uint8_t w1_port = 1;
+	const uint8_t w3_port = 3;
+	const uint8_t tx_port = 4;
+	int err;
+	int i;
+	uint32_t deq_pkts, j;
+	struct rte_mbuf *mbufs[3];
+	struct rte_mbuf *mbufs_out[3];
+	const uint32_t MAGIC_SEQN = 1234;
+
+	/* Create qm instance with 4 ports */
+	if (init(t, 2, tx_port + 1) < 0 ||
+			create_ports(t, tx_port + 1) < 0 ||
+			(check_order ?  create_ordered_qids(t, 1) :
+				create_unordered_qids(t, 1)) < 0 ||
+			create_directed_qids(t, 1, &tx_port)) {
+		printf("%d: Error initializing device\n", __LINE__);
+		return -1;
+	}
+
+	/*
+	 * CQ mapping to QID
+	 * We need three ports, all mapped to the same ordered qid0. Then we'll
+	 * take a packet out to each port, re-enqueue in reverse order,
+	 * then make sure the reordering has taken place properly when we
+	 * dequeue from the tx_port.
+	 *
+	 * Simplified test setup diagram:
+	 *
+	 * rx_port        w1_port
+	 *        \     /         \
+	 *         qid0 - w2_port - qid1
+	 *              \         /     \
+	 *                w3_port        tx_port
+	 */
+	/* CQ mapping to QID for load balanced ports (directed mapped on create) */
+	for (i = w1_port; i <= w3_port; i++) {
+		err = rte_event_port_link(evdev, t->port[i], &t->qid[0], NULL, 1);
+		if (err != 1) {
+			printf("%d: error mapping lb qid\n", __LINE__);
+			cleanup(t);
+			return -1;
+		}
+	}
+
+	if (rte_event_dev_start(evdev) < 0) {
+		printf("%d: Error with start call\n", __LINE__);
+		return -1;
+	}
+
+	/* Enqueue 3 packets to the rx port */
+	for (i = 0; i < 3; i++) {
+		struct rte_event ev;
+		mbufs[i] = rte_gen_arp(0, t->mbuf_pool);
+		if (!mbufs[i]) {
+			printf("%d: gen of pkt failed\n", __LINE__);
+			return -1;
+		}
+
+		ev.queue_id = t->qid[0];
+		ev.op = RTE_EVENT_OP_NEW;
+		ev.mbuf = mbufs[i];
+		mbufs[i]->seqn = MAGIC_SEQN + i;
+
+		/* generate pkt and enqueue */
+		err = rte_event_enqueue_burst(evdev, t->port[rx_port], &ev, 1);
+		if (err != 1) {
+			printf("%d: Failed to enqueue pkt %u, retval = %u\n",
+					__LINE__, i, err);
+			return -1;
+		}
+	}
+
+	rte_event_schedule(evdev);
+
+	/* use extra slot to make logic in loops easier */
+	struct rte_event deq_ev[w3_port + 1];
+
+	/* Dequeue the 3 packets, one from each worker port */
+	for (i = w1_port; i <= w3_port; i++) {
+		deq_pkts = rte_event_dequeue_burst(evdev, t->port[i], &deq_ev[i],
+				1, 0);
+		if (deq_pkts != 1) {
+			printf("%d: Failed to deq\n", __LINE__);
+			rte_event_dev_dump(evdev, stdout);
+			return -1;
+		}
+	}
+
+	/* Enqueue each packet in reverse order, flushing after each one */
+	for (i = w3_port; i >= w1_port; i--) {
+
+		deq_ev[i].op = RTE_EVENT_OP_FORWARD;
+		deq_ev[i].queue_id = t->qid[1];
+		err = rte_event_enqueue_burst(evdev, t->port[i], &deq_ev[i], 1);
+		if (err != 1) {
+			printf("%d: Failed to enqueue\n", __LINE__);
+			return -1;
+		}
+	}
+	rte_event_schedule(evdev);
+
+	/* dequeue from the tx ports, we should get 3 packets */
+	deq_pkts = rte_event_dequeue_burst(evdev, t->port[tx_port], deq_ev, 3, 0);
+
+	/* Check to see if we've got all 3 packets */
+	if (deq_pkts != 3) {
+		printf("%d: expected 3 packets at tx port got %d from port %d\n",
+			__LINE__, deq_pkts, tx_port);
+		rte_event_dev_dump(evdev, stdout);
+		return 1;
+	}
+
+	/* Check to see if the sequence numbers are in expected order */
+	if (check_order) {
+		for (j = 0 ; j < deq_pkts ; j++) {
+			if (deq_ev[j].mbuf->seqn != MAGIC_SEQN + j) {
+				printf("%d: Incorrect sequence number(%d) from port %d\n",
+					__LINE__, mbufs_out[j]->seqn, tx_port);
+				return -1;
+			}
+		}
+	}
+
+	/* Destroy the qm instance */
+	cleanup(t);
+	return 0;
+}
+
+static int
+ordered_basic(struct test *t)
+{
+	return parallel_basic(t, 1);
+}
+
+static int
+unordered_basic(struct test *t)
+{
+	return parallel_basic(t, 0);
+}
+
+static int
+holb(struct test *t) /* test to check we avoid basic head-of-line blocking */
+{
+	const struct rte_event new_ev = {
+			.op = RTE_EVENT_OP_NEW
+			/* all other fields zero */
+	};
+	struct rte_event ev = new_ev;
+	unsigned int rx_port = 0; /* port we get the first flow on */
+	char rx_port_used_stat[64], rx_port_free_stat[64], other_port_used_stat[64];
+
+	if (init(t, 1, 2) < 0 ||
+			create_ports(t, 2) < 0 ||
+			create_atomic_qids(t, 1) < 0) {
+		printf("%d: Error initializing device\n", __LINE__);
+		return -1;
+	}
+	if (rte_event_port_link(evdev, t->port[0], NULL, NULL, 0) != 1 ||
+			rte_event_port_link(evdev, t->port[1], NULL, NULL, 0) != 1) {
+		printf("%d: Error links queue to ports\n", __LINE__);
+		goto err;
+	}
+	if (rte_event_dev_start(evdev) < 0) {
+		printf("%d: Error with start call\n", __LINE__);
+		goto err;
+	}
+
+	/* send one packet and see where it goes, port 0 or 1 */
+	if (rte_event_enqueue_burst(evdev, t->port[0], &ev, 1) != 1) {
+		printf("%d: Error doing first enqueue\n", __LINE__);
+		goto err;
+	}
+	rte_event_schedule(evdev);
+
+	if (rte_event_dev_get_xstat_by_name(evdev, "port_0_cq_ring_used", NULL) != 1)
+		rx_port = 1;
+	snprintf(rx_port_used_stat, sizeof(rx_port_used_stat),
+			"port_%u_cq_ring_used", rx_port);
+	snprintf(rx_port_free_stat, sizeof(rx_port_free_stat),
+			"port_%u_cq_ring_free", rx_port);
+	snprintf(other_port_used_stat, sizeof(other_port_used_stat),
+			"port_%u_cq_ring_used", rx_port ^ 1);
+	if (rte_event_dev_get_xstat_by_name(evdev, rx_port_used_stat, NULL) != 1) {
+		printf("%d: Error, first event not scheduled\n", __LINE__);
+		goto err;
+	}
+
+	/* now fill up the rx port's queue with one flow to cause HOLB */
+	do {
+		ev = new_ev;
+		if (rte_event_enqueue_burst(evdev, t->port[0], &ev, 1) != 1) {
+			printf("%d: Error with enqueue\n", __LINE__);
+			goto err;
+		}
+		rte_event_schedule(evdev);
+	} while (rte_event_dev_get_xstat_by_name(evdev, rx_port_free_stat, NULL) != 0);
+
+	/* one more packet, which needs to stay in IQ - i.e. HOLB */
+	ev = new_ev;
+	if (rte_event_enqueue_burst(evdev, t->port[0], &ev, 1) != 1) {
+		printf("%d: Error with enqueue\n", __LINE__);
+		goto err;
+	}
+	rte_event_schedule(evdev);
+
+	/* check that the other port still has an empty CQ */
+	if (rte_event_dev_get_xstat_by_name(evdev, other_port_used_stat, NULL) != 0) {
+		printf("%d: Error, second port CQ is not empty\n", __LINE__);
+		goto err;
+	}
+	/* check IQ now has one packet */
+	if (rte_event_dev_get_xstat_by_name(evdev, "qid_0_iq_0_used", NULL) != 1) {
+		printf("%d: Error, QID does not have exactly 1 packet\n", __LINE__);
+		goto err;
+	}
+
+	/* send another flow, which should pass the other IQ entry */
+	ev = new_ev;
+	ev.flow_id = 1;
+	if (rte_event_enqueue_burst(evdev, t->port[0], &ev, 1) != 1) {
+		printf("%d: Error with enqueue\n", __LINE__);
+		goto err;
+	}
+	rte_event_schedule(evdev);
+
+	if (rte_event_dev_get_xstat_by_name(evdev, other_port_used_stat, NULL) != 1) {
+		printf("%d: Error, second flow did not pass out first\n", __LINE__);
+		goto err;
+	}
+
+	if (rte_event_dev_get_xstat_by_name(evdev, "qid_0_iq_0_used", NULL) != 1) {
+		printf("%d: Error, QID does not have exactly 1 packet\n", __LINE__);
+		goto err;
+	}
+	cleanup(t);
+	return 0;
+err:
+	rte_event_dev_dump(evdev, stdout);
+	cleanup(t);
+	return -1;
+}
+
+static int
+worker_loopback_worker_fn(void *arg)
+{
+	struct test *t = arg;
+	uint8_t port = t->port[1];
+	int count = 0;
+	int enqd;
+
+	/*
+	 * Takes packets from the input port and then loops them back through
+	 * the Queue Manager. Each packet gets looped through QIDs 0-8, 16 times,
+	 * so each packet goes through 8*16 = 128 times.
+	 */
+	printf("%d: \tWorker function started\n", __LINE__);
+	while (count < NUM_PACKETS) {
+#define BURST_SIZE 32
+		struct rte_event ev[BURST_SIZE];
+		uint16_t i, nb_rx = rte_event_dequeue_burst(evdev, port, ev, BURST_SIZE, 0);
+		if (nb_rx == 0) {
+			rte_pause();
+			continue;
+		}
+
+		for (i = 0; i < nb_rx; i++) {
+			ev[i].queue_id++;
+			if (ev[i].queue_id != 8) {
+				ev[i].op = RTE_EVENT_OP_FORWARD;
+				enqd = rte_event_enqueue_burst(evdev, port, &ev[i], 1);
+				if (enqd != 1) {
+					printf("%d: Can't enqueue FWD!!\n", __LINE__);
+					return -1;
+				}
+				continue;
+			}
+
+			ev[i].queue_id = 0;
+			ev[i].mbuf->udata64++;
+			if (ev[i].mbuf->udata64 != 16) {
+				ev[i].op = RTE_EVENT_OP_FORWARD;
+				enqd = rte_event_enqueue_burst(evdev, port, &ev[i], 1);
+				if (enqd != 1) {
+					printf("%d: Can't enqueue FWD!!\n", __LINE__);
+					return -1;
+				}
+				continue;
+			}
+			/* we have hit 16 iterations through system - drop */
+			rte_pktmbuf_free(ev[i].mbuf);
+			count++;
+			ev[i].op = RTE_EVENT_OP_RELEASE;
+			enqd = rte_event_enqueue_burst(evdev, port, &ev[i], 1);
+			if (enqd != 1) {
+				printf("%d drop enqueue failed\n", __LINE__);
+				return -1;
+			}
+		}
+	}
+
+	return 0;
+}
+
+static int
+worker_loopback_producer_fn(void *arg)
+{
+	struct test *t = arg;
+	uint8_t port = t->port[0];
+	uint64_t count = 0;
+
+	printf("%d: \tProducer function started\n", __LINE__);
+	while (count < NUM_PACKETS) {
+		struct rte_mbuf *m = 0;
+		do {
+			m = rte_pktmbuf_alloc(t->mbuf_pool);
+		} while (m == NULL);
+
+		m->udata64 = 0;
+
+		struct rte_event ev = {
+				.op = RTE_EVENT_OP_NEW,
+				.queue_id = t->qid[0],
+				.flow_id = (uintptr_t)m & 0xFFFF,
+				.mbuf = m,
+		};
+
+		if (rte_event_enqueue_burst(evdev, port, &ev, 1) != 1) {
+			while (rte_event_enqueue_burst(evdev, port, &ev, 1) != 1)
+				rte_pause();
+		}
+
+		count++;
+	}
+
+	return 0;
+}
+
+static int
+worker_loopback(struct test *t)
+{
+	/* use a single producer core, and a worker core to see what happens
+	 * if the worker loops packets back multiple times
+	 */
+	struct test_event_dev_stats stats;
+	uint64_t print_cycles = 0, cycles = 0;
+	uint64_t tx_pkts = 0;
+	int err;
+	int w_lcore, p_lcore;
+
+	if (init(t, 8, 2) < 0 ||
+			create_atomic_qids(t, 8) < 0) {
+		printf("%d: Error initializing device\n", __LINE__);
+		return -1;
+	}
+
+	/* RX with low max events */
+	static struct rte_event_port_conf conf = {
+			.new_event_threshold = 512,
+			.dequeue_depth = 32,
+			.enqueue_depth = 64,
+	};
+	if (rte_event_port_setup(evdev, 0, &conf) < 0) {
+		printf("Error setting up RX port\n");
+		return -1;
+	}
+	t->port[0] = 0;
+	/* TX with higher max events */
+	conf.new_event_threshold = 4096;
+	if (rte_event_port_setup(evdev, 1, &conf) < 0) {
+		printf("Error setting up TX port\n");
+		return -1;
+	}
+	t->port[1] = 1;
+
+	/* CQ mapping to QID */
+	err = rte_event_port_link(evdev, t->port[1], NULL, NULL, 0);
+	if (err != 8) { /* should have mapped all queues*/
+		printf("%d: error mapping port 2 to all qids\n", __LINE__);
+		return -1;
+	}
+
+	if (rte_event_dev_start(evdev) < 0) {
+		printf("%d: Error with start call\n", __LINE__);
+		return -1;
+	}
+
+	p_lcore = rte_get_next_lcore(
+			/* start core */ -1,
+			/* skip master */ 1,
+			/* wrap */ 0);
+	w_lcore = rte_get_next_lcore(p_lcore, 1, 0);
+
+	rte_eal_remote_launch(worker_loopback_producer_fn, t, p_lcore);
+	rte_eal_remote_launch(worker_loopback_worker_fn, t, w_lcore);
+
+	print_cycles = cycles = rte_get_timer_cycles();
+	while (rte_eal_get_lcore_state(p_lcore) != FINISHED ||
+			rte_eal_get_lcore_state(w_lcore) != FINISHED) {
+
+		rte_event_schedule(evdev);
+
+		uint64_t new_cycles = rte_get_timer_cycles();
+
+		if (new_cycles - print_cycles > rte_get_timer_hz()) {
+			test_event_dev_stats_get(evdev, &stats);
+			printf("%d: \tSched Rx = %" PRIu64 ", Tx = %" PRIu64 "\n",
+					__LINE__, stats.rx_pkts, stats.tx_pkts);
+
+			print_cycles = new_cycles;
+		}
+		if (new_cycles - cycles > rte_get_timer_hz() * 3) {
+			test_event_dev_stats_get(evdev, &stats);
+			if (stats.tx_pkts == tx_pkts) {
+				rte_event_dev_dump(evdev, stdout);
+				printf("%d: No schedules for seconds, deadlock\n", __LINE__);
+				return -1;
+			}
+			tx_pkts = stats.tx_pkts;
+			cycles = new_cycles;
+		}
+	}
+	rte_event_schedule(evdev); /* ensure all completions are flushed */
+
+	rte_eal_mp_wait_lcore();
+
+	cleanup(t);
+	return 0;
+}
+
+static struct rte_mempool *eventdev_func_mempool;
+
+static int
+test_sw_eventdev(void)
+{
+	struct test *t = malloc(sizeof(struct test));
+	int ret;
+
+	const char *eventdev_name = "event_sw0";
+	evdev = rte_event_dev_get_dev_id(eventdev_name);
+	if (evdev < 0) {
+		printf("%d: Eventdev %s not found - creating.\n",
+				__LINE__, eventdev_name);
+		if (rte_eal_vdev_init(eventdev_name, NULL) < 0) {
+			printf("Error creating eventdev\n");
+			return -1;
+		}
+		evdev = rte_event_dev_get_dev_id(eventdev_name);
+		if (evdev < 0) {
+			printf("Error finding newly created eventdev\n");
+			return -1;
+		}
+	}
+
+	/* Only create mbuf pool once, reuse for each test run */
+	if (!eventdev_func_mempool) {
+		eventdev_func_mempool = rte_pktmbuf_pool_create("QM_SA_MBUF_POOL",
+				(1<<12), /* 4k buffers */
+				32 /*MBUF_CACHE_SIZE*/,
+				0,
+				512, /* use very small mbufs */
+				rte_socket_id());
+		if (!eventdev_func_mempool) {
+			printf("ERROR creating mempool\n");
+			return -1;
+		}
+	}
+	t->mbuf_pool = eventdev_func_mempool;
+
+	printf("*** Running Single Directed Packet test...\n");
+	ret = test_single_directed_packet(t);
+	if (ret != 0) {
+		printf("ERROR - Single Directed Packet test FAILED.\n");
+		return ret;
+	}
+	printf("*** Running Single Load Balanced Packet test...\n");
+	ret = single_packet(t);
+	if (ret != 0) {
+		printf("ERROR - Single Packet test FAILED.\n");
+		return ret;
+	}
+	printf("*** Running Unordered Basic test...\n");
+	ret = unordered_basic(t);
+	if (ret != 0) {
+		printf("ERROR -  Unordered Basic test FAILED.\n");
+		return ret;
+	}
+	printf("*** Running Ordered Basic test...\n");
+	ret = ordered_basic(t);
+	if (ret != 0) {
+		printf("ERROR -  Ordered Basic test FAILED.\n");
+		return ret;
+	}
+	printf("*** Running Burst Packets test...\n");
+	ret = burst_packets(t);
+	if (ret != 0) {
+		printf("ERROR - Burst Packets test FAILED.\n");
+		return ret;
+	}
+	printf("*** Running Load Balancing test...\n");
+	ret = load_balancing(t);
+	if (ret != 0) {
+		printf("ERROR - Load Balancing test FAILED.\n");
+		return ret;
+	}
+	printf("*** Running Prioritized Directed test...\n");
+	ret = test_priority_directed(t);
+	if (ret != 0) {
+		printf("ERROR - Prioritized Directed test FAILED.\n");
+		return ret;
+	}
+	printf("*** Running Prioritized Atomic test...\n");
+	ret = test_priority_atomic(t);
+	if (ret != 0) {
+		printf("ERROR - Prioritized Atomic test FAILED.\n");
+		return ret;
+	}
+
+	printf("*** Running Prioritized Ordered test...\n");
+	ret = test_priority_ordered(t);
+	if (ret != 0) {
+		printf("ERROR - Prioritized Ordered test FAILED.\n");
+		return ret;
+	}
+	printf("*** Running Prioritized Unordered test...\n");
+	ret = test_priority_unordered(t);
+	if (ret != 0) {
+		printf("ERROR - Prioritized Unordered test FAILED.\n");
+		return ret;
+	}
+	printf("*** Running Invalid QID test...\n");
+	ret = invalid_qid(t);
+	if (ret != 0) {
+		printf("ERROR - Invalid QID test FAILED.\n");
+		return ret;
+	}
+	printf("*** Running Load Balancing History test...\n");
+	ret = load_balancing_history(t);
+	if (ret != 0) {
+		printf("ERROR - Load Balancing History test FAILED.\n");
+		return ret;
+	}
+	printf("*** Running Inflight Count test...\n");
+	ret = inflight_counts(t);
+	if (ret != 0) {
+		printf("ERROR - Inflight Count test FAILED.\n");
+		return ret;
+	}
+	printf("*** Running Abuse Inflights test...\n");
+	ret = abuse_inflights(t);
+	if (ret != 0) {
+		printf("ERROR - Abuse Inflights test FAILED.\n");
+		return ret;
+	}
+	printf("*** Running QID Priority test...\n");
+	ret = qid_priorities(t);
+	if (ret != 0) {
+		printf("ERROR - QID Priority test FAILED.\n");
+		return ret;
+	}
+	printf("*** Running Head-of-line-blocking test...\n");
+	ret = holb(t);
+	if (ret != 0) {
+		printf("ERROR - Head-of-line-blocking test FAILED.\n");
+		return ret;
+	}
+	if (rte_lcore_count() >= 3) {
+		printf("*** Running Worker loopback test...\n");
+		ret = worker_loopback(t);
+		if (ret != 0) {
+			printf("ERROR - Worker loopback test FAILED.\n");
+			return ret;
+		}
+	} else {
+		printf("### Not enough cores for worker loopback test.\n");
+		printf("### Need at least 3 cores for test.\n");
+	}
+	/*
+	 * Free test instance, leaving mempool initialized, and a pointer to it
+	 * in the static eventdev_func_mempool variable. It is re-used on re-runs
+	 */
+	free(t);
+
+	return 0;
+}
+
+REGISTER_TEST_COMMAND(eventdev_sw_autotest, test_sw_eventdev);
-- 
2.7.4

Re: [dpdk-dev] [PATCH 00/15] next-eventdev: event/sw Software Eventdev

[Jerin Jacob]

On Mon, Jan 16, 2017 at 03:40:40PM +0000, Harry van Haaren wrote:
> The following patchset adds software eventdev implementation
> to the next-eventdev tree, and applies to current git HEAD.
> 
> This implementation is based on the previous software eventdev
> RFC patchset[1], updated to integrate with the latest rte_eventdev.h
> API.
> 
> The first two patches make changes to the eventdev API,
> then the software implementation is added, and finally
> tests are added for the sw eventdev implementation.
> 
> This patchset contains the work of multiple developers,
> please see signoffs on each patch.
> 
> Signed-off-by: Harry van Haaren <harry.van.haaren@intel.com>
> 
> [1] http://dpdk.org/ml/archives/dev/2016-November/050285.html
>

A few git-am, check-git-log.sh and checkpatches.sh errors in the patch
series.

1) [dpdk-next-eventdev-review] $ pwclient git-am 19436
Applying patch #19436 using 'git am'
Description: [dpdk-dev,03/15] event/sw: add new software-only eventdev
driver
Applying: event/sw: add new software-only eventdev driver
/export/dpdk-next-eventdev-review/.git/rebase-apply/patch:114: new blank
line at EOF.
+
warning: 1 line adds whitespace errors.

2) [dpdk-next-eventdev-review] $ ./devtools/check-git-log.sh
Wrong headline format:
	event/sw: add start, stop and close functions

3) [dpdk-next-eventdev-review] $ ./devtools/checkpatches.sh *.patch

### [PATCH 03/12] event/sw: add support for event queues

ERROR:COMPLEX_MACRO: Macros with complex values should be enclosed in parentheses
#83: FILE: drivers/event/sw/iq_ring.h:60:
+#define force_inline inline __attribute__((always_inline))

ERROR:SPACING: spaces required around that '>' (ctx:OxW)
#163: FILE: drivers/event/sw/iq_ring.h:140:
+	for (i = nb_qes; i --> 0; )
 	                     ^

WARNING:LONG_LINE_STRING: line over 80 characters
#257: FILE: drivers/event/sw/sw_evdev.c:90:
+			SW_LOG_DBG("invalid reorder_window_size for ordered queue\n");

WARNING:LONG_LINE_COMMENT: line over 80 characters
#454: FILE: drivers/event/sw/sw_evdev.h:116:
+	struct reorder_buffer_entry *reorder_buffer; /* packets awaiting reordering */

total: 2 errors, 2 warnings, 437 lines checked

### [PATCH 04/12] event/sw: add support for event ports

ERROR:COMPLEX_MACRO: Macros with complex values should be enclosed in parentheses
#89: FILE: drivers/event/sw/event_ring.h:65:
+#define force_inline inline __attribute__((always_inline))

WARNING:LONG_LINE: line over 80 characters
#235: FILE: drivers/event/sw/sw_evdev.c:65:
+	uint8_t enq_oversize = conf->enqueue_depth > info.max_event_port_enqueue_depth;

WARNING:LONG_LINE: line over 80 characters
#236: FILE: drivers/event/sw/sw_evdev.c:66:
+	uint8_t deq_oversize = conf->dequeue_depth > info.max_event_port_dequeue_depth;

WARNING:LONG_LINE_STRING: line over 80 characters
#244: FILE: drivers/event/sw/sw_evdev.c:74:
+	snprintf(buf, sizeof(buf), "sw%d_%s", dev->data->dev_id, "rx_worker_ring");

WARNING:LONG_LINE_STRING: line over 80 characters
#255: FILE: drivers/event/sw/sw_evdev.c:85:
+	snprintf(buf, sizeof(buf), "sw%d_%s", dev->data->dev_id, "cq_worker_ring");

WARNING:LONG_LINE_COMMENT: line over 80 characters
#309: FILE: drivers/event/sw/sw_evdev.h:53:
+#define SCHED_DEQUEUE_BURST_SIZE 32 /* how many packets pulled from port by sched */

WARNING:LONG_LINE_COMMENT: line over 80 characters
#360: FILE: drivers/event/sw/sw_evdev.h:169:
+	uint64_t avg_pkt_ticks;      /* tracks average over NUM_SAMPLES bursts */

total: 1 errors, 6 warnings, 368 lines checked

### [PATCH 05/12] event/sw: add support for linking queues to ports

WARNING:LONG_LINE: line over 80 characters
#74: FILE: drivers/event/sw/sw_evdev.c:108:
+				q->cq_map[j] = q->cq_map[q->cq_num_mapped_cqs - 1];

total: 0 errors, 1 warnings, 79 lines checked

### [PATCH 06/12] event/sw: add worker core functions

WARNING:LONG_LINE_COMMENT: line over 80 characters
#73: FILE: drivers/event/sw/sw_evdev.h:67:
+#define QE_FLAG_VALID    (1 << QE_FLAG_VALID_SHIFT)  /* set for NEW, FWD, FRAG */

WARNING:LONG_LINE: line over 80 characters
#191: FILE: drivers/event/sw/sw_evdev_worker.c:80:
+		int failed_to_get = rte_atomic32_dec_and_test(&sw->inflight_quanta);

WARNING:LONG_LINE: line over 80 characters
#217: FILE: drivers/event/sw/sw_evdev_worker.c:106:
+		uint64_t burst_pkt_ticks = burst_ticks / p->last_dequeue_burst_sz;

total: 0 errors, 3 warnings, 233 lines checked

### [PATCH 07/12] event/sw: add scheduling logic

WARNING:LONG_LINE_COMMENT: line over 80 characters
#156: FILE: drivers/event/sw/sw_evdev_scheduler.c:76:
+		/* use cheap bit mixing, since we only need to lose a few bits */

WARNING:LONG_LINE: line over 80 characters
#170: FILE: drivers/event/sw/sw_evdev_scheduler.c:90:
+			for (cq_idx = 0; cq_idx < qid->cq_num_mapped_cqs; cq_idx++) {

WARNING:LONG_LINE: line over 80 characters
#174: FILE: drivers/event/sw/sw_evdev_scheduler.c:94:
+					cq = test_cq, cq_free_cnt = test_cq_free;

WARNING:LONG_LINE: line over 80 characters
#204: FILE: drivers/event/sw/sw_evdev_scheduler.c:124:
+			qe_ring_enqueue_burst(worker, p->cq_buf, p->cq_buf_count,

WARNING:LONG_LINE: line over 80 characters
#231: FILE: drivers/event/sw/sw_evdev_scheduler.c:151:
+		count = RTE_MIN(count, rte_ring_count(qid->reorder_buffer_freelist));

WARNING:LONG_LINE: line over 80 characters
#249: FILE: drivers/event/sw/sw_evdev_scheduler.c:169:
+		} while (qe_ring_free_count(sw->ports[cq].cq_worker_ring) == 0 ||

WARNING:LONG_LINE: line over 80 characters
#342: FILE: drivers/event/sw/sw_evdev_scheduler.c:262:
+						(type == RTE_SCHED_TYPE_ORDERED));

WARNING:LONG_LINE: line over 80 characters
#391: FILE: drivers/event/sw/sw_evdev_scheduler.c:311:
+				qe = &entry->fragments[entry->fragment_index + j];

WARNING:LONG_LINE: line over 80 characters
#401: FILE: drivers/event/sw/sw_evdev_scheduler.c:321:
+				struct sw_qid *dest_qid_ptr = &sw->qids[dest_qid];

WARNING:LONG_LINE: line over 80 characters
#402: FILE: drivers/event/sw/sw_evdev_scheduler.c:322:
+				const struct iq_ring *dest_iq_ptr = dest_qid_ptr->iq[dest_iq];

WARNING:LONG_LINE: line over 80 characters
#427: FILE: drivers/event/sw/sw_evdev_scheduler.c:347:
+				rte_ring_sp_enqueue(qid->reorder_buffer_freelist,

WARNING:LONG_LINE: line over 80 characters
#495: FILE: drivers/event/sw/sw_evdev_scheduler.c:415:
+			struct sw_fid_t *fid = &sw->qids[hist_qid].fids[hist_fid];

WARNING:LONG_LINE: line over 80 characters
#502: FILE: drivers/event/sw/sw_evdev_scheduler.c:422:
+				uintptr_t rob_ptr = (uintptr_t)hist_entry->rob_entry;

WARNING:LONG_LINE: line over 80 characters
#505: FILE: drivers/event/sw/sw_evdev_scheduler.c:425:
+				rob_ptr |= ((valid - 1) & (uintptr_t)&dummy_rob);

WARNING:LONG_LINE: line over 80 characters
#506: FILE: drivers/event/sw/sw_evdev_scheduler.c:426:
+				((struct reorder_buffer_entry *)rob_ptr)->ready =

WARNING:LONG_LINE_COMMENT: line over 80 characters
#520: FILE: drivers/event/sw/sw_evdev_scheduler.c:440:
+				/* Although fragmentation not currently supported

WARNING:LONG_LINE: line over 80 characters
#525: FILE: drivers/event/sw/sw_evdev_scheduler.c:445:
+				if (rob_entry->num_fragments == SW_FRAGMENTS_MAX)

WARNING:LONG_LINE: line over 80 characters
#528: FILE: drivers/event/sw/sw_evdev_scheduler.c:448:
+					rob_entry->fragments[rob_entry->num_fragments++] = *qe;

WARNING:LONG_LINE_COMMENT: line over 80 characters
#625: FILE: drivers/event/sw/sw_evdev_scheduler.c:545:
+				/* TODO: use a function pointer in the port itself */

WARNING:LONG_LINE: line over 80 characters
#627: FILE: drivers/event/sw/sw_evdev_scheduler.c:547:
+					in_pkts += sw_schedule_pull_port_dir(sw, i);

WARNING:LONG_LINE: line over 80 characters
#629: FILE: drivers/event/sw/sw_evdev_scheduler.c:549:
+					in_pkts += sw_schedule_pull_port_lb(sw, i);

WARNING:LONG_LINE: line over 80 characters
#631: FILE: drivers/event/sw/sw_evdev_scheduler.c:551:
+					in_pkts += sw_schedule_pull_port_no_reorder(sw, i);

total: 0 errors, 22 warnings, 623 lines checked

### [PATCH 09/12] event/sw: add dump function for easier debugging

WARNING:LONG_LINE: line over 80 characters
#39: FILE: drivers/event/sw/sw_evdev.c:422:
+	uint32_t credits = sw->nb_events_limit - (quanta * SW_INFLIGHT_QUANTA_SIZE);

WARNING:LONG_LINE_STRING: line over 80 characters
#50: FILE: drivers/event/sw/sw_evdev.c:433:
+		fprintf(f, "\trx   %"PRIu64"\tdrop %"PRIu64"\ttx   %"PRIu64"\tinflight %d\n",

WARNING:LONG_LINE: line over 80 characters
#56: FILE: drivers/event/sw/sw_evdev.c:439:
+			sw->ports[i].avg_pkt_ticks, sw->ports[i].inflight_credits);

ERROR:SPACING: spaces required around that '>' (ctx:OxW)
#61: FILE: drivers/event/sw/sw_evdev.c:444:
+		for (max = (int)RTE_DIM(p->poll_buckets); max --> 0;)
 		                                                ^

WARNING:LONG_LINE: line over 80 characters
#67: FILE: drivers/event/sw/sw_evdev.c:450:
+						((j << SW_DEQ_STAT_BUCKET_SHIFT) + 1),

WARNING:LONG_LINE: line over 80 characters
#68: FILE: drivers/event/sw/sw_evdev.c:451:
+						((j+1) << SW_DEQ_STAT_BUCKET_SHIFT),

WARNING:LONG_LINE: line over 80 characters
#69: FILE: drivers/event/sw/sw_evdev.c:452:
+						p->poll_buckets[j] * 100.0 / p->total_polls);

WARNING:LONG_LINE: line over 80 characters
#75: FILE: drivers/event/sw/sw_evdev.c:458:
+		fprintf(f, "\t%srx ring used: %4"PRIu64"\tfree: %4"PRIu64 COL_RESET"\n",

WARNING:LONG_LINE: line over 80 characters
#81: FILE: drivers/event/sw/sw_evdev.c:464:
+		fprintf(f, "\t%scq ring used: %4"PRIu64"\tfree: %4"PRIu64 COL_RESET"\n",

WARNING:LONG_LINE: line over 80 characters
#96: FILE: drivers/event/sw/sw_evdev.c:479:
+				rte_ring_free_count(qid->reorder_buffer_freelist));

WARNING:LONG_LINE: line over 80 characters
#106: FILE: drivers/event/sw/sw_evdev.c:489:
+		fprintf(f, "\tInflights: %u\tFlows pinned per port: ", inflights);

WARNING:LONG_LINE: line over 80 characters
#118: FILE: drivers/event/sw/sw_evdev.c:501:
+				fprintf(f, "\t%siq %d: Used %d\tFree %d"COL_RESET"\n",

total: 1 errors, 11 warnings, 121 lines checked

### [PATCH 10/12] event/sw: add xstats support

WARNING:LONG_LINE: line over 80 characters
#340: FILE: drivers/event/sw/sw_evdev_xstats.c:243:
+	RTE_BUILD_BUG_ON(RTE_DIM(port_bucket_stats) != RTE_DIM(port_bucket_types));

WARNING:LONG_LINE: line over 80 characters
#349: FILE: drivers/event/sw/sw_evdev_xstats.c:252:
+				((MAX_SW_CONS_Q_DEPTH >> SW_DEQ_STAT_BUCKET_SHIFT) + 1) +

WARNING:LONG_LINE: line over 80 characters
#352: FILE: drivers/event/sw/sw_evdev_xstats.c:255:
+			sw->qid_count * sw->port_count * RTE_DIM(qid_port_stats);

WARNING:LONG_LINE: line over 80 characters
#382: FILE: drivers/event/sw/sw_evdev_xstats.c:285:
+			for (i = 0; i < RTE_DIM(port_bucket_stats); i++, stat++) {

WARNING:LONG_LINE_STRING: line over 80 characters
#389: FILE: drivers/event/sw/sw_evdev_xstats.c:292:
+				snprintf(sname, sizeof(sname), "port_%u_%s_%u-%u",

WARNING:LONG_LINE: line over 80 characters
#391: FILE: drivers/event/sw/sw_evdev_xstats.c:294:
+						(bkt << SW_DEQ_STAT_BUCKET_SHIFT) + 1,

WARNING:LONG_LINE: line over 80 characters
#392: FILE: drivers/event/sw/sw_evdev_xstats.c:295:
+						(bkt + 1) << SW_DEQ_STAT_BUCKET_SHIFT);

total: 0 errors, 7 warnings, 454 lines checked

### [PATCH 11/12] app/test: add unit tests for SW eventdev driver

WARNING:BLOCK_COMMENT_STYLE: Block comments should align the * on each line
#114: FILE: app/test/test_sw_eventdev.c:73:
+	/*
+	* len = 14 + 46

WARNING:LONG_LINE: line over 80 characters
#405: FILE: app/test/test_sw_eventdev.c:364:
+		printf("%d: error stats incorrect for directed port\n", __LINE__);

WARNING:LONG_LINE: line over 80 characters
#494: FILE: app/test/test_sw_eventdev.c:453:
+		printf("%d: error stats incorrect for directed port\n", __LINE__);

WARNING:LONG_LINE: line over 80 characters
#513: FILE: app/test/test_sw_eventdev.c:472:
+		printf("%d: error magic sequence number not dequeued\n", __LINE__);

WARNING:LONG_LINE: line over 80 characters
#697: FILE: app/test/test_sw_eventdev.c:656:
+		printf("%d: Half of NUM_PKTS didn't arrive at port 1\n", __LINE__);

WARNING:LONG_LINE: line over 80 characters
#710: FILE: app/test/test_sw_eventdev.c:669:
+		printf("%d: Half of NUM_PKTS didn't arrive at port 2\n", __LINE__);

WARNING:LONG_LINE: line over 80 characters
#767: FILE: app/test/test_sw_eventdev.c:726:
+		printf("%d: Sched core didn't handle pkt as expected\n", __LINE__);

WARNING:LONG_LINE: line over 80 characters
#793: FILE: app/test/test_sw_eventdev.c:752:
+				.event_queue_cfg = RTE_EVENT_QUEUE_CFG_ATOMIC_ONLY,

WARNING:LONG_LINE_COMMENT: line over 80 characters
#815: FILE: app/test/test_sw_eventdev.c:774:
+	/* enqueue 3 packets, setting seqn and QID as needed to check priority */

WARNING:LONG_LINE: line over 80 characters
#839: FILE: app/test/test_sw_eventdev.c:798:
+	uint32_t deq_pkts = rte_event_dequeue_burst(evdev, t->port[0], ev, 32, 0);

WARNING:LONG_LINE_STRING: line over 80 characters
#847: FILE: app/test/test_sw_eventdev.c:806:
+			printf("%d: qid priority test: seqn %d incorrectly prioritized\n",

WARNING:LONG_LINE: line over 80 characters
#874: FILE: app/test/test_sw_eventdev.c:833:
+			printf("%d: error mapping qid to port %d\n", __LINE__, i);

WARNING:LONG_LINE: line over 80 characters
#923: FILE: app/test/test_sw_eventdev.c:882:
+		printf("%d:%s: port 1 inflight not correct\n", __LINE__, __func__);

WARNING:LONG_LINE: line over 80 characters
#927: FILE: app/test/test_sw_eventdev.c:886:
+		printf("%d:%s: port 2 inflight not correct\n", __LINE__, __func__);

WARNING:LONG_LINE: line over 80 characters
#931: FILE: app/test/test_sw_eventdev.c:890:
+		printf("%d:%s: port 3 inflight not correct\n", __LINE__, __func__);

WARNING:LONG_LINE_COMMENT: line over 80 characters
#947: FILE: app/test/test_sw_eventdev.c:906:
+	/* Create qm instance with 1 atomic QID going to 3 ports + 1 prod port */

WARNING:LONG_LINE: line over 80 characters
#1073: FILE: app/test/test_sw_eventdev.c:1032:
+		printf("%d:%s: port 1 inflight not correct\n", __LINE__, __func__);

WARNING:LONG_LINE: line over 80 characters
#1081: FILE: app/test/test_sw_eventdev.c:1040:
+		printf("%d:%s: port 2 inflight not correct\n", __LINE__, __func__);

WARNING:LONG_LINE: line over 80 characters
#1089: FILE: app/test/test_sw_eventdev.c:1048:
+		printf("%d:%s: port 3 inflight not correct\n", __LINE__, __func__);

WARNING:LONG_LINE: line over 80 characters
#1125: FILE: app/test/test_sw_eventdev.c:1084:
+		err = rte_event_port_link(evdev, t->port[i], &t->qid[0], NULL, 1);

WARNING:LONG_LINE: line over 80 characters
#1179: FILE: app/test/test_sw_eventdev.c:1138:
+		printf("%d:%s: port 1 inflight count not correct\n", __LINE__, __func__);

WARNING:LONG_LINE: line over 80 characters
#1190: FILE: app/test/test_sw_eventdev.c:1149:
+		printf("%d:%s: port 1 dropped count not correct\n", __LINE__, __func__);

WARNING:LONG_LINE: line over 80 characters
#1261: FILE: app/test/test_sw_eventdev.c:1220:
+		printf("%d: Sched core didn't handle pkt as expected\n", __LINE__);

WARNING:LONG_LINE: line over 80 characters
#1390: FILE: app/test/test_sw_eventdev.c:1349:
+		printf("%d: Sched core didn't handle pkt as expected\n", __LINE__);

WARNING:LONG_LINE: line over 80 characters
#1395: FILE: app/test/test_sw_eventdev.c:1354:
+		printf("%d: %s port 1 inflight not correct\n", __LINE__, __func__);

WARNING:LONG_LINE: line over 80 characters
#1399: FILE: app/test/test_sw_eventdev.c:1358:
+		printf("%d: %s port 2 inflight not correct\n", __LINE__, __func__);

WARNING:LONG_LINE: line over 80 characters
#1415: FILE: app/test/test_sw_eventdev.c:1374:
+		printf("%d: port 1 inflight decrement after DEQ != 0\n", __LINE__);

WARNING:LONG_LINE: line over 80 characters
#1419: FILE: app/test/test_sw_eventdev.c:1378:
+		err = rte_event_enqueue_burst(evdev, t->port[p1], &release_ev, 1);

WARNING:LONG_LINE: line over 80 characters
#1449: FILE: app/test/test_sw_eventdev.c:1408:
+		printf("%d: port 1 inflight decrement after DEQ != 0\n", __LINE__);

WARNING:LONG_LINE: line over 80 characters
#1453: FILE: app/test/test_sw_eventdev.c:1412:
+		err = rte_event_enqueue_burst(evdev, t->port[p2], &release_ev, 1);

WARNING:LONG_LINE_COMMENT: line over 80 characters
#1521: FILE: app/test/test_sw_eventdev.c:1480:
+	/* CQ mapping to QID for load balanced ports (directed mapped on create) */

WARNING:LONG_LINE: line over 80 characters
#1523: FILE: app/test/test_sw_eventdev.c:1482:
+		err = rte_event_port_link(evdev, t->port[i], &t->qid[0], NULL, 1);

WARNING:LONG_LINE: line over 80 characters
#1566: FILE: app/test/test_sw_eventdev.c:1525:
+		deq_pkts = rte_event_dequeue_burst(evdev, t->port[i], &deq_ev[i],

WARNING:LONG_LINE: line over 80 characters
#1589: FILE: app/test/test_sw_eventdev.c:1548:
+	deq_pkts = rte_event_dequeue_burst(evdev, t->port[tx_port], deq_ev, 3, 0);

WARNING:LONG_LINE_STRING: line over 80 characters
#1593: FILE: app/test/test_sw_eventdev.c:1552:
+		printf("%d: expected 3 packets at tx port got %d from port %d\n",

WARNING:LONG_LINE_STRING: line over 80 characters
#1603: FILE: app/test/test_sw_eventdev.c:1562:
+				printf("%d: Incorrect sequence number(%d) from port %d\n",

WARNING:LONG_LINE: line over 80 characters
#1636: FILE: app/test/test_sw_eventdev.c:1595:
+	char rx_port_used_stat[64], rx_port_free_stat[64], other_port_used_stat[64];

WARNING:LONG_LINE: line over 80 characters
#1645: FILE: app/test/test_sw_eventdev.c:1604:
+			rte_event_port_link(evdev, t->port[1], NULL, NULL, 0) != 1) {

WARNING:LONG_LINE: line over 80 characters
#1661: FILE: app/test/test_sw_eventdev.c:1620:
+	if (rte_event_dev_get_xstat_by_name(evdev, "port_0_cq_ring_used", NULL) != 1)

WARNING:LONG_LINE: line over 80 characters
#1669: FILE: app/test/test_sw_eventdev.c:1628:
+	if (rte_event_dev_get_xstat_by_name(evdev, rx_port_used_stat, NULL) != 1) {

WARNING:LONG_LINE: line over 80 characters
#1682: FILE: app/test/test_sw_eventdev.c:1641:
+	} while (rte_event_dev_get_xstat_by_name(evdev, rx_port_free_stat, NULL) != 0);

WARNING:LONG_LINE: line over 80 characters
#1693: FILE: app/test/test_sw_eventdev.c:1652:
+	if (rte_event_dev_get_xstat_by_name(evdev, other_port_used_stat, NULL) != 0) {

WARNING:LONG_LINE: line over 80 characters
#1698: FILE: app/test/test_sw_eventdev.c:1657:
+	if (rte_event_dev_get_xstat_by_name(evdev, "qid_0_iq_0_used", NULL) != 1) {

WARNING:LONG_LINE: line over 80 characters
#1699: FILE: app/test/test_sw_eventdev.c:1658:
+		printf("%d: Error, QID does not have exactly 1 packet\n", __LINE__);

WARNING:LONG_LINE: line over 80 characters
#1712: FILE: app/test/test_sw_eventdev.c:1671:
+	if (rte_event_dev_get_xstat_by_name(evdev, other_port_used_stat, NULL) != 1) {

WARNING:LONG_LINE: line over 80 characters
#1713: FILE: app/test/test_sw_eventdev.c:1672:
+		printf("%d: Error, second flow did not pass out first\n", __LINE__);

WARNING:LONG_LINE: line over 80 characters
#1717: FILE: app/test/test_sw_eventdev.c:1676:
+	if (rte_event_dev_get_xstat_by_name(evdev, "qid_0_iq_0_used", NULL) != 1) {

WARNING:LONG_LINE: line over 80 characters
#1718: FILE: app/test/test_sw_eventdev.c:1677:
+		printf("%d: Error, QID does not have exactly 1 packet\n", __LINE__);

WARNING:LONG_LINE_COMMENT: line over 80 characters
#1739: FILE: app/test/test_sw_eventdev.c:1698:
+	 * the Queue Manager. Each packet gets looped through QIDs 0-8, 16 times,

WARNING:LONG_LINE: line over 80 characters
#1746: FILE: app/test/test_sw_eventdev.c:1705:
+		uint16_t i, nb_rx = rte_event_dequeue_burst(evdev, port, ev, BURST_SIZE, 0);

WARNING:LONG_LINE: line over 80 characters
#1756: FILE: app/test/test_sw_eventdev.c:1715:
+				enqd = rte_event_enqueue_burst(evdev, port, &ev[i], 1);

WARNING:LONG_LINE: line over 80 characters
#1758: FILE: app/test/test_sw_eventdev.c:1717:
+					printf("%d: Can't enqueue FWD!!\n", __LINE__);

WARNING:LONG_LINE: line over 80 characters
#1768: FILE: app/test/test_sw_eventdev.c:1727:
+				enqd = rte_event_enqueue_burst(evdev, port, &ev[i], 1);

WARNING:LONG_LINE: line over 80 characters
#1770: FILE: app/test/test_sw_eventdev.c:1729:
+					printf("%d: Can't enqueue FWD!!\n", __LINE__);

WARNING:LONG_LINE: line over 80 characters
#1814: FILE: app/test/test_sw_eventdev.c:1773:
+			while (rte_event_enqueue_burst(evdev, port, &ev, 1) != 1)

WARNING:LONG_LINE_STRING: line over 80 characters
#1892: FILE: app/test/test_sw_eventdev.c:1851:
+			printf("%d: \tSched Rx = %" PRIu64 ", Tx = %" PRIu64 "\n",

WARNING:LONG_LINE: line over 80 characters
#1901: FILE: app/test/test_sw_eventdev.c:1860:
+				printf("%d: No schedules for seconds, deadlock\n", __LINE__);

WARNING:LONG_LINE_STRING: line over 80 characters
#1942: FILE: app/test/test_sw_eventdev.c:1901:
+		eventdev_func_mempool = rte_pktmbuf_pool_create("QM_SA_MBUF_POOL",

WARNING:LONG_LINE_COMMENT: line over 80 characters
#2065: FILE: app/test/test_sw_eventdev.c:2024:
+	 * in the static eventdev_func_mempool variable. It is re-used on re-runs

total: 0 errors, 59 warnings, 2042 lines checked

3/11 valid patches