{
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;
}
}
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");
# 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
#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",
/**** 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;
}
}
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;
}
}
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)
{
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;
/* 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)
* 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)
{
* 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;
* 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))
/* 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;
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)) {
/* 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;
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);
}
* 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;
* 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);
/* 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 */
}
/* 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;
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 */
/*-
* 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
* 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
#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)
{
void
-find_match_vec(struct rte_distributor_v1705 *d,
+find_match_vec(struct rte_distributor *d,
uint16_t *data_ptr,
uint16_t *output_ptr)
{
+++ /dev/null
-/*-
- * 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
* 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;
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 */
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;
};
* 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 */
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 */
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);
#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",
/**** 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))
}
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;
}
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;
/* 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 */
}
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);
* 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;
}
}
* 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;
/* 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;
/* 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;
/* 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;
/* 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__
}
/* 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;
#define RTE_DISTRIBUTOR_NAMESIZE 32 /**< Length of name for instance */
-struct rte_distributor;
+struct rte_distributor_v20;
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 *** */
* 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);
/**
* 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);
/**
* 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
* 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 *** */
/*
* 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);
/**
* 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.
* 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);
/**
* 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
/*-
* 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
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;
-
- unsigned count = 0;
- unsigned id = __sync_fetch_and_add(&worker_idx, 1);
-
- pkt = rte_distributor_get_pkt(d, id, NULL);
+ struct rte_distributor *db = wp->dist;
+ unsigned int count = 0, num = 0;
+ unsigned int id = __sync_fetch_and_add(&worker_idx, 1);
+ int i;
+
+ 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;
}
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();
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",
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) {
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",
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;
-
- rte_distributor_process(d, bufs, BURST);
- rte_distributor_flush(d);
+ bufs[i]->hash.usr = i+1;
+
+ 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",
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;
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);
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);
-
- pkt = rte_distributor_get_pkt(d, id, NULL);
+ unsigned int count = 0;
+ unsigned int i;
+ unsigned int num = 0;
+ unsigned int id = __sync_fetch_and_add(&worker_idx, 1);
+
+ 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;
}
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;
}
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(),
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
*/
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;
}
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.
*/
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;
/* 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",
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;
}
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) {
/* 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",
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;
}
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;
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;
}
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 {
}
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)
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;
}
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;
git.droids-corp.org / dpdk.git / commitdiff