1 /* SPDX-License-Identifier: BSD-3-Clause
2 * Copyright(c) 2017 Intel Corporation
9 #include <rte_memory.h>
10 #include <rte_cycles.h>
11 #include <rte_function_versioning.h>
12 #include <rte_memzone.h>
13 #include <rte_errno.h>
14 #include <rte_string_fns.h>
15 #include <rte_eal_memconfig.h>
16 #include <rte_pause.h>
17 #include <rte_tailq.h>
19 #include "rte_distributor.h"
20 #include "rte_distributor_single.h"
21 #include "distributor_private.h"
23 TAILQ_HEAD(rte_dist_burst_list, rte_distributor);
25 static struct rte_tailq_elem rte_dist_burst_tailq = {
26 .name = "RTE_DIST_BURST",
28 EAL_REGISTER_TAILQ(rte_dist_burst_tailq)
30 /**** APIs called by workers ****/
32 /**** Burst Packet APIs called by workers ****/
35 rte_distributor_request_pkt(struct rte_distributor *d,
36 unsigned int worker_id, struct rte_mbuf **oldpkt,
39 struct rte_distributor_buffer *buf = &(d->bufs[worker_id]);
42 volatile int64_t *retptr64;
44 if (unlikely(d->alg_type == RTE_DIST_ALG_SINGLE)) {
45 rte_distributor_request_pkt_single(d->d_single,
46 worker_id, oldpkt[0]);
50 retptr64 = &(buf->retptr64[0]);
51 /* Spin while handshake bits are set (scheduler clears it).
52 * Sync with worker on GET_BUF flag.
54 while (unlikely(__atomic_load_n(retptr64, __ATOMIC_ACQUIRE)
55 & RTE_DISTRIB_GET_BUF)) {
57 uint64_t t = rte_rdtsc()+100;
59 while (rte_rdtsc() < t)
64 * OK, if we've got here, then the scheduler has just cleared the
65 * handshake bits. Populate the retptrs with returning packets.
68 for (i = count; i < RTE_DIST_BURST_SIZE; i++)
71 /* Set Return bit for each packet returned */
72 for (i = count; i-- > 0; )
74 (((int64_t)(uintptr_t)(oldpkt[i])) <<
75 RTE_DISTRIB_FLAG_BITS) | RTE_DISTRIB_RETURN_BUF;
78 * Finally, set the GET_BUF to signal to distributor that cache
79 * line is ready for processing
80 * Sync with distributor to release retptrs
82 __atomic_store_n(retptr64, *retptr64 | RTE_DISTRIB_GET_BUF,
87 rte_distributor_poll_pkt(struct rte_distributor *d,
88 unsigned int worker_id, struct rte_mbuf **pkts)
90 struct rte_distributor_buffer *buf = &d->bufs[worker_id];
95 if (unlikely(d->alg_type == RTE_DIST_ALG_SINGLE)) {
96 pkts[0] = rte_distributor_poll_pkt_single(d->d_single,
98 return (pkts[0]) ? 1 : 0;
101 /* If bit is set, return
102 * Sync with distributor to acquire bufptrs
104 if (__atomic_load_n(&(buf->bufptr64[0]), __ATOMIC_ACQUIRE)
105 & RTE_DISTRIB_GET_BUF)
108 /* since bufptr64 is signed, this should be an arithmetic shift */
109 for (i = 0; i < RTE_DIST_BURST_SIZE; i++) {
110 if (likely(buf->bufptr64[i] & RTE_DISTRIB_VALID_BUF)) {
111 ret = buf->bufptr64[i] >> RTE_DISTRIB_FLAG_BITS;
112 pkts[count++] = (struct rte_mbuf *)((uintptr_t)(ret));
117 * so now we've got the contents of the cacheline into an array of
118 * mbuf pointers, so toggle the bit so scheduler can start working
119 * on the next cacheline while we're working.
120 * Sync with distributor on GET_BUF flag. Release bufptrs.
122 __atomic_store_n(&(buf->bufptr64[0]),
123 buf->bufptr64[0] | RTE_DISTRIB_GET_BUF, __ATOMIC_RELEASE);
129 rte_distributor_get_pkt(struct rte_distributor *d,
130 unsigned int worker_id, struct rte_mbuf **pkts,
131 struct rte_mbuf **oldpkt, unsigned int return_count)
135 if (unlikely(d->alg_type == RTE_DIST_ALG_SINGLE)) {
136 if (return_count <= 1) {
137 pkts[0] = rte_distributor_get_pkt_single(d->d_single,
138 worker_id, oldpkt[0]);
139 return (pkts[0]) ? 1 : 0;
144 rte_distributor_request_pkt(d, worker_id, oldpkt, return_count);
146 count = rte_distributor_poll_pkt(d, worker_id, pkts);
147 while (count == -1) {
148 uint64_t t = rte_rdtsc() + 100;
150 while (rte_rdtsc() < t)
153 count = rte_distributor_poll_pkt(d, worker_id, pkts);
159 rte_distributor_return_pkt(struct rte_distributor *d,
160 unsigned int worker_id, struct rte_mbuf **oldpkt, int num)
162 struct rte_distributor_buffer *buf = &d->bufs[worker_id];
165 if (unlikely(d->alg_type == RTE_DIST_ALG_SINGLE)) {
167 return rte_distributor_return_pkt_single(d->d_single,
168 worker_id, oldpkt[0]);
173 /* Sync with distributor to acquire retptrs */
174 __atomic_thread_fence(__ATOMIC_ACQUIRE);
175 for (i = 0; i < RTE_DIST_BURST_SIZE; i++)
176 /* Switch off the return bit first */
177 buf->retptr64[i] &= ~RTE_DISTRIB_RETURN_BUF;
179 for (i = num; i-- > 0; )
180 buf->retptr64[i] = (((int64_t)(uintptr_t)oldpkt[i]) <<
181 RTE_DISTRIB_FLAG_BITS) | RTE_DISTRIB_RETURN_BUF;
183 /* set the GET_BUF but even if we got no returns.
184 * Sync with distributor on GET_BUF flag. Release retptrs.
186 __atomic_store_n(&(buf->retptr64[0]),
187 buf->retptr64[0] | RTE_DISTRIB_GET_BUF, __ATOMIC_RELEASE);
192 /**** APIs called on distributor core ***/
194 /* stores a packet returned from a worker inside the returns array */
196 store_return(uintptr_t oldbuf, struct rte_distributor *d,
197 unsigned int *ret_start, unsigned int *ret_count)
201 /* store returns in a circular buffer */
202 d->returns.mbufs[(*ret_start + *ret_count) & RTE_DISTRIB_RETURNS_MASK]
204 *ret_start += (*ret_count == RTE_DISTRIB_RETURNS_MASK);
205 *ret_count += (*ret_count != RTE_DISTRIB_RETURNS_MASK);
209 * Match then flow_ids (tags) of the incoming packets to the flow_ids
210 * of the inflight packets (both inflight on the workers and in each worker
211 * backlog). This will then allow us to pin those packets to the relevant
212 * workers to give us our atomic flow pinning.
215 find_match_scalar(struct rte_distributor *d,
217 uint16_t *output_ptr)
219 struct rte_distributor_backlog *bl;
224 * 1. Loop through all worker ID's
225 * 2. Compare the current inflights to the incoming tags
226 * 3. Compare the current backlog to the incoming tags
227 * 4. Add any matches to the output
230 for (j = 0 ; j < RTE_DIST_BURST_SIZE; j++)
233 for (i = 0; i < d->num_workers; i++) {
236 for (j = 0; j < RTE_DIST_BURST_SIZE ; j++)
237 for (w = 0; w < RTE_DIST_BURST_SIZE; w++)
238 if (d->in_flight_tags[i][j] == data_ptr[w]) {
242 for (j = 0; j < RTE_DIST_BURST_SIZE; j++)
243 for (w = 0; w < RTE_DIST_BURST_SIZE; w++)
244 if (bl->tags[j] == data_ptr[w]) {
251 * At this stage, the output contains 8 16-bit values, with
252 * each non-zero value containing the worker ID on which the
253 * corresponding flow is pinned to.
259 * When the handshake bits indicate that there are packets coming
260 * back from the worker, this function is called to copy and store
261 * the valid returned pointers (store_return).
264 handle_returns(struct rte_distributor *d, unsigned int wkr)
266 struct rte_distributor_buffer *buf = &(d->bufs[wkr]);
268 unsigned int ret_start = d->returns.start,
269 ret_count = d->returns.count;
270 unsigned int count = 0;
273 /* Sync on GET_BUF flag. Acquire retptrs. */
274 if (__atomic_load_n(&(buf->retptr64[0]), __ATOMIC_ACQUIRE)
275 & RTE_DISTRIB_GET_BUF) {
276 for (i = 0; i < RTE_DIST_BURST_SIZE; i++) {
277 if (buf->retptr64[i] & RTE_DISTRIB_RETURN_BUF) {
278 oldbuf = ((uintptr_t)(buf->retptr64[i] >>
279 RTE_DISTRIB_FLAG_BITS));
280 /* store returns in a circular buffer */
281 store_return(oldbuf, d, &ret_start, &ret_count);
283 buf->retptr64[i] &= ~RTE_DISTRIB_RETURN_BUF;
286 d->returns.start = ret_start;
287 d->returns.count = ret_count;
288 /* Clear for the worker to populate with more returns.
289 * Sync with distributor on GET_BUF flag. Release retptrs.
291 __atomic_store_n(&(buf->retptr64[0]), 0, __ATOMIC_RELEASE);
297 * This function releases a burst (cache line) to a worker.
298 * It is called from the process function when a cacheline is
299 * full to make room for more packets for that worker, or when
300 * all packets have been assigned to bursts and need to be flushed
302 * It also needs to wait for any outstanding packets from the worker
303 * before sending out new packets.
306 release(struct rte_distributor *d, unsigned int wkr)
308 struct rte_distributor_buffer *buf = &(d->bufs[wkr]);
311 /* Sync with worker on GET_BUF flag */
312 while (!(__atomic_load_n(&(d->bufs[wkr].bufptr64[0]), __ATOMIC_ACQUIRE)
313 & RTE_DISTRIB_GET_BUF))
316 handle_returns(d, wkr);
320 for (i = 0; i < d->backlog[wkr].count; i++) {
321 d->bufs[wkr].bufptr64[i] = d->backlog[wkr].pkts[i] |
322 RTE_DISTRIB_GET_BUF | RTE_DISTRIB_VALID_BUF;
323 d->in_flight_tags[wkr][i] = d->backlog[wkr].tags[i];
326 for ( ; i < RTE_DIST_BURST_SIZE ; i++) {
327 buf->bufptr64[i] = RTE_DISTRIB_GET_BUF;
328 d->in_flight_tags[wkr][i] = 0;
331 d->backlog[wkr].count = 0;
333 /* Clear the GET bit.
334 * Sync with worker on GET_BUF flag. Release bufptrs.
336 __atomic_store_n(&(buf->bufptr64[0]),
337 buf->bufptr64[0] & ~RTE_DISTRIB_GET_BUF, __ATOMIC_RELEASE);
343 /* process a set of packets to distribute them to workers */
345 rte_distributor_process(struct rte_distributor *d,
346 struct rte_mbuf **mbufs, unsigned int num_mbufs)
348 unsigned int next_idx = 0;
349 static unsigned int wkr;
350 struct rte_mbuf *next_mb = NULL;
351 int64_t next_value = 0;
352 uint16_t new_tag = 0;
353 uint16_t flows[RTE_DIST_BURST_SIZE] __rte_cache_aligned;
354 unsigned int i, j, w, wid;
356 if (d->alg_type == RTE_DIST_ALG_SINGLE) {
357 /* Call the old API */
358 return rte_distributor_process_single(d->d_single,
362 if (unlikely(num_mbufs == 0)) {
363 /* Flush out all non-full cache-lines to workers. */
364 for (wid = 0 ; wid < d->num_workers; wid++) {
365 /* Sync with worker on GET_BUF flag. */
366 if (__atomic_load_n(&(d->bufs[wid].bufptr64[0]),
367 __ATOMIC_ACQUIRE) & RTE_DISTRIB_GET_BUF) {
369 handle_returns(d, wid);
375 while (next_idx < num_mbufs) {
376 uint16_t matches[RTE_DIST_BURST_SIZE];
379 /* Sync with worker on GET_BUF flag. */
380 if (__atomic_load_n(&(d->bufs[wkr].bufptr64[0]),
381 __ATOMIC_ACQUIRE) & RTE_DISTRIB_GET_BUF)
382 d->bufs[wkr].count = 0;
384 if ((num_mbufs - next_idx) < RTE_DIST_BURST_SIZE)
385 pkts = num_mbufs - next_idx;
387 pkts = RTE_DIST_BURST_SIZE;
389 for (i = 0; i < pkts; i++) {
390 if (mbufs[next_idx + i]) {
391 /* flows have to be non-zero */
392 flows[i] = mbufs[next_idx + i]->hash.usr | 1;
396 for (; i < RTE_DIST_BURST_SIZE; i++)
399 switch (d->dist_match_fn) {
400 case RTE_DIST_MATCH_VECTOR:
401 find_match_vec(d, &flows[0], &matches[0]);
404 find_match_scalar(d, &flows[0], &matches[0]);
408 * Matches array now contain the intended worker ID (+1) of
409 * the incoming packets. Any zeroes need to be assigned
413 for (j = 0; j < pkts; j++) {
415 next_mb = mbufs[next_idx++];
416 next_value = (((int64_t)(uintptr_t)next_mb) <<
417 RTE_DISTRIB_FLAG_BITS);
419 * User is advocated to set tag value for each
420 * mbuf before calling rte_distributor_process.
421 * User defined tags are used to identify flows,
424 /* flows MUST be non-zero */
425 new_tag = (uint16_t)(next_mb->hash.usr) | 1;
428 * Uncommenting the next line will cause the find_match
429 * function to be optimized out, making this function
430 * do parallel (non-atomic) distribution
432 /* matches[j] = 0; */
435 struct rte_distributor_backlog *bl =
436 &d->backlog[matches[j]-1];
437 if (unlikely(bl->count ==
438 RTE_DIST_BURST_SIZE)) {
439 release(d, matches[j]-1);
442 /* Add to worker that already has flow */
443 unsigned int idx = bl->count++;
445 bl->tags[idx] = new_tag;
446 bl->pkts[idx] = next_value;
449 struct rte_distributor_backlog *bl =
451 if (unlikely(bl->count ==
452 RTE_DIST_BURST_SIZE)) {
456 /* Add to current worker worker */
457 unsigned int idx = bl->count++;
459 bl->tags[idx] = new_tag;
460 bl->pkts[idx] = next_value;
462 * Now that we've just added an unpinned flow
463 * to a worker, we need to ensure that all
464 * other packets with that same flow will go
465 * to the same worker in this burst.
467 for (w = j; w < pkts; w++)
468 if (flows[w] == new_tag)
473 if (wkr >= d->num_workers)
477 /* Flush out all non-full cache-lines to workers. */
478 for (wid = 0 ; wid < d->num_workers; wid++)
479 /* Sync with worker on GET_BUF flag. */
480 if ((__atomic_load_n(&(d->bufs[wid].bufptr64[0]),
481 __ATOMIC_ACQUIRE) & RTE_DISTRIB_GET_BUF))
487 /* return to the caller, packets returned from workers */
489 rte_distributor_returned_pkts(struct rte_distributor *d,
490 struct rte_mbuf **mbufs, unsigned int max_mbufs)
492 struct rte_distributor_returned_pkts *returns = &d->returns;
493 unsigned int retval = (max_mbufs < returns->count) ?
494 max_mbufs : returns->count;
497 if (d->alg_type == RTE_DIST_ALG_SINGLE) {
498 /* Call the old API */
499 return rte_distributor_returned_pkts_single(d->d_single,
503 for (i = 0; i < retval; i++) {
504 unsigned int idx = (returns->start + i) &
505 RTE_DISTRIB_RETURNS_MASK;
507 mbufs[i] = returns->mbufs[idx];
516 * Return the number of packets in-flight in a distributor, i.e. packets
517 * being worked on or queued up in a backlog.
519 static inline unsigned int
520 total_outstanding(const struct rte_distributor *d)
522 unsigned int wkr, total_outstanding = 0;
524 for (wkr = 0; wkr < d->num_workers; wkr++)
525 total_outstanding += d->backlog[wkr].count;
527 return total_outstanding;
531 * Flush the distributor, so that there are no outstanding packets in flight or
535 rte_distributor_flush(struct rte_distributor *d)
537 unsigned int flushed;
540 if (d->alg_type == RTE_DIST_ALG_SINGLE) {
541 /* Call the old API */
542 return rte_distributor_flush_single(d->d_single);
545 flushed = total_outstanding(d);
547 while (total_outstanding(d) > 0)
548 rte_distributor_process(d, NULL, 0);
550 /* wait 10ms to allow all worker drain the pkts */
554 * Send empty burst to all workers to allow them to exit
555 * gracefully, should they need to.
557 rte_distributor_process(d, NULL, 0);
559 for (wkr = 0; wkr < d->num_workers; wkr++)
560 handle_returns(d, wkr);
565 /* clears the internal returns array in the distributor */
567 rte_distributor_clear_returns(struct rte_distributor *d)
571 if (d->alg_type == RTE_DIST_ALG_SINGLE) {
572 /* Call the old API */
573 rte_distributor_clear_returns_single(d->d_single);
577 /* throw away returns, so workers can exit */
578 for (wkr = 0; wkr < d->num_workers; wkr++)
579 /* Sync with worker. Release retptrs. */
580 __atomic_store_n(&(d->bufs[wkr].retptr64[0]), 0,
584 /* creates a distributor instance */
585 struct rte_distributor *
586 rte_distributor_create(const char *name,
587 unsigned int socket_id,
588 unsigned int num_workers,
589 unsigned int alg_type)
591 struct rte_distributor *d;
592 struct rte_dist_burst_list *dist_burst_list;
593 char mz_name[RTE_MEMZONE_NAMESIZE];
594 const struct rte_memzone *mz;
597 /* TODO Reorganise function properly around RTE_DIST_ALG_SINGLE/BURST */
599 /* compilation-time checks */
600 RTE_BUILD_BUG_ON((sizeof(*d) & RTE_CACHE_LINE_MASK) != 0);
601 RTE_BUILD_BUG_ON((RTE_DISTRIB_MAX_WORKERS & 7) != 0);
603 if (name == NULL || num_workers >=
604 (unsigned int)RTE_MIN(RTE_DISTRIB_MAX_WORKERS, RTE_MAX_LCORE)) {
609 if (alg_type == RTE_DIST_ALG_SINGLE) {
610 d = malloc(sizeof(struct rte_distributor));
615 d->d_single = rte_distributor_create_single(name,
616 socket_id, num_workers);
617 if (d->d_single == NULL) {
619 /* rte_errno will have been set */
622 d->alg_type = alg_type;
626 snprintf(mz_name, sizeof(mz_name), RTE_DISTRIB_PREFIX"%s", name);
627 mz = rte_memzone_reserve(mz_name, sizeof(*d), socket_id, NO_FLAGS);
634 strlcpy(d->name, name, sizeof(d->name));
635 d->num_workers = num_workers;
636 d->alg_type = alg_type;
638 d->dist_match_fn = RTE_DIST_MATCH_SCALAR;
639 #if defined(RTE_ARCH_X86)
640 d->dist_match_fn = RTE_DIST_MATCH_VECTOR;
644 * Set up the backlog tags so they're pointing at the second cache
645 * line for performance during flow matching
647 for (i = 0 ; i < num_workers ; i++)
648 d->backlog[i].tags = &d->in_flight_tags[i][RTE_DIST_BURST_SIZE];
650 dist_burst_list = RTE_TAILQ_CAST(rte_dist_burst_tailq.head,
651 rte_dist_burst_list);
654 rte_mcfg_tailq_write_lock();
655 TAILQ_INSERT_TAIL(dist_burst_list, d, next);
656 rte_mcfg_tailq_write_unlock();