* Re: [PATCH v12 1/6] ethdev: add Tx preparation
From: Thomas Monjalon @ 2016-12-01 16:28 UTC (permalink / raw)
To: Tomasz Kulasek; +Cc: dev, konstantin.ananyev, olivier.matz
In-Reply-To: <1479922585-8640-2-git-send-email-tomaszx.kulasek@intel.com>
2016-11-23 18:36, Tomasz Kulasek:
> +/**
> + * Process a burst of output packets on a transmit queue of an Ethernet device.
> + *
> + * The rte_eth_tx_prepare() function is invoked to prepare output packets to be
> + * transmitted on the output queue *queue_id* of the Ethernet device designated
> + * by its *port_id*.
> + * The *nb_pkts* parameter is the number of packets to be prepared which are
> + * supplied in the *tx_pkts* array of *rte_mbuf* structures, each of them
> + * allocated from a pool created with rte_pktmbuf_pool_create().
> + * For each packet to send, the rte_eth_tx_prepare() function performs
> + * the following operations:
> + *
> + * - Check if packet meets devices requirements for tx offloads.
> + *
> + * - Check limitations about number of segments.
> + *
> + * - Check additional requirements when debug is enabled.
> + *
> + * - Update and/or reset required checksums when tx offload is set for packet.
> + *
> + * Since this function can modify packet data, provided mbufs must be safely
> + * writable (e.g. modified data cannot be in shared segment).
I think we will have to remove this limitation in next releases.
As we don't know how it could affect the API, I suggest to declare this
API EXPERIMENTAL.
^ permalink raw reply
* Re: [PATCH v12 1/6] ethdev: add Tx preparation
From: Thomas Monjalon @ 2016-12-01 16:26 UTC (permalink / raw)
To: Tomasz Kulasek; +Cc: dev, konstantin.ananyev, olivier.matz
In-Reply-To: <1479922585-8640-2-git-send-email-tomaszx.kulasek@intel.com>
2016-11-23 18:36, Tomasz Kulasek:
> Added fields to the `struct rte_eth_desc_lim`:
>
> uint16_t nb_seg_max;
> /**< Max number of segments per whole packet. */
>
> uint16_t nb_mtu_seg_max;
> /**< Max number of segments per one MTU */
How (and when) an application is supposed to use these fields?
Is it useful to expose them if we make tx_prepare() mandatory?
^ permalink raw reply
* Re: [PATCH v12 1/6] ethdev: add Tx preparation
From: Thomas Monjalon @ 2016-12-01 16:24 UTC (permalink / raw)
To: Tomasz Kulasek; +Cc: dev, konstantin.ananyev, olivier.matz, bruce.richardson
In-Reply-To: <15420823.7ghYzVRe3h@xps13>
Please, a reply to this question would be greatly appreciated.
2016-11-28 11:54, Thomas Monjalon:
> Hi,
>
> 2016-11-23 18:36, Tomasz Kulasek:
> > --- a/config/common_base
> > +++ b/config/common_base
> > @@ -120,6 +120,7 @@ CONFIG_RTE_MAX_QUEUES_PER_PORT=1024
> > CONFIG_RTE_LIBRTE_IEEE1588=n
> > CONFIG_RTE_ETHDEV_QUEUE_STAT_CNTRS=16
> > CONFIG_RTE_ETHDEV_RXTX_CALLBACKS=y
> > +CONFIG_RTE_ETHDEV_TX_PREPARE=y
>
> Please, remind me why is there a configuration here.
> It should be the responsibility of the application to call tx_prepare
> or not. If the application choose to use this new API but it is
> disabled, then the packets won't be prepared and there is no error code:
>
> > +#else
> > +
> > +static inline uint16_t
> > +rte_eth_tx_prepare(__rte_unused uint8_t port_id, __rte_unused uint16_t queue_id,
> > + __rte_unused struct rte_mbuf **tx_pkts, uint16_t nb_pkts)
> > +{
> > + return nb_pkts;
> > +}
> > +
> > +#endif
>
> So the application is not aware of the issue and it will not use
> any fallback.
^ permalink raw reply
* [PATCH] vhost: allow for many vhost user ports
From: Jan Wickbom @ 2016-12-01 15:26 UTC (permalink / raw)
To: huawei.xie, yuanhan.liu; +Cc: dev, patrik.r.andersson, Jan Wickbom
Currently select() is used to monitor file descriptors for vhostuser
ports. This limits the number of ports possible to create since the
fd number is used as index in the fd_set and we have seen fds > 1023.
This patch changes select() to poll(). This way we can keep an
packed (pollfd) array for the fds, e.g. as many fds as the size of
the array.
Also see:
http://dpdk.org/ml/archives/dev/2016-April/037024.html
Signed-off-by: Jan Wickbom <jan.wickbom@ericsson.com>
Reported-by: Patrik Andersson <patrik.r.andersson@ericsson.com>
---
lib/librte_vhost/fd_man.c | 140 +++++++++++++++++++++++++++-------------------
lib/librte_vhost/fd_man.h | 2 +-
2 files changed, 83 insertions(+), 59 deletions(-)
diff --git a/lib/librte_vhost/fd_man.c b/lib/librte_vhost/fd_man.c
index 2d3eeb7..dc4aa3f 100644
--- a/lib/librte_vhost/fd_man.c
+++ b/lib/librte_vhost/fd_man.c
@@ -35,16 +35,40 @@
#include <stdio.h>
#include <stdlib.h>
#include <sys/socket.h>
-#include <sys/select.h>
#include <sys/time.h>
#include <sys/types.h>
+#include <poll.h>
#include <unistd.h>
#include <rte_common.h>
+#include <rte_malloc.h>
#include <rte_log.h>
#include "fd_man.h"
+#define FDPOLLERR (POLLERR | POLLHUP | POLLNVAL)
+
+
+/**
+ * Adjusts the highest index populated in the array of fds
+ * @return
+ * Index of highest position populated + 1.
+ */
+static int
+fdset_adjust_num(struct fdset *pfdset)
+{
+ int idx;
+
+ for (idx = pfdset->num - 1;
+ idx >= 0 && pfdset->fd[idx].fd == -1;
+ idx--)
+ ;
+
+ pfdset->num = idx + 1;
+
+ return pfdset->num;
+}
+
/**
* Returns the index in the fdset for a given fd.
* If fd is -1, it means to search for a free entry.
@@ -56,72 +80,62 @@
{
int i;
- if (pfdset == NULL)
- return -1;
-
- for (i = 0; i < MAX_FDS && pfdset->fd[i].fd != fd; i++)
+ for (i = 0; i < pfdset->num && pfdset->fd[i].fd != fd; i++)
;
- return i == MAX_FDS ? -1 : i;
+ return i == pfdset->num ? -1 : i;
}
static int
fdset_find_free_slot(struct fdset *pfdset)
{
- return fdset_find_fd(pfdset, -1);
+ if (pfdset->num < MAX_FDS)
+ return pfdset->num;
+ else
+ return fdset_find_fd(pfdset, -1);
}
-static int
-fdset_add_fd(struct fdset *pfdset, int idx, int fd,
+static void
+fdset_add_fd(struct fdset *pfdset, int idx, int fd,
fd_cb rcb, fd_cb wcb, void *dat)
{
struct fdentry *pfdentry;
- if (pfdset == NULL || idx >= MAX_FDS || fd >= FD_SETSIZE)
- return -1;
-
pfdentry = &pfdset->fd[idx];
pfdentry->fd = fd;
pfdentry->rcb = rcb;
pfdentry->wcb = wcb;
pfdentry->dat = dat;
-
- return 0;
}
/**
- * Fill the read/write fd_set with the fds in the fdset.
+ * Compact the fdset and fill the read/write fds with the fds in the fdset.
* @return
- * the maximum fds filled in the read/write fd_set.
+ * the number of fds filled in the read/write fds.
*/
static int
-fdset_fill(fd_set *rfset, fd_set *wfset, struct fdset *pfdset)
+fdset_fill(struct pollfd *rwfds, struct fdset *pfdset)
{
struct fdentry *pfdentry;
- int i, maxfds = -1;
- int num = MAX_FDS;
-
- if (pfdset == NULL)
- return -1;
+ int i;
+ int num;
- for (i = 0; i < num; i++) {
+ for (i = 0, num = pfdset->num; i < num; i++) {
pfdentry = &pfdset->fd[i];
- if (pfdentry->fd != -1) {
- int added = 0;
- if (pfdentry->rcb && rfset) {
- FD_SET(pfdentry->fd, rfset);
- added = 1;
- }
- if (pfdentry->wcb && wfset) {
- FD_SET(pfdentry->fd, wfset);
- added = 1;
- }
- if (added)
- maxfds = pfdentry->fd < maxfds ?
- maxfds : pfdentry->fd;
+
+ if (pfdentry->fd < 0) {
+ /* Hole in the list. Move the last one here */
+
+ *pfdentry = pfdset->fd[num - 1];
+ pfdset->fd[num - 1].fd = -1;
+ num = fdset_adjust_num(pfdset);
}
+ rwfds[i].fd = pfdentry->fd;
+ rwfds[i].events = pfdentry->rcb ? POLLIN : 0;
+ rwfds[i].events |= pfdentry->wcb ? POLLOUT : 0;
}
- return maxfds;
+
+ return i;
}
void
@@ -132,6 +146,8 @@
if (pfdset == NULL)
return;
+ pthread_mutex_init(&pfdset->fd_mutex, NULL);
+
for (i = 0; i < MAX_FDS; i++) {
pfdset->fd[i].fd = -1;
pfdset->fd[i].dat = NULL;
@@ -152,14 +168,15 @@
pthread_mutex_lock(&pfdset->fd_mutex);
- /* Find a free slot in the list. */
i = fdset_find_free_slot(pfdset);
- if (i == -1 || fdset_add_fd(pfdset, i, fd, rcb, wcb, dat) < 0) {
+ if (i == -1) {
pthread_mutex_unlock(&pfdset->fd_mutex);
return -2;
}
- pfdset->num++;
+ fdset_add_fd(pfdset, i, fd, rcb, wcb, dat);
+ if (i == pfdset->num)
+ pfdset->num++;
pthread_mutex_unlock(&pfdset->fd_mutex);
@@ -189,7 +206,7 @@
pfdset->fd[i].fd = -1;
pfdset->fd[i].rcb = pfdset->fd[i].wcb = NULL;
pfdset->fd[i].dat = NULL;
- pfdset->num--;
+ (void) fdset_adjust_num(pfdset);
i = -1;
}
pthread_mutex_unlock(&pfdset->fd_mutex);
@@ -211,12 +228,12 @@
pfdset->fd[index].fd = -1;
pfdset->fd[index].rcb = pfdset->fd[index].wcb = NULL;
- pfdset->fd[index].dat = NULL;
- pfdset->num--;
+ (void) fdset_adjust_num(pfdset);
pthread_mutex_unlock(&pfdset->fd_mutex);
}
+
/**
* This functions runs in infinite blocking loop until there is no fd in
* pfdset. It calls corresponding r/w handler if there is event on the fd.
@@ -229,44 +246,48 @@
void
fdset_event_dispatch(struct fdset *pfdset)
{
- fd_set rfds, wfds;
- int i, maxfds;
+ int i;
struct fdentry *pfdentry;
- int num = MAX_FDS;
+ int numfds;
fd_cb rcb, wcb;
void *dat;
int fd;
int remove1, remove2;
int ret;
+ int handled;
if (pfdset == NULL)
return;
+ struct pollfd * const rwfds =
+ rte_malloc("struct pollfd", MAX_FDS * sizeof(*rwfds), 0);
+
while (1) {
- struct timeval tv;
- tv.tv_sec = 1;
- tv.tv_usec = 0;
- FD_ZERO(&rfds);
- FD_ZERO(&wfds);
pthread_mutex_lock(&pfdset->fd_mutex);
- maxfds = fdset_fill(&rfds, &wfds, pfdset);
+ numfds = fdset_fill(rwfds, pfdset);
pthread_mutex_unlock(&pfdset->fd_mutex);
/*
- * When select is blocked, other threads might unregister
+ * When poll is blocked, other threads might unregister
* listenfds from and register new listenfds into fdset.
- * When select returns, the entries for listenfds in the fdset
+ * When poll returns, the entries for listenfds in the fdset
* might have been updated. It is ok if there is unwanted call
* for new listenfds.
*/
- ret = select(maxfds + 1, &rfds, &wfds, NULL, &tv);
+ ret = poll(rwfds, numfds, 1000 /* millisecs */);
+
if (ret <= 0)
continue;
- for (i = 0; i < num; i++) {
+ for (i = handled = 0; i < numfds && handled < ret; i++) {
+ if (!rwfds[i].revents)
+ continue;
+
+ handled++;
remove1 = remove2 = 0;
+
pthread_mutex_lock(&pfdset->fd_mutex);
pfdentry = &pfdset->fd[i];
fd = pfdentry->fd;
@@ -275,9 +296,12 @@
dat = pfdentry->dat;
pfdentry->busy = 1;
pthread_mutex_unlock(&pfdset->fd_mutex);
- if (fd >= 0 && FD_ISSET(fd, &rfds) && rcb)
+
+ if (fd >= 0 && rcb &&
+ rwfds[i].revents & (POLLIN | FDPOLLERR))
rcb(fd, dat, &remove1);
- if (fd >= 0 && FD_ISSET(fd, &wfds) && wcb)
+ if (fd >= 0 && wcb &&
+ rwfds[i].revents & (POLLOUT | FDPOLLERR))
wcb(fd, dat, &remove2);
pfdentry->busy = 0;
/*
diff --git a/lib/librte_vhost/fd_man.h b/lib/librte_vhost/fd_man.h
index bd66ed1..b5ba688 100644
--- a/lib/librte_vhost/fd_man.h
+++ b/lib/librte_vhost/fd_man.h
@@ -51,7 +51,7 @@ struct fdentry {
struct fdset {
struct fdentry fd[MAX_FDS];
pthread_mutex_t fd_mutex;
- int num; /* current fd number of this fdset */
+ int num; /* highest index occupied in fd array + 1 */
};
--
1.9.1
^ permalink raw reply related
* Re: [PATCH 3/3] maintainers: add stable mailing list
From: Mcnamara, John @ 2016-12-01 15:18 UTC (permalink / raw)
To: Yuanhan Liu, dev@dpdk.org; +Cc: Thomas Monjalon
In-Reply-To: <1480575999-14453-4-git-send-email-yuanhan.liu@linux.intel.com>
> -----Original Message-----
> From: Yuanhan Liu [mailto:yuanhan.liu@linux.intel.com]
> Sent: Thursday, December 1, 2016 7:07 AM
> To: dev@dpdk.org
> Cc: Thomas Monjalon <thomas.monjalon@6wind.com>; Mcnamara, John
> <john.mcnamara@intel.com>; Yuanhan Liu <yuanhan.liu@linux.intel.com>
> Subject: [PATCH 3/3] maintainers: add stable mailing list
>
> Signed-off-by: Yuanhan Liu <yuanhan.liu@linux.intel.com>
Acked-by: John McNamara <john.mcnamara@intel.com>
^ permalink raw reply
* Re: [PATCH 2/3] maintainers: update virtio section name
From: Mcnamara, John @ 2016-12-01 15:18 UTC (permalink / raw)
To: Yuanhan Liu, dev@dpdk.org; +Cc: Thomas Monjalon
In-Reply-To: <1480575999-14453-3-git-send-email-yuanhan.liu@linux.intel.com>
> -----Original Message-----
> From: Yuanhan Liu [mailto:yuanhan.liu@linux.intel.com]
> Sent: Thursday, December 1, 2016 7:07 AM
> To: dev@dpdk.org
> Cc: Thomas Monjalon <thomas.monjalon@6wind.com>; Mcnamara, John
> <john.mcnamara@intel.com>; Yuanhan Liu <yuanhan.liu@linux.intel.com>
> Subject: [PATCH 2/3] maintainers: update virtio section name
>
> Signed-off-by: Yuanhan Liu <yuanhan.liu@linux.intel.com>
Acked-by: John McNamara <john.mcnamara@intel.com>
^ permalink raw reply
* Re: [PATCH] pmdinfogen: Fix pmdinfogen to select proper endianess on cross-compile
From: Thomas Monjalon @ 2016-12-01 15:17 UTC (permalink / raw)
To: Neil Horman; +Cc: Bruce Richardson, dev, Hemant Agrawal, Jerin Jacob, stable
In-Reply-To: <20161121101133.GC16124@bricha3-MOBL3.ger.corp.intel.com>
2016-11-21 10:11, Bruce Richardson:
> On Fri, Nov 18, 2016 at 01:47:52PM -0500, Neil Horman wrote:
> > pmdinfogen has a bug in which, during build, it pulls in rte_byteorder.h to
> > obtain the rte macros for byteswapping between the cpu byte order and big or
> > little endian. Unfortunately, pmdinfogen is a tool that is only meant to be run
> > during the build of dpdk components, and so, it runs on the host. In cross
> > compile environments however, the rte_byteorder.h is configured using a target
> > cpu, who's endianess may differ from that of the host, leading to improper
> > swapping.
> >
> > The fix is to use host system defined byte swapping routines rather than the
> > dpdk provided routines. Note that we are using non posix compliant routines, as
> > the posix compliant api only addresses 16 and 32 bit swaps, and we also need 64
> > bit swaps. Those macros exist (via endian.h), but BSD and Linux put that header
> > in different locations so some ifdeffery is required.
> >
> > Tested successfully by myself on Linux and BSD systems.
> >
> > Signed-off-by: Neil Horman <nhorman@tuxdriver.com>
> > CC: Hemant Agrawal <hemant.agrawal@nxp.com>
> > CC: Jerin Jacob <Jerin.Jacob@cavium.com>
> > CC: Bruce Richardson <bruce.richardson@intel.com>
> > CC: Thomas Monjalon <thomas.monjalon@6wind.com>
> > ---
> > buildtools/pmdinfogen/pmdinfogen.h | 10 +++++++---
> > 1 file changed, 7 insertions(+), 3 deletions(-)
>
> Compiles fine on FreeBSD with clang.
>
> Tested-by: Bruce Richardson <bruce.richardson@intel.com>
Fixed "endianness" typo, headline, added Fixes:, CC: stable@dpdk.org and removed a trailing whitespace, then
applied, thanks
^ permalink raw reply
* Re: [PATCH 1/3] maintainers: update virtio maintainer
From: Mcnamara, John @ 2016-12-01 15:17 UTC (permalink / raw)
To: Yuanhan Liu, dev@dpdk.org; +Cc: Thomas Monjalon, Xie, Huawei
In-Reply-To: <1480575999-14453-2-git-send-email-yuanhan.liu@linux.intel.com>
> -----Original Message-----
> From: Yuanhan Liu [mailto:yuanhan.liu@linux.intel.com]
> Sent: Thursday, December 1, 2016 7:07 AM
> To: dev@dpdk.org
> Cc: Thomas Monjalon <thomas.monjalon@6wind.com>; Mcnamara, John
> <john.mcnamara@intel.com>; Yuanhan Liu <yuanhan.liu@linux.intel.com>; Xie,
> Huawei <huawei.xie@intel.com>
> Subject: [PATCH 1/3] maintainers: update virtio maintainer
>
> Huawei has left DPDK team for months, and he hasn't showed up since then.
> Remove him.
>
> Cc: Huawei Xie <huawei.xie@intel.com>
> Signed-off-by: Yuanhan Liu <yuanhan.liu@linux.intel.com>
Acked-by: John McNamara <john.mcnamara@intel.com>
^ permalink raw reply
* Re: [PATCH 2/3] maintainers: update virtio section name
From: Thomas Monjalon @ 2016-12-01 15:08 UTC (permalink / raw)
To: Yuanhan Liu; +Cc: dev, John McNamara
In-Reply-To: <1480575999-14453-3-git-send-email-yuanhan.liu@linux.intel.com>
2016-12-01 15:06, Yuanhan Liu:
> Signed-off-by: Yuanhan Liu <yuanhan.liu@linux.intel.com>
> ---
>
> hmm.., maybe we could seperate lib vhost and virtio pmd, into two
> different sections?
Yes we can :)
> -RedHat virtio
> +Virtio PMD and vhost lib
> M: Yuanhan Liu <yuanhan.liu@linux.intel.com>
> T: git://dpdk.org/next/dpdk-next-virtio
> F: drivers/net/virtio/
>
^ permalink raw reply
* Re: [PATCH v2] scripts: check cc stable mailing list in commit
From: Thomas Monjalon @ 2016-12-01 15:03 UTC (permalink / raw)
To: Ferruh Yigit; +Cc: dev
In-Reply-To: <bb9337aa-9987-7472-c15a-73f11b4f9b34@intel.com>
2016-12-01 15:00, Ferruh Yigit:
> On 12/1/2016 1:43 PM, Thomas Monjalon wrote:
> > Add a check for commits fixing a released bug.
> > Such commits are found thanks to scripts/git-log-fixes.sh.
> > They must be sent CC: stable@dpdk.org.
> > In order to avoid forgetting CC, this mail header can be written
> > in the git commit message.
> >
> > Signed-off-by: Thomas Monjalon <thomas.monjalon@6wind.com>
>
> Tested-by: Ferruh Yigit <ferruh.yigit@intel.com>
Applied
^ permalink raw reply
* Re: [PATCH v2] scripts: check cc stable mailing list in commit
From: Ferruh Yigit @ 2016-12-01 15:00 UTC (permalink / raw)
To: Thomas Monjalon, dev
In-Reply-To: <1480599818-14911-1-git-send-email-thomas.monjalon@6wind.com>
On 12/1/2016 1:43 PM, Thomas Monjalon wrote:
> Add a check for commits fixing a released bug.
> Such commits are found thanks to scripts/git-log-fixes.sh.
> They must be sent CC: stable@dpdk.org.
> In order to avoid forgetting CC, this mail header can be written
> in the git commit message.
>
> Signed-off-by: Thomas Monjalon <thomas.monjalon@6wind.com>
Tested-by: Ferruh Yigit <ferruh.yigit@intel.com>
^ permalink raw reply
* Re: [PATCH v2] ethdev: check number of queues less than RTE_ETHDEV_QUEUE_STAT_CNTRS
From: Alejandro Lucero @ 2016-12-01 14:44 UTC (permalink / raw)
To: Thomas Monjalon; +Cc: Olivier Matz, dev, Bert van Leeuwen
In-Reply-To: <7874818.0HvcekTQN4@xps13>
On Mon, Nov 28, 2016 at 11:13 AM, Thomas Monjalon <thomas.monjalon@6wind.com
> wrote:
> 2016-11-24 17:59, Olivier Matz:
> > Hi,
> >
> > On Mon, 2016-11-21 at 09:59 +0000, Alejandro Lucero wrote:
> > > From: Bert van Leeuwen <bert.vanleeuwen@netronome.com>
> > >
> > > Arrays inside rte_eth_stats have size=RTE_ETHDEV_QUEUE_STAT_CNTRS.
> > > Some devices report more queues than that and this code blindly uses
> > > the reported number of queues by the device to fill those arrays up.
> > > This patch fixes the problem using MIN between the reported number of
> > > queues and RTE_ETHDEV_QUEUE_STAT_CNTRS.
> > >
> > > Signed-off-by: Alejandro Lucero <alejandro.lucero@netronome.com>
> > >
> >
> > Reviewed-by: Olivier Matz <olivier.matz@6wind.com>
> >
> >
> > As a next step, I'm wondering if it would be possible to remove
> > this limitation. We could replace the tables in struct rte_eth_stats
> > by a pointer to an array allocated dynamically at pmd setup.
>
> Yes that's definitely the right way to handle these statistics.
>
>
Agree.
> > It would break the API, so it should be announced first. I'm thinking
> > of something like:
> >
> > struct rte_eth_generic_stats {
> > uint64_t ipackets;
> > uint64_t opackets;
> > uint64_t ibytes;
> > uint64_t obytes;
> > uint64_t imissed;
> > uint64_t ierrors;
> > uint64_t oerrors;
> > uint64_t rx_nombuf
> > };
> >
> > struct rte_eth_stats {
> > struct rte_eth_generic_stats port_stats;
> > struct rte_eth_generic_stats *queue_stats;
> > };
> >
> > The queue_stats array would always be indexed by queue_id.
> > The xstats would continue to report the generic stats per-port and
> > per-queue.
> >
> > About the mapping API, either we keep it as-is, or it could
> > become a driver-specific API.
>
> Yes I agree to remove the queue statistics mapping which is very specific.
> I will send a patch with a deprecation notice to move the mapping API
> to a driver-specific API.
>
> Any objection?
>
No from my side.
^ permalink raw reply
* [PATCH v2] scripts: check cc stable mailing list in commit
From: Thomas Monjalon @ 2016-12-01 13:43 UTC (permalink / raw)
To: dev
In-Reply-To: <1479768194-6255-1-git-send-email-thomas.monjalon@6wind.com>
Add a check for commits fixing a released bug.
Such commits are found thanks to scripts/git-log-fixes.sh.
They must be sent CC: stable@dpdk.org.
In order to avoid forgetting CC, this mail header can be written
in the git commit message.
Signed-off-by: Thomas Monjalon <thomas.monjalon@6wind.com>
---
v2: fix option -N
---
scripts/check-git-log.sh | 13 +++++++++++++
1 file changed, 13 insertions(+)
diff --git a/scripts/check-git-log.sh b/scripts/check-git-log.sh
index 5f8a9fc..f79f0a2 100755
--- a/scripts/check-git-log.sh
+++ b/scripts/check-git-log.sh
@@ -47,12 +47,18 @@ if [ "$1" = '-h' -o "$1" = '--help' ] ; then
exit
fi
+selfdir=$(dirname $(readlink -e $0))
range=${1:-origin/master..}
+# convert -N to HEAD~N.. in order to comply with git-log-fixes.sh getopts
+if printf -- $range | grep -q '^-[0-9]\+' ; then
+ range="HEAD$(printf -- $range | sed 's,^-,~,').."
+fi
commits=$(git log --format='%h' --reverse $range)
headlines=$(git log --format='%s' --reverse $range)
bodylines=$(git log --format='%b' --reverse $range)
fixes=$(git log --format='%h %s' --reverse $range | grep -i ': *fix' | cut -d' ' -f1)
+stablefixes=$($selfdir/git-log-fixes.sh $range | sed '/(N\/A)$/d' | cut -d' ' -f2)
tags=$(git log --format='%b' --reverse $range | grep -i -e 'by *:' -e 'fix.*:')
bytag='\(Reported\|Suggested\|Signed-off\|Acked\|Reviewed\|Tested\)-by:'
@@ -191,3 +197,10 @@ bad=$(for fixtag in $fixtags ; do
printf "$fixtag" | grep -v "^$good$"
done | sed 's,^,\t,')
[ -z "$bad" ] || printf "Wrong 'Fixes' reference:\n$bad\n"
+
+# check CC:stable for fixes
+bad=$(for fix in $stablefixes ; do
+ git log --format='%b' -1 $fix | grep -qi '^CC: *stable@dpdk.org' ||
+ git log --format='\t%s' -1 $fix
+done)
+[ -z "$bad" ] || printf "Should CC: stable@dpdk.org\n$bad\n"
--
2.7.0
^ permalink raw reply related
* Re: [PATCH v2] doc: add pdump library to API doxygen
From: Mcnamara, John @ 2016-12-01 13:28 UTC (permalink / raw)
To: Yigit, Ferruh, dev@dpdk.org; +Cc: Pattan, Reshma
In-Reply-To: <20161201110210.10601-1-ferruh.yigit@intel.com>
> -----Original Message-----
> From: dev [mailto:dev-bounces@dpdk.org] On Behalf Of Ferruh Yigit
> Sent: Thursday, December 1, 2016 11:02 AM
> To: dev@dpdk.org
> Cc: Pattan, Reshma <reshma.pattan@intel.com>
> Subject: [dpdk-dev] [PATCH v2] doc: add pdump library to API doxygen
>
> From: Reshma Pattan <reshma.pattan@intel.com>
>
> Add pdump library to API doxygen.
>
> Signed-off-by: Reshma Pattan <reshma.pattan@intel.com>
Acked-by: John McNamara <john.mcnamara@intel.com>
^ permalink raw reply
* [PATCH v1 2/2] example: distributor app modified to use burstAPI
From: David Hunt @ 2016-12-01 4:50 UTC (permalink / raw)
To: dev; +Cc: bruce.richardson, David Hunt
In-Reply-To: <1480567821-70846-1-git-send-email-david.hunt@intel.com>
New stats to show details of throughput per second. Runs on a separate
core to the rx thread so as not to affect performance.
Thread for Stats, rx, tx, and distributor, all other cores in coremask
will be used for workers.
There's some "#if 0" lines in the code that allow the rx + dist to be
run on the same core, and a few more commented out lines to allow
skipping of flow matching algo, etc. Code is comment in the appropriate
location
Signed-off-by: David Hunt <david.hunt@intel.com>
---
examples/distributor/main.c | 489 ++++++++++++++++++++++++++++++++++----------
1 file changed, 380 insertions(+), 109 deletions(-)
diff --git a/examples/distributor/main.c b/examples/distributor/main.c
index 537cee1..fcac807 100644
--- a/examples/distributor/main.c
+++ b/examples/distributor/main.c
@@ -1,8 +1,7 @@
/*-
* BSD LICENSE
*
- * Copyright(c) 2010-2014 Intel Corporation. All rights reserved.
- * All rights reserved.
+ * Copyright(c) 2010-2016 Intel Corporation. All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions
@@ -31,6 +30,8 @@
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*/
+#define BURST_API 1
+
#include <stdint.h>
#include <inttypes.h>
#include <unistd.h>
@@ -43,39 +44,87 @@
#include <rte_malloc.h>
#include <rte_debug.h>
#include <rte_prefetch.h>
+#if BURST_API
+#include <rte_distributor_burst.h>
+#else
#include <rte_distributor.h>
+#endif
-#define RX_RING_SIZE 256
-#define TX_RING_SIZE 512
+#define RX_QUEUE_SIZE 512
+#define TX_QUEUE_SIZE 512
#define NUM_MBUFS ((64*1024)-1)
-#define MBUF_CACHE_SIZE 250
+#define MBUF_CACHE_SIZE 128
+#if BURST_API
+#define BURST_SIZE 64
+#define SCHED_RX_RING_SZ 8192
+#define SCHED_TX_RING_SZ 65536
+#else
#define BURST_SIZE 32
-#define RTE_RING_SZ 1024
+#define SCHED_RX_RING_SZ 1024
+#define SCHED_TX_RING_SZ 1024
+#endif
+#define BURST_SIZE_TX 32
#define RTE_LOGTYPE_DISTRAPP RTE_LOGTYPE_USER1
+#define ANSI_COLOR_RED "\x1b[31m"
+#define ANSI_COLOR_RESET "\x1b[0m"
+
/* mask of enabled ports */
static uint32_t enabled_port_mask;
volatile uint8_t quit_signal;
volatile uint8_t quit_signal_rx;
+volatile uint8_t quit_signal_dist;
+volatile uint8_t quit_signal_work;
static volatile struct app_stats {
struct {
uint64_t rx_pkts;
uint64_t returned_pkts;
uint64_t enqueued_pkts;
+ uint64_t enqdrop_pkts;
} rx __rte_cache_aligned;
+ int pad1 __rte_cache_aligned;
+
+ struct {
+ uint64_t in_pkts;
+ uint64_t ret_pkts;
+ uint64_t sent_pkts;
+ uint64_t enqdrop_pkts;
+ } dist __rte_cache_aligned;
+ int pad2 __rte_cache_aligned;
struct {
uint64_t dequeue_pkts;
uint64_t tx_pkts;
+ uint64_t enqdrop_pkts;
} tx __rte_cache_aligned;
+ int pad3 __rte_cache_aligned;
+
+ uint64_t worker_pkts[64] __rte_cache_aligned;
+
+ int pad4 __rte_cache_aligned;
+
+ uint64_t worker_bursts[64][8] __rte_cache_aligned;
+
+ int pad5 __rte_cache_aligned;
+
+ uint64_t port_rx_pkts[64] __rte_cache_aligned;
+ uint64_t port_tx_pkts[64] __rte_cache_aligned;
} app_stats;
+struct app_stats prev_app_stats;
+
static const struct rte_eth_conf port_conf_default = {
.rxmode = {
.mq_mode = ETH_MQ_RX_RSS,
.max_rx_pkt_len = ETHER_MAX_LEN,
+ .split_hdr_size = 0,
+ .header_split = 0, /**< Header Split disabled */
+ .hw_ip_checksum = 1, /**< IP checksum offload enabled */
+ .hw_vlan_filter = 0, /**< VLAN filtering disabled */
+ .jumbo_frame = 0, /**< Jumbo Frame Support disabled */
+ .hw_strip_crc = 0, /**< CRC stripped by hardware */
},
.txmode = {
.mq_mode = ETH_MQ_TX_NONE,
@@ -93,6 +142,8 @@ struct output_buffer {
struct rte_mbuf *mbufs[BURST_SIZE];
};
+static void print_stats(void);
+
/*
* Initialises a given port using global settings and with the rx buffers
* coming from the mbuf_pool passed as parameter
@@ -101,9 +152,13 @@ static inline int
port_init(uint8_t port, struct rte_mempool *mbuf_pool)
{
struct rte_eth_conf port_conf = port_conf_default;
- const uint16_t rxRings = 1, txRings = rte_lcore_count() - 1;
- int retval;
+ const uint16_t rxRings = 1;
+ uint16_t txRings = rte_lcore_count() - 1;
uint16_t q;
+ int retval;
+
+ if (txRings > RTE_MAX_ETHPORTS)
+ txRings = RTE_MAX_ETHPORTS;
if (port >= rte_eth_dev_count())
return -1;
@@ -113,7 +168,7 @@ port_init(uint8_t port, struct rte_mempool *mbuf_pool)
return retval;
for (q = 0; q < rxRings; q++) {
- retval = rte_eth_rx_queue_setup(port, q, RX_RING_SIZE,
+ retval = rte_eth_rx_queue_setup(port, q, RX_QUEUE_SIZE,
rte_eth_dev_socket_id(port),
NULL, mbuf_pool);
if (retval < 0)
@@ -121,7 +176,7 @@ port_init(uint8_t port, struct rte_mempool *mbuf_pool)
}
for (q = 0; q < txRings; q++) {
- retval = rte_eth_tx_queue_setup(port, q, TX_RING_SIZE,
+ retval = rte_eth_tx_queue_setup(port, q, TX_QUEUE_SIZE,
rte_eth_dev_socket_id(port),
NULL);
if (retval < 0)
@@ -134,7 +189,8 @@ port_init(uint8_t port, struct rte_mempool *mbuf_pool)
struct rte_eth_link link;
rte_eth_link_get_nowait(port, &link);
- if (!link.link_status) {
+ while (!link.link_status) {
+ printf("Waiting for Link up on port %"PRIu8"\n", port);
sleep(1);
rte_eth_link_get_nowait(port, &link);
}
@@ -161,40 +217,51 @@ port_init(uint8_t port, struct rte_mempool *mbuf_pool)
struct lcore_params {
unsigned worker_id;
struct rte_distributor *d;
- struct rte_ring *r;
+ struct rte_ring *rx_dist_ring;
+ struct rte_ring *dist_tx_ring;
struct rte_mempool *mem_pool;
};
-static int
-quit_workers(struct rte_distributor *d, struct rte_mempool *p)
+static inline void
+flush_one_port(struct output_buffer *outbuf, uint8_t outp)
{
- const unsigned num_workers = rte_lcore_count() - 2;
- unsigned i;
- struct rte_mbuf *bufs[num_workers];
+ unsigned int nb_tx = rte_eth_tx_burst(outp, 0,
+ outbuf->mbufs, outbuf->count);
+ app_stats.tx.tx_pkts += outbuf->count;
- if (rte_mempool_get_bulk(p, (void *)bufs, num_workers) != 0) {
- printf("line %d: Error getting mbufs from pool\n", __LINE__);
- return -1;
+ if (unlikely(nb_tx < outbuf->count)) {
+ app_stats.tx.enqdrop_pkts += outbuf->count - nb_tx;
+ do {
+ rte_pktmbuf_free(outbuf->mbufs[nb_tx]);
+ } while (++nb_tx < outbuf->count);
}
+ outbuf->count = 0;
+}
- for (i = 0; i < num_workers; i++)
- bufs[i]->hash.rss = i << 1;
+static inline void
+flush_all_ports(struct output_buffer *tx_buffers, uint8_t nb_ports)
+{
+ uint8_t outp;
- rte_distributor_process(d, bufs, num_workers);
- rte_mempool_put_bulk(p, (void *)bufs, num_workers);
+ for (outp = 0; outp < nb_ports; outp++) {
+ /* skip ports that are not enabled */
+ if ((enabled_port_mask & (1 << outp)) == 0)
+ continue;
- return 0;
+ if (tx_buffers[outp].count == 0)
+ continue;
+
+ flush_one_port(&tx_buffers[outp], outp);
+ }
}
static int
lcore_rx(struct lcore_params *p)
{
- struct rte_distributor *d = p->d;
- struct rte_mempool *mem_pool = p->mem_pool;
- struct rte_ring *r = p->r;
const uint8_t nb_ports = rte_eth_dev_count();
const int socket_id = rte_socket_id();
uint8_t port;
+ struct rte_mbuf *bufs[BURST_SIZE*2];
for (port = 0; port < nb_ports; port++) {
/* skip ports that are not enabled */
@@ -210,6 +277,7 @@ lcore_rx(struct lcore_params *p)
printf("\nCore %u doing packet RX.\n", rte_lcore_id());
port = 0;
+
while (!quit_signal_rx) {
/* skip ports that are not enabled */
@@ -218,7 +286,7 @@ lcore_rx(struct lcore_params *p)
port = 0;
continue;
}
- struct rte_mbuf *bufs[BURST_SIZE*2];
+
const uint16_t nb_rx = rte_eth_rx_burst(port, 0, bufs,
BURST_SIZE);
if (unlikely(nb_rx == 0)) {
@@ -228,19 +296,46 @@ lcore_rx(struct lcore_params *p)
}
app_stats.rx.rx_pkts += nb_rx;
+/*
+ * You can run the distributor on the rx core with this code. Returned
+ * packets are then send straight to the tx core.
+ */
+#if 0
+
+#if BURST_API
+ rte_distributor_process_burst(d, bufs, nb_rx);
+ const uint16_t nb_ret = rte_distributor_returned_pkts_burst(d,
+ bufs, BURST_SIZE*2);
+#else
rte_distributor_process(d, bufs, nb_rx);
const uint16_t nb_ret = rte_distributor_returned_pkts(d,
bufs, BURST_SIZE*2);
+#endif
+
app_stats.rx.returned_pkts += nb_ret;
if (unlikely(nb_ret == 0)) {
if (++port == nb_ports)
port = 0;
continue;
}
-
- uint16_t sent = rte_ring_enqueue_burst(r, (void *)bufs, nb_ret);
+ struct rte_ring *tx_ring = p->dist_tx_ring;
+ uint16_t sent = rte_ring_enqueue_burst(tx_ring,
+ (void *)bufs, nb_ret);
+#else
+ uint16_t nb_ret = nb_rx;
+ /*
+ * Swap the following two lines if you want the rx traffic
+ * to go directly to tx, no distribution.
+ */
+ struct rte_ring *out_ring = p->rx_dist_ring;
+ //struct rte_ring *out_ring = p->dist_tx_ring;
+
+ uint16_t sent = rte_ring_enqueue_burst(out_ring,
+ (void *)bufs, nb_ret);
+#endif
app_stats.rx.enqueued_pkts += sent;
if (unlikely(sent < nb_ret)) {
+ app_stats.rx.enqdrop_pkts += nb_ret - sent;
RTE_LOG(DEBUG, DISTRAPP,
"%s:Packet loss due to full ring\n", __func__);
while (sent < nb_ret)
@@ -249,54 +344,86 @@ lcore_rx(struct lcore_params *p)
if (++port == nb_ports)
port = 0;
}
- rte_distributor_process(d, NULL, 0);
- /* flush distributor to bring to known state */
- rte_distributor_flush(d);
/* set worker & tx threads quit flag */
+ printf("\nCore %u exiting rx task.\n", rte_lcore_id());
quit_signal = 1;
- /*
- * worker threads may hang in get packet as
- * distributor process is not running, just make sure workers
- * get packets till quit_signal is actually been
- * received and they gracefully shutdown
- */
- if (quit_workers(d, mem_pool) != 0)
- return -1;
- /* rx thread should quit at last */
return 0;
}
-static inline void
-flush_one_port(struct output_buffer *outbuf, uint8_t outp)
+static int
+lcore_distributor(struct lcore_params *p)
{
- unsigned nb_tx = rte_eth_tx_burst(outp, 0, outbuf->mbufs,
- outbuf->count);
- app_stats.tx.tx_pkts += nb_tx;
+ struct rte_ring *in_r = p->rx_dist_ring;
+ struct rte_ring *out_r = p->dist_tx_ring;
+ struct rte_mbuf *bufs[BURST_SIZE * 4];
+ struct rte_distributor *d = p->d;
- if (unlikely(nb_tx < outbuf->count)) {
- RTE_LOG(DEBUG, DISTRAPP,
- "%s:Packet loss with tx_burst\n", __func__);
- do {
- rte_pktmbuf_free(outbuf->mbufs[nb_tx]);
- } while (++nb_tx < outbuf->count);
- }
- outbuf->count = 0;
-}
+ printf("\nCore %u acting as distributor core.\n", rte_lcore_id());
+ while (!quit_signal_dist) {
-static inline void
-flush_all_ports(struct output_buffer *tx_buffers, uint8_t nb_ports)
-{
- uint8_t outp;
- for (outp = 0; outp < nb_ports; outp++) {
- /* skip ports that are not enabled */
- if ((enabled_port_mask & (1 << outp)) == 0)
- continue;
+ const uint16_t nb_rx = rte_ring_dequeue_burst(in_r,
+ (void *)bufs, BURST_SIZE*1);
- if (tx_buffers[outp].count == 0)
- continue;
+ if (nb_rx) {
+ app_stats.dist.in_pkts += nb_rx;
+/*
+ * This #if allows you to bypass the distributor. Incoming packets may be
+ * sent straight to the tx ring.
+ */
+#if 1
+
+#if BURST_API
+ /* Distribute the packets */
+ rte_distributor_process_burst(d, bufs, nb_rx);
+ /* Handle Returns */
+ const uint16_t nb_ret =
+ rte_distributor_returned_pkts_burst(d,
+ bufs, BURST_SIZE*2);
+#else
+ /* Distribute the packets */
+ rte_distributor_process(d, bufs, nb_rx);
+ /* Handle Returns */
+ const uint16_t nb_ret =
+ rte_distributor_returned_pkts(d,
+ bufs, BURST_SIZE*2);
+#endif
+
+#else
+ /* Bypass the distributor */
+ const unsigned int xor_val = (rte_eth_dev_count() > 1);
+ /* Touch the mbuf by xor'ing the port */
+ for (unsigned int i = 0; i < nb_rx; i++)
+ bufs[i]->port ^= xor_val;
+
+ const uint16_t nb_ret = nb_rx;
+#endif
+ if (unlikely(nb_ret == 0))
+ continue;
- flush_one_port(&tx_buffers[outp], outp);
+ app_stats.dist.ret_pkts += nb_ret;
+
+ uint16_t sent = rte_ring_enqueue_burst(out_r,
+ (void *)bufs, nb_ret);
+ app_stats.dist.sent_pkts += sent;
+ if (unlikely(sent < nb_ret)) {
+ app_stats.dist.enqdrop_pkts += nb_ret - sent;
+ RTE_LOG(DEBUG, DISTRAPP,
+ "%s:Packet loss due to full out ring\n",
+ __func__);
+ while (sent < nb_ret)
+ rte_pktmbuf_free(bufs[sent++]);
+ }
+ }
}
+ printf("\nCore %u exiting distributor task.\n", rte_lcore_id());
+ quit_signal_work = 1;
+
+#if BURST_API
+ /* Unblock any returns so workers can exit */
+ rte_distributor_clear_returns_burst(d);
+#endif
+ quit_signal_rx = 1;
+ return 0;
}
static int
@@ -327,9 +454,9 @@ lcore_tx(struct rte_ring *in_r)
if ((enabled_port_mask & (1 << port)) == 0)
continue;
- struct rte_mbuf *bufs[BURST_SIZE];
+ struct rte_mbuf *bufs[BURST_SIZE_TX];
const uint16_t nb_rx = rte_ring_dequeue_burst(in_r,
- (void *)bufs, BURST_SIZE);
+ (void *)bufs, BURST_SIZE_TX);
app_stats.tx.dequeue_pkts += nb_rx;
/* if we get no traffic, flush anything we have */
@@ -358,11 +485,12 @@ lcore_tx(struct rte_ring *in_r)
outbuf = &tx_buffers[outp];
outbuf->mbufs[outbuf->count++] = bufs[i];
- if (outbuf->count == BURST_SIZE)
+ if (outbuf->count == BURST_SIZE_TX)
flush_one_port(outbuf, outp);
}
}
}
+ printf("\nCore %u exiting tx task.\n", rte_lcore_id());
return 0;
}
@@ -371,7 +499,7 @@ int_handler(int sig_num)
{
printf("Exiting on signal %d\n", sig_num);
/* set quit flag for rx thread to exit */
- quit_signal_rx = 1;
+ quit_signal_dist = 1;
}
static void
@@ -379,24 +507,88 @@ print_stats(void)
{
struct rte_eth_stats eth_stats;
unsigned i;
-
- printf("\nRX thread stats:\n");
- printf(" - Received: %"PRIu64"\n", app_stats.rx.rx_pkts);
- printf(" - Processed: %"PRIu64"\n", app_stats.rx.returned_pkts);
- printf(" - Enqueued: %"PRIu64"\n", app_stats.rx.enqueued_pkts);
-
- printf("\nTX thread stats:\n");
- printf(" - Dequeued: %"PRIu64"\n", app_stats.tx.dequeue_pkts);
- printf(" - Transmitted: %"PRIu64"\n", app_stats.tx.tx_pkts);
+ const unsigned int num_workers = rte_lcore_count() - 4;
for (i = 0; i < rte_eth_dev_count(); i++) {
rte_eth_stats_get(i, ð_stats);
- printf("\nPort %u stats:\n", i);
- printf(" - Pkts in: %"PRIu64"\n", eth_stats.ipackets);
- printf(" - Pkts out: %"PRIu64"\n", eth_stats.opackets);
- printf(" - In Errs: %"PRIu64"\n", eth_stats.ierrors);
- printf(" - Out Errs: %"PRIu64"\n", eth_stats.oerrors);
- printf(" - Mbuf Errs: %"PRIu64"\n", eth_stats.rx_nombuf);
+ app_stats.port_rx_pkts[i] = eth_stats.ipackets;
+ app_stats.port_tx_pkts[i] = eth_stats.opackets;
+ }
+
+ printf("\n\nRX Thread:\n");
+ for (i = 0; i < rte_eth_dev_count(); i++) {
+ printf("Port %u Pktsin : %5.2f\n", i,
+ (app_stats.port_rx_pkts[i] -
+ prev_app_stats.port_rx_pkts[i])/1000000.0);
+ prev_app_stats.port_rx_pkts[i] = app_stats.port_rx_pkts[i];
+ }
+ printf(" - Received: %5.2f\n",
+ (app_stats.rx.rx_pkts -
+ prev_app_stats.rx.rx_pkts)/1000000.0);
+ printf(" - Returned: %5.2f\n",
+ (app_stats.rx.returned_pkts -
+ prev_app_stats.rx.returned_pkts)/1000000.0);
+ printf(" - Enqueued: %5.2f\n",
+ (app_stats.rx.enqueued_pkts -
+ prev_app_stats.rx.enqueued_pkts)/1000000.0);
+ printf(" - Dropped: %s%5.2f%s\n", ANSI_COLOR_RED,
+ (app_stats.rx.enqdrop_pkts -
+ prev_app_stats.rx.enqdrop_pkts)/1000000.0,
+ ANSI_COLOR_RESET);
+
+ printf("Distributor thread:\n");
+ printf(" - In: %5.2f\n",
+ (app_stats.dist.in_pkts -
+ prev_app_stats.dist.in_pkts)/1000000.0);
+ printf(" - Returned: %5.2f\n",
+ (app_stats.dist.ret_pkts -
+ prev_app_stats.dist.ret_pkts)/1000000.0);
+ printf(" - Sent: %5.2f\n",
+ (app_stats.dist.sent_pkts -
+ prev_app_stats.dist.sent_pkts)/1000000.0);
+ printf(" - Dropped %s%5.2f%s\n", ANSI_COLOR_RED,
+ (app_stats.dist.enqdrop_pkts -
+ prev_app_stats.dist.enqdrop_pkts)/1000000.0,
+ ANSI_COLOR_RESET);
+
+ printf("TX thread:\n");
+ printf(" - Dequeued: %5.2f\n",
+ (app_stats.tx.dequeue_pkts -
+ prev_app_stats.tx.dequeue_pkts)/1000000.0);
+ for (i = 0; i < rte_eth_dev_count(); i++) {
+ printf("Port %u Pktsout: %5.2f\n",
+ i, (app_stats.port_tx_pkts[i] -
+ prev_app_stats.port_tx_pkts[i])/1000000.0);
+ prev_app_stats.port_tx_pkts[i] = app_stats.port_tx_pkts[i];
+ }
+ printf(" - Transmitted: %5.2f\n",
+ (app_stats.tx.tx_pkts -
+ prev_app_stats.tx.tx_pkts)/1000000.0);
+ printf(" - Dropped: %s%5.2f%s\n", ANSI_COLOR_RED,
+ (app_stats.tx.enqdrop_pkts -
+ prev_app_stats.tx.enqdrop_pkts)/1000000.0,
+ ANSI_COLOR_RESET);
+
+ prev_app_stats.rx.rx_pkts = app_stats.rx.rx_pkts;
+ prev_app_stats.rx.returned_pkts = app_stats.rx.returned_pkts;
+ prev_app_stats.rx.enqueued_pkts = app_stats.rx.enqueued_pkts;
+ prev_app_stats.rx.enqdrop_pkts = app_stats.rx.enqdrop_pkts;
+ prev_app_stats.dist.in_pkts = app_stats.dist.in_pkts;
+ prev_app_stats.dist.ret_pkts = app_stats.dist.ret_pkts;
+ prev_app_stats.dist.sent_pkts = app_stats.dist.sent_pkts;
+ prev_app_stats.dist.enqdrop_pkts = app_stats.dist.enqdrop_pkts;
+ prev_app_stats.tx.dequeue_pkts = app_stats.tx.dequeue_pkts;
+ prev_app_stats.tx.tx_pkts = app_stats.tx.tx_pkts;
+ prev_app_stats.tx.enqdrop_pkts = app_stats.tx.enqdrop_pkts;
+
+ for (i = 0; i < num_workers; i++) {
+ printf("Worker %02u Pkts: %5.2f. Bursts(1-8): ", i,
+ (app_stats.worker_pkts[i] -
+ prev_app_stats.worker_pkts[i])/1000000.0);
+ for (int j = 0; j < 8; j++)
+ printf("%ld ", app_stats.worker_bursts[i][j]);
+ printf("\n");
+ prev_app_stats.worker_pkts[i] = app_stats.worker_pkts[i];
}
}
@@ -405,18 +597,48 @@ lcore_worker(struct lcore_params *p)
{
struct rte_distributor *d = p->d;
const unsigned id = p->worker_id;
+ unsigned int num = 0;
+
/*
* for single port, xor_val will be zero so we won't modify the output
* port, otherwise we send traffic from 0 to 1, 2 to 3, and vice versa
*/
const unsigned xor_val = (rte_eth_dev_count() > 1);
- struct rte_mbuf *buf = NULL;
+ struct rte_mbuf *buf[8] __rte_cache_aligned;
+
+ for (int i = 0; i < 8; i++)
+ buf[i] = NULL;
+
+ app_stats.worker_pkts[p->worker_id] = 1;
+
printf("\nCore %u acting as worker core.\n", rte_lcore_id());
- while (!quit_signal) {
- buf = rte_distributor_get_pkt(d, id, buf);
- buf->port ^= xor_val;
+ while (!quit_signal_work) {
+
+#if BURST_API
+ num = rte_distributor_get_pkt_burst(d, id, buf, buf, num);
+ /* Do a little bit of work for each packet */
+ for (unsigned int i = 0; i < num; i++) {
+ uint64_t t = __rdtsc()+100;
+
+ while (__rdtsc() < t)
+ rte_pause();
+ buf[i]->port ^= xor_val;
+ }
+#else
+ buf[0] = rte_distributor_get_pkt(d, id, buf[0]);
+ uint64_t t = __rdtsc() + 10;
+
+ while (__rdtsc() < t)
+ rte_pause();
+ buf[0]->port ^= xor_val;
+#endif
+
+ app_stats.worker_pkts[p->worker_id] += num;
+ if (num > 0)
+ app_stats.worker_bursts[p->worker_id][num-1]++;
}
+ printf("\nCore %u exiting worker task.\n", rte_lcore_id());
return 0;
}
@@ -497,11 +719,13 @@ main(int argc, char *argv[])
{
struct rte_mempool *mbuf_pool;
struct rte_distributor *d;
- struct rte_ring *output_ring;
+ struct rte_ring *dist_tx_ring;
+ struct rte_ring *rx_dist_ring;
unsigned lcore_id, worker_id = 0;
unsigned nb_ports;
uint8_t portid;
uint8_t nb_ports_available;
+ uint64_t t, freq;
/* catch ctrl-c so we can print on exit */
signal(SIGINT, int_handler);
@@ -518,10 +742,12 @@ main(int argc, char *argv[])
if (ret < 0)
rte_exit(EXIT_FAILURE, "Invalid distributor parameters\n");
- if (rte_lcore_count() < 3)
+ if (rte_lcore_count() < 5)
rte_exit(EXIT_FAILURE, "Error, This application needs at "
- "least 3 logical cores to run:\n"
- "1 lcore for packet RX and distribution\n"
+ "least 5 logical cores to run:\n"
+ "1 lcore for stats (can be core 0)\n"
+ "1 lcore for packet RX\n"
+ "1 lcore for distribution\n"
"1 lcore for packet TX\n"
"and at least 1 lcore for worker threads\n");
@@ -560,41 +786,86 @@ main(int argc, char *argv[])
"All available ports are disabled. Please set portmask.\n");
}
+#if BURST_API
+ d = rte_distributor_create_burst("PKT_DIST", rte_socket_id(),
+ rte_lcore_count() - 4);
+#else
d = rte_distributor_create("PKT_DIST", rte_socket_id(),
- rte_lcore_count() - 2);
+ rte_lcore_count() - 4);
+#endif
if (d == NULL)
rte_exit(EXIT_FAILURE, "Cannot create distributor\n");
/*
- * scheduler ring is read only by the transmitter core, but written to
- * by multiple threads
+ * scheduler ring is read by the transmitter core, and written to
+ * by scheduler core
*/
- output_ring = rte_ring_create("Output_ring", RTE_RING_SZ,
- rte_socket_id(), RING_F_SC_DEQ);
- if (output_ring == NULL)
+ dist_tx_ring = rte_ring_create("Output_ring", SCHED_TX_RING_SZ,
+ rte_socket_id(), RING_F_SC_DEQ | RING_F_SP_ENQ);
+ if (dist_tx_ring == NULL)
+ rte_exit(EXIT_FAILURE, "Cannot create output ring\n");
+
+ rx_dist_ring = rte_ring_create("Input_ring", SCHED_RX_RING_SZ,
+ rte_socket_id(), RING_F_SC_DEQ | RING_F_SP_ENQ);
+ if (rx_dist_ring == NULL)
rte_exit(EXIT_FAILURE, "Cannot create output ring\n");
RTE_LCORE_FOREACH_SLAVE(lcore_id) {
- if (worker_id == rte_lcore_count() - 2)
+ if (worker_id == rte_lcore_count() - 3) {
+ printf("Starting distributor on lcore_id %d\n",
+ lcore_id);
+ /* distributor core */
+ struct lcore_params *p =
+ rte_malloc(NULL, sizeof(*p), 0);
+ if (!p)
+ rte_panic("malloc failure\n");
+ *p = (struct lcore_params){worker_id, d,
+ rx_dist_ring, dist_tx_ring, mbuf_pool};
+ rte_eal_remote_launch(
+ (lcore_function_t *)lcore_distributor,
+ p, lcore_id);
+ } else if (worker_id == rte_lcore_count() - 4) {
+ printf("Starting tx on worker_id %d, lcore_id %d\n",
+ worker_id, lcore_id);
+ /* tx core */
rte_eal_remote_launch((lcore_function_t *)lcore_tx,
- output_ring, lcore_id);
- else {
+ dist_tx_ring, lcore_id);
+ } else if (worker_id == rte_lcore_count() - 2) {
+ printf("Starting rx on worker_id %d, lcore_id %d\n",
+ worker_id, lcore_id);
+ /* rx core */
+ struct lcore_params *p =
+ rte_malloc(NULL, sizeof(*p), 0);
+ if (!p)
+ rte_panic("malloc failure\n");
+ *p = (struct lcore_params){worker_id, d, rx_dist_ring,
+ dist_tx_ring, mbuf_pool};
+ rte_eal_remote_launch((lcore_function_t *)lcore_rx,
+ p, lcore_id);
+ } else {
+ printf("Starting worker on worker_id %d, lcore_id %d\n",
+ worker_id, lcore_id);
struct lcore_params *p =
rte_malloc(NULL, sizeof(*p), 0);
if (!p)
rte_panic("malloc failure\n");
- *p = (struct lcore_params){worker_id, d, output_ring, mbuf_pool};
+ *p = (struct lcore_params){worker_id, d, rx_dist_ring,
+ dist_tx_ring, mbuf_pool};
rte_eal_remote_launch((lcore_function_t *)lcore_worker,
p, lcore_id);
}
worker_id++;
}
- /* call lcore_main on master core only */
- struct lcore_params p = { 0, d, output_ring, mbuf_pool};
- if (lcore_rx(&p) != 0)
- return -1;
+ freq = rte_get_timer_hz();
+ t = __rdtsc() + freq;
+ while (!quit_signal_dist) {
+ if (t < __rdtsc()) {
+ print_stats();
+ t = _rdtsc() + freq;
+ }
+ }
RTE_LCORE_FOREACH_SLAVE(lcore_id) {
if (rte_eal_wait_lcore(lcore_id) < 0)
--
2.7.4
^ permalink raw reply related
* [PATCH v1 1/2] lib: distributor performance enhancements
From: David Hunt @ 2016-12-01 4:50 UTC (permalink / raw)
To: dev; +Cc: bruce.richardson, David Hunt
In-Reply-To: <1480567821-70846-1-git-send-email-david.hunt@intel.com>
Now sends bursts of up to 8 mbufs to each worker, and tracks
the in-flight flow-ids (atomic scheduling)
New file with a new api, similar to the old API except with _burst
at the end of the function names.
Signed-off-by: David Hunt <david.hunt@intel.com>
---
lib/librte_distributor/Makefile | 2 +
lib/librte_distributor/rte_distributor.c | 27 +-
lib/librte_distributor/rte_distributor_burst.c | 617 ++++++++++++++++++++++++
lib/librte_distributor/rte_distributor_burst.h | 255 ++++++++++
lib/librte_distributor/rte_distributor_common.h | 75 +++
5 files changed, 950 insertions(+), 26 deletions(-)
create mode 100644 lib/librte_distributor/rte_distributor_burst.c
create mode 100644 lib/librte_distributor/rte_distributor_burst.h
create mode 100644 lib/librte_distributor/rte_distributor_common.h
diff --git a/lib/librte_distributor/Makefile b/lib/librte_distributor/Makefile
index 4c9af17..2acc54d 100644
--- a/lib/librte_distributor/Makefile
+++ b/lib/librte_distributor/Makefile
@@ -43,9 +43,11 @@ LIBABIVER := 1
# all source are stored in SRCS-y
SRCS-$(CONFIG_RTE_LIBRTE_DISTRIBUTOR) := rte_distributor.c
+SRCS-$(CONFIG_RTE_LIBRTE_DISTRIBUTOR) += rte_distributor_burst.c
# install this header file
SYMLINK-$(CONFIG_RTE_LIBRTE_DISTRIBUTOR)-include := rte_distributor.h
+SYMLINK-$(CONFIG_RTE_LIBRTE_DISTRIBUTOR)-include += rte_distributor_burst.h
# this lib needs eal
DEPDIRS-$(CONFIG_RTE_LIBRTE_DISTRIBUTOR) += lib/librte_eal
diff --git a/lib/librte_distributor/rte_distributor.c b/lib/librte_distributor/rte_distributor.c
index f3f778c..cfd187c 100644
--- a/lib/librte_distributor/rte_distributor.c
+++ b/lib/librte_distributor/rte_distributor.c
@@ -40,34 +40,9 @@
#include <rte_errno.h>
#include <rte_string_fns.h>
#include <rte_eal_memconfig.h>
+#include "rte_distributor_common.h"
#include "rte_distributor.h"
-#define NO_FLAGS 0
-#define RTE_DISTRIB_PREFIX "DT_"
-
-/* we will use the bottom four bits of pointer for flags, shifting out
- * the top four bits to make room (since a 64-bit pointer actually only uses
- * 48 bits). An arithmetic-right-shift will then appropriately restore the
- * original pointer value with proper sign extension into the top bits. */
-#define RTE_DISTRIB_FLAG_BITS 4
-#define RTE_DISTRIB_FLAGS_MASK (0x0F)
-#define RTE_DISTRIB_NO_BUF 0 /**< empty flags: no buffer requested */
-#define RTE_DISTRIB_GET_BUF (1) /**< worker requests a buffer, returns old */
-#define RTE_DISTRIB_RETURN_BUF (2) /**< worker returns a buffer, no request */
-
-#define RTE_DISTRIB_BACKLOG_SIZE 8
-#define RTE_DISTRIB_BACKLOG_MASK (RTE_DISTRIB_BACKLOG_SIZE - 1)
-
-#define RTE_DISTRIB_MAX_RETURNS 128
-#define RTE_DISTRIB_RETURNS_MASK (RTE_DISTRIB_MAX_RETURNS - 1)
-
-/**
- * Maximum number of workers allowed.
- * Be aware of increasing the limit, becaus it is limited by how we track
- * in-flight tags. See @in_flight_bitmask and @rte_distributor_process
- */
-#define RTE_DISTRIB_MAX_WORKERS 64
-
/**
* Buffer structure used to pass the pointer data between cores. This is cache
* line aligned, but to improve performance and prevent adjacent cache-line
diff --git a/lib/librte_distributor/rte_distributor_burst.c b/lib/librte_distributor/rte_distributor_burst.c
new file mode 100644
index 0000000..1e32c0f
--- /dev/null
+++ b/lib/librte_distributor/rte_distributor_burst.c
@@ -0,0 +1,617 @@
+/*-
+ * BSD LICENSE
+ *
+ * Copyright(c) 2016 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 <stdio.h>
+#include <sys/queue.h>
+#include <string.h>
+#include <rte_mbuf.h>
+#include <rte_memory.h>
+#include <rte_cycles.h>
+#include <rte_memzone.h>
+#include <rte_errno.h>
+#include <rte_string_fns.h>
+#include <rte_eal_memconfig.h>
+#include "rte_distributor_common.h"
+#include "rte_distributor_burst.h"
+#include "smmintrin.h"
+
+/**
+ * Number of packets to deal with in bursts. Needs to be 8 so as to
+ * fit in one cache line.
+ */
+#define RTE_DIST_BURST_SIZE (sizeof(__m128i) / sizeof(uint16_t))
+
+/**
+ * Buffer structure used to pass the pointer data between cores. This is cache
+ * line aligned, but to improve performance and prevent adjacent cache-line
+ * prefetches of buffers for other workers, e.g. when worker 1's buffer is on
+ * the next cache line to worker 0, we pad this out to two cache lines.
+ * We can pass up to 8 mbufs at a time in one cacheline.
+ * There is a separate cacheline for returns.
+ */
+struct rte_distributor_buffer {
+ volatile int64_t bufptr64[RTE_DIST_BURST_SIZE]
+ __rte_cache_aligned; /* <= outgoing to worker */
+
+ int64_t pad1 __rte_cache_aligned; /* <= one cache line */
+
+ volatile int64_t retptr64[RTE_DIST_BURST_SIZE]
+ __rte_cache_aligned; /* <= incoming from worker */
+
+ int64_t pad2 __rte_cache_aligned; /* <= one cache line */
+
+ int count __rte_cache_aligned; /* <= number of current mbufs */
+};
+
+struct rte_distributor_backlog {
+ unsigned int start;
+ unsigned int count;
+ int64_t pkts[RTE_DIST_BURST_SIZE] __rte_cache_aligned;
+ uint16_t *tags; /* will point to second cacheline of inflights */
+} __rte_cache_aligned;
+
+struct rte_distributor_returned_pkts {
+ unsigned int start;
+ unsigned int count;
+ struct rte_mbuf *mbufs[RTE_DISTRIB_MAX_RETURNS];
+};
+
+struct rte_distributor {
+ TAILQ_ENTRY(rte_distributor) next; /**< Next in list. */
+
+ char name[RTE_DISTRIBUTOR_NAMESIZE]; /**< Name of the ring. */
+ unsigned int num_workers; /**< Number of workers polling */
+
+ /**>
+ * First cache line in the this array are the tags inflight
+ * on the worker core. Second cache line are the backlog
+ * that are going to go to the worker core.
+ */
+ uint16_t in_flight_tags[RTE_DISTRIB_MAX_WORKERS][RTE_DIST_BURST_SIZE*2]
+ __rte_cache_aligned;
+
+ struct rte_distributor_backlog backlog[RTE_DISTRIB_MAX_WORKERS]
+ __rte_cache_aligned;
+
+ struct rte_distributor_buffer bufs[RTE_DISTRIB_MAX_WORKERS];
+
+ struct rte_distributor_returned_pkts returns;
+};
+
+TAILQ_HEAD(rte_distributor_list, rte_distributor);
+
+static struct rte_tailq_elem rte_distributor_tailq = {
+ .name = "RTE_DISTRIBUTOR",
+};
+EAL_REGISTER_TAILQ(rte_distributor_tailq)
+
+/**** APIs called by workers ****/
+
+/**** Burst Packet APIs called by workers ****/
+
+/* This function should really be called return_pkt_burst() */
+void
+rte_distributor_request_pkt_burst(struct rte_distributor *d,
+ unsigned int worker_id, struct rte_mbuf **oldpkt,
+ unsigned int count)
+{
+ struct rte_distributor_buffer *buf = &(d->bufs[worker_id]);
+ unsigned int i;
+
+ volatile int64_t *retptr64;
+
+ retptr64 = &(buf->retptr64[0]);
+ /* Spin while handshake bits are set (scheduler clears it) */
+ while (unlikely(*retptr64 & RTE_DISTRIB_GET_BUF)) {
+ rte_pause();
+ uint64_t t = __rdtsc()+100;
+
+ while (__rdtsc() < t)
+ rte_pause();
+ }
+
+ /*
+ * OK, if we've got here, then the scheduler has just cleared the
+ * handshake bits. Populate the retptrs with returning packets.
+ */
+
+ for (i = count; i < RTE_DIST_BURST_SIZE; i++)
+ buf->retptr64[i] = 0;
+
+ /* Set Return bit for each packet returned */
+ for (i = count; i-- > 0; )
+ buf->retptr64[i] =
+ (((int64_t)(uintptr_t)(oldpkt[i])) <<
+ RTE_DISTRIB_FLAG_BITS) | RTE_DISTRIB_RETURN_BUF;
+
+ /*
+ * Finally, set the GET_BUF to signal to distributor that cache
+ * line is ready for processing
+ */
+ *retptr64 |= RTE_DISTRIB_GET_BUF;
+}
+
+int
+rte_distributor_poll_pkt_burst(struct rte_distributor *d,
+ unsigned int worker_id, struct rte_mbuf **pkts)
+{
+ struct rte_distributor_buffer *buf = &d->bufs[worker_id];
+ uint64_t ret;
+ int count = 0;
+
+ /* If bit is set, return */
+ if (buf->bufptr64[0] & RTE_DISTRIB_GET_BUF)
+ return 0;
+
+ /* since bufptr64 is signed, this should be an arithmetic shift */
+ for (unsigned int i = 0; i < RTE_DIST_BURST_SIZE; i++) {
+ if (likely(buf->bufptr64[i] & RTE_DISTRIB_VALID_BUF)) {
+ ret = buf->bufptr64[i] >> RTE_DISTRIB_FLAG_BITS;
+ pkts[count++] = (struct rte_mbuf *)((uintptr_t)(ret));
+ }
+ }
+
+ /*
+ * so now we've got the contents of the cacheline into an array of
+ * mbuf pointers, so toggle the bit so scheduler can start working
+ * on the next cacheline while we're working.
+ */
+ buf->bufptr64[0] |= RTE_DISTRIB_GET_BUF;
+
+
+ return count;
+}
+
+int
+rte_distributor_get_pkt_burst(struct rte_distributor *d,
+ unsigned int worker_id, struct rte_mbuf **pkts,
+ struct rte_mbuf **oldpkt, unsigned int return_count)
+{
+ unsigned int count;
+ uint64_t retries = 0;
+
+ rte_distributor_request_pkt_burst(d, worker_id, oldpkt, return_count);
+
+ count = rte_distributor_poll_pkt_burst(d, worker_id, pkts);
+ while (count == 0) {
+ rte_pause();
+ retries++;
+ if (retries > 1000) {
+ retries = 0;
+ return 0;
+ }
+ uint64_t t = __rdtsc()+100;
+
+ while (__rdtsc() < t)
+ rte_pause();
+
+ count = rte_distributor_poll_pkt_burst(d, worker_id, pkts);
+ }
+ return count;
+}
+
+int
+rte_distributor_return_pkt_burst(struct rte_distributor *d,
+ unsigned int worker_id, struct rte_mbuf **oldpkt, int num)
+{
+ struct rte_distributor_buffer *buf = &d->bufs[worker_id];
+ unsigned int i;
+
+ for (i = 0; i < RTE_DIST_BURST_SIZE; i++)
+ /* Switch off the return bit first */
+ buf->retptr64[i] &= ~RTE_DISTRIB_RETURN_BUF;
+
+ for (i = num; i-- > 0; )
+ buf->retptr64[i] = (((int64_t)(uintptr_t)oldpkt[i]) <<
+ RTE_DISTRIB_FLAG_BITS) | RTE_DISTRIB_RETURN_BUF;
+
+ /* set the GET_BUF but even if we got no returns */
+ buf->retptr64[0] |= RTE_DISTRIB_GET_BUF;
+
+ return 0;
+}
+
+/**** APIs called on distributor core ***/
+
+/* stores a packet returned from a worker inside the returns array */
+static inline void
+store_return(uintptr_t oldbuf, struct rte_distributor *d,
+ unsigned int *ret_start, unsigned int *ret_count)
+{
+ if (!oldbuf)
+ return;
+ /* store returns in a circular buffer */
+ d->returns.mbufs[(*ret_start + *ret_count) & RTE_DISTRIB_RETURNS_MASK]
+ = (void *)oldbuf;
+ *ret_start += (*ret_count == RTE_DISTRIB_RETURNS_MASK);
+ *ret_count += (*ret_count != RTE_DISTRIB_RETURNS_MASK);
+}
+
+
+static void
+find_match(struct rte_distributor *d,
+ uint16_t *data_ptr,
+ uint16_t *output_ptr)
+{
+ /* Setup */
+ __m128i incoming_fids;
+ __m128i inflight_fids;
+ __m128i preflight_fids;
+ __m128i wkr;
+ __m128i mask1;
+ __m128i mask2;
+ __m128i output;
+ struct rte_distributor_backlog *bl;
+
+ /*
+ * Function overview:
+ * 2. Loop through all worker ID's
+ * 2a. Load the current inflights for that worker into an xmm reg
+ * 2b. Load the current backlog for that worker into an xmm reg
+ * 2c. use cmpestrm to intersect flow_ids with backlog and inflights
+ * 2d. Add any matches to the output
+ * 3. Write the output xmm (matching worker ids).
+ */
+
+
+ output = _mm_set1_epi16(0);
+ incoming_fids = _mm_load_si128((__m128i *)data_ptr);
+
+ for (uint16_t i = 0; i < d->num_workers; i++) {
+ bl = &d->backlog[i];
+
+ inflight_fids =
+ _mm_load_si128((__m128i *)&(d->in_flight_tags[i]));
+ preflight_fids =
+ _mm_load_si128((__m128i *)(bl->tags));
+
+ /*
+ * Any incoming_fid that exists anywhere in inflight_fids will
+ * have 0xffff in same position of the mask as the incoming fid
+ * Example (shortened to bytes for brevity):
+ * incoming_fids 0x01 0x02 0x03 0x04 0x05 0x06 0x07 0x08
+ * inflight_fids 0x03 0x05 0x07 0x00 0x00 0x00 0x00 0x00
+ * mask 0x00 0x00 0xff 0x00 0xff 0x00 0xff 0x00
+ */
+
+ mask1 = _mm_cmpestrm(inflight_fids, 8, incoming_fids, 8,
+ _SIDD_UWORD_OPS |
+ _SIDD_CMP_EQUAL_ANY |
+ _SIDD_UNIT_MASK);
+ mask2 = _mm_cmpestrm(preflight_fids, 8, incoming_fids, 8,
+ _SIDD_UWORD_OPS |
+ _SIDD_CMP_EQUAL_ANY |
+ _SIDD_UNIT_MASK);
+
+ mask1 = _mm_or_si128(mask1, mask2);
+ /*
+ * Now mask contains 0xffff where there's a match.
+ * Next we need to store the worker_id in the relevant position
+ * in the output.
+ */
+
+ wkr = _mm_set1_epi16(i+1);
+ mask1 = _mm_and_si128(mask1, wkr);
+ output = _mm_or_si128(mask1, output);
+ }
+
+ /*
+ * At this stage, the output 128-bit contains 8 16-bit values, with
+ * each non-zero value containing the worker ID on which the
+ * corresponding flow is pinned to.
+ */
+ _mm_store_si128((__m128i *)output_ptr, output);
+}
+
+
+static unsigned int
+release(struct rte_distributor *d, unsigned int wkr)
+{
+ uintptr_t oldbuf;
+ unsigned int ret_start = d->returns.start,
+ ret_count = d->returns.count;
+ struct rte_distributor_buffer *buf = &(d->bufs[wkr]);
+ unsigned int i;
+
+ if (d->backlog[wkr].count == 0)
+ return 0;
+
+ /*
+ * wait for the GET_BUF bit to go high, otherwise we can't send
+ * the packets to the worker
+ */
+ while (!(d->bufs[wkr].bufptr64[0] & RTE_DISTRIB_GET_BUF))
+ rte_pause();
+
+ if (buf->retptr64[0] & RTE_DISTRIB_GET_BUF) {
+ for (unsigned int i = 0; i < RTE_DIST_BURST_SIZE; i++) {
+ if (buf->retptr64[i] & RTE_DISTRIB_RETURN_BUF) {
+ oldbuf = ((uintptr_t)(buf->retptr64[i] >>
+ RTE_DISTRIB_FLAG_BITS));
+ /* store returns in a circular buffer */
+ store_return(oldbuf, d, &ret_start, &ret_count);
+ buf->retptr64[i] &= ~RTE_DISTRIB_RETURN_BUF;
+ }
+ }
+ d->returns.start = ret_start;
+ d->returns.count = ret_count;
+ /* Clear for the worker to populate with more returns */
+ buf->retptr64[0] = 0;
+ }
+
+ buf->count = 0;
+
+ for (i = 0; i < d->backlog[wkr].count; i++) {
+ d->bufs[wkr].bufptr64[i] = d->backlog[wkr].pkts[i] |
+ RTE_DISTRIB_GET_BUF | RTE_DISTRIB_VALID_BUF;
+ d->in_flight_tags[wkr][i] = d->backlog[wkr].tags[i];
+ }
+ buf->count = i;
+ for ( ; i < RTE_DIST_BURST_SIZE ; i++) {
+ buf->bufptr64[i] = RTE_DISTRIB_GET_BUF;
+ d->in_flight_tags[wkr][i] = 0;
+ }
+ d->backlog[wkr].count = 0;
+
+ /* Clear the GET bit */
+ buf->bufptr64[0] &= ~RTE_DISTRIB_GET_BUF;
+ return buf->count;
+
+}
+
+
+/* process a set of packets to distribute them to workers */
+int
+rte_distributor_process_burst(struct rte_distributor *d,
+ struct rte_mbuf **mbufs, unsigned int num_mbufs)
+{
+ unsigned int next_idx = 0;
+ static unsigned int wkr;
+ struct rte_mbuf *next_mb = NULL;
+ int64_t next_value = 0;
+ uint16_t new_tag = 0;
+ uint16_t flows[8] __rte_cache_aligned;
+
+ if (unlikely(num_mbufs == 0)) {
+ /* Flush out all non-full cache-lines to workers. */
+ for (unsigned int wid = 0 ; wid < d->num_workers; wid++) {
+ if ((d->bufs[wid].bufptr64[0] & RTE_DISTRIB_GET_BUF))
+ release(d, wid);
+ }
+ return 0;
+ }
+
+ while (next_idx < num_mbufs) {
+ uint16_t matches[8];
+ int pkts;
+
+ if (d->bufs[wkr].bufptr64[0] & RTE_DISTRIB_GET_BUF)
+ d->bufs[wkr].count = 0;
+
+ for (unsigned int i = 0; i < RTE_DIST_BURST_SIZE; i++) {
+ if (mbufs[next_idx + i]) {
+ /* flows have to be non-zero */
+ flows[i] = mbufs[next_idx + i]->hash.usr | 1;
+ } else
+ flows[i] = 0;
+ }
+
+ find_match(d, &flows[0], &matches[0]);
+
+ /*
+ * Matches array now contain the intended worker ID (+1) of
+ * the incoming packets. Any zeroes need to be assigned
+ * workers.
+ */
+
+ if ((num_mbufs - next_idx) < RTE_DIST_BURST_SIZE)
+ pkts = num_mbufs - next_idx;
+ else
+ pkts = RTE_DIST_BURST_SIZE;
+
+ for (int j = 0; j < pkts; j++) {
+
+ next_mb = mbufs[next_idx++];
+ next_value = (((int64_t)(uintptr_t)next_mb) <<
+ RTE_DISTRIB_FLAG_BITS);
+ /*
+ * User is advocated to set tag vaue for each
+ * mbuf before calling rte_distributor_process.
+ * User defined tags are used to identify flows,
+ * or sessions.
+ */
+ /* flows MUST be non-zero */
+ new_tag = (uint16_t)(next_mb->hash.usr) | 1;
+
+ /*
+ * Using the next line will cause the find_match
+ * function to be optimised out, making this function
+ * do parallel (non-atomic) distribution
+ */
+ //matches[j] = 0;
+
+ if (matches[j]) {
+ struct rte_distributor_backlog *bl =
+ &d->backlog[matches[j]-1];
+ if (unlikely(bl->count == RTE_DIST_BURST_SIZE))
+ release(d, matches[j]-1);
+
+ /* Add to worker that already has flow */
+ unsigned int idx = bl->count++;
+
+ bl->tags[idx] = new_tag;
+ bl->pkts[idx] = next_value;
+
+ } else {
+ struct rte_distributor_backlog *bl =
+ &d->backlog[wkr];
+ if (unlikely(bl->count == RTE_DIST_BURST_SIZE))
+ release(d, wkr);
+
+ /* Add to current worker worker */
+ unsigned int idx = bl->count++;
+
+ bl->tags[idx] = new_tag;
+ bl->pkts[idx] = next_value;
+ /*
+ * Now that we've just added an unpinned flow
+ * to a worker, we need to ensure that all
+ * other packets with that same flow will go
+ * to the same worker in this burst.
+ */
+ for (int w = j; w < pkts; w++)
+ if (flows[w] == new_tag)
+ matches[w] = wkr+1;
+ }
+ }
+ wkr++;
+ if (wkr >= d->num_workers)
+ wkr = 0;
+ }
+
+ /* Flush out all non-full cache-lines to workers. */
+ for (unsigned int wid = 0 ; wid < d->num_workers; wid++) {
+ if ((d->bufs[wid].bufptr64[0] & RTE_DISTRIB_GET_BUF))
+ release(d, wid);
+ }
+
+ return num_mbufs;
+}
+
+/* return to the caller, packets returned from workers */
+int
+rte_distributor_returned_pkts_burst(struct rte_distributor *d,
+ struct rte_mbuf **mbufs, unsigned int max_mbufs)
+{
+ struct rte_distributor_returned_pkts *returns = &d->returns;
+ unsigned int retval = (max_mbufs < returns->count) ?
+ max_mbufs : returns->count;
+ unsigned int i;
+
+ for (i = 0; i < retval; i++) {
+ unsigned int idx = (returns->start + i) &
+ RTE_DISTRIB_RETURNS_MASK;
+
+ mbufs[i] = returns->mbufs[idx];
+ }
+ returns->start += i;
+ returns->count -= i;
+
+ return retval;
+}
+
+/*
+ * Return the number of packets in-flight in a distributor, i.e. packets
+ * being workered on or queued up in a backlog.
+ */
+static inline unsigned int
+total_outstanding(const struct rte_distributor *d)
+{
+ unsigned int wkr, total_outstanding = 0;
+
+ for (wkr = 0; wkr < d->num_workers; wkr++)
+ total_outstanding += d->backlog[wkr].count;
+
+ return total_outstanding;
+}
+
+/*
+ * Flush the distributor, so that there are no outstanding packets in flight or
+ * queued up.
+ */
+int
+rte_distributor_flush_burst(struct rte_distributor *d)
+{
+ const unsigned int flushed = total_outstanding(d);
+
+ while (total_outstanding(d) > 0)
+ rte_distributor_process_burst(d, NULL, 0);
+
+ return flushed;
+}
+
+/* clears the internal returns array in the distributor */
+void
+rte_distributor_clear_returns_burst(struct rte_distributor *d)
+{
+ /* throw away returns, so workers can exit */
+ for (unsigned int wkr = 0; wkr < d->num_workers; wkr++)
+ d->bufs[wkr].retptr64[0] = 0;
+}
+
+/* creates a distributor instance */
+struct rte_distributor *
+rte_distributor_create_burst(const char *name,
+ unsigned int socket_id,
+ unsigned int num_workers)
+{
+ struct rte_distributor *d;
+ struct rte_distributor_list *distributor_list;
+ char mz_name[RTE_MEMZONE_NAMESIZE];
+ const struct rte_memzone *mz;
+
+ /* compilation-time checks */
+ RTE_BUILD_BUG_ON((sizeof(*d) & RTE_CACHE_LINE_MASK) != 0);
+ RTE_BUILD_BUG_ON((RTE_DISTRIB_MAX_WORKERS & 7) != 0);
+
+ if (name == NULL || num_workers >= RTE_DISTRIB_MAX_WORKERS) {
+ rte_errno = EINVAL;
+ return NULL;
+ }
+
+ snprintf(mz_name, sizeof(mz_name), RTE_DISTRIB_PREFIX"%s", name);
+ mz = rte_memzone_reserve(mz_name, sizeof(*d), socket_id, NO_FLAGS);
+ if (mz == NULL) {
+ rte_errno = ENOMEM;
+ return NULL;
+ }
+
+ d = mz->addr;
+ snprintf(d->name, sizeof(d->name), "%s", name);
+ d->num_workers = num_workers;
+
+ /*
+ * Set up the backog tags so they're pointing at the second cache
+ * line for performance during flow matching
+ */
+ for (unsigned int i = 0 ; i < num_workers ; i++)
+ d->backlog[i].tags = &d->in_flight_tags[i][RTE_DIST_BURST_SIZE];
+
+ distributor_list = RTE_TAILQ_CAST(rte_distributor_tailq.head,
+ rte_distributor_list);
+
+ rte_rwlock_write_lock(RTE_EAL_TAILQ_RWLOCK);
+ TAILQ_INSERT_TAIL(distributor_list, d, next);
+ rte_rwlock_write_unlock(RTE_EAL_TAILQ_RWLOCK);
+
+ return d;
+}
diff --git a/lib/librte_distributor/rte_distributor_burst.h b/lib/librte_distributor/rte_distributor_burst.h
new file mode 100644
index 0000000..1437657
--- /dev/null
+++ b/lib/librte_distributor/rte_distributor_burst.h
@@ -0,0 +1,255 @@
+/*-
+ * BSD LICENSE
+ *
+ * Copyright(c) 2016 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 _RTE_DISTRIBUTE_H_
+#define _RTE_DISTRIBUTE_H_
+
+/**
+ * @file
+ * RTE distributor
+ *
+ * The distributor is a component which is designed to pass packets
+ * one-at-a-time to workers, with dynamic load balancing.
+ */
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+struct rte_distributor;
+struct rte_mbuf;
+
+/**
+ * Function to create a new distributor instance
+ *
+ * Reserves the memory needed for the distributor operation and
+ * initializes the distributor to work with the configured number of workers.
+ *
+ * @param name
+ * The name to be given to the distributor instance.
+ * @param socket_id
+ * The NUMA node on which the memory is to be allocated
+ * @param num_workers
+ * The maximum number of workers that will request packets from this
+ * distributor
+ * @return
+ * The newly created distributor instance
+ */
+struct rte_distributor *
+rte_distributor_create_burst(const char *name, unsigned int socket_id,
+ unsigned int num_workers);
+
+/* *** APIS to be called on the distributor lcore *** */
+/*
+ * The following APIs are the public APIs which are designed for use on a
+ * single lcore which acts as the distributor lcore for a given distributor
+ * instance. These functions cannot be called on multiple cores simultaneously
+ * without using locking to protect access to the internals of the distributor.
+ *
+ * NOTE: a given lcore cannot act as both a distributor lcore and a worker lcore
+ * for the same distributor instance, otherwise deadlock will result.
+ */
+
+/**
+ * Process a set of packets by distributing them among workers that request
+ * packets. The distributor will ensure that no two packets that have the
+ * same flow id, or tag, in the mbuf will be processed on different cores at
+ * the same time.
+ *
+ * The user is advocated to set tag for each mbuf before calling this function.
+ * If user doesn't set the tag, the tag value can be various values depending on
+ * driver implementation and configuration.
+ *
+ * This is not multi-thread safe and should only be called on a single lcore.
+ *
+ * @param d
+ * The distributor instance to be used
+ * @param mbufs
+ * The mbufs to be distributed
+ * @param num_mbufs
+ * The number of mbufs in the mbufs array
+ * @return
+ * The number of mbufs processed.
+ */
+int
+rte_distributor_process_burst(struct rte_distributor *d,
+ struct rte_mbuf **mbufs, unsigned int num_mbufs);
+
+/**
+ * Get a set of mbufs that have been returned to the distributor by workers
+ *
+ * This should only be called on the same lcore as rte_distributor_process()
+ *
+ * @param d
+ * The distributor instance to be used
+ * @param mbufs
+ * The mbufs pointer array to be filled in
+ * @param max_mbufs
+ * The size of the mbufs array
+ * @return
+ * The number of mbufs returned in the mbufs array.
+ */
+int
+rte_distributor_returned_pkts_burst(struct rte_distributor *d,
+ struct rte_mbuf **mbufs, unsigned int max_mbufs);
+
+/**
+ * Flush the distributor component, so that there are no in-flight or
+ * backlogged packets awaiting processing
+ *
+ * This should only be called on the same lcore as rte_distributor_process()
+ *
+ * @param d
+ * The distributor instance to be used
+ * @return
+ * The number of queued/in-flight packets that were completed by this call.
+ */
+int
+rte_distributor_flush_burst(struct rte_distributor *d);
+
+/**
+ * Clears the array of returned packets used as the source for the
+ * rte_distributor_returned_pkts() API call.
+ *
+ * This should only be called on the same lcore as rte_distributor_process()
+ *
+ * @param d
+ * The distributor instance to be used
+ */
+void
+rte_distributor_clear_returns_burst(struct rte_distributor *d);
+
+/* *** APIS to be called on the worker lcores *** */
+/*
+ * The following APIs are the public APIs which are designed for use on
+ * multiple lcores which act as workers for a distributor. Each lcore should use
+ * a unique worker id when requesting packets.
+ *
+ * NOTE: a given lcore cannot act as both a distributor lcore and a worker lcore
+ * for the same distributor instance, otherwise deadlock will result.
+ */
+
+/**
+ * API called by a worker to get new packets to process. Any previous packets
+ * given to the worker is assumed to have completed processing, and may be
+ * optionally returned to the distributor via the oldpkt parameter.
+ *
+ * @param d
+ * The distributor instance to be used
+ * @param worker_id
+ * The worker instance number to use - must be less that num_workers passed
+ * at distributor creation time.
+ * @param pkts
+ * The mbufs pointer array to be filled in (up to 8 packets)
+ * @param oldpkt
+ * The previous packet, if any, being processed by the worker
+ * @param retcount
+ * The number of packets being returned
+ *
+ * @return
+ * The number of packets in the pkts array
+ */
+int
+rte_distributor_get_pkt_burst(struct rte_distributor *d,
+ unsigned int worker_id, struct rte_mbuf **pkts,
+ struct rte_mbuf **oldpkt, unsigned int retcount);
+
+/**
+ * API called by a worker to return a completed packet without requesting a
+ * new packet, for example, because a worker thread is shutting down
+ *
+ * @param d
+ * The distributor instance to be used
+ * @param worker_id
+ * The worker instance number to use - must be less that num_workers passed
+ * at distributor creation time.
+ * @param mbuf
+ * The previous packet being processed by the worker
+ */
+int
+rte_distributor_return_pkt_burst(struct rte_distributor *d,
+ unsigned int worker_id, struct rte_mbuf **oldpkt, int num);
+
+/**
+ * API called by a worker to request a new packet to process.
+ * Any previous packet given to the worker is assumed to have completed
+ * processing, and may be optionally returned to the distributor via
+ * the oldpkt parameter.
+ * Unlike rte_distributor_get_pkt_burst(), this function does not wait for a
+ * new packet to be provided by the distributor.
+ *
+ * NOTE: after calling this function, rte_distributor_poll_pkt_burst() should
+ * be used to poll for the packet requested. The rte_distributor_get_pkt_burst()
+ * API should *not* be used to try and retrieve the new packet.
+ *
+ * @param d
+ * The distributor instance to be used
+ * @param worker_id
+ * The worker instance number to use - must be less that num_workers passed
+ * at distributor creation time.
+ * @param oldpkt
+ * The returning packets, if any, processed by the worker
+ * @param count
+ * The number of returning packets
+ */
+void
+rte_distributor_request_pkt_burst(struct rte_distributor *d,
+ unsigned int worker_id, struct rte_mbuf **oldpkt,
+ unsigned int count);
+
+/**
+ * API called by a worker to check for a new packet that was previously
+ * requested by a call to rte_distributor_request_pkt(). It does not wait
+ * for the new packet to be available, but returns NULL if the request has
+ * not yet been fulfilled by the distributor.
+ *
+ * @param d
+ * The distributor instance to be used
+ * @param worker_id
+ * The worker instance number to use - must be less that num_workers passed
+ * at distributor creation time.
+ * @param mbufs
+ * The array of mbufs being given to the worker
+ *
+ * @return
+ * The number of packets being given to the worker thread, zero if no
+ * packet is yet available.
+ */
+int
+rte_distributor_poll_pkt_burst(struct rte_distributor *d,
+ unsigned int worker_id, struct rte_mbuf **mbufs);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
diff --git a/lib/librte_distributor/rte_distributor_common.h b/lib/librte_distributor/rte_distributor_common.h
new file mode 100644
index 0000000..e2a9c3e
--- /dev/null
+++ b/lib/librte_distributor/rte_distributor_common.h
@@ -0,0 +1,75 @@
+/*-
+ * BSD LICENSE
+ *
+ * Copyright(c) 2016 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 _RTE_DIST_COMMON_H_
+#define _RTE_DIST_COMMON_H_
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define NO_FLAGS 0
+#define RTE_DISTRIB_PREFIX "DT_"
+
+/*
+ * We will use the bottom four bits of pointer for flags, shifting out
+ * the top four bits to make room (since a 64-bit pointer actually only uses
+ * 48 bits). An arithmetic-right-shift will then appropriately restore the
+ * original pointer value with proper sign extension into the top bits.
+ */
+#define RTE_DISTRIB_FLAG_BITS 4
+#define RTE_DISTRIB_FLAGS_MASK (0x0F)
+#define RTE_DISTRIB_NO_BUF 0 /**< empty flags: no buffer requested */
+#define RTE_DISTRIB_GET_BUF (1) /**< worker requests a buffer, returns old */
+#define RTE_DISTRIB_RETURN_BUF (2) /**< worker returns a buffer, no request */
+#define RTE_DISTRIB_VALID_BUF (4) /**< set if bufptr contains ptr */
+
+#define RTE_DISTRIB_BACKLOG_SIZE 8
+#define RTE_DISTRIB_BACKLOG_MASK (RTE_DISTRIB_BACKLOG_SIZE - 1)
+
+#define RTE_DISTRIB_MAX_RETURNS 128
+#define RTE_DISTRIB_RETURNS_MASK (RTE_DISTRIB_MAX_RETURNS - 1)
+
+/**
+ * Maximum number of workers allowed.
+ * Be aware of increasing the limit, becaus it is limited by how we track
+ * in-flight tags. See @in_flight_bitmask and @rte_distributor_process
+ */
+#define RTE_DISTRIB_MAX_WORKERS 64
+
+#define RTE_DISTRIBUTOR_NAMESIZE 32 /**< Length of name for instance */
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
--
2.7.4
^ permalink raw reply related
* [PATCH v1 1/2] distributor lib performance enhancements
From: David Hunt @ 2016-12-01 4:50 UTC (permalink / raw)
To: dev; +Cc: bruce.richardson
This patch aims to improve the throughput of the distributor library.
It uses a similar handshake mechanism to the previous version of
the library, in that bits are used to indicate when packets are ready
to be sent to a worker and ready to be returned from a worker. One main
difference is that instead of sending one packet in a cache line, it makes
use of the 7 free spaces in the same cache line in order to send up to
8 packets at a time to/from a worker.
The flow matching algorithm has had significant re-work, and now keeps an
array of inflight flows and an array of backlog flows, and matches incoming
flows to the inflight/backlog flows of all workers so that flow pinning to
workers can be maintained.
Notes:
Apps must now work in bursts, as up to 8 are given to a worker at a time
For performance in matching, Flow ID's are 15-bits
Original API (and code) is kept for backward compatibility
Performance Gains
2.2GHz Intel(R) Xeon(R) CPU E5-2699 v4 @ 2.20GHz
2 x XL710 40GbE NICS to 2 x 40Gbps traffic generator channels 64b packets
separate cores for rx, tx, distributor
1 worker - 4.8x
4 workers - 2.9x
8 workers - 1.8x
12 workers - 2.1x
16 workers - 1.8x
[PATCH v1 1/2] lib: distributor performance enhancements
[PATCH v1 2/2] example: distributor app modified to use burstAPI
^ permalink raw reply
* [PATCH v2] vhost: fix add_guest_pages bug
From: Haifeng Lin @ 2016-12-01 11:42 UTC (permalink / raw)
To: dev, yuanhan.liu, maxime.coquelin; +Cc: jerry.lilijun
When reg_size < page_size the function read in
rte_mem_virt2phy would not return, becausue
host_user_addr is invalid.
Signed-off-by: Haifeng Lin <haifeng.lin@huawei.com>
---
v2:
fix TYPO_SPELLING warning
---
lib/librte_vhost/vhost_user.c | 10 +++++-----
1 file changed, 5 insertions(+), 5 deletions(-)
diff --git a/lib/librte_vhost/vhost_user.c b/lib/librte_vhost/vhost_user.c
index 6b83c15..ce55e85 100644
--- a/lib/librte_vhost/vhost_user.c
+++ b/lib/librte_vhost/vhost_user.c
@@ -447,14 +447,14 @@ add_guest_pages(struct virtio_net *dev, struct virtio_memory_region *reg,
reg_size -= size;
while (reg_size > 0) {
+ size = reg_size >= page_size ? page_size : reg_size;
host_phys_addr = rte_mem_virt2phy((void *)(uintptr_t)
host_user_addr);
- add_one_guest_page(dev, guest_phys_addr, host_phys_addr,
- page_size);
+ add_one_guest_page(dev, guest_phys_addr, host_phys_addr, size);
- host_user_addr += page_size;
- guest_phys_addr += page_size;
- reg_size -= page_size;
+ host_user_addr += size;
+ guest_phys_addr += size;
+ reg_size -= size;
}
}
--
1.8.3.1
^ permalink raw reply related
* Re: [PATCH v2] i40e: Fix eth_i40e_dev_init sequence on ThunderX
From: Ananyev, Konstantin @ 2016-12-01 11:38 UTC (permalink / raw)
To: Jerin Jacob
Cc: Richardson, Bruce, Satha Rao, Zhang, Helin, Wu, Jingjing,
jianbo.liu@linaro.org, dev@dpdk.org
In-Reply-To: <20161130205455.GA5535@svelivela-lt.caveonetworks.com>
Hi Jerin,
> > > > > > >
> > > > > > > i40e_asq_send_command: rd32 & wr32 under ThunderX gives unpredictable
> > > > > > > results. To solve this include rte memory barriers
> > > > > > >
> > > > > > > Signed-off-by: Satha Rao <skoteshwar@caviumnetworks.com>
> > > > > > > ---
> > > > > > > drivers/net/i40e/base/i40e_osdep.h | 14 ++++++++++++++
> > > > > > > 1 file changed, 14 insertions(+)
> > > > > > >
> > > > > > > diff --git a/drivers/net/i40e/base/i40e_osdep.h b/drivers/net/i40e/base/i40e_osdep.h
> > > > > > > index 38e7ba5..ffa3160 100644
> > > > > > > --- a/drivers/net/i40e/base/i40e_osdep.h
> > > > > > > +++ b/drivers/net/i40e/base/i40e_osdep.h
> > > > > > > @@ -158,7 +158,13 @@ do { \
> > > > > > > ((volatile uint32_t *)((char *)(a)->hw_addr + (reg)))
> > > > > > > static inline uint32_t i40e_read_addr(volatile void *addr)
> > > > > > > {
> > > > > > > +#if defined(RTE_ARCH_ARM64)
> > > > > > > + uint32_t val = rte_le_to_cpu_32(I40E_PCI_REG(addr));
> > > > > > > + rte_rmb();
> > > > > > > + return val;
> > > > > >
> > > > > > If you really need an rmb/wmb with MMIO read/writes on ARM,
> > > > > > I think you can avoid #ifdefs here and use rte_smp_rmb/rte_smp_wmb.
> > > > > > BTW, I suppose if you need it for i40e, you would need it for other devices too.
> > > > >
> > > > > Yes. ARM would need for all devices(typically, the devices on external PCI bus).
> > > > > I guess rte_smp_rmb may not be the correct abstraction. So we need more of
> > > > > rte_rmb() as we need only non smp variant on IO side. I guess then it make sense to
> > > > > create new abstraction in eal with following variants so that each arch
> > > > > gets opportunity to make what it makes sense that specific platform
> > > > >
> > > > > rte_readb_relaxed
> > > > > rte_readw_relaxed
> > > > > rte_readl_relaxed
> > > > > rte_readq_relaxed
> > > > > rte_writeb_relaxed
> > > > > rte_writew_relaxed
> > > > > rte_writel_relaxed
> > > > > rte_writeq_relaxed
> > > > > rte_readb
> > > > > rte_readw
> > > > > rte_readl
> > > > > rte_readq
> > > > > rte_writeb
> > > > > rte_writew
> > > > > rte_writel
> > > > > rte_writeq
> > > > >
> > > > > Thoughts ?
> > > > >
> > > >
> > > > That seems like a lot of API calls!
> > > > Perhaps you can clarify - why would the rte_smp_rmb() not work for you?
> > >
> > > Currently arm64 mapped DMB as rte_smp_rmb() for smp case.
> > >
> > > Ideally for io barrier and non smp case, we need to map it as DSB and it is
> > > bit heavier than DMB
> >
> > Ok, so you need some new macro, like rte_io_(r|w)mb or so, that would expand into dmb
> > for ARM, correct?
>
> The io barrier expands to dsb.
> http://lxr.free-electrons.com/source/arch/arm64/include/asm/io.h#L110
Sorry, yes I meant DSB here.
>
> >
> > >
> > > The linux kernel arm64 mappings
> > > http://lxr.free-electrons.com/source/arch/arm64/include/asm/io.h#L142
> > >
> > > DMB vs DSB
> > > https://community.arm.com/thread/3833
> > >
> > > The relaxed one are without any barriers.(the use case like accessing on
> > > chip peripherals may need only relaxed versions)
> > >
> > > Thoughts on new rte EAL abstraction?
> >
> > Looks like a lot of macros but if you guys think that would help - NP with that :)
>
> I don't have strong opinion here. If there is concern on a lot of macros
> then, I can introduce only "rte_io_(r|w)mb" instead of read[b|w|l|q]/write[b|w|l|q]/relaxed.
> let me know?
I think we can have both.
The question is in the amount of work need to be done.
>
> > Again, in that case we probably can get rid of driver specific pci reg read/write defines.
> Yes. But, That's going to have a lot of change :-(
Yes I agree, the changes would be quite significant.
>
> If there is no objection then I will introduce
> "read[b|w|l|q]/write[b|w|l|q]/relaxed" and then change all external pcie drivers
> with new macros.
That seems like a good idea to me.
Though as you said that seems quite a significant change.
Probably make sense to do it in 2 stages (just a suggestion):
First introduce rte_io_(r|w)mb and fix with it existing issues in the particular drivers.
Second replace existing PMD specific xxx_read/write_addr() with your new generic
Konstantin
^ permalink raw reply
* Re: Hyper-v support
From: Thomas Monjalon @ 2016-12-01 11:25 UTC (permalink / raw)
To: Varun, Stephen Hemminger; +Cc: dev
In-Reply-To: <2343421.PKDcrM7tSE@xps13>
(fixed the email, sorry)
2016-12-01 12:21, Thomas Monjalon:
> 2016-11-30 14:34, Varun:
> > Hi,
> >
> > I would like to know if the latest DPDK (16.11) supports hyper-v?
> >
> > I couldn't find any conclusive evidence online or in dpdk roadmap. Is it
> > likely that we see it in 17.05?
>
> Stephen did a presentation at the last DPDK userspace summit:
> https://dpdksummit.com/Archive/pdf/2016Userspace/Day01-Session03-StephenHemminger-Userspace2016.pdf
>
> Stephen, please, could you confirm the expected release for Hyper-V support?
> A patch for the roadmap page would be great:
> http://dpdk.org/dev/roadmap
^ permalink raw reply
* Re: Hyper-v support
From: Thomas Monjalon @ 2016-12-01 11:21 UTC (permalink / raw)
To: Varun, Stephen Hemminger; +Cc: dev
In-Reply-To: <CAG+bKvq5wvgCj6VWK_EgVqpDu6_xiYN8jyQLt1eeZ2PbjZdsoA@mail.gmail.com>
2016-11-30 14:34, Varun:
> Hi,
>
> I would like to know if the latest DPDK (16.11) supports hyper-v?
>
> I couldn't find any conclusive evidence online or in dpdk roadmap. Is it
> likely that we see it in 17.05?
Stephen did a presentation at the last DPDK userspace summit:
https://dpdksummit.com/Archive/pdf/2016Userspace/Day01-Session03-StephenHemminger-Userspace2016.pdf
Stephen, please, could you confirm the expected release for Hyper-V support?
A patch for the roadmap page would be great:
http://dpdk.org/dev/roadmap
^ permalink raw reply
* [PATCH v2] doc: add pdump library to API doxygen
From: Ferruh Yigit @ 2016-12-01 11:02 UTC (permalink / raw)
To: dev; +Cc: Reshma Pattan
In-Reply-To: <1479220879-7914-1-git-send-email-reshma.pattan@intel.com>
From: Reshma Pattan <reshma.pattan@intel.com>
Add pdump library to API doxygen.
Signed-off-by: Reshma Pattan <reshma.pattan@intel.com>
---
v2:
* Move pdump higher position in the index
---
doc/api/doxy-api-index.md | 1 +
doc/api/doxy-api.conf | 1 +
2 files changed, 2 insertions(+)
diff --git a/doc/api/doxy-api-index.md b/doc/api/doxy-api-index.md
index 6675f96..ed1a204 100644
--- a/doc/api/doxy-api-index.md
+++ b/doc/api/doxy-api-index.md
@@ -136,6 +136,7 @@ There are many libraries, so their headers may be grouped by topics:
- **debug**:
[jobstats] (@ref rte_jobstats.h),
+ [pdump] (@ref rte_pdump.h),
[hexdump] (@ref rte_hexdump.h),
[debug] (@ref rte_debug.h),
[log] (@ref rte_log.h),
diff --git a/doc/api/doxy-api.conf b/doc/api/doxy-api.conf
index 9dc7ae5..b340fcf 100644
--- a/doc/api/doxy-api.conf
+++ b/doc/api/doxy-api.conf
@@ -51,6 +51,7 @@ INPUT = doc/api/doxy-api-index.md \
lib/librte_mempool \
lib/librte_meter \
lib/librte_net \
+ lib/librte_pdump \
lib/librte_pipeline \
lib/librte_port \
lib/librte_power \
--
2.9.3
^ permalink raw reply related
* Re: [PATCH v12 0/6] add Tx preparation
From: Ferruh Yigit @ 2016-12-01 10:50 UTC (permalink / raw)
To: Thomas Monjalon, Ananyev, Konstantin
Cc: dev, Jan Medala, Jakub Palider, Alejandro Lucero, Yuanhan Liu,
Yong Wang, Kulasek, TomaszX, Netanel Belgazal, Evgeny Schemeilin
In-Reply-To: <3517413.XL3bTbAyaC@xps13>
On 11/30/2016 6:26 PM, Thomas Monjalon wrote:
> 2016-11-30 17:42, Ananyev, Konstantin:
>>>> Please, we need a comment for each driver saying
>>>> "it is OK, we do not need any checksum preparation for TSO"
>>>> or
>>>> "yes we have to implement tx_prepare or TSO will not work in this mode"
>>>>
>>>
>>> qede PMD doesn’t currently support TSO yet, it only supports Tx TCP/UDP/IP
>>> csum offloads.
>>> So Tx preparation isn’t applicable. So as of now -
>>> "it is OK, we do not need any checksum preparation for TSO"
>>
>> Thanks for the answer.
>> Though please note that it not only for TSO.
>
> Oh yes, sorry, my wording was incorrect.
> We need to know if any checksum preparation is needed prior
> offloading its final computation to the hardware or driver.
> So the question applies to TSO and simple checksum offload.
>
> We are still waiting answers for
> bnxt, cxgbe, ena, nfp, thunderx, virtio and vmxnet3.
>
Remaining ones:
ena
nfp
virtio
vmxnet3
^ permalink raw reply
* Re: apply commit e30a0178d290a4e83dc01f9c2170d4859339c9cf "kni: support RHEL 7.3" to dpdk-stable?
From: Ferruh Yigit @ 2016-12-01 10:40 UTC (permalink / raw)
To: Roberts, Lee A., dev@dpdk.org, dpdk stable; +Cc: Yuanhan Liu
In-Reply-To: <DF4PR84MB0012FCE5E310ABB94415426EE58C0@DF4PR84MB0012.NAMPRD84.PROD.OUTLOOK.COM>
On 11/30/2016 8:54 PM, Roberts, Lee A. wrote:
> Does it make sense to apply the commit for "kni: support RHEL 7.3" (http://www.dpdk.org/browse/dpdk/commit/lib/librte_eal/linuxapp/kni/ethtool/igb/kcompat.h?id=e30a0178d290a4e83dc01f9c2170d4859339c9cf)
> to the stable tree to enable clean compilation on RHEL 7.3?
Yes, good idea, but these requests should go into stable mail list:
CC: stable@dpdk.org
As far as I can see that patch can be applied to stable tree without
modification, so I guess request is enough, no patch required, but
Yuanhan please correct me if this is wrong.
And 16.07.2 released yesterday, so this patch missed that release, not
sure when next stable release is scheduled.
>
> - Lee Roberts
>
^ permalink raw reply
* [PATCH] vhost: fix add_guest_pages bug
From: Haifeng Lin @ 2016-12-01 10:22 UTC (permalink / raw)
To: dev, yuanhan.liu, maxime.coquelin; +Cc: Haifeng Lin
When reg_size < page_size the function read in
rte_mem_virt2phy would not return, becasue gpa
is invalid.
Signed-off-by: Haifeng Lin <haifeng.lin@huawei.com>
---
lib/librte_vhost/vhost_user.c | 10 +++++-----
1 file changed, 5 insertions(+), 5 deletions(-)
diff --git a/lib/librte_vhost/vhost_user.c b/lib/librte_vhost/vhost_user.c
index 6b83c15..ce55e85 100644
--- a/lib/librte_vhost/vhost_user.c
+++ b/lib/librte_vhost/vhost_user.c
@@ -447,14 +447,14 @@ add_guest_pages(struct virtio_net *dev, struct virtio_memory_region *reg,
reg_size -= size;
while (reg_size > 0) {
+ size = reg_size >= page_size ? page_size : reg_size;
host_phys_addr = rte_mem_virt2phy((void *)(uintptr_t)
host_user_addr);
- add_one_guest_page(dev, guest_phys_addr, host_phys_addr,
- page_size);
+ add_one_guest_page(dev, guest_phys_addr, host_phys_addr, size);
- host_user_addr += page_size;
- guest_phys_addr += page_size;
- reg_size -= page_size;
+ host_user_addr += size;
+ guest_phys_addr += size;
+ reg_size -= size;
}
}
--
1.8.3.1
^ permalink raw reply related
page: next (older) | prev (newer) | latest
- recent:[subjects (threaded)|topics (new)|topics (active)]
This is a public inbox, see mirroring instructions
for how to clone and mirror all data and code used for this inbox