lib/librte_distributor/rte_distributor.c

   1 /* SPDX-License-Identifier: BSD-3-Clause
   2  * Copyright(c) 2017 Intel Corporation
   3  */
   4
   5 #include <stdio.h>
   6 #include <sys/queue.h>
   7 #include <string.h>
   8 #include <rte_mbuf.h>
   9 #include <rte_memory.h>
  10 #include <rte_cycles.h>
  11 #include <rte_memzone.h>
  12 #include <rte_errno.h>
  13 #include <rte_string_fns.h>
  14 #include <rte_eal_memconfig.h>
  15 #include <rte_pause.h>
  16 #include <rte_tailq.h>
  17
  18 #include "rte_distributor.h"
  19 #include "rte_distributor_single.h"
  20 #include "distributor_private.h"
  21
  22 TAILQ_HEAD(rte_dist_burst_list, rte_distributor);
  23
  24 static struct rte_tailq_elem rte_dist_burst_tailq = {
  25         .name = "RTE_DIST_BURST",
  26 };
  27 EAL_REGISTER_TAILQ(rte_dist_burst_tailq)
  28
  29 /**** APIs called by workers ****/
  30
  31 /**** Burst Packet APIs called by workers ****/
  32
  33 void
  34 rte_distributor_request_pkt(struct rte_distributor *d,
  35                 unsigned int worker_id, struct rte_mbuf **oldpkt,
  36                 unsigned int count)
  37 {
  38         struct rte_distributor_buffer *buf = &(d->bufs[worker_id]);
  39         unsigned int i;
  40
  41         volatile int64_t *retptr64;
  42
  43         if (unlikely(d->alg_type == RTE_DIST_ALG_SINGLE)) {
  44                 rte_distributor_request_pkt_single(d->d_single,
  45                         worker_id, count ? oldpkt[0] : NULL);
  46                 return;
  47         }
  48
  49         retptr64 = &(buf->retptr64[0]);
  50         /* Spin while handshake bits are set (scheduler clears it).
  51          * Sync with worker on GET_BUF flag.
  52          */
  53         while (unlikely(__atomic_load_n(retptr64, __ATOMIC_ACQUIRE)
  54                         & (RTE_DISTRIB_GET_BUF | RTE_DISTRIB_RETURN_BUF))) {
  55                 rte_pause();
  56                 uint64_t t = rte_rdtsc()+100;
  57
  58                 while (rte_rdtsc() < t)
  59                         rte_pause();
  60         }
  61
  62         /*
  63          * OK, if we've got here, then the scheduler has just cleared the
  64          * handshake bits. Populate the retptrs with returning packets.
  65          */
  66
  67         for (i = count; i < RTE_DIST_BURST_SIZE; i++)
  68                 buf->retptr64[i] = 0;
  69
  70         /* Set VALID_BUF bit for each packet returned */
  71         for (i = count; i-- > 0; )
  72                 buf->retptr64[i] =
  73                         (((int64_t)(uintptr_t)(oldpkt[i])) <<
  74                         RTE_DISTRIB_FLAG_BITS) | RTE_DISTRIB_VALID_BUF;
  75
  76         /*
  77          * Finally, set the GET_BUF  to signal to distributor that cache
  78          * line is ready for processing
  79          * Sync with distributor to release retptrs
  80          */
  81         __atomic_store_n(retptr64, *retptr64 | RTE_DISTRIB_GET_BUF,
  82                         __ATOMIC_RELEASE);
  83 }
  84
  85 int
  86 rte_distributor_poll_pkt(struct rte_distributor *d,
  87                 unsigned int worker_id, struct rte_mbuf **pkts)
  88 {
  89         struct rte_distributor_buffer *buf = &d->bufs[worker_id];
  90         uint64_t ret;
  91         int count = 0;
  92         unsigned int i;
  93
  94         if (unlikely(d->alg_type == RTE_DIST_ALG_SINGLE)) {
  95                 pkts[0] = rte_distributor_poll_pkt_single(d->d_single,
  96                         worker_id);
  97                 return (pkts[0]) ? 1 : 0;
  98         }
  99
 100         /* If any of below bits is set, return.
 101          * GET_BUF is set when distributor hasn't sent any packets yet
 102          * RETURN_BUF is set when distributor must retrieve in-flight packets
 103          * Sync with distributor to acquire bufptrs
 104          */
 105         if (__atomic_load_n(&(buf->bufptr64[0]), __ATOMIC_ACQUIRE)
 106                 & (RTE_DISTRIB_GET_BUF | RTE_DISTRIB_RETURN_BUF))
 107                 return -1;
 108
 109         /* since bufptr64 is signed, this should be an arithmetic shift */
 110         for (i = 0; i < RTE_DIST_BURST_SIZE; i++) {
 111                 if (likely(buf->bufptr64[i] & RTE_DISTRIB_VALID_BUF)) {
 112                         ret = buf->bufptr64[i] >> RTE_DISTRIB_FLAG_BITS;
 113                         pkts[count++] = (struct rte_mbuf *)((uintptr_t)(ret));
 114                 }
 115         }
 116
 117         /*
 118          * so now we've got the contents of the cacheline into an array of
 119          * mbuf pointers, so toggle the bit so scheduler can start working
 120          * on the next cacheline while we're working.
 121          * Sync with distributor on GET_BUF flag. Release bufptrs.
 122          */
 123         __atomic_store_n(&(buf->bufptr64[0]),
 124                 buf->bufptr64[0] | RTE_DISTRIB_GET_BUF, __ATOMIC_RELEASE);
 125
 126         return count;
 127 }
 128
 129 int
 130 rte_distributor_get_pkt(struct rte_distributor *d,
 131                 unsigned int worker_id, struct rte_mbuf **pkts,
 132                 struct rte_mbuf **oldpkt, unsigned int return_count)
 133 {
 134         int count;
 135
 136         if (unlikely(d->alg_type == RTE_DIST_ALG_SINGLE)) {
 137                 if (return_count <= 1) {
 138                         pkts[0] = rte_distributor_get_pkt_single(d->d_single,
 139                                 worker_id, return_count ? oldpkt[0] : NULL);
 140                         return (pkts[0]) ? 1 : 0;
 141                 } else
 142                         return -EINVAL;
 143         }
 144
 145         rte_distributor_request_pkt(d, worker_id, oldpkt, return_count);
 146
 147         count = rte_distributor_poll_pkt(d, worker_id, pkts);
 148         while (count == -1) {
 149                 uint64_t t = rte_rdtsc() + 100;
 150
 151                 while (rte_rdtsc() < t)
 152                         rte_pause();
 153
 154                 count = rte_distributor_poll_pkt(d, worker_id, pkts);
 155         }
 156         return count;
 157 }
 158
 159 int
 160 rte_distributor_return_pkt(struct rte_distributor *d,
 161                 unsigned int worker_id, struct rte_mbuf **oldpkt, int num)
 162 {
 163         struct rte_distributor_buffer *buf = &d->bufs[worker_id];
 164         unsigned int i;
 165
 166         if (unlikely(d->alg_type == RTE_DIST_ALG_SINGLE)) {
 167                 if (num == 1)
 168                         return rte_distributor_return_pkt_single(d->d_single,
 169                                 worker_id, oldpkt[0]);
 170                 else if (num == 0)
 171                         return rte_distributor_return_pkt_single(d->d_single,
 172                                 worker_id, NULL);
 173                 else
 174                         return -EINVAL;
 175         }
 176
 177         /* Spin while handshake bits are set (scheduler clears it).
 178          * Sync with worker on GET_BUF flag.
 179          */
 180         while (unlikely(__atomic_load_n(&(buf->retptr64[0]), __ATOMIC_RELAXED)
 181                         & (RTE_DISTRIB_GET_BUF | RTE_DISTRIB_RETURN_BUF))) {
 182                 rte_pause();
 183                 uint64_t t = rte_rdtsc()+100;
 184
 185                 while (rte_rdtsc() < t)
 186                         rte_pause();
 187         }
 188
 189         /* Sync with distributor to acquire retptrs */
 190         __atomic_thread_fence(__ATOMIC_ACQUIRE);
 191         for (i = 0; i < RTE_DIST_BURST_SIZE; i++)
 192                 /* Switch off the return bit first */
 193                 buf->retptr64[i] = 0;
 194
 195         for (i = num; i-- > 0; )
 196                 buf->retptr64[i] = (((int64_t)(uintptr_t)oldpkt[i]) <<
 197                         RTE_DISTRIB_FLAG_BITS) | RTE_DISTRIB_VALID_BUF;
 198
 199         /* Use RETURN_BUF on bufptr64 to notify distributor that
 200          * we won't read any mbufs from there even if GET_BUF is set.
 201          * This allows distributor to retrieve in-flight already sent packets.
 202          */
 203         __atomic_or_fetch(&(buf->bufptr64[0]), RTE_DISTRIB_RETURN_BUF,
 204                 __ATOMIC_ACQ_REL);
 205
 206         /* set the RETURN_BUF on retptr64 even if we got no returns.
 207          * Sync with distributor on RETURN_BUF flag. Release retptrs.
 208          * Notify distributor that we don't request more packets any more.
 209          */
 210         __atomic_store_n(&(buf->retptr64[0]),
 211                 buf->retptr64[0] | RTE_DISTRIB_RETURN_BUF, __ATOMIC_RELEASE);
 212
 213         return 0;
 214 }
 215
 216 /**** APIs called on distributor core ***/
 217
 218 /* stores a packet returned from a worker inside the returns array */
 219 static inline void
 220 store_return(uintptr_t oldbuf, struct rte_distributor *d,
 221                 unsigned int *ret_start, unsigned int *ret_count)
 222 {
 223         if (!oldbuf)
 224                 return;
 225         /* store returns in a circular buffer */
 226         d->returns.mbufs[(*ret_start + *ret_count) & RTE_DISTRIB_RETURNS_MASK]
 227                         = (void *)oldbuf;
 228         *ret_start += (*ret_count == RTE_DISTRIB_RETURNS_MASK);
 229         *ret_count += (*ret_count != RTE_DISTRIB_RETURNS_MASK);
 230 }
 231
 232 /*
 233  * Match then flow_ids (tags) of the incoming packets to the flow_ids
 234  * of the inflight packets (both inflight on the workers and in each worker
 235  * backlog). This will then allow us to pin those packets to the relevant
 236  * workers to give us our atomic flow pinning.
 237  */
 238 void
 239 find_match_scalar(struct rte_distributor *d,
 240                         uint16_t *data_ptr,
 241                         uint16_t *output_ptr)
 242 {
 243         struct rte_distributor_backlog *bl;
 244         uint16_t i, j, w;
 245
 246         /*
 247          * Function overview:
 248          * 1. Loop through all worker ID's
 249          * 2. Compare the current inflights to the incoming tags
 250          * 3. Compare the current backlog to the incoming tags
 251          * 4. Add any matches to the output
 252          */
 253
 254         for (j = 0 ; j < RTE_DIST_BURST_SIZE; j++)
 255                 output_ptr[j] = 0;
 256
 257         for (i = 0; i < d->num_workers; i++) {
 258                 bl = &d->backlog[i];
 259
 260                 for (j = 0; j < RTE_DIST_BURST_SIZE ; j++)
 261                         for (w = 0; w < RTE_DIST_BURST_SIZE; w++)
 262                                 if (d->in_flight_tags[i][w] == data_ptr[j]) {
 263                                         output_ptr[j] = i+1;
 264                                         break;
 265                                 }
 266                 for (j = 0; j < RTE_DIST_BURST_SIZE; j++)
 267                         for (w = 0; w < RTE_DIST_BURST_SIZE; w++)
 268                                 if (bl->tags[w] == data_ptr[j]) {
 269                                         output_ptr[j] = i+1;
 270                                         break;
 271                                 }
 272         }
 273
 274         /*
 275          * At this stage, the output contains 8 16-bit values, with
 276          * each non-zero value containing the worker ID on which the
 277          * corresponding flow is pinned to.
 278          */
 279 }
 280
 281 /*
 282  * When worker called rte_distributor_return_pkt()
 283  * and passed RTE_DISTRIB_RETURN_BUF handshake through retptr64,
 284  * distributor must retrieve both inflight and backlog packets assigned
 285  * to the worker and reprocess them to another worker.
 286  */
 287 static void
 288 handle_worker_shutdown(struct rte_distributor *d, unsigned int wkr)
 289 {
 290         struct rte_distributor_buffer *buf = &(d->bufs[wkr]);
 291         /* double BURST size for storing both inflights and backlog */
 292         struct rte_mbuf *pkts[RTE_DIST_BURST_SIZE * 2];
 293         unsigned int pkts_count = 0;
 294         unsigned int i;
 295
 296         /* If GET_BUF is cleared there are in-flight packets sent
 297          * to worker which does not require new packets.
 298          * They must be retrieved and assigned to another worker.
 299          */
 300         if (!(__atomic_load_n(&(buf->bufptr64[0]), __ATOMIC_ACQUIRE)
 301                 & RTE_DISTRIB_GET_BUF))
 302                 for (i = 0; i < RTE_DIST_BURST_SIZE; i++)
 303                         if (buf->bufptr64[i] & RTE_DISTRIB_VALID_BUF)
 304                                 pkts[pkts_count++] = (void *)((uintptr_t)
 305                                         (buf->bufptr64[i]
 306                                                 >> RTE_DISTRIB_FLAG_BITS));
 307
 308         /* Make following operations on handshake flags on bufptr64:
 309          * - set GET_BUF to indicate that distributor can overwrite buffer
 310          *     with new packets if worker will make a new request.
 311          * - clear RETURN_BUF to unlock reads on worker side.
 312          */
 313         __atomic_store_n(&(buf->bufptr64[0]), RTE_DISTRIB_GET_BUF,
 314                 __ATOMIC_RELEASE);
 315
 316         /* Collect backlog packets from worker */
 317         for (i = 0; i < d->backlog[wkr].count; i++)
 318                 pkts[pkts_count++] = (void *)((uintptr_t)
 319                         (d->backlog[wkr].pkts[i] >> RTE_DISTRIB_FLAG_BITS));
 320
 321         d->backlog[wkr].count = 0;
 322
 323         /* Clear both inflight and backlog tags */
 324         for (i = 0; i < RTE_DIST_BURST_SIZE; i++) {
 325                 d->in_flight_tags[wkr][i] = 0;
 326                 d->backlog[wkr].tags[i] = 0;
 327         }
 328
 329         /* Recursive call */
 330         if (pkts_count > 0)
 331                 rte_distributor_process(d, pkts, pkts_count);
 332 }
 333
 334
 335 /*
 336  * When the handshake bits indicate that there are packets coming
 337  * back from the worker, this function is called to copy and store
 338  * the valid returned pointers (store_return).
 339  */
 340 static unsigned int
 341 handle_returns(struct rte_distributor *d, unsigned int wkr)
 342 {
 343         struct rte_distributor_buffer *buf = &(d->bufs[wkr]);
 344         uintptr_t oldbuf;
 345         unsigned int ret_start = d->returns.start,
 346                         ret_count = d->returns.count;
 347         unsigned int count = 0;
 348         unsigned int i;
 349
 350         /* Sync on GET_BUF flag. Acquire retptrs. */
 351         if (__atomic_load_n(&(buf->retptr64[0]), __ATOMIC_ACQUIRE)
 352                 & (RTE_DISTRIB_GET_BUF | RTE_DISTRIB_RETURN_BUF)) {
 353                 for (i = 0; i < RTE_DIST_BURST_SIZE; i++) {
 354                         if (buf->retptr64[i] & RTE_DISTRIB_VALID_BUF) {
 355                                 oldbuf = ((uintptr_t)(buf->retptr64[i] >>
 356                                         RTE_DISTRIB_FLAG_BITS));
 357                                 /* store returns in a circular buffer */
 358                                 store_return(oldbuf, d, &ret_start, &ret_count);
 359                                 count++;
 360                                 buf->retptr64[i] &= ~RTE_DISTRIB_VALID_BUF;
 361                         }
 362                 }
 363                 d->returns.start = ret_start;
 364                 d->returns.count = ret_count;
 365
 366                 /* If worker requested packets with GET_BUF, set it to active
 367                  * otherwise (RETURN_BUF), set it to not active.
 368                  */
 369                 d->activesum -= d->active[wkr];
 370                 d->active[wkr] = !!(buf->retptr64[0] & RTE_DISTRIB_GET_BUF);
 371                 d->activesum += d->active[wkr];
 372
 373                 /* If worker returned packets without requesting new ones,
 374                  * handle all in-flights and backlog packets assigned to it.
 375                  */
 376                 if (unlikely(buf->retptr64[0] & RTE_DISTRIB_RETURN_BUF))
 377                         handle_worker_shutdown(d, wkr);
 378
 379                 /* Clear for the worker to populate with more returns.
 380                  * Sync with distributor on GET_BUF flag. Release retptrs.
 381                  */
 382                 __atomic_store_n(&(buf->retptr64[0]), 0, __ATOMIC_RELEASE);
 383         }
 384         return count;
 385 }
 386
 387 /*
 388  * This function releases a burst (cache line) to a worker.
 389  * It is called from the process function when a cacheline is
 390  * full to make room for more packets for that worker, or when
 391  * all packets have been assigned to bursts and need to be flushed
 392  * to the workers.
 393  * It also needs to wait for any outstanding packets from the worker
 394  * before sending out new packets.
 395  */
 396 static unsigned int
 397 release(struct rte_distributor *d, unsigned int wkr)
 398 {
 399         struct rte_distributor_buffer *buf = &(d->bufs[wkr]);
 400         unsigned int i;
 401
 402         handle_returns(d, wkr);
 403         if (unlikely(!d->active[wkr]))
 404                 return 0;
 405
 406         /* Sync with worker on GET_BUF flag */
 407         while (!(__atomic_load_n(&(d->bufs[wkr].bufptr64[0]), __ATOMIC_ACQUIRE)
 408                 & RTE_DISTRIB_GET_BUF)) {
 409                 handle_returns(d, wkr);
 410                 if (unlikely(!d->active[wkr]))
 411                         return 0;
 412                 rte_pause();
 413         }
 414
 415         buf->count = 0;
 416
 417         for (i = 0; i < d->backlog[wkr].count; i++) {
 418                 d->bufs[wkr].bufptr64[i] = d->backlog[wkr].pkts[i] |
 419                                 RTE_DISTRIB_GET_BUF | RTE_DISTRIB_VALID_BUF;
 420                 d->in_flight_tags[wkr][i] = d->backlog[wkr].tags[i];
 421         }
 422         buf->count = i;
 423         for ( ; i < RTE_DIST_BURST_SIZE ; i++) {
 424                 buf->bufptr64[i] = RTE_DISTRIB_GET_BUF;
 425                 d->in_flight_tags[wkr][i] = 0;
 426         }
 427
 428         d->backlog[wkr].count = 0;
 429
 430         /* Clear the GET bit.
 431          * Sync with worker on GET_BUF flag. Release bufptrs.
 432          */
 433         __atomic_store_n(&(buf->bufptr64[0]),
 434                 buf->bufptr64[0] & ~RTE_DISTRIB_GET_BUF, __ATOMIC_RELEASE);
 435         return  buf->count;
 436
 437 }
 438
 439
 440 /* process a set of packets to distribute them to workers */
 441 int
 442 rte_distributor_process(struct rte_distributor *d,
 443                 struct rte_mbuf **mbufs, unsigned int num_mbufs)
 444 {
 445         unsigned int next_idx = 0;
 446         static unsigned int wkr;
 447         struct rte_mbuf *next_mb = NULL;
 448         int64_t next_value = 0;
 449         uint16_t new_tag = 0;
 450         uint16_t flows[RTE_DIST_BURST_SIZE] __rte_cache_aligned;
 451         unsigned int i, j, w, wid, matching_required;
 452
 453         if (d->alg_type == RTE_DIST_ALG_SINGLE) {
 454                 /* Call the old API */
 455                 return rte_distributor_process_single(d->d_single,
 456                         mbufs, num_mbufs);
 457         }
 458
 459         for (wid = 0 ; wid < d->num_workers; wid++)
 460                 handle_returns(d, wid);
 461
 462         if (unlikely(num_mbufs == 0)) {
 463                 /* Flush out all non-full cache-lines to workers. */
 464                 for (wid = 0 ; wid < d->num_workers; wid++) {
 465                         /* Sync with worker on GET_BUF flag. */
 466                         if (__atomic_load_n(&(d->bufs[wid].bufptr64[0]),
 467                                 __ATOMIC_ACQUIRE) & RTE_DISTRIB_GET_BUF) {
 468                                 release(d, wid);
 469                                 handle_returns(d, wid);
 470                         }
 471                 }
 472                 return 0;
 473         }
 474
 475         if (unlikely(!d->activesum))
 476                 return 0;
 477
 478         while (next_idx < num_mbufs) {
 479                 uint16_t matches[RTE_DIST_BURST_SIZE];
 480                 unsigned int pkts;
 481
 482                 /* Sync with worker on GET_BUF flag. */
 483                 if (__atomic_load_n(&(d->bufs[wkr].bufptr64[0]),
 484                         __ATOMIC_ACQUIRE) & RTE_DISTRIB_GET_BUF)
 485                         d->bufs[wkr].count = 0;
 486
 487                 if ((num_mbufs - next_idx) < RTE_DIST_BURST_SIZE)
 488                         pkts = num_mbufs - next_idx;
 489                 else
 490                         pkts = RTE_DIST_BURST_SIZE;
 491
 492                 for (i = 0; i < pkts; i++) {
 493                         if (mbufs[next_idx + i]) {
 494                                 /* flows have to be non-zero */
 495                                 flows[i] = mbufs[next_idx + i]->hash.usr | 1;
 496                         } else
 497                                 flows[i] = 0;
 498                 }
 499                 for (; i < RTE_DIST_BURST_SIZE; i++)
 500                         flows[i] = 0;
 501
 502                 matching_required = 1;
 503
 504                 for (j = 0; j < pkts; j++) {
 505                         if (unlikely(!d->activesum))
 506                                 return next_idx;
 507
 508                         if (unlikely(matching_required)) {
 509                                 switch (d->dist_match_fn) {
 510                                 case RTE_DIST_MATCH_VECTOR:
 511                                         find_match_vec(d, &flows[0],
 512                                                 &matches[0]);
 513                                         break;
 514                                 default:
 515                                         find_match_scalar(d, &flows[0],
 516                                                 &matches[0]);
 517                                 }
 518                                 matching_required = 0;
 519                         }
 520                 /*
 521                  * Matches array now contain the intended worker ID (+1) of
 522                  * the incoming packets. Any zeroes need to be assigned
 523                  * workers.
 524                  */
 525
 526                         next_mb = mbufs[next_idx++];
 527                         next_value = (((int64_t)(uintptr_t)next_mb) <<
 528                                         RTE_DISTRIB_FLAG_BITS);
 529                         /*
 530                          * User is advocated to set tag value for each
 531                          * mbuf before calling rte_distributor_process.
 532                          * User defined tags are used to identify flows,
 533                          * or sessions.
 534                          */
 535                         /* flows MUST be non-zero */
 536                         new_tag = (uint16_t)(next_mb->hash.usr) | 1;
 537
 538                         /*
 539                          * Uncommenting the next line will cause the find_match
 540                          * function to be optimized out, making this function
 541                          * do parallel (non-atomic) distribution
 542                          */
 543                         /* matches[j] = 0; */
 544
 545                         if (matches[j] && d->active[matches[j]-1]) {
 546                                 struct rte_distributor_backlog *bl =
 547                                                 &d->backlog[matches[j]-1];
 548                                 if (unlikely(bl->count ==
 549                                                 RTE_DIST_BURST_SIZE)) {
 550                                         release(d, matches[j]-1);
 551                                         if (!d->active[matches[j]-1]) {
 552                                                 j--;
 553                                                 next_idx--;
 554                                                 matching_required = 1;
 555                                                 continue;
 556                                         }
 557                                 }
 558
 559                                 /* Add to worker that already has flow */
 560                                 unsigned int idx = bl->count++;
 561
 562                                 bl->tags[idx] = new_tag;
 563                                 bl->pkts[idx] = next_value;
 564
 565                         } else {
 566                                 struct rte_distributor_backlog *bl;
 567
 568                                 while (unlikely(!d->active[wkr]))
 569                                         wkr = (wkr + 1) % d->num_workers;
 570                                 bl = &d->backlog[wkr];
 571
 572                                 if (unlikely(bl->count ==
 573                                                 RTE_DIST_BURST_SIZE)) {
 574                                         release(d, wkr);
 575                                         if (!d->active[wkr]) {
 576                                                 j--;
 577                                                 next_idx--;
 578                                                 matching_required = 1;
 579                                                 continue;
 580                                         }
 581                                 }
 582
 583                                 /* Add to current worker worker */
 584                                 unsigned int idx = bl->count++;
 585
 586                                 bl->tags[idx] = new_tag;
 587                                 bl->pkts[idx] = next_value;
 588                                 /*
 589                                  * Now that we've just added an unpinned flow
 590                                  * to a worker, we need to ensure that all
 591                                  * other packets with that same flow will go
 592                                  * to the same worker in this burst.
 593                                  */
 594                                 for (w = j; w < pkts; w++)
 595                                         if (flows[w] == new_tag)
 596                                                 matches[w] = wkr+1;
 597                         }
 598                 }
 599                 wkr = (wkr + 1) % d->num_workers;
 600         }
 601
 602         /* Flush out all non-full cache-lines to workers. */
 603         for (wid = 0 ; wid < d->num_workers; wid++)
 604                 /* Sync with worker on GET_BUF flag. */
 605                 if ((__atomic_load_n(&(d->bufs[wid].bufptr64[0]),
 606                         __ATOMIC_ACQUIRE) & RTE_DISTRIB_GET_BUF))
 607                         release(d, wid);
 608
 609         return num_mbufs;
 610 }
 611
 612 /* return to the caller, packets returned from workers */
 613 int
 614 rte_distributor_returned_pkts(struct rte_distributor *d,
 615                 struct rte_mbuf **mbufs, unsigned int max_mbufs)
 616 {
 617         struct rte_distributor_returned_pkts *returns = &d->returns;
 618         unsigned int retval = (max_mbufs < returns->count) ?
 619                         max_mbufs : returns->count;
 620         unsigned int i;
 621
 622         if (d->alg_type == RTE_DIST_ALG_SINGLE) {
 623                 /* Call the old API */
 624                 return rte_distributor_returned_pkts_single(d->d_single,
 625                                 mbufs, max_mbufs);
 626         }
 627
 628         for (i = 0; i < retval; i++) {
 629                 unsigned int idx = (returns->start + i) &
 630                                 RTE_DISTRIB_RETURNS_MASK;
 631
 632                 mbufs[i] = returns->mbufs[idx];
 633         }
 634         returns->start += i;
 635         returns->count -= i;
 636
 637         return retval;
 638 }
 639
 640 /*
 641  * Return the number of packets in-flight in a distributor, i.e. packets
 642  * being worked on or queued up in a backlog.
 643  */
 644 static inline unsigned int
 645 total_outstanding(const struct rte_distributor *d)
 646 {
 647         unsigned int wkr, total_outstanding = 0;
 648
 649         for (wkr = 0; wkr < d->num_workers; wkr++)
 650                 total_outstanding += d->backlog[wkr].count;
 651
 652         return total_outstanding;
 653 }
 654
 655 /*
 656  * Flush the distributor, so that there are no outstanding packets in flight or
 657  * queued up.
 658  */
 659 int
 660 rte_distributor_flush(struct rte_distributor *d)
 661 {
 662         unsigned int flushed;
 663         unsigned int wkr;
 664
 665         if (d->alg_type == RTE_DIST_ALG_SINGLE) {
 666                 /* Call the old API */
 667                 return rte_distributor_flush_single(d->d_single);
 668         }
 669
 670         flushed = total_outstanding(d);
 671
 672         while (total_outstanding(d) > 0)
 673                 rte_distributor_process(d, NULL, 0);
 674
 675         /* wait 10ms to allow all worker drain the pkts */
 676         rte_delay_us(10000);
 677
 678         /*
 679          * Send empty burst to all workers to allow them to exit
 680          * gracefully, should they need to.
 681          */
 682         rte_distributor_process(d, NULL, 0);
 683
 684         for (wkr = 0; wkr < d->num_workers; wkr++)
 685                 handle_returns(d, wkr);
 686
 687         return flushed;
 688 }
 689
 690 /* clears the internal returns array in the distributor */
 691 void
 692 rte_distributor_clear_returns(struct rte_distributor *d)
 693 {
 694         unsigned int wkr;
 695
 696         if (d->alg_type == RTE_DIST_ALG_SINGLE) {
 697                 /* Call the old API */
 698                 rte_distributor_clear_returns_single(d->d_single);
 699                 return;
 700         }
 701
 702         /* throw away returns, so workers can exit */
 703         for (wkr = 0; wkr < d->num_workers; wkr++)
 704                 /* Sync with worker. Release retptrs. */
 705                 __atomic_store_n(&(d->bufs[wkr].retptr64[0]), 0,
 706                                 __ATOMIC_RELEASE);
 707 }
 708
 709 /* creates a distributor instance */
 710 struct rte_distributor *
 711 rte_distributor_create(const char *name,
 712                 unsigned int socket_id,
 713                 unsigned int num_workers,
 714                 unsigned int alg_type)
 715 {
 716         struct rte_distributor *d;
 717         struct rte_dist_burst_list *dist_burst_list;
 718         char mz_name[RTE_MEMZONE_NAMESIZE];
 719         const struct rte_memzone *mz;
 720         unsigned int i;
 721
 722         /* TODO Reorganise function properly around RTE_DIST_ALG_SINGLE/BURST */
 723
 724         /* compilation-time checks */
 725         RTE_BUILD_BUG_ON((sizeof(*d) & RTE_CACHE_LINE_MASK) != 0);
 726         RTE_BUILD_BUG_ON((RTE_DISTRIB_MAX_WORKERS & 7) != 0);
 727
 728         if (name == NULL || num_workers >=
 729                 (unsigned int)RTE_MIN(RTE_DISTRIB_MAX_WORKERS, RTE_MAX_LCORE)) {
 730                 rte_errno = EINVAL;
 731                 return NULL;
 732         }
 733
 734         if (alg_type == RTE_DIST_ALG_SINGLE) {
 735                 d = malloc(sizeof(struct rte_distributor));
 736                 if (d == NULL) {
 737                         rte_errno = ENOMEM;
 738                         return NULL;
 739                 }
 740                 d->d_single = rte_distributor_create_single(name,
 741                                 socket_id, num_workers);
 742                 if (d->d_single == NULL) {
 743                         free(d);
 744                         /* rte_errno will have been set */
 745                         return NULL;
 746                 }
 747                 d->alg_type = alg_type;
 748                 return d;
 749         }
 750
 751         snprintf(mz_name, sizeof(mz_name), RTE_DISTRIB_PREFIX"%s", name);
 752         mz = rte_memzone_reserve(mz_name, sizeof(*d), socket_id, NO_FLAGS);
 753         if (mz == NULL) {
 754                 rte_errno = ENOMEM;
 755                 return NULL;
 756         }
 757
 758         d = mz->addr;
 759         strlcpy(d->name, name, sizeof(d->name));
 760         d->num_workers = num_workers;
 761         d->alg_type = alg_type;
 762
 763         d->dist_match_fn = RTE_DIST_MATCH_SCALAR;
 764 #if defined(RTE_ARCH_X86)
 765         d->dist_match_fn = RTE_DIST_MATCH_VECTOR;
 766 #endif
 767
 768         /*
 769          * Set up the backlog tags so they're pointing at the second cache
 770          * line for performance during flow matching
 771          */
 772         for (i = 0 ; i < num_workers ; i++)
 773                 d->backlog[i].tags = &d->in_flight_tags[i][RTE_DIST_BURST_SIZE];
 774
 775         memset(d->active, 0, sizeof(d->active));
 776         d->activesum = 0;
 777
 778         dist_burst_list = RTE_TAILQ_CAST(rte_dist_burst_tailq.head,
 779                                           rte_dist_burst_list);
 780
 781
 782         rte_mcfg_tailq_write_lock();
 783         TAILQ_INSERT_TAIL(dist_burst_list, d, next);
 784         rte_mcfg_tailq_write_unlock();
 785
 786         return d;
 787 }