d/numam-dpdk
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

distributor: switch over to new API

Browse Source 
This is the main switch over between the legacy API and the new
burst API. We rename all the functions in rte_distributor.c to remove
the _v1705, and we add in _v20 in the rte_distributor_v20.c

We also rename the rte_distributor_next.h as rte_distributor.h, as
this is now the public header.

At the same time, we need the autotests and sample app to compile
properly, hence those changes are in this patch also.

Signed-off-by: David Hunt <david.hunt@intel.com>
Acked-by: Bruce Richardson <bruce.richardson@intel.com>
This commit is contained in:
David Hunt 2017-03-20 10:08:30 +00:00 committed by Thomas Monjalon
parent 063b886701
commit c0de0eb82e
12 changed files with 509 additions and 450 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
examples/distributor/main.c
View File

@ -405,17 +405,30 @@ lcore_worker(struct lcore_params *p)
{
struct rte_distributor *d = p->d;
const unsigned id = p->worker_id;
unsigned int num = 0;
unsigned int i;
/*
* 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 (i = 0; i < 8; i++)
buf[i] = NULL;
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;
num = rte_distributor_get_pkt(d, id, buf, buf, num);
/* Do a little bit of work for each packet */
for (i = 0; i < num; i++) {
uint64_t t = rte_rdtsc()+100;
while (rte_rdtsc() < t)
rte_pause();
buf[i]->port ^= xor_val;
}
}
return 0;
}
@ -561,7 +574,8 @@ main(int argc, char *argv[])
}
d = rte_distributor_create("PKT_DIST", rte_socket_id(),
rte_lcore_count() - 2);
rte_lcore_count() - 2,
RTE_DIST_ALG_BURST);
if (d == NULL)
rte_exit(EXIT_FAILURE, "Cannot create distributor\n");

1
lib/librte_distributor/Makefile
View File

@ -57,6 +57,5 @@ endif
# install this header file
SYMLINK-$(CONFIG_RTE_LIBRTE_DISTRIBUTOR)-include := rte_distributor.h
SYMLINK-$(CONFIG_RTE_LIBRTE_DISTRIBUTOR)-include += rte_distributor_v20.h
include $(RTE_SDK)/mk/rte.lib.mk

76
lib/librte_distributor/rte_distributor.c
View File

@ -42,10 +42,10 @@
#include <rte_eal_memconfig.h>
#include <rte_compat.h>
#include "rte_distributor_private.h"
#include "rte_distributor_next.h"
#include "rte_distributor.h"
#include "rte_distributor_v20.h"
TAILQ_HEAD(rte_dist_burst_list, rte_distributor_v1705);
TAILQ_HEAD(rte_dist_burst_list, rte_distributor);
static struct rte_tailq_elem rte_dist_burst_tailq = {
.name = "RTE_DIST_BURST",
@ -57,17 +57,17 @@ EAL_REGISTER_TAILQ(rte_dist_burst_tailq)
/**** Burst Packet APIs called by workers ****/
void
rte_distributor_request_pkt_v1705(struct rte_distributor_v1705 *d,
rte_distributor_request_pkt(struct rte_distributor *d,
unsigned int worker_id, struct rte_mbuf **oldpkt,
unsigned int count)
{
struct rte_distributor_buffer_v1705 *buf = &(d->bufs[worker_id]);
struct rte_distributor_buffer *buf = &(d->bufs[worker_id]);
unsigned int i;
volatile int64_t *retptr64;
if (unlikely(d->alg_type == RTE_DIST_ALG_SINGLE)) {
rte_distributor_request_pkt(d->d_v20,
rte_distributor_request_pkt_v20(d->d_v20,
worker_id, oldpkt[0]);
return;
}
@ -104,16 +104,16 @@ rte_distributor_request_pkt_v1705(struct rte_distributor_v1705 *d,
}
int
rte_distributor_poll_pkt_v1705(struct rte_distributor_v1705 *d,
rte_distributor_poll_pkt(struct rte_distributor *d,
unsigned int worker_id, struct rte_mbuf **pkts)
{
struct rte_distributor_buffer_v1705 *buf = &d->bufs[worker_id];
struct rte_distributor_buffer *buf = &d->bufs[worker_id];
uint64_t ret;
int count = 0;
unsigned int i;
if (unlikely(d->alg_type == RTE_DIST_ALG_SINGLE)) {
pkts[0] = rte_distributor_poll_pkt(d->d_v20, worker_id);
pkts[0] = rte_distributor_poll_pkt_v20(d->d_v20, worker_id);
return (pkts[0]) ? 1 : 0;
}
@ -140,7 +140,7 @@ rte_distributor_poll_pkt_v1705(struct rte_distributor_v1705 *d,
}
int
rte_distributor_get_pkt_v1705(struct rte_distributor_v1705 *d,
rte_distributor_get_pkt(struct rte_distributor *d,
unsigned int worker_id, struct rte_mbuf **pkts,
struct rte_mbuf **oldpkt, unsigned int return_count)
{
@ -148,37 +148,37 @@ rte_distributor_get_pkt_v1705(struct rte_distributor_v1705 *d,
if (unlikely(d->alg_type == RTE_DIST_ALG_SINGLE)) {
if (return_count <= 1) {
pkts[0] = rte_distributor_get_pkt(d->d_v20,
pkts[0] = rte_distributor_get_pkt_v20(d->d_v20,
worker_id, oldpkt[0]);
return (pkts[0]) ? 1 : 0;
} else
return -EINVAL;
}
rte_distributor_request_pkt_v1705(d, worker_id, oldpkt, return_count);
rte_distributor_request_pkt(d, worker_id, oldpkt, return_count);
count = rte_distributor_poll_pkt_v1705(d, worker_id, pkts);
count = rte_distributor_poll_pkt(d, worker_id, pkts);
while (count == -1) {
uint64_t t = rte_rdtsc() + 100;
while (rte_rdtsc() < t)
rte_pause();
count = rte_distributor_poll_pkt_v1705(d, worker_id, pkts);
count = rte_distributor_poll_pkt(d, worker_id, pkts);
}
return count;
}
int
rte_distributor_return_pkt_v1705(struct rte_distributor_v1705 *d,
rte_distributor_return_pkt(struct rte_distributor *d,
unsigned int worker_id, struct rte_mbuf **oldpkt, int num)
{
struct rte_distributor_buffer_v1705 *buf = &d->bufs[worker_id];
struct rte_distributor_buffer *buf = &d->bufs[worker_id];
unsigned int i;
if (unlikely(d->alg_type == RTE_DIST_ALG_SINGLE)) {
if (num == 1)
return rte_distributor_return_pkt(d->d_v20,
return rte_distributor_return_pkt_v20(d->d_v20,
worker_id, oldpkt[0]);
else
return -EINVAL;
@ -202,7 +202,7 @@ rte_distributor_return_pkt_v1705(struct rte_distributor_v1705 *d,
/* stores a packet returned from a worker inside the returns array */
static inline void
store_return(uintptr_t oldbuf, struct rte_distributor_v1705 *d,
store_return(uintptr_t oldbuf, struct rte_distributor *d,
unsigned int *ret_start, unsigned int *ret_count)
{
if (!oldbuf)
@ -221,7 +221,7 @@ store_return(uintptr_t oldbuf, struct rte_distributor_v1705 *d,
* workers to give us our atomic flow pinning.
*/
void
find_match_scalar(struct rte_distributor_v1705 *d,
find_match_scalar(struct rte_distributor *d,
uint16_t *data_ptr,
uint16_t *output_ptr)
{
@ -270,9 +270,9 @@ find_match_scalar(struct rte_distributor_v1705 *d,
* the valid returned pointers (store_return).
*/
static unsigned int
handle_returns(struct rte_distributor_v1705 *d, unsigned int wkr)
handle_returns(struct rte_distributor *d, unsigned int wkr)
{
struct rte_distributor_buffer_v1705 *buf = &(d->bufs[wkr]);
struct rte_distributor_buffer *buf = &(d->bufs[wkr]);
uintptr_t oldbuf;
unsigned int ret_start = d->returns.start,
ret_count = d->returns.count;
@ -308,9 +308,9 @@ handle_returns(struct rte_distributor_v1705 *d, unsigned int wkr)
* before sending out new packets.
*/
static unsigned int
release(struct rte_distributor_v1705 *d, unsigned int wkr)
release(struct rte_distributor *d, unsigned int wkr)
{
struct rte_distributor_buffer_v1705 *buf = &(d->bufs[wkr]);
struct rte_distributor_buffer *buf = &(d->bufs[wkr]);
unsigned int i;
while (!(d->bufs[wkr].bufptr64[0] & RTE_DISTRIB_GET_BUF))
@ -342,7 +342,7 @@ release(struct rte_distributor_v1705 *d, unsigned int wkr)
/* process a set of packets to distribute them to workers */
int
rte_distributor_process_v1705(struct rte_distributor_v1705 *d,
rte_distributor_process(struct rte_distributor *d,
struct rte_mbuf **mbufs, unsigned int num_mbufs)
{
unsigned int next_idx = 0;
@ -355,7 +355,7 @@ rte_distributor_process_v1705(struct rte_distributor_v1705 *d,
if (d->alg_type == RTE_DIST_ALG_SINGLE) {
/* Call the old API */
return rte_distributor_process(d->d_v20, mbufs, num_mbufs);
return rte_distributor_process_v20(d->d_v20, mbufs, num_mbufs);
}
if (unlikely(num_mbufs == 0)) {
@ -479,7 +479,7 @@ rte_distributor_process_v1705(struct rte_distributor_v1705 *d,
/* return to the caller, packets returned from workers */
int
rte_distributor_returned_pkts_v1705(struct rte_distributor_v1705 *d,
rte_distributor_returned_pkts(struct rte_distributor *d,
struct rte_mbuf **mbufs, unsigned int max_mbufs)
{
struct rte_distributor_returned_pkts *returns = &d->returns;
@ -489,7 +489,7 @@ rte_distributor_returned_pkts_v1705(struct rte_distributor_v1705 *d,
if (d->alg_type == RTE_DIST_ALG_SINGLE) {
/* Call the old API */
return rte_distributor_returned_pkts(d->d_v20,
return rte_distributor_returned_pkts_v20(d->d_v20,
mbufs, max_mbufs);
}
@ -510,7 +510,7 @@ rte_distributor_returned_pkts_v1705(struct rte_distributor_v1705 *d,
* being workered on or queued up in a backlog.
*/
static inline unsigned int
total_outstanding(const struct rte_distributor_v1705 *d)
total_outstanding(const struct rte_distributor *d)
{
unsigned int wkr, total_outstanding = 0;
@ -525,24 +525,24 @@ total_outstanding(const struct rte_distributor_v1705 *d)
* queued up.
*/
int
rte_distributor_flush_v1705(struct rte_distributor_v1705 *d)
rte_distributor_flush(struct rte_distributor *d)
{
const unsigned int flushed = total_outstanding(d);
unsigned int wkr;
if (d->alg_type == RTE_DIST_ALG_SINGLE) {
/* Call the old API */
return rte_distributor_flush(d->d_v20);
return rte_distributor_flush_v20(d->d_v20);
}
while (total_outstanding(d) > 0)
rte_distributor_process_v1705(d, NULL, 0);
rte_distributor_process(d, NULL, 0);
/*
* Send empty burst to all workers to allow them to exit
* gracefully, should they need to.
*/
rte_distributor_process_v1705(d, NULL, 0);
rte_distributor_process(d, NULL, 0);
for (wkr = 0; wkr < d->num_workers; wkr++)
handle_returns(d, wkr);
@ -552,13 +552,13 @@ rte_distributor_flush_v1705(struct rte_distributor_v1705 *d)
/* clears the internal returns array in the distributor */
void
rte_distributor_clear_returns_v1705(struct rte_distributor_v1705 *d)
rte_distributor_clear_returns(struct rte_distributor *d)
{
unsigned int wkr;
if (d->alg_type == RTE_DIST_ALG_SINGLE) {
/* Call the old API */
rte_distributor_clear_returns(d->d_v20);
rte_distributor_clear_returns_v20(d->d_v20);
}
/* throw away returns, so workers can exit */
@ -567,13 +567,13 @@ rte_distributor_clear_returns_v1705(struct rte_distributor_v1705 *d)
}
/* creates a distributor instance */
struct rte_distributor_v1705 *
rte_distributor_create_v1705(const char *name,
struct rte_distributor *
rte_distributor_create(const char *name,
unsigned int socket_id,
unsigned int num_workers,
unsigned int alg_type)
{
struct rte_distributor_v1705 *d;
struct rte_distributor *d;
struct rte_dist_burst_list *dist_burst_list;
char mz_name[RTE_MEMZONE_NAMESIZE];
const struct rte_memzone *mz;
@ -586,8 +586,8 @@ rte_distributor_create_v1705(const char *name,
RTE_BUILD_BUG_ON((RTE_DISTRIB_MAX_WORKERS & 7) != 0);
if (alg_type == RTE_DIST_ALG_SINGLE) {
d = malloc(sizeof(struct rte_distributor_v1705));
d->d_v20 = rte_distributor_create(name,
d = malloc(sizeof(struct rte_distributor));
d->d_v20 = rte_distributor_create_v20(name,
socket_id, num_workers);
if (d->d_v20 == NULL) {
/* rte_errno will have been set */

240
lib/librte_distributor/rte_distributor.h
View File

@ -1,8 +1,7 @@
/*-
* BSD LICENSE
*
* Copyright(c) 2010-2014 Intel Corporation. All rights reserved.
* All rights reserved.
* Copyright(c) 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
@ -31,9 +30,240 @@
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*/
#ifndef _RTE_DISTRIBUTE_H_
#define _RTE_DISTRIBUTE_H_
#ifndef _RTE_DISTRIBUTOR_H_
#define _RTE_DISTRIBUTOR_H_
#include <rte_distributor_v20.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
/* Type of distribution (burst/single) */
enum rte_distributor_alg_type {
RTE_DIST_ALG_BURST = 0,
RTE_DIST_ALG_SINGLE,
RTE_DIST_NUM_ALG_TYPES
};
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
* @param alg_type
* Call the legacy API, or use the new burst API. legacy uses 32-bit
* flow ID, and works on a single packet at a time. Latest uses 15-
* bit flow ID and works on up to 8 packets at a time to worers.
* @return
* The newly created distributor instance
*/
struct rte_distributor *
rte_distributor_create(const char *name, unsigned int socket_id,
unsigned int num_workers,
unsigned int alg_type);
/* *** 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(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(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(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(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(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 oldpkt
* The previous packets being processed by the worker
* @param num
* The number of packets in the oldpkt array
*/
int
rte_distributor_return_pkt(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(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(struct rte_distributor *d,
unsigned int worker_id, struct rte_mbuf **mbufs);
#ifdef __cplusplus
}
#endif
#endif

2
lib/librte_distributor/rte_distributor_match_generic.c
View File

@ -35,7 +35,7 @@
#include "rte_distributor.h"
void
find_match_vec(struct rte_distributor_v1705 *d,
find_match_vec(struct rte_distributor *d,
uint16_t *data_ptr,
uint16_t *output_ptr)
{

2
lib/librte_distributor/rte_distributor_match_sse.c
View File

@ -38,7 +38,7 @@
void
find_match_vec(struct rte_distributor_v1705 *d,
find_match_vec(struct rte_distributor *d,
uint16_t *data_ptr,
uint16_t *output_ptr)
{

269
lib/librte_distributor/rte_distributor_next.h
View File

@ -1,269 +0,0 @@
/*-
* BSD LICENSE
*
* Copyright(c) 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 _RTE_DISTRIBUTOR_H_
#define _RTE_DISTRIBUTOR_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
/* Type of distribution (burst/single) */
enum rte_distributor_alg_type {
RTE_DIST_ALG_BURST = 0,
RTE_DIST_ALG_SINGLE,
RTE_DIST_NUM_ALG_TYPES
};
struct rte_distributor_v1705;
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
* @param alg_type
* Call the legacy API, or use the new burst API. legacy uses 32-bit
* flow ID, and works on a single packet at a time. Latest uses 15-
* bit flow ID and works on up to 8 packets at a time to worers.
* @return
* The newly created distributor instance
*/
struct rte_distributor_v1705 *
rte_distributor_create_v1705(const char *name, unsigned int socket_id,
unsigned int num_workers,
unsigned int alg_type);
/* *** 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_v1705(struct rte_distributor_v1705 *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_v1705(struct rte_distributor_v1705 *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_v1705(struct rte_distributor_v1705 *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_v1705(struct rte_distributor_v1705 *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_v1705(struct rte_distributor_v1705 *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 oldpkt
* The previous packets being processed by the worker
* @param num
* The number of packets in the oldpkt array
*/
int
rte_distributor_return_pkt_v1705(struct rte_distributor_v1705 *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_v1705(struct rte_distributor_v1705 *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_v1705(struct rte_distributor_v1705 *d,
unsigned int worker_id, struct rte_mbuf **mbufs);
#ifdef __cplusplus
}
#endif
#endif

22
lib/librte_distributor/rte_distributor_private.h
View File

@ -83,7 +83,7 @@ extern "C" {
* the next cache line to worker 0, we pad this out to three cache lines.
* Only 64-bits of the memory is actually used though.
*/
union rte_distributor_buffer {
union rte_distributor_buffer_v20 {
volatile int64_t bufptr64;
char pad[RTE_CACHE_LINE_SIZE*3];
} __rte_cache_aligned;
@ -108,8 +108,8 @@ struct rte_distributor_returned_pkts {
struct rte_mbuf *mbufs[RTE_DISTRIB_MAX_RETURNS];
};
struct rte_distributor {
TAILQ_ENTRY(rte_distributor) next; /**< Next in list. */
struct rte_distributor_v20 {
TAILQ_ENTRY(rte_distributor_v20) next; /**< Next in list. */
char name[RTE_DISTRIBUTOR_NAMESIZE]; /**< Name of the ring. */
unsigned int num_workers; /**< Number of workers polling */
@ -124,7 +124,7 @@ struct rte_distributor {
struct rte_distributor_backlog backlog[RTE_DISTRIB_MAX_WORKERS];
union rte_distributor_buffer bufs[RTE_DISTRIB_MAX_WORKERS];
union rte_distributor_buffer_v20 bufs[RTE_DISTRIB_MAX_WORKERS];
struct rte_distributor_returned_pkts returns;
};
@ -144,7 +144,7 @@ enum rte_distributor_match_function {
* We can pass up to 8 mbufs at a time in one cacheline.
* There is a separate cacheline for returns in the burst API.
*/
struct rte_distributor_buffer_v1705 {
struct rte_distributor_buffer {
volatile int64_t bufptr64[RTE_DIST_BURST_SIZE]
__rte_cache_aligned; /* <= outgoing to worker */
@ -158,8 +158,8 @@ struct rte_distributor_buffer_v1705 {
int count __rte_cache_aligned; /* <= number of current mbufs */
};
struct rte_distributor_v1705 {
TAILQ_ENTRY(rte_distributor_v1705) next; /**< Next in list. */
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 */
@ -176,22 +176,22 @@ struct rte_distributor_v1705 {
struct rte_distributor_backlog backlog[RTE_DISTRIB_MAX_WORKERS]
__rte_cache_aligned;
struct rte_distributor_buffer_v1705 bufs[RTE_DISTRIB_MAX_WORKERS];
struct rte_distributor_buffer bufs[RTE_DISTRIB_MAX_WORKERS];
struct rte_distributor_returned_pkts returns;
enum rte_distributor_match_function dist_match_fn;
struct rte_distributor *d_v20;
struct rte_distributor_v20 *d_v20;
};
void
find_match_scalar(struct rte_distributor_v1705 *d,
find_match_scalar(struct rte_distributor *d,
uint16_t *data_ptr,
uint16_t *output_ptr);
void
find_match_vec(struct rte_distributor_v1705 *d,
find_match_vec(struct rte_distributor *d,
uint16_t *data_ptr,
uint16_t *output_ptr);

46
lib/librte_distributor/rte_distributor_v20.c
View File

@ -43,7 +43,7 @@
#include "rte_distributor_v20.h"
#include "rte_distributor_private.h"
TAILQ_HEAD(rte_distributor_list, rte_distributor);
TAILQ_HEAD(rte_distributor_list, rte_distributor_v20);
static struct rte_tailq_elem rte_distributor_tailq = {
.name = "RTE_DISTRIBUTOR",
@ -53,10 +53,10 @@ EAL_REGISTER_TAILQ(rte_distributor_tailq)
/**** APIs called by workers ****/
void
rte_distributor_request_pkt(struct rte_distributor *d,
rte_distributor_request_pkt_v20(struct rte_distributor_v20 *d,
unsigned worker_id, struct rte_mbuf *oldpkt)
{
union rte_distributor_buffer *buf = &d->bufs[worker_id];
union rte_distributor_buffer_v20 *buf = &d->bufs[worker_id];
int64_t req = (((int64_t)(uintptr_t)oldpkt) << RTE_DISTRIB_FLAG_BITS)
| RTE_DISTRIB_GET_BUF;
while (unlikely(buf->bufptr64 & RTE_DISTRIB_FLAGS_MASK))
@ -65,10 +65,10 @@ rte_distributor_request_pkt(struct rte_distributor *d,
}
struct rte_mbuf *
rte_distributor_poll_pkt(struct rte_distributor *d,
rte_distributor_poll_pkt_v20(struct rte_distributor_v20 *d,
unsigned worker_id)
{
union rte_distributor_buffer *buf = &d->bufs[worker_id];
union rte_distributor_buffer_v20 *buf = &d->bufs[worker_id];
if (buf->bufptr64 & RTE_DISTRIB_GET_BUF)
return NULL;
@ -78,21 +78,21 @@ rte_distributor_poll_pkt(struct rte_distributor *d,
}
struct rte_mbuf *
rte_distributor_get_pkt(struct rte_distributor *d,
rte_distributor_get_pkt_v20(struct rte_distributor_v20 *d,
unsigned worker_id, struct rte_mbuf *oldpkt)
{
struct rte_mbuf *ret;
rte_distributor_request_pkt(d, worker_id, oldpkt);
while ((ret = rte_distributor_poll_pkt(d, worker_id)) == NULL)
rte_distributor_request_pkt_v20(d, worker_id, oldpkt);
while ((ret = rte_distributor_poll_pkt_v20(d, worker_id)) == NULL)
rte_pause();
return ret;
}
int
rte_distributor_return_pkt(struct rte_distributor *d,
rte_distributor_return_pkt_v20(struct rte_distributor_v20 *d,
unsigned worker_id, struct rte_mbuf *oldpkt)
{
union rte_distributor_buffer *buf = &d->bufs[worker_id];
union rte_distributor_buffer_v20 *buf = &d->bufs[worker_id];
uint64_t req = (((int64_t)(uintptr_t)oldpkt) << RTE_DISTRIB_FLAG_BITS)
| RTE_DISTRIB_RETURN_BUF;
buf->bufptr64 = req;
@ -123,7 +123,7 @@ backlog_pop(struct rte_distributor_backlog *bl)
/* stores a packet returned from a worker inside the returns array */
static inline void
store_return(uintptr_t oldbuf, struct rte_distributor *d,
store_return(uintptr_t oldbuf, struct rte_distributor_v20 *d,
unsigned *ret_start, unsigned *ret_count)
{
/* store returns in a circular buffer - code is branch-free */
@ -134,7 +134,7 @@ store_return(uintptr_t oldbuf, struct rte_distributor *d,
}
static inline void
handle_worker_shutdown(struct rte_distributor *d, unsigned wkr)
handle_worker_shutdown(struct rte_distributor_v20 *d, unsigned int wkr)
{
d->in_flight_tags[wkr] = 0;
d->in_flight_bitmask &= ~(1UL << wkr);
@ -164,7 +164,7 @@ handle_worker_shutdown(struct rte_distributor *d, unsigned wkr)
* Note that the tags were set before first level call
* to rte_distributor_process.
*/
rte_distributor_process(d, pkts, i);
rte_distributor_process_v20(d, pkts, i);
bl->count = bl->start = 0;
}
}
@ -174,7 +174,7 @@ handle_worker_shutdown(struct rte_distributor *d, unsigned wkr)
* to do a partial flush.
*/
static int
process_returns(struct rte_distributor *d)
process_returns(struct rte_distributor_v20 *d)
{
unsigned wkr;
unsigned flushed = 0;
@ -213,7 +213,7 @@ process_returns(struct rte_distributor *d)
/* process a set of packets to distribute them to workers */
int
rte_distributor_process(struct rte_distributor *d,
rte_distributor_process_v20(struct rte_distributor_v20 *d,
struct rte_mbuf **mbufs, unsigned num_mbufs)
{
unsigned next_idx = 0;
@ -317,7 +317,7 @@ rte_distributor_process(struct rte_distributor *d,
/* return to the caller, packets returned from workers */
int
rte_distributor_returned_pkts(struct rte_distributor *d,
rte_distributor_returned_pkts_v20(struct rte_distributor_v20 *d,
struct rte_mbuf **mbufs, unsigned max_mbufs)
{
struct rte_distributor_returned_pkts *returns = &d->returns;
@ -338,7 +338,7 @@ rte_distributor_returned_pkts(struct rte_distributor *d,
/* 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
total_outstanding(const struct rte_distributor *d)
total_outstanding(const struct rte_distributor_v20 *d)
{
unsigned wkr, total_outstanding;
@ -353,19 +353,19 @@ total_outstanding(const struct rte_distributor *d)
/* flush the distributor, so that there are no outstanding packets in flight or
* queued up. */
int
rte_distributor_flush(struct rte_distributor *d)
rte_distributor_flush_v20(struct rte_distributor_v20 *d)
{
const unsigned flushed = total_outstanding(d);
while (total_outstanding(d) > 0)
rte_distributor_process(d, NULL, 0);
rte_distributor_process_v20(d, NULL, 0);
return flushed;
}
/* clears the internal returns array in the distributor */
void
rte_distributor_clear_returns(struct rte_distributor *d)
rte_distributor_clear_returns_v20(struct rte_distributor_v20 *d)
{
d->returns.start = d->returns.count = 0;
#ifndef __OPTIMIZE__
@ -374,12 +374,12 @@ rte_distributor_clear_returns(struct rte_distributor *d)
}
/* creates a distributor instance */
struct rte_distributor *
rte_distributor_create(const char *name,
struct rte_distributor_v20 *
rte_distributor_create_v20(const char *name,
unsigned socket_id,
unsigned num_workers)
{
struct rte_distributor *d;
struct rte_distributor_v20 *d;
struct rte_distributor_list *distributor_list;
char mz_name[RTE_MEMZONE_NAMESIZE];
const struct rte_memzone *mz;

24
lib/librte_distributor/rte_distributor_v20.h
View File

@ -48,7 +48,7 @@ extern "C" {
#define RTE_DISTRIBUTOR_NAMESIZE 32 /**< Length of name for instance */
struct rte_distributor;
struct rte_distributor_v20;
struct rte_mbuf;
/**
@ -67,8 +67,8 @@ struct rte_mbuf;
* @return
* The newly created distributor instance
*/
struct rte_distributor *
rte_distributor_create(const char *name, unsigned int socket_id,
struct rte_distributor_v20 *
rte_distributor_create_v20(const char *name, unsigned int socket_id,
unsigned int num_workers);
/* *** APIS to be called on the distributor lcore *** */
@ -103,7 +103,7 @@ rte_distributor_create(const char *name, unsigned int socket_id,
* The number of mbufs processed.
*/
int
rte_distributor_process(struct rte_distributor *d,
rte_distributor_process_v20(struct rte_distributor_v20 *d,
struct rte_mbuf **mbufs, unsigned int num_mbufs);
/**
@ -121,7 +121,7 @@ rte_distributor_process(struct rte_distributor *d,
* The number of mbufs returned in the mbufs array.
*/
int
rte_distributor_returned_pkts(struct rte_distributor *d,
rte_distributor_returned_pkts_v20(struct rte_distributor_v20 *d,
struct rte_mbuf **mbufs, unsigned int max_mbufs);
/**
@ -136,7 +136,7 @@ rte_distributor_returned_pkts(struct rte_distributor *d,
* The number of queued/in-flight packets that were completed by this call.
*/
int
rte_distributor_flush(struct rte_distributor *d);
rte_distributor_flush_v20(struct rte_distributor_v20 *d);
/**
* Clears the array of returned packets used as the source for the
@ -148,7 +148,7 @@ rte_distributor_flush(struct rte_distributor *d);
* The distributor instance to be used
*/
void
rte_distributor_clear_returns(struct rte_distributor *d);
rte_distributor_clear_returns_v20(struct rte_distributor_v20 *d);
/* *** APIS to be called on the worker lcores *** */
/*
@ -177,7 +177,7 @@ rte_distributor_clear_returns(struct rte_distributor *d);
* A new packet to be processed by the worker thread.
*/
struct rte_mbuf *
rte_distributor_get_pkt(struct rte_distributor *d,
rte_distributor_get_pkt_v20(struct rte_distributor_v20 *d,
unsigned int worker_id, struct rte_mbuf *oldpkt);
/**
@ -193,8 +193,8 @@ rte_distributor_get_pkt(struct rte_distributor *d,
* The previous packet being processed by the worker
*/
int
rte_distributor_return_pkt(struct rte_distributor *d, unsigned int worker_id,
struct rte_mbuf *mbuf);
rte_distributor_return_pkt_v20(struct rte_distributor_v20 *d,
unsigned int worker_id, struct rte_mbuf *mbuf);
/**
* API called by a worker to request a new packet to process.
@ -217,7 +217,7 @@ rte_distributor_return_pkt(struct rte_distributor *d, unsigned int worker_id,
* The previous packet, if any, being processed by the worker
*/
void
rte_distributor_request_pkt(struct rte_distributor *d,
rte_distributor_request_pkt_v20(struct rte_distributor_v20 *d,
unsigned int worker_id, struct rte_mbuf *oldpkt);
/**
@ -237,7 +237,7 @@ rte_distributor_request_pkt(struct rte_distributor *d,
* packet is yet available.
*/
struct rte_mbuf *
rte_distributor_poll_pkt(struct rte_distributor *d,
rte_distributor_poll_pkt_v20(struct rte_distributor_v20 *d,
unsigned int worker_id);
#ifdef __cplusplus

229
test/test/test_distributor.c
View File

@ -1,7 +1,7 @@
/*-
* BSD LICENSE
*
* Copyright(c) 2010-2014 Intel Corporation. All rights reserved.
* Copyright(c) 2010-2017 Intel Corporation. All rights reserved.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
@ -87,20 +87,25 @@ clear_packet_count(void)
static int
handle_work(void *arg)
{
struct rte_mbuf *pkt = NULL;
struct rte_mbuf *buf[8] __rte_cache_aligned;
struct worker_params *wp = arg;
struct rte_distributor *d = wp->dist;
struct rte_distributor *db = wp->dist;
unsigned int count = 0, num = 0;
unsigned int id = __sync_fetch_and_add(&worker_idx, 1);
int i;
unsigned count = 0;
unsigned id = __sync_fetch_and_add(&worker_idx, 1);
pkt = rte_distributor_get_pkt(d, id, NULL);
for (i = 0; i < 8; i++)
buf[i] = NULL;
num = rte_distributor_get_pkt(db, id, buf, buf, num);
while (!quit) {
worker_stats[id].handled_packets++, count++;
pkt = rte_distributor_get_pkt(d, id, pkt);
worker_stats[id].handled_packets += num;
count += num;
num = rte_distributor_get_pkt(db, id,
buf, buf, num);
}
worker_stats[id].handled_packets++, count++;
rte_distributor_return_pkt(d, id, pkt);
worker_stats[id].handled_packets += num;
count += num;
rte_distributor_return_pkt(db, id, buf, num);
return 0;
}
@ -118,9 +123,11 @@ handle_work(void *arg)
static int
sanity_test(struct worker_params *wp, struct rte_mempool *p)
{
struct rte_distributor *d = wp->dist;
struct rte_distributor *db = wp->dist;
struct rte_mbuf *bufs[BURST];
unsigned i;
struct rte_mbuf *returns[BURST*2];
unsigned int i, count;
unsigned int retries;
printf("=== Basic distributor sanity tests ===\n");
clear_packet_count();
@ -134,8 +141,15 @@ sanity_test(struct worker_params *wp, struct rte_mempool *p)
for (i = 0; i < BURST; i++)
bufs[i]->hash.usr = 0;
rte_distributor_process(d, bufs, BURST);
rte_distributor_flush(d);
rte_distributor_process(db, bufs, BURST);
count = 0;
do {
rte_distributor_flush(db);
count += rte_distributor_returned_pkts(db,
returns, BURST*2);
} while (count < BURST);
if (total_packet_count() != BURST) {
printf("Line %d: Error, not all packets flushed. "
"Expected %u, got %u\n",
@ -147,8 +161,6 @@ sanity_test(struct worker_params *wp, struct rte_mempool *p)
printf("Worker %u handled %u packets\n", i,
worker_stats[i].handled_packets);
printf("Sanity test with all zero hashes done.\n");
if (worker_stats[0].handled_packets != BURST)
return -1;
/* pick two flows and check they go correctly */
if (rte_lcore_count() >= 3) {
@ -156,8 +168,13 @@ sanity_test(struct worker_params *wp, struct rte_mempool *p)
for (i = 0; i < BURST; i++)
bufs[i]->hash.usr = (i & 1) << 8;
rte_distributor_process(d, bufs, BURST);
rte_distributor_flush(d);
rte_distributor_process(db, bufs, BURST);
count = 0;
do {
rte_distributor_flush(db);
count += rte_distributor_returned_pkts(db,
returns, BURST*2);
} while (count < BURST);
if (total_packet_count() != BURST) {
printf("Line %d: Error, not all packets flushed. "
"Expected %u, got %u\n",
@ -169,20 +186,21 @@ sanity_test(struct worker_params *wp, struct rte_mempool *p)
printf("Worker %u handled %u packets\n", i,
worker_stats[i].handled_packets);
printf("Sanity test with two hash values done\n");
if (worker_stats[0].handled_packets != 16 ||
worker_stats[1].handled_packets != 16)
return -1;
}
/* give a different hash value to each packet,
* so load gets distributed */
clear_packet_count();
for (i = 0; i < BURST; i++)
bufs[i]->hash.usr = i;
bufs[i]->hash.usr = i+1;
rte_distributor_process(d, bufs, BURST);
rte_distributor_flush(d);
rte_distributor_process(db, bufs, BURST);
count = 0;
do {
rte_distributor_flush(db);
count += rte_distributor_returned_pkts(db,
returns, BURST*2);
} while (count < BURST);
if (total_packet_count() != BURST) {
printf("Line %d: Error, not all packets flushed. "
"Expected %u, got %u\n",
@ -204,8 +222,9 @@ sanity_test(struct worker_params *wp, struct rte_mempool *p)
unsigned num_returned = 0;
/* flush out any remaining packets */
rte_distributor_flush(d);
rte_distributor_clear_returns(d);
rte_distributor_flush(db);
rte_distributor_clear_returns(db);
if (rte_mempool_get_bulk(p, (void *)many_bufs, BIG_BATCH) != 0) {
printf("line %d: Error getting mbufs from pool\n", __LINE__);
return -1;
@ -213,28 +232,44 @@ sanity_test(struct worker_params *wp, struct rte_mempool *p)
for (i = 0; i < BIG_BATCH; i++)
many_bufs[i]->hash.usr = i << 2;
printf("=== testing big burst (%s) ===\n", wp->name);
for (i = 0; i < BIG_BATCH/BURST; i++) {
rte_distributor_process(d, &many_bufs[i*BURST], BURST);
num_returned += rte_distributor_returned_pkts(d,
rte_distributor_process(db,
&many_bufs[i*BURST], BURST);
count = rte_distributor_returned_pkts(db,
&return_bufs[num_returned],
BIG_BATCH - num_returned);
num_returned += count;
}
rte_distributor_flush(d);
num_returned += rte_distributor_returned_pkts(d,
&return_bufs[num_returned], BIG_BATCH - num_returned);
rte_distributor_flush(db);
count = rte_distributor_returned_pkts(db,
&return_bufs[num_returned],
BIG_BATCH - num_returned);
num_returned += count;
retries = 0;
do {
rte_distributor_flush(db);
count = rte_distributor_returned_pkts(db,
&return_bufs[num_returned],
BIG_BATCH - num_returned);
num_returned += count;
retries++;
} while ((num_returned < BIG_BATCH) && (retries < 100));
if (num_returned != BIG_BATCH) {
printf("line %d: Number returned is not the same as "
"number sent\n", __LINE__);
printf("line %d: Missing packets, expected %d\n",
__LINE__, num_returned);
return -1;
}
/* big check - make sure all packets made it back!! */
for (i = 0; i < BIG_BATCH; i++) {
unsigned j;
struct rte_mbuf *src = many_bufs[i];
for (j = 0; j < BIG_BATCH; j++)
for (j = 0; j < BIG_BATCH; j++) {
if (return_bufs[j] == src)
break;
}
if (j == BIG_BATCH) {
printf("Error: could not find source packet #%u\n", i);
@ -258,20 +293,28 @@ sanity_test(struct worker_params *wp, struct rte_mempool *p)
static int
handle_work_with_free_mbufs(void *arg)
{
struct rte_mbuf *pkt = NULL;
struct rte_mbuf *buf[8] __rte_cache_aligned;
struct worker_params *wp = arg;
struct rte_distributor *d = wp->dist;
unsigned count = 0;
unsigned id = __sync_fetch_and_add(&worker_idx, 1);
unsigned int count = 0;
unsigned int i;
unsigned int num = 0;
unsigned int id = __sync_fetch_and_add(&worker_idx, 1);
pkt = rte_distributor_get_pkt(d, id, NULL);
for (i = 0; i < 8; i++)
buf[i] = NULL;
num = rte_distributor_get_pkt(d, id, buf, buf, num);
while (!quit) {
worker_stats[id].handled_packets++, count++;
rte_pktmbuf_free(pkt);
pkt = rte_distributor_get_pkt(d, id, pkt);
worker_stats[id].handled_packets += num;
count += num;
for (i = 0; i < num; i++)
rte_pktmbuf_free(buf[i]);
num = rte_distributor_get_pkt(d,
id, buf, buf, num);
}
worker_stats[id].handled_packets++, count++;
rte_distributor_return_pkt(d, id, pkt);
worker_stats[id].handled_packets += num;
count += num;
rte_distributor_return_pkt(d, id, buf, num);
return 0;
}
@ -287,7 +330,8 @@ sanity_test_with_mbuf_alloc(struct worker_params *wp, struct rte_mempool *p)
unsigned i;
struct rte_mbuf *bufs[BURST];
printf("=== Sanity test with mbuf alloc/free ===\n");
printf("=== Sanity test with mbuf alloc/free (%s) ===\n", wp->name);
clear_packet_count();
for (i = 0; i < ((1<<ITER_POWER)); i += BURST) {
unsigned j;
@ -302,6 +346,9 @@ sanity_test_with_mbuf_alloc(struct worker_params *wp, struct rte_mempool *p)
}
rte_distributor_flush(d);
rte_delay_us(10000);
if (total_packet_count() < (1<<ITER_POWER)) {
printf("Line %u: Packet count is incorrect, %u, expected %u\n",
__LINE__, total_packet_count(),
@ -317,21 +364,32 @@ static int
handle_work_for_shutdown_test(void *arg)
{
struct rte_mbuf *pkt = NULL;
struct rte_mbuf *buf[8] __rte_cache_aligned;
struct worker_params *wp = arg;
struct rte_distributor *d = wp->dist;
unsigned count = 0;
const unsigned id = __sync_fetch_and_add(&worker_idx, 1);
unsigned int count = 0;
unsigned int num = 0;
unsigned int total = 0;
unsigned int i;
unsigned int returned = 0;
const unsigned int id = __sync_fetch_and_add(&worker_idx, 1);
num = rte_distributor_get_pkt(d, id, buf, buf, num);
pkt = rte_distributor_get_pkt(d, id, NULL);
/* wait for quit single globally, or for worker zero, wait
* for zero_quit */
while (!quit && !(id == 0 && zero_quit)) {
worker_stats[id].handled_packets++, count++;
rte_pktmbuf_free(pkt);
pkt = rte_distributor_get_pkt(d, id, NULL);
worker_stats[id].handled_packets += num;
count += num;
for (i = 0; i < num; i++)
rte_pktmbuf_free(buf[i]);
num = rte_distributor_get_pkt(d,
id, buf, buf, num);
total += num;
}
worker_stats[id].handled_packets++, count++;
rte_distributor_return_pkt(d, id, pkt);
worker_stats[id].handled_packets += num;
count += num;
returned = rte_distributor_return_pkt(d, id, buf, num);
if (id == 0) {
/* for worker zero, allow it to restart to pick up last packet
@ -339,13 +397,18 @@ handle_work_for_shutdown_test(void *arg)
*/
while (zero_quit)
usleep(100);
pkt = rte_distributor_get_pkt(d, id, NULL);
num = rte_distributor_get_pkt(d,
id, buf, buf, num);
while (!quit) {
worker_stats[id].handled_packets++, count++;
rte_pktmbuf_free(pkt);
pkt = rte_distributor_get_pkt(d, id, NULL);
num = rte_distributor_get_pkt(d, id, buf, buf, num);
}
rte_distributor_return_pkt(d, id, pkt);
returned = rte_distributor_return_pkt(d,
id, buf, num);
printf("Num returned = %d\n", returned);
}
return 0;
}
@ -367,17 +430,22 @@ sanity_test_with_worker_shutdown(struct worker_params *wp,
printf("=== Sanity test of worker shutdown ===\n");
clear_packet_count();
if (rte_mempool_get_bulk(p, (void *)bufs, BURST) != 0) {
printf("line %d: Error getting mbufs from pool\n", __LINE__);
return -1;
}
/* now set all hash values in all buffers to zero, so all pkts go to the
* one worker thread */
/*
* Now set all hash values in all buffers to same value so all
* pkts go to the one worker thread
*/
for (i = 0; i < BURST; i++)
bufs[i]->hash.usr = 0;
bufs[i]->hash.usr = 1;
rte_distributor_process(d, bufs, BURST);
rte_distributor_flush(d);
/* at this point, we will have processed some packets and have a full
* backlog for the other ones at worker 0.
*/
@ -388,7 +456,7 @@ sanity_test_with_worker_shutdown(struct worker_params *wp,
return -1;
}
for (i = 0; i < BURST; i++)
bufs[i]->hash.usr = 0;
bufs[i]->hash.usr = 1;
/* get worker zero to quit */
zero_quit = 1;
@ -396,6 +464,12 @@ sanity_test_with_worker_shutdown(struct worker_params *wp,
/* flush the distributor */
rte_distributor_flush(d);
rte_delay_us(10000);
for (i = 0; i < rte_lcore_count() - 1; i++)
printf("Worker %u handled %u packets\n", i,
worker_stats[i].handled_packets);
if (total_packet_count() != BURST * 2) {
printf("Line %d: Error, not all packets flushed. "
"Expected %u, got %u\n",
@ -403,10 +477,6 @@ sanity_test_with_worker_shutdown(struct worker_params *wp,
return -1;
}
for (i = 0; i < rte_lcore_count() - 1; i++)
printf("Worker %u handled %u packets\n", i,
worker_stats[i].handled_packets);
printf("Sanity test with worker shutdown passed\n\n");
return 0;
}
@ -422,7 +492,7 @@ test_flush_with_worker_shutdown(struct worker_params *wp,
struct rte_mbuf *bufs[BURST];
unsigned i;
printf("=== Test flush fn with worker shutdown ===\n");
printf("=== Test flush fn with worker shutdown (%s) ===\n", wp->name);
clear_packet_count();
if (rte_mempool_get_bulk(p, (void *)bufs, BURST) != 0) {
@ -446,7 +516,13 @@ test_flush_with_worker_shutdown(struct worker_params *wp,
/* flush the distributor */
rte_distributor_flush(d);
rte_delay_us(10000);
zero_quit = 0;
for (i = 0; i < rte_lcore_count() - 1; i++)
printf("Worker %u handled %u packets\n", i,
worker_stats[i].handled_packets);
if (total_packet_count() != BURST) {
printf("Line %d: Error, not all packets flushed. "
"Expected %u, got %u\n",
@ -454,10 +530,6 @@ test_flush_with_worker_shutdown(struct worker_params *wp,
return -1;
}
for (i = 0; i < rte_lcore_count() - 1; i++)
printf("Worker %u handled %u packets\n", i,
worker_stats[i].handled_packets);
printf("Flush test with worker shutdown passed\n\n");
return 0;
}
@ -469,7 +541,9 @@ int test_error_distributor_create_name(void)
char *name = NULL;
d = rte_distributor_create(name, rte_socket_id(),
rte_lcore_count() - 1);
rte_lcore_count() - 1,
RTE_DIST_ALG_BURST);
if (d != NULL || rte_errno != EINVAL) {
printf("ERROR: No error on create() with NULL name param\n");
return -1;
@ -483,8 +557,10 @@ static
int test_error_distributor_create_numworkers(void)
{
struct rte_distributor *d = NULL;
d = rte_distributor_create("test_numworkers", rte_socket_id(),
RTE_MAX_LCORE + 10);
RTE_MAX_LCORE + 10,
RTE_DIST_ALG_BURST);
if (d != NULL || rte_errno != EINVAL) {
printf("ERROR: No error on create() with num_workers > MAX\n");
return -1;
@ -530,10 +606,11 @@ test_distributor(void)
}
if (d == NULL) {
d = rte_distributor_create("Test_distributor", rte_socket_id(),
rte_lcore_count() - 1);
d = rte_distributor_create("Test_dist_burst", rte_socket_id(),
rte_lcore_count() - 1,
RTE_DIST_ALG_BURST);
if (d == NULL) {
printf("Error creating distributor\n");
printf("Error creating burst distributor\n");
return -1;
}
} else {
@ -553,7 +630,7 @@ test_distributor(void)
}
worker_params.dist = d;
sprintf(worker_params.name, "single");
sprintf(worker_params.name, "burst");
rte_eal_mp_remote_launch(handle_work, &worker_params, SKIP_MASTER);
if (sanity_test(&worker_params, p) < 0)

26
test/test/test_distributor_perf.c
View File

@ -129,18 +129,25 @@ clear_packet_count(void)
static int
handle_work(void *arg)
{
struct rte_mbuf *pkt = NULL;
struct rte_distributor *d = arg;
unsigned count = 0;
unsigned id = __sync_fetch_and_add(&worker_idx, 1);
unsigned int count = 0;
unsigned int num = 0;
int i;
unsigned int id = __sync_fetch_and_add(&worker_idx, 1);
struct rte_mbuf *buf[8] __rte_cache_aligned;
pkt = rte_distributor_get_pkt(d, id, NULL);
for (i = 0; i < 8; i++)
buf[i] = NULL;
num = rte_distributor_get_pkt(d, id, buf, buf, num);
while (!quit) {
worker_stats[id].handled_packets++, count++;
pkt = rte_distributor_get_pkt(d, id, pkt);
worker_stats[id].handled_packets += num;
count += num;
num = rte_distributor_get_pkt(d, id, buf, buf, num);
}
worker_stats[id].handled_packets++, count++;
rte_distributor_return_pkt(d, id, pkt);
worker_stats[id].handled_packets += num;
count += num;
rte_distributor_return_pkt(d, id, buf, num);
return 0;
}
@ -228,7 +235,8 @@ test_distributor_perf(void)
if (d == NULL) {
d = rte_distributor_create("Test_perf", rte_socket_id(),
rte_lcore_count() - 1);
rte_lcore_count() - 1,
RTE_DIST_ALG_SINGLE);
if (d == NULL) {
printf("Error creating distributor\n");
return -1;