X-Git-Url: http://git.droids-corp.org/?a=blobdiff_plain;f=lib%2Flibrte_distributor%2Frte_distributor.c;h=1c047f065a10e62fe82d2122f7ebf501af02bb2d;hb=c47463272f572a4d78109763399b5e3817b848a9;hp=d50598377c25d962c8440bf5c953f809f3e939f6;hpb=5bc53b786b4f83792a25b37dc6d45614e0afba03;p=dpdk.git diff --git a/lib/librte_distributor/rte_distributor.c b/lib/librte_distributor/rte_distributor.c index d50598377c..1c047f065a 100644 --- a/lib/librte_distributor/rte_distributor.c +++ b/lib/librte_distributor/rte_distributor.c @@ -8,17 +8,16 @@ #include #include #include -#include #include #include #include #include #include +#include -#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); @@ -32,7 +31,7 @@ EAL_REGISTER_TAILQ(rte_dist_burst_tailq) /**** 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) { @@ -42,14 +41,17 @@ rte_distributor_request_pkt_v1705(struct rte_distributor *d, 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; @@ -74,17 +76,14 @@ rte_distributor_request_pkt_v1705(struct rte_distributor *d, /* * 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]; @@ -93,12 +92,16 @@ rte_distributor_poll_pkt_v1705(struct rte_distributor *d, 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 */ @@ -113,18 +116,16 @@ rte_distributor_poll_pkt_v1705(struct rte_distributor *d, * 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) { @@ -132,7 +133,7 @@ rte_distributor_get_pkt_v1705(struct rte_distributor *d, 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 @@ -152,14 +153,9 @@ rte_distributor_get_pkt_v1705(struct rte_distributor *d, } 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]; @@ -167,12 +163,14 @@ rte_distributor_return_pkt_v1705(struct rte_distributor *d, 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; @@ -181,15 +179,14 @@ rte_distributor_return_pkt_v1705(struct rte_distributor *d, 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 ***/ @@ -272,7 +269,9 @@ handle_returns(struct rte_distributor *d, unsigned int wkr) 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] >> @@ -285,8 +284,10 @@ handle_returns(struct rte_distributor *d, unsigned int wkr) } 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; } @@ -306,7 +307,9 @@ release(struct rte_distributor *d, unsigned int wkr) 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); @@ -326,8 +329,11 @@ release(struct rte_distributor *d, unsigned int 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; } @@ -335,7 +341,7 @@ release(struct rte_distributor *d, unsigned int wkr) /* 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; @@ -348,13 +354,16 @@ rte_distributor_process_v1705(struct rte_distributor *d, 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); } @@ -366,7 +375,9 @@ rte_distributor_process_v1705(struct rte_distributor *d, 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) @@ -464,19 +475,17 @@ rte_distributor_process_v1705(struct rte_distributor *d, /* 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; @@ -486,7 +495,7 @@ rte_distributor_returned_pkts_v1705(struct rte_distributor *d, 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); } @@ -501,10 +510,6 @@ rte_distributor_returned_pkts_v1705(struct rte_distributor *d, 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 @@ -526,14 +531,14 @@ total_outstanding(const struct rte_distributor *d) * 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); @@ -541,6 +546,9 @@ rte_distributor_flush_v1705(struct rte_distributor *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. @@ -552,33 +560,29 @@ rte_distributor_flush_v1705(struct rte_distributor *d) 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) @@ -595,15 +599,21 @@ rte_distributor_create_v1705(const char *name, 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; @@ -612,11 +622,6 @@ rte_distributor_create_v1705(const char *name, 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) { @@ -625,7 +630,7 @@ rte_distributor_create_v1705(const char *name, } 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; @@ -645,14 +650,9 @@ rte_distributor_create_v1705(const char *name, 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);