X-Git-Url: http://git.droids-corp.org/?a=blobdiff_plain;f=lib%2Flibrte_distributor%2Frte_distributor.c;h=07e385a25953bcc7ed132d852232f4ea133c8cc3;hb=2b9a66e1b606d3813d72dd81c626949e09706e27;hp=75b0d47329b002cb830af180ab207fa0a755acbd;hpb=775003ad2f96c9881a00a6534518a875723978c8;p=dpdk.git diff --git a/lib/librte_distributor/rte_distributor.c b/lib/librte_distributor/rte_distributor.c index 75b0d47329..07e385a259 100644 --- a/lib/librte_distributor/rte_distributor.c +++ b/lib/librte_distributor/rte_distributor.c @@ -1,33 +1,5 @@ -/*- - * 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. +/* SPDX-License-Identifier: BSD-3-Clause + * Copyright(c) 2017 Intel Corporation */ #include @@ -40,12 +12,15 @@ #include #include #include -#include -#include "rte_distributor_private.h" -#include "rte_distributor_next.h" -#include "rte_distributor_v20.h" +#include +#include +#include -TAILQ_HEAD(rte_dist_burst_list, rte_distributor_v1705); +#include "rte_distributor.h" +#include "rte_distributor_single.h" +#include "distributor_private.h" + +TAILQ_HEAD(rte_dist_burst_list, rte_distributor); static struct rte_tailq_elem rte_dist_burst_tailq = { .name = "RTE_DIST_BURST", @@ -57,24 +32,27 @@ 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, - worker_id, oldpkt[0]); + rte_distributor_request_pkt_single(d->d_single, + worker_id, count ? oldpkt[0] : NULL); 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_DISTRIB_RETURN_BUF))) { rte_pause(); uint64_t t = rte_rdtsc()+100; @@ -90,35 +68,43 @@ rte_distributor_request_pkt_v1705(struct rte_distributor_v1705 *d, for (i = count; i < RTE_DIST_BURST_SIZE; i++) buf->retptr64[i] = 0; - /* Set Return bit for each packet returned */ + /* Set VALID_BUF bit for each packet returned */ for (i = count; i-- > 0; ) buf->retptr64[i] = (((int64_t)(uintptr_t)(oldpkt[i])) << - RTE_DISTRIB_FLAG_BITS) | RTE_DISTRIB_RETURN_BUF; + RTE_DISTRIB_FLAG_BITS) | RTE_DISTRIB_VALID_BUF; /* * 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); } 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_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 any of below bits is set, return. + * GET_BUF is set when distributor hasn't sent any packets yet + * RETURN_BUF is set when distributor must retrieve in-flight packets + * Sync with distributor to acquire bufptrs + */ + if (__atomic_load_n(&(buf->bufptr64[0]), __ATOMIC_ACQUIRE) + & (RTE_DISTRIB_GET_BUF | RTE_DISTRIB_RETURN_BUF)) return -1; /* since bufptr64 is signed, this should be an arithmetic shift */ @@ -130,17 +116,19 @@ rte_distributor_poll_pkt_v1705(struct rte_distributor_v1705 *d, } /* - * so now we've got the contents of the cacheline into an array of + * 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; } 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,52 +136,80 @@ 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, - worker_id, oldpkt[0]); + pkts[0] = rte_distributor_get_pkt_single(d->d_single, + worker_id, return_count ? oldpkt[0] : NULL); 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_single(d->d_single, worker_id, oldpkt[0]); + else if (num == 0) + return rte_distributor_return_pkt_single(d->d_single, + worker_id, NULL); else return -EINVAL; } + /* Spin while handshake bits are set (scheduler clears it). + * Sync with worker on GET_BUF flag. + */ + while (unlikely(__atomic_load_n(&(buf->retptr64[0]), __ATOMIC_RELAXED) + & (RTE_DISTRIB_GET_BUF | RTE_DISTRIB_RETURN_BUF))) { + rte_pause(); + uint64_t t = rte_rdtsc()+100; + + while (rte_rdtsc() < t) + rte_pause(); + } + + /* 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] = 0; for (i = num; i-- > 0; ) buf->retptr64[i] = (((int64_t)(uintptr_t)oldpkt[i]) << - RTE_DISTRIB_FLAG_BITS) | RTE_DISTRIB_RETURN_BUF; + RTE_DISTRIB_FLAG_BITS) | RTE_DISTRIB_VALID_BUF; + + /* Use RETURN_BUF on bufptr64 to notify distributor that + * we won't read any mbufs from there even if GET_BUF is set. + * This allows distributor to retrieve in-flight already sent packets. + */ + __atomic_or_fetch(&(buf->bufptr64[0]), RTE_DISTRIB_RETURN_BUF, + __ATOMIC_ACQ_REL); - /* set the GET_BUF but even if we got no returns */ - buf->retptr64[0] |= RTE_DISTRIB_GET_BUF; + /* set the RETURN_BUF on retptr64 even if we got no returns. + * Sync with distributor on RETURN_BUF flag. Release retptrs. + * Notify distributor that we don't request more packets any more. + */ + __atomic_store_n(&(buf->retptr64[0]), + buf->retptr64[0] | RTE_DISTRIB_RETURN_BUF, __ATOMIC_RELEASE); return 0; } @@ -202,7 +218,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 +237,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) { @@ -244,13 +260,13 @@ find_match_scalar(struct rte_distributor_v1705 *d, for (j = 0; j < RTE_DIST_BURST_SIZE ; j++) for (w = 0; w < RTE_DIST_BURST_SIZE; w++) - if (d->in_flight_tags[i][j] == data_ptr[w]) { + if (d->in_flight_tags[i][w] == data_ptr[j]) { output_ptr[j] = i+1; break; } for (j = 0; j < RTE_DIST_BURST_SIZE; j++) for (w = 0; w < RTE_DIST_BURST_SIZE; w++) - if (bl->tags[j] == data_ptr[w]) { + if (bl->tags[w] == data_ptr[j]) { output_ptr[j] = i+1; break; } @@ -263,6 +279,59 @@ find_match_scalar(struct rte_distributor_v1705 *d, */ } +/* + * When worker called rte_distributor_return_pkt() + * and passed RTE_DISTRIB_RETURN_BUF handshake through retptr64, + * distributor must retrieve both inflight and backlog packets assigned + * to the worker and reprocess them to another worker. + */ +static void +handle_worker_shutdown(struct rte_distributor *d, unsigned int wkr) +{ + struct rte_distributor_buffer *buf = &(d->bufs[wkr]); + /* double BURST size for storing both inflights and backlog */ + struct rte_mbuf *pkts[RTE_DIST_BURST_SIZE * 2]; + unsigned int pkts_count = 0; + unsigned int i; + + /* If GET_BUF is cleared there are in-flight packets sent + * to worker which does not require new packets. + * They must be retrieved and assigned to another worker. + */ + if (!(__atomic_load_n(&(buf->bufptr64[0]), __ATOMIC_ACQUIRE) + & RTE_DISTRIB_GET_BUF)) + for (i = 0; i < RTE_DIST_BURST_SIZE; i++) + if (buf->bufptr64[i] & RTE_DISTRIB_VALID_BUF) + pkts[pkts_count++] = (void *)((uintptr_t) + (buf->bufptr64[i] + >> RTE_DISTRIB_FLAG_BITS)); + + /* Make following operations on handshake flags on bufptr64: + * - set GET_BUF to indicate that distributor can overwrite buffer + * with new packets if worker will make a new request. + * - clear RETURN_BUF to unlock reads on worker side. + */ + __atomic_store_n(&(buf->bufptr64[0]), RTE_DISTRIB_GET_BUF, + __ATOMIC_RELEASE); + + /* Collect backlog packets from worker */ + for (i = 0; i < d->backlog[wkr].count; i++) + pkts[pkts_count++] = (void *)((uintptr_t) + (d->backlog[wkr].pkts[i] >> RTE_DISTRIB_FLAG_BITS)); + + d->backlog[wkr].count = 0; + + /* Clear both inflight and backlog tags */ + for (i = 0; i < RTE_DIST_BURST_SIZE; i++) { + d->in_flight_tags[wkr][i] = 0; + d->backlog[wkr].tags[i] = 0; + } + + /* Recursive call */ + if (pkts_count > 0) + rte_distributor_process(d, pkts, pkts_count); +} + /* * When the handshake bits indicate that there are packets coming @@ -270,30 +339,48 @@ 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; 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 | RTE_DISTRIB_RETURN_BUF)) { for (i = 0; i < RTE_DIST_BURST_SIZE; i++) { - if (buf->retptr64[i] & RTE_DISTRIB_RETURN_BUF) { + if (buf->retptr64[i] & RTE_DISTRIB_VALID_BUF) { oldbuf = ((uintptr_t)(buf->retptr64[i] >> RTE_DISTRIB_FLAG_BITS)); /* store returns in a circular buffer */ store_return(oldbuf, d, &ret_start, &ret_count); count++; - buf->retptr64[i] &= ~RTE_DISTRIB_RETURN_BUF; + buf->retptr64[i] &= ~RTE_DISTRIB_VALID_BUF; } } d->returns.start = ret_start; d->returns.count = ret_count; - /* Clear for the worker to populate with more returns */ - buf->retptr64[0] = 0; + + /* If worker requested packets with GET_BUF, set it to active + * otherwise (RETURN_BUF), set it to not active. + */ + d->activesum -= d->active[wkr]; + d->active[wkr] = !!(buf->retptr64[0] & RTE_DISTRIB_GET_BUF); + d->activesum += d->active[wkr]; + + /* If worker returned packets without requesting new ones, + * handle all in-flights and backlog packets assigned to it. + */ + if (unlikely(buf->retptr64[0] & RTE_DISTRIB_RETURN_BUF)) + handle_worker_shutdown(d, wkr); + + /* 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; } @@ -308,15 +395,23 @@ 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)) - rte_pause(); - handle_returns(d, wkr); + if (unlikely(!d->active[wkr])) + return 0; + + /* Sync with worker on GET_BUF flag */ + while (!(__atomic_load_n(&(d->bufs[wkr].bufptr64[0]), __ATOMIC_ACQUIRE) + & RTE_DISTRIB_GET_BUF)) { + handle_returns(d, wkr); + if (unlikely(!d->active[wkr])) + return 0; + rte_pause(); + } buf->count = 0; @@ -333,8 +428,11 @@ release(struct rte_distributor_v1705 *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; } @@ -342,7 +440,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; @@ -351,17 +449,24 @@ rte_distributor_process_v1705(struct rte_distributor_v1705 *d, int64_t next_value = 0; uint16_t new_tag = 0; uint16_t flows[RTE_DIST_BURST_SIZE] __rte_cache_aligned; - unsigned int i, j, w, wid; + unsigned int i, j, w, wid, matching_required; 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_single(d->d_single, + mbufs, num_mbufs); } + for (wid = 0 ; wid < d->num_workers; wid++) + handle_returns(d, wid); + 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) { + d->bufs[wid].count = 0; release(d, wid); handle_returns(d, wid); } @@ -369,13 +474,13 @@ rte_distributor_process_v1705(struct rte_distributor_v1705 *d, return 0; } + if (unlikely(!d->activesum)) + return 0; + while (next_idx < num_mbufs) { uint16_t matches[RTE_DIST_BURST_SIZE]; unsigned int pkts; - if (d->bufs[wkr].bufptr64[0] & RTE_DISTRIB_GET_BUF) - d->bufs[wkr].count = 0; - if ((num_mbufs - next_idx) < RTE_DIST_BURST_SIZE) pkts = num_mbufs - next_idx; else @@ -391,21 +496,35 @@ rte_distributor_process_v1705(struct rte_distributor_v1705 *d, for (; i < RTE_DIST_BURST_SIZE; i++) flows[i] = 0; - find_match_scalar(d, &flows[0], &matches[0]); + matching_required = 1; + for (j = 0; j < pkts; j++) { + if (unlikely(!d->activesum)) + return next_idx; + + if (unlikely(matching_required)) { + switch (d->dist_match_fn) { + case RTE_DIST_MATCH_VECTOR: + find_match_vec(d, &flows[0], + &matches[0]); + break; + default: + find_match_scalar(d, &flows[0], + &matches[0]); + } + matching_required = 0; + } /* * Matches array now contain the intended worker ID (+1) of * the incoming packets. Any zeroes need to be assigned * workers. */ - for (j = 0; j < pkts; j++) { - next_mb = mbufs[next_idx++]; next_value = (((int64_t)(uintptr_t)next_mb) << RTE_DISTRIB_FLAG_BITS); /* - * User is advocated to set tag vaue for each + * User is advocated to set tag value for each * mbuf before calling rte_distributor_process. * User defined tags are used to identify flows, * or sessions. @@ -415,17 +534,23 @@ rte_distributor_process_v1705(struct rte_distributor_v1705 *d, /* * Uncommenting the next line will cause the find_match - * function to be optimised out, making this function + * function to be optimized out, making this function * do parallel (non-atomic) distribution */ /* matches[j] = 0; */ - if (matches[j]) { + if (matches[j] && d->active[matches[j]-1]) { struct rte_distributor_backlog *bl = &d->backlog[matches[j]-1]; if (unlikely(bl->count == RTE_DIST_BURST_SIZE)) { release(d, matches[j]-1); + if (!d->active[matches[j]-1]) { + j--; + next_idx--; + matching_required = 1; + continue; + } } /* Add to worker that already has flow */ @@ -435,11 +560,21 @@ rte_distributor_process_v1705(struct rte_distributor_v1705 *d, bl->pkts[idx] = next_value; } else { - struct rte_distributor_backlog *bl = - &d->backlog[wkr]; + struct rte_distributor_backlog *bl; + + while (unlikely(!d->active[wkr])) + wkr = (wkr + 1) % d->num_workers; + bl = &d->backlog[wkr]; + if (unlikely(bl->count == RTE_DIST_BURST_SIZE)) { release(d, wkr); + if (!d->active[wkr]) { + j--; + next_idx--; + matching_required = 1; + continue; + } } /* Add to current worker worker */ @@ -458,22 +593,24 @@ rte_distributor_process_v1705(struct rte_distributor_v1705 *d, matches[w] = wkr+1; } } - wkr++; - if (wkr >= d->num_workers) - wkr = 0; + wkr = (wkr + 1) % d->num_workers; } /* 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)) { + d->bufs[wid].count = 0; release(d, wid); + } return num_mbufs; } /* 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; @@ -483,7 +620,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_single(d->d_single, mbufs, max_mbufs); } @@ -501,15 +638,15 @@ rte_distributor_returned_pkts_v1705(struct rte_distributor_v1705 *d, /* * Return the number of packets in-flight in a distributor, i.e. packets - * being workered on or queued up in a backlog. + * being worked 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; for (wkr = 0; wkr < d->num_workers; wkr++) - total_outstanding += d->backlog[wkr].count; + total_outstanding += d->backlog[wkr].count + d->bufs[wkr].count; return total_outstanding; } @@ -519,24 +656,29 @@ 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 flushed; 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_single(d->d_single); } + flushed = total_outstanding(d); + while (total_outstanding(d) > 0) - rte_distributor_process_v1705(d, NULL, 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. */ - 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); @@ -546,28 +688,33 @@ 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_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); + + d->returns.start = d->returns.count = 0; } /* 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; @@ -579,11 +726,22 @@ 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_v1705)); - d->d_v20 = rte_distributor_create(name, + d = malloc(sizeof(struct rte_distributor)); + if (d == NULL) { + rte_errno = ENOMEM; + return NULL; + } + 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; } @@ -591,11 +749,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) { @@ -604,25 +757,33 @@ 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; + d->dist_match_fn = RTE_DIST_MATCH_SCALAR; +#if defined(RTE_ARCH_X86) + if (rte_vect_get_max_simd_bitwidth() >= RTE_VECT_SIMD_128) + d->dist_match_fn = RTE_DIST_MATCH_VECTOR; +#endif /* - * Set up the backog tags so they're pointing at the second cache + * Set up the backlog tags so they're pointing at the second cache * line for performance during flow matching */ for (i = 0 ; i < num_workers ; i++) d->backlog[i].tags = &d->in_flight_tags[i][RTE_DIST_BURST_SIZE]; + memset(d->active, 0, sizeof(d->active)); + d->activesum = 0; + dist_burst_list = RTE_TAILQ_CAST(rte_dist_burst_tailq.head, 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; }