#include <rte_mbuf.h>
#include <rte_memory.h>
#include <rte_cycles.h>
-#include <rte_compat.h>
#include <rte_memzone.h>
#include <rte_errno.h>
#include <rte_string_fns.h>
#include <rte_eal_memconfig.h>
#include <rte_pause.h>
+#include <rte_tailq.h>
-#include "rte_distributor_private.h"
#include "rte_distributor.h"
-#include "rte_distributor_v20.h"
-#include "rte_distributor_v1705.h"
+#include "rte_distributor_single.h"
+#include "distributor_private.h"
TAILQ_HEAD(rte_dist_burst_list, rte_distributor);
/**** Burst Packet APIs called by workers ****/
void
-rte_distributor_request_pkt_v1705(struct rte_distributor *d,
+rte_distributor_request_pkt(struct rte_distributor *d,
unsigned int worker_id, struct rte_mbuf **oldpkt,
unsigned int count)
{
volatile int64_t *retptr64;
if (unlikely(d->alg_type == RTE_DIST_ALG_SINGLE)) {
- rte_distributor_request_pkt_v20(d->d_v20,
+ rte_distributor_request_pkt_single(d->d_single,
worker_id, oldpkt[0]);
return;
}
retptr64 = &(buf->retptr64[0]);
- /* Spin while handshake bits are set (scheduler clears it) */
- while (unlikely(*retptr64 & RTE_DISTRIB_GET_BUF)) {
+ /* Spin while handshake bits are set (scheduler clears it).
+ * Sync with worker on GET_BUF flag.
+ */
+ while (unlikely(__atomic_load_n(retptr64, __ATOMIC_ACQUIRE)
+ & RTE_DISTRIB_GET_BUF)) {
rte_pause();
uint64_t t = rte_rdtsc()+100;
/*
* Finally, set the GET_BUF to signal to distributor that cache
* line is ready for processing
+ * Sync with distributor to release retptrs
*/
- *retptr64 |= RTE_DISTRIB_GET_BUF;
+ __atomic_store_n(retptr64, *retptr64 | RTE_DISTRIB_GET_BUF,
+ __ATOMIC_RELEASE);
}
-BIND_DEFAULT_SYMBOL(rte_distributor_request_pkt, _v1705, 17.05);
-MAP_STATIC_SYMBOL(void rte_distributor_request_pkt(struct rte_distributor *d,
- unsigned int worker_id, struct rte_mbuf **oldpkt,
- unsigned int count),
- rte_distributor_request_pkt_v1705);
int
-rte_distributor_poll_pkt_v1705(struct rte_distributor *d,
+rte_distributor_poll_pkt(struct rte_distributor *d,
unsigned int worker_id, struct rte_mbuf **pkts)
{
struct rte_distributor_buffer *buf = &d->bufs[worker_id];
unsigned int i;
if (unlikely(d->alg_type == RTE_DIST_ALG_SINGLE)) {
- pkts[0] = rte_distributor_poll_pkt_v20(d->d_v20, worker_id);
+ pkts[0] = rte_distributor_poll_pkt_single(d->d_single,
+ worker_id);
return (pkts[0]) ? 1 : 0;
}
- /* If bit is set, return */
- if (buf->bufptr64[0] & RTE_DISTRIB_GET_BUF)
+ /* If bit is set, return
+ * Sync with distributor to acquire bufptrs
+ */
+ if (__atomic_load_n(&(buf->bufptr64[0]), __ATOMIC_ACQUIRE)
+ & RTE_DISTRIB_GET_BUF)
return -1;
/* since bufptr64 is signed, this should be an arithmetic shift */
* so now we've got the contents of the cacheline into an array of
* mbuf pointers, so toggle the bit so scheduler can start working
* on the next cacheline while we're working.
+ * Sync with distributor on GET_BUF flag. Release bufptrs.
*/
- buf->bufptr64[0] |= RTE_DISTRIB_GET_BUF;
+ __atomic_store_n(&(buf->bufptr64[0]),
+ buf->bufptr64[0] | RTE_DISTRIB_GET_BUF, __ATOMIC_RELEASE);
return count;
}
-BIND_DEFAULT_SYMBOL(rte_distributor_poll_pkt, _v1705, 17.05);
-MAP_STATIC_SYMBOL(int rte_distributor_poll_pkt(struct rte_distributor *d,
- unsigned int worker_id, struct rte_mbuf **pkts),
- rte_distributor_poll_pkt_v1705);
int
-rte_distributor_get_pkt_v1705(struct rte_distributor *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_v20(d->d_v20,
+ pkts[0] = rte_distributor_get_pkt_single(d->d_single,
worker_id, oldpkt[0]);
return (pkts[0]) ? 1 : 0;
} else
}
return count;
}
-BIND_DEFAULT_SYMBOL(rte_distributor_get_pkt, _v1705, 17.05);
-MAP_STATIC_SYMBOL(int rte_distributor_get_pkt(struct rte_distributor *d,
- unsigned int worker_id, struct rte_mbuf **pkts,
- struct rte_mbuf **oldpkt, unsigned int return_count),
- rte_distributor_get_pkt_v1705);
int
-rte_distributor_return_pkt_v1705(struct rte_distributor *d,
+rte_distributor_return_pkt(struct rte_distributor *d,
unsigned int worker_id, struct rte_mbuf **oldpkt, int num)
{
struct rte_distributor_buffer *buf = &d->bufs[worker_id];
if (unlikely(d->alg_type == RTE_DIST_ALG_SINGLE)) {
if (num == 1)
- return rte_distributor_return_pkt_v20(d->d_v20,
+ return rte_distributor_return_pkt_single(d->d_single,
worker_id, oldpkt[0]);
else
return -EINVAL;
}
+ /* Sync with distributor to acquire retptrs */
+ __atomic_thread_fence(__ATOMIC_ACQUIRE);
for (i = 0; i < RTE_DIST_BURST_SIZE; i++)
/* Switch off the return bit first */
buf->retptr64[i] &= ~RTE_DISTRIB_RETURN_BUF;
buf->retptr64[i] = (((int64_t)(uintptr_t)oldpkt[i]) <<
RTE_DISTRIB_FLAG_BITS) | RTE_DISTRIB_RETURN_BUF;
- /* set the GET_BUF but even if we got no returns */
- buf->retptr64[0] |= RTE_DISTRIB_GET_BUF;
+ /* set the GET_BUF but even if we got no returns.
+ * Sync with distributor on GET_BUF flag. Release retptrs.
+ */
+ __atomic_store_n(&(buf->retptr64[0]),
+ buf->retptr64[0] | RTE_DISTRIB_GET_BUF, __ATOMIC_RELEASE);
return 0;
}
-BIND_DEFAULT_SYMBOL(rte_distributor_return_pkt, _v1705, 17.05);
-MAP_STATIC_SYMBOL(int rte_distributor_return_pkt(struct rte_distributor *d,
- unsigned int worker_id, struct rte_mbuf **oldpkt, int num),
- rte_distributor_return_pkt_v1705);
/**** APIs called on distributor core ***/
unsigned int count = 0;
unsigned int i;
- if (buf->retptr64[0] & RTE_DISTRIB_GET_BUF) {
+ /* Sync on GET_BUF flag. Acquire retptrs. */
+ if (__atomic_load_n(&(buf->retptr64[0]), __ATOMIC_ACQUIRE)
+ & RTE_DISTRIB_GET_BUF) {
for (i = 0; i < RTE_DIST_BURST_SIZE; i++) {
if (buf->retptr64[i] & RTE_DISTRIB_RETURN_BUF) {
oldbuf = ((uintptr_t)(buf->retptr64[i] >>
}
d->returns.start = ret_start;
d->returns.count = ret_count;
- /* Clear for the worker to populate with more returns */
- buf->retptr64[0] = 0;
+ /* Clear for the worker to populate with more returns.
+ * Sync with distributor on GET_BUF flag. Release retptrs.
+ */
+ __atomic_store_n(&(buf->retptr64[0]), 0, __ATOMIC_RELEASE);
}
return count;
}
struct rte_distributor_buffer *buf = &(d->bufs[wkr]);
unsigned int i;
- while (!(d->bufs[wkr].bufptr64[0] & RTE_DISTRIB_GET_BUF))
+ /* Sync with worker on GET_BUF flag */
+ while (!(__atomic_load_n(&(d->bufs[wkr].bufptr64[0]), __ATOMIC_ACQUIRE)
+ & RTE_DISTRIB_GET_BUF))
rte_pause();
handle_returns(d, wkr);
d->backlog[wkr].count = 0;
- /* Clear the GET bit */
- buf->bufptr64[0] &= ~RTE_DISTRIB_GET_BUF;
+ /* Clear the GET bit.
+ * Sync with worker on GET_BUF flag. Release bufptrs.
+ */
+ __atomic_store_n(&(buf->bufptr64[0]),
+ buf->bufptr64[0] & ~RTE_DISTRIB_GET_BUF, __ATOMIC_RELEASE);
return buf->count;
}
/* process a set of packets to distribute them to workers */
int
-rte_distributor_process_v1705(struct rte_distributor *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_v20(d->d_v20, mbufs, num_mbufs);
+ return rte_distributor_process_single(d->d_single,
+ mbufs, num_mbufs);
}
if (unlikely(num_mbufs == 0)) {
/* Flush out all non-full cache-lines to workers. */
for (wid = 0 ; wid < d->num_workers; wid++) {
- if (d->bufs[wid].bufptr64[0] & RTE_DISTRIB_GET_BUF) {
+ /* Sync with worker on GET_BUF flag. */
+ if (__atomic_load_n(&(d->bufs[wid].bufptr64[0]),
+ __ATOMIC_ACQUIRE) & RTE_DISTRIB_GET_BUF) {
release(d, wid);
handle_returns(d, wid);
}
uint16_t matches[RTE_DIST_BURST_SIZE];
unsigned int pkts;
- if (d->bufs[wkr].bufptr64[0] & RTE_DISTRIB_GET_BUF)
+ /* Sync with worker on GET_BUF flag. */
+ if (__atomic_load_n(&(d->bufs[wkr].bufptr64[0]),
+ __ATOMIC_ACQUIRE) & RTE_DISTRIB_GET_BUF)
d->bufs[wkr].count = 0;
if ((num_mbufs - next_idx) < RTE_DIST_BURST_SIZE)
/* Flush out all non-full cache-lines to workers. */
for (wid = 0 ; wid < d->num_workers; wid++)
- if ((d->bufs[wid].bufptr64[0] & RTE_DISTRIB_GET_BUF))
+ /* Sync with worker on GET_BUF flag. */
+ if ((__atomic_load_n(&(d->bufs[wid].bufptr64[0]),
+ __ATOMIC_ACQUIRE) & RTE_DISTRIB_GET_BUF))
release(d, wid);
return num_mbufs;
}
-BIND_DEFAULT_SYMBOL(rte_distributor_process, _v1705, 17.05);
-MAP_STATIC_SYMBOL(int rte_distributor_process(struct rte_distributor *d,
- struct rte_mbuf **mbufs, unsigned int num_mbufs),
- rte_distributor_process_v1705);
/* return to the caller, packets returned from workers */
int
-rte_distributor_returned_pkts_v1705(struct rte_distributor *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_v20(d->d_v20,
+ return rte_distributor_returned_pkts_single(d->d_single,
mbufs, max_mbufs);
}
return retval;
}
-BIND_DEFAULT_SYMBOL(rte_distributor_returned_pkts, _v1705, 17.05);
-MAP_STATIC_SYMBOL(int rte_distributor_returned_pkts(struct rte_distributor *d,
- struct rte_mbuf **mbufs, unsigned int max_mbufs),
- rte_distributor_returned_pkts_v1705);
/*
* Return the number of packets in-flight in a distributor, i.e. packets
* queued up.
*/
int
-rte_distributor_flush_v1705(struct rte_distributor *d)
+rte_distributor_flush(struct rte_distributor *d)
{
unsigned int flushed;
unsigned int wkr;
if (d->alg_type == RTE_DIST_ALG_SINGLE) {
/* Call the old API */
- return rte_distributor_flush_v20(d->d_v20);
+ return rte_distributor_flush_single(d->d_single);
}
flushed = total_outstanding(d);
while (total_outstanding(d) > 0)
rte_distributor_process(d, NULL, 0);
+ /* wait 10ms to allow all worker drain the pkts */
+ rte_delay_us(10000);
+
/*
* Send empty burst to all workers to allow them to exit
* gracefully, should they need to.
return flushed;
}
-BIND_DEFAULT_SYMBOL(rte_distributor_flush, _v1705, 17.05);
-MAP_STATIC_SYMBOL(int rte_distributor_flush(struct rte_distributor *d),
- rte_distributor_flush_v1705);
/* clears the internal returns array in the distributor */
void
-rte_distributor_clear_returns_v1705(struct rte_distributor *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_v20(d->d_v20);
+ rte_distributor_clear_returns_single(d->d_single);
return;
}
/* throw away returns, so workers can exit */
for (wkr = 0; wkr < d->num_workers; wkr++)
- d->bufs[wkr].retptr64[0] = 0;
+ /* Sync with worker. Release retptrs. */
+ __atomic_store_n(&(d->bufs[wkr].retptr64[0]), 0,
+ __ATOMIC_RELEASE);
}
-BIND_DEFAULT_SYMBOL(rte_distributor_clear_returns, _v1705, 17.05);
-MAP_STATIC_SYMBOL(void rte_distributor_clear_returns(struct rte_distributor *d),
- rte_distributor_clear_returns_v1705);
/* creates a distributor instance */
struct rte_distributor *
-rte_distributor_create_v1705(const char *name,
+rte_distributor_create(const char *name,
unsigned int socket_id,
unsigned int num_workers,
unsigned int alg_type)
RTE_BUILD_BUG_ON((sizeof(*d) & RTE_CACHE_LINE_MASK) != 0);
RTE_BUILD_BUG_ON((RTE_DISTRIB_MAX_WORKERS & 7) != 0);
+ if (name == NULL || num_workers >=
+ (unsigned int)RTE_MIN(RTE_DISTRIB_MAX_WORKERS, RTE_MAX_LCORE)) {
+ rte_errno = EINVAL;
+ return NULL;
+ }
+
if (alg_type == RTE_DIST_ALG_SINGLE) {
d = malloc(sizeof(struct rte_distributor));
if (d == NULL) {
rte_errno = ENOMEM;
return NULL;
}
- d->d_v20 = rte_distributor_create_v20(name,
+ d->d_single = rte_distributor_create_single(name,
socket_id, num_workers);
- if (d->d_v20 == NULL) {
+ if (d->d_single == NULL) {
free(d);
/* rte_errno will have been set */
return NULL;
return d;
}
- if (name == NULL || num_workers >= RTE_DISTRIB_MAX_WORKERS) {
- rte_errno = EINVAL;
- return NULL;
- }
-
snprintf(mz_name, sizeof(mz_name), RTE_DISTRIB_PREFIX"%s", name);
mz = rte_memzone_reserve(mz_name, sizeof(*d), socket_id, NO_FLAGS);
if (mz == NULL) {
}
d = mz->addr;
- snprintf(d->name, sizeof(d->name), "%s", name);
+ strlcpy(d->name, name, sizeof(d->name));
d->num_workers = num_workers;
d->alg_type = alg_type;
rte_dist_burst_list);
- rte_rwlock_write_lock(RTE_EAL_TAILQ_RWLOCK);
+ rte_mcfg_tailq_write_lock();
TAILQ_INSERT_TAIL(dist_burst_list, d, next);
- rte_rwlock_write_unlock(RTE_EAL_TAILQ_RWLOCK);
+ rte_mcfg_tailq_write_unlock();
return d;
}
-BIND_DEFAULT_SYMBOL(rte_distributor_create, _v1705, 17.05);
-MAP_STATIC_SYMBOL(struct rte_distributor *rte_distributor_create(
- const char *name, unsigned int socket_id,
- unsigned int num_workers, unsigned int alg_type),
- rte_distributor_create_v1705);
git.droids-corp.org / dpdk.git / blobdiff