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, oldpkt[0]);
  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)) {
  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 Return 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_RETURN_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 bit is set, return
 101          * Sync with distributor to acquire bufptrs
 102          */
 103         if (__atomic_load_n(&(buf->bufptr64[0]), __ATOMIC_ACQUIRE)
 104                 & RTE_DISTRIB_GET_BUF)
 105                 return -1;
 106
 107         /* since bufptr64 is signed, this should be an arithmetic shift */
 108         for (i = 0; i < RTE_DIST_BURST_SIZE; i++) {
 109                 if (likely(buf->bufptr64[i] & RTE_DISTRIB_VALID_BUF)) {
 110                         ret = buf->bufptr64[i] >> RTE_DISTRIB_FLAG_BITS;
 111                         pkts[count++] = (struct rte_mbuf *)((uintptr_t)(ret));
 112                 }
 113         }
 114
 115         /*
 116          * so now we've got the contents of the cacheline into an  array of
 117          * mbuf pointers, so toggle the bit so scheduler can start working
 118          * on the next cacheline while we're working.
 119          * Sync with distributor on GET_BUF flag. Release bufptrs.
 120          */
 121         __atomic_store_n(&(buf->bufptr64[0]),
 122                 buf->bufptr64[0] | RTE_DISTRIB_GET_BUF, __ATOMIC_RELEASE);
 123
 124         return count;
 125 }
 126
 127 int
 128 rte_distributor_get_pkt(struct rte_distributor *d,
 129                 unsigned int worker_id, struct rte_mbuf **pkts,
 130                 struct rte_mbuf **oldpkt, unsigned int return_count)
 131 {
 132         int count;
 133
 134         if (unlikely(d->alg_type == RTE_DIST_ALG_SINGLE)) {
 135                 if (return_count <= 1) {
 136                         pkts[0] = rte_distributor_get_pkt_single(d->d_single,
 137                                 worker_id, oldpkt[0]);
 138                         return (pkts[0]) ? 1 : 0;
 139                 } else
 140                         return -EINVAL;
 141         }
 142
 143         rte_distributor_request_pkt(d, worker_id, oldpkt, return_count);
 144
 145         count = rte_distributor_poll_pkt(d, worker_id, pkts);
 146         while (count == -1) {
 147                 uint64_t t = rte_rdtsc() + 100;
 148
 149                 while (rte_rdtsc() < t)
 150                         rte_pause();
 151
 152                 count = rte_distributor_poll_pkt(d, worker_id, pkts);
 153         }
 154         return count;
 155 }
 156
 157 int
 158 rte_distributor_return_pkt(struct rte_distributor *d,
 159                 unsigned int worker_id, struct rte_mbuf **oldpkt, int num)
 160 {
 161         struct rte_distributor_buffer *buf = &d->bufs[worker_id];
 162         unsigned int i;
 163
 164         if (unlikely(d->alg_type == RTE_DIST_ALG_SINGLE)) {
 165                 if (num == 1)
 166                         return rte_distributor_return_pkt_single(d->d_single,
 167                                 worker_id, oldpkt[0]);
 168                 else
 169                         return -EINVAL;
 170         }
 171
 172         /* Spin while handshake bits are set (scheduler clears it).
 173          * Sync with worker on GET_BUF flag.
 174          */
 175         while (unlikely(__atomic_load_n(&(buf->retptr64[0]), __ATOMIC_RELAXED)
 176                         & RTE_DISTRIB_GET_BUF)) {
 177                 rte_pause();
 178                 uint64_t t = rte_rdtsc()+100;
 179
 180                 while (rte_rdtsc() < t)
 181                         rte_pause();
 182         }
 183
 184         /* Sync with distributor to acquire retptrs */
 185         __atomic_thread_fence(__ATOMIC_ACQUIRE);
 186         for (i = 0; i < RTE_DIST_BURST_SIZE; i++)
 187                 /* Switch off the return bit first */
 188                 buf->retptr64[i] &= ~RTE_DISTRIB_RETURN_BUF;
 189
 190         for (i = num; i-- > 0; )
 191                 buf->retptr64[i] = (((int64_t)(uintptr_t)oldpkt[i]) <<
 192                         RTE_DISTRIB_FLAG_BITS) | RTE_DISTRIB_RETURN_BUF;
 193
 194         /* set the GET_BUF but even if we got no returns.
 195          * Sync with distributor on GET_BUF flag. Release retptrs.
 196          */
 197         __atomic_store_n(&(buf->retptr64[0]),
 198                 buf->retptr64[0] | RTE_DISTRIB_GET_BUF, __ATOMIC_RELEASE);
 199
 200         return 0;
 201 }
 202
 203 /**** APIs called on distributor core ***/
 204
 205 /* stores a packet returned from a worker inside the returns array */
 206 static inline void
 207 store_return(uintptr_t oldbuf, struct rte_distributor *d,
 208                 unsigned int *ret_start, unsigned int *ret_count)
 209 {
 210         if (!oldbuf)
 211                 return;
 212         /* store returns in a circular buffer */
 213         d->returns.mbufs[(*ret_start + *ret_count) & RTE_DISTRIB_RETURNS_MASK]
 214                         = (void *)oldbuf;
 215         *ret_start += (*ret_count == RTE_DISTRIB_RETURNS_MASK);
 216         *ret_count += (*ret_count != RTE_DISTRIB_RETURNS_MASK);
 217 }
 218
 219 /*
 220  * Match then flow_ids (tags) of the incoming packets to the flow_ids
 221  * of the inflight packets (both inflight on the workers and in each worker
 222  * backlog). This will then allow us to pin those packets to the relevant
 223  * workers to give us our atomic flow pinning.
 224  */
 225 void
 226 find_match_scalar(struct rte_distributor *d,
 227                         uint16_t *data_ptr,
 228                         uint16_t *output_ptr)
 229 {
 230         struct rte_distributor_backlog *bl;
 231         uint16_t i, j, w;
 232
 233         /*
 234          * Function overview:
 235          * 1. Loop through all worker ID's
 236          * 2. Compare the current inflights to the incoming tags
 237          * 3. Compare the current backlog to the incoming tags
 238          * 4. Add any matches to the output
 239          */
 240
 241         for (j = 0 ; j < RTE_DIST_BURST_SIZE; j++)
 242                 output_ptr[j] = 0;
 243
 244         for (i = 0; i < d->num_workers; i++) {
 245                 bl = &d->backlog[i];
 246
 247                 for (j = 0; j < RTE_DIST_BURST_SIZE ; j++)
 248                         for (w = 0; w < RTE_DIST_BURST_SIZE; w++)
 249                                 if (d->in_flight_tags[i][j] == data_ptr[w]) {
 250                                         output_ptr[j] = i+1;
 251                                         break;
 252                                 }
 253                 for (j = 0; j < RTE_DIST_BURST_SIZE; j++)
 254                         for (w = 0; w < RTE_DIST_BURST_SIZE; w++)
 255                                 if (bl->tags[j] == data_ptr[w]) {
 256                                         output_ptr[j] = i+1;
 257                                         break;
 258                                 }
 259         }
 260
 261         /*
 262          * At this stage, the output contains 8 16-bit values, with
 263          * each non-zero value containing the worker ID on which the
 264          * corresponding flow is pinned to.
 265          */
 266 }
 267
 268
 269 /*
 270  * When the handshake bits indicate that there are packets coming
 271  * back from the worker, this function is called to copy and store
 272  * the valid returned pointers (store_return).
 273  */
 274 static unsigned int
 275 handle_returns(struct rte_distributor *d, unsigned int wkr)
 276 {
 277         struct rte_distributor_buffer *buf = &(d->bufs[wkr]);
 278         uintptr_t oldbuf;
 279         unsigned int ret_start = d->returns.start,
 280                         ret_count = d->returns.count;
 281         unsigned int count = 0;
 282         unsigned int i;
 283
 284         /* Sync on GET_BUF flag. Acquire retptrs. */
 285         if (__atomic_load_n(&(buf->retptr64[0]), __ATOMIC_ACQUIRE)
 286                 & RTE_DISTRIB_GET_BUF) {
 287                 for (i = 0; i < RTE_DIST_BURST_SIZE; i++) {
 288                         if (buf->retptr64[i] & RTE_DISTRIB_RETURN_BUF) {
 289                                 oldbuf = ((uintptr_t)(buf->retptr64[i] >>
 290                                         RTE_DISTRIB_FLAG_BITS));
 291                                 /* store returns in a circular buffer */
 292                                 store_return(oldbuf, d, &ret_start, &ret_count);
 293                                 count++;
 294                                 buf->retptr64[i] &= ~RTE_DISTRIB_RETURN_BUF;
 295                         }
 296                 }
 297                 d->returns.start = ret_start;
 298                 d->returns.count = ret_count;
 299                 /* Clear for the worker to populate with more returns.
 300                  * Sync with distributor on GET_BUF flag. Release retptrs.
 301                  */
 302                 __atomic_store_n(&(buf->retptr64[0]), 0, __ATOMIC_RELEASE);
 303         }
 304         return count;
 305 }
 306
 307 /*
 308  * This function releases a burst (cache line) to a worker.
 309  * It is called from the process function when a cacheline is
 310  * full to make room for more packets for that worker, or when
 311  * all packets have been assigned to bursts and need to be flushed
 312  * to the workers.
 313  * It also needs to wait for any outstanding packets from the worker
 314  * before sending out new packets.
 315  */
 316 static unsigned int
 317 release(struct rte_distributor *d, unsigned int wkr)
 318 {
 319         struct rte_distributor_buffer *buf = &(d->bufs[wkr]);
 320         unsigned int i;
 321
 322         handle_returns(d, wkr);
 323
 324         /* Sync with worker on GET_BUF flag */
 325         while (!(__atomic_load_n(&(d->bufs[wkr].bufptr64[0]), __ATOMIC_ACQUIRE)
 326                 & RTE_DISTRIB_GET_BUF)) {
 327                 handle_returns(d, wkr);
 328                 rte_pause();
 329         }
 330
 331         buf->count = 0;
 332
 333         for (i = 0; i < d->backlog[wkr].count; i++) {
 334                 d->bufs[wkr].bufptr64[i] = d->backlog[wkr].pkts[i] |
 335                                 RTE_DISTRIB_GET_BUF | RTE_DISTRIB_VALID_BUF;
 336                 d->in_flight_tags[wkr][i] = d->backlog[wkr].tags[i];
 337         }
 338         buf->count = i;
 339         for ( ; i < RTE_DIST_BURST_SIZE ; i++) {
 340                 buf->bufptr64[i] = RTE_DISTRIB_GET_BUF;
 341                 d->in_flight_tags[wkr][i] = 0;
 342         }
 343
 344         d->backlog[wkr].count = 0;
 345
 346         /* Clear the GET bit.
 347          * Sync with worker on GET_BUF flag. Release bufptrs.
 348          */
 349         __atomic_store_n(&(buf->bufptr64[0]),
 350                 buf->bufptr64[0] & ~RTE_DISTRIB_GET_BUF, __ATOMIC_RELEASE);
 351         return  buf->count;
 352
 353 }
 354
 355
 356 /* process a set of packets to distribute them to workers */
 357 int
 358 rte_distributor_process(struct rte_distributor *d,
 359                 struct rte_mbuf **mbufs, unsigned int num_mbufs)
 360 {
 361         unsigned int next_idx = 0;
 362         static unsigned int wkr;
 363         struct rte_mbuf *next_mb = NULL;
 364         int64_t next_value = 0;
 365         uint16_t new_tag = 0;
 366         uint16_t flows[RTE_DIST_BURST_SIZE] __rte_cache_aligned;
 367         unsigned int i, j, w, wid;
 368
 369         if (d->alg_type == RTE_DIST_ALG_SINGLE) {
 370                 /* Call the old API */
 371                 return rte_distributor_process_single(d->d_single,
 372                         mbufs, num_mbufs);
 373         }
 374
 375         if (unlikely(num_mbufs == 0)) {
 376                 /* Flush out all non-full cache-lines to workers. */
 377                 for (wid = 0 ; wid < d->num_workers; wid++) {
 378                         /* Sync with worker on GET_BUF flag. */
 379                         handle_returns(d, wid);
 380                         if (__atomic_load_n(&(d->bufs[wid].bufptr64[0]),
 381                                 __ATOMIC_ACQUIRE) & RTE_DISTRIB_GET_BUF) {
 382                                 release(d, wid);
 383                                 handle_returns(d, wid);
 384                         }
 385                 }
 386                 return 0;
 387         }
 388
 389         while (next_idx < num_mbufs) {
 390                 uint16_t matches[RTE_DIST_BURST_SIZE];
 391                 unsigned int pkts;
 392
 393                 /* Sync with worker on GET_BUF flag. */
 394                 if (__atomic_load_n(&(d->bufs[wkr].bufptr64[0]),
 395                         __ATOMIC_ACQUIRE) & RTE_DISTRIB_GET_BUF)
 396                         d->bufs[wkr].count = 0;
 397
 398                 if ((num_mbufs - next_idx) < RTE_DIST_BURST_SIZE)
 399                         pkts = num_mbufs - next_idx;
 400                 else
 401                         pkts = RTE_DIST_BURST_SIZE;
 402
 403                 for (i = 0; i < pkts; i++) {
 404                         if (mbufs[next_idx + i]) {
 405                                 /* flows have to be non-zero */
 406                                 flows[i] = mbufs[next_idx + i]->hash.usr | 1;
 407                         } else
 408                                 flows[i] = 0;
 409                 }
 410                 for (; i < RTE_DIST_BURST_SIZE; i++)
 411                         flows[i] = 0;
 412
 413                 switch (d->dist_match_fn) {
 414                 case RTE_DIST_MATCH_VECTOR:
 415                         find_match_vec(d, &flows[0], &matches[0]);
 416                         break;
 417                 default:
 418                         find_match_scalar(d, &flows[0], &matches[0]);
 419                 }
 420
 421                 /*
 422                  * Matches array now contain the intended worker ID (+1) of
 423                  * the incoming packets. Any zeroes need to be assigned
 424                  * workers.
 425                  */
 426
 427                 for (j = 0; j < pkts; j++) {
 428
 429                         next_mb = mbufs[next_idx++];
 430                         next_value = (((int64_t)(uintptr_t)next_mb) <<
 431                                         RTE_DISTRIB_FLAG_BITS);
 432                         /*
 433                          * User is advocated to set tag value for each
 434                          * mbuf before calling rte_distributor_process.
 435                          * User defined tags are used to identify flows,
 436                          * or sessions.
 437                          */
 438                         /* flows MUST be non-zero */
 439                         new_tag = (uint16_t)(next_mb->hash.usr) | 1;
 440
 441                         /*
 442                          * Uncommenting the next line will cause the find_match
 443                          * function to be optimized out, making this function
 444                          * do parallel (non-atomic) distribution
 445                          */
 446                         /* matches[j] = 0; */
 447
 448                         if (matches[j]) {
 449                                 struct rte_distributor_backlog *bl =
 450                                                 &d->backlog[matches[j]-1];
 451                                 if (unlikely(bl->count ==
 452                                                 RTE_DIST_BURST_SIZE)) {
 453                                         release(d, matches[j]-1);
 454                                 }
 455
 456                                 /* Add to worker that already has flow */
 457                                 unsigned int idx = bl->count++;
 458
 459                                 bl->tags[idx] = new_tag;
 460                                 bl->pkts[idx] = next_value;
 461
 462                         } else {
 463                                 struct rte_distributor_backlog *bl =
 464                                                 &d->backlog[wkr];
 465                                 if (unlikely(bl->count ==
 466                                                 RTE_DIST_BURST_SIZE)) {
 467                                         release(d, wkr);
 468                                 }
 469
 470                                 /* Add to current worker worker */
 471                                 unsigned int idx = bl->count++;
 472
 473                                 bl->tags[idx] = new_tag;
 474                                 bl->pkts[idx] = next_value;
 475                                 /*
 476                                  * Now that we've just added an unpinned flow
 477                                  * to a worker, we need to ensure that all
 478                                  * other packets with that same flow will go
 479                                  * to the same worker in this burst.
 480                                  */
 481                                 for (w = j; w < pkts; w++)
 482                                         if (flows[w] == new_tag)
 483                                                 matches[w] = wkr+1;
 484                         }
 485                 }
 486                 wkr++;
 487                 if (wkr >= d->num_workers)
 488                         wkr = 0;
 489         }
 490
 491         /* Flush out all non-full cache-lines to workers. */
 492         for (wid = 0 ; wid < d->num_workers; wid++)
 493                 /* Sync with worker on GET_BUF flag. */
 494                 if ((__atomic_load_n(&(d->bufs[wid].bufptr64[0]),
 495                         __ATOMIC_ACQUIRE) & RTE_DISTRIB_GET_BUF))
 496                         release(d, wid);
 497
 498         return num_mbufs;
 499 }
 500
 501 /* return to the caller, packets returned from workers */
 502 int
 503 rte_distributor_returned_pkts(struct rte_distributor *d,
 504                 struct rte_mbuf **mbufs, unsigned int max_mbufs)
 505 {
 506         struct rte_distributor_returned_pkts *returns = &d->returns;
 507         unsigned int retval = (max_mbufs < returns->count) ?
 508                         max_mbufs : returns->count;
 509         unsigned int i;
 510
 511         if (d->alg_type == RTE_DIST_ALG_SINGLE) {
 512                 /* Call the old API */
 513                 return rte_distributor_returned_pkts_single(d->d_single,
 514                                 mbufs, max_mbufs);
 515         }
 516
 517         for (i = 0; i < retval; i++) {
 518                 unsigned int idx = (returns->start + i) &
 519                                 RTE_DISTRIB_RETURNS_MASK;
 520
 521                 mbufs[i] = returns->mbufs[idx];
 522         }
 523         returns->start += i;
 524         returns->count -= i;
 525
 526         return retval;
 527 }
 528
 529 /*
 530  * Return the number of packets in-flight in a distributor, i.e. packets
 531  * being worked on or queued up in a backlog.
 532  */
 533 static inline unsigned int
 534 total_outstanding(const struct rte_distributor *d)
 535 {
 536         unsigned int wkr, total_outstanding = 0;
 537
 538         for (wkr = 0; wkr < d->num_workers; wkr++)
 539                 total_outstanding += d->backlog[wkr].count;
 540
 541         return total_outstanding;
 542 }
 543
 544 /*
 545  * Flush the distributor, so that there are no outstanding packets in flight or
 546  * queued up.
 547  */
 548 int
 549 rte_distributor_flush(struct rte_distributor *d)
 550 {
 551         unsigned int flushed;
 552         unsigned int wkr;
 553
 554         if (d->alg_type == RTE_DIST_ALG_SINGLE) {
 555                 /* Call the old API */
 556                 return rte_distributor_flush_single(d->d_single);
 557         }
 558
 559         flushed = total_outstanding(d);
 560
 561         while (total_outstanding(d) > 0)
 562                 rte_distributor_process(d, NULL, 0);
 563
 564         /* wait 10ms to allow all worker drain the pkts */
 565         rte_delay_us(10000);
 566
 567         /*
 568          * Send empty burst to all workers to allow them to exit
 569          * gracefully, should they need to.
 570          */
 571         rte_distributor_process(d, NULL, 0);
 572
 573         for (wkr = 0; wkr < d->num_workers; wkr++)
 574                 handle_returns(d, wkr);
 575
 576         return flushed;
 577 }
 578
 579 /* clears the internal returns array in the distributor */
 580 void
 581 rte_distributor_clear_returns(struct rte_distributor *d)
 582 {
 583         unsigned int wkr;
 584
 585         if (d->alg_type == RTE_DIST_ALG_SINGLE) {
 586                 /* Call the old API */
 587                 rte_distributor_clear_returns_single(d->d_single);
 588                 return;
 589         }
 590
 591         /* throw away returns, so workers can exit */
 592         for (wkr = 0; wkr < d->num_workers; wkr++)
 593                 /* Sync with worker. Release retptrs. */
 594                 __atomic_store_n(&(d->bufs[wkr].retptr64[0]), 0,
 595                                 __ATOMIC_RELEASE);
 596 }
 597
 598 /* creates a distributor instance */
 599 struct rte_distributor *
 600 rte_distributor_create(const char *name,
 601                 unsigned int socket_id,
 602                 unsigned int num_workers,
 603                 unsigned int alg_type)
 604 {
 605         struct rte_distributor *d;
 606         struct rte_dist_burst_list *dist_burst_list;
 607         char mz_name[RTE_MEMZONE_NAMESIZE];
 608         const struct rte_memzone *mz;
 609         unsigned int i;
 610
 611         /* TODO Reorganise function properly around RTE_DIST_ALG_SINGLE/BURST */
 612
 613         /* compilation-time checks */
 614         RTE_BUILD_BUG_ON((sizeof(*d) & RTE_CACHE_LINE_MASK) != 0);
 615         RTE_BUILD_BUG_ON((RTE_DISTRIB_MAX_WORKERS & 7) != 0);
 616
 617         if (name == NULL || num_workers >=
 618                 (unsigned int)RTE_MIN(RTE_DISTRIB_MAX_WORKERS, RTE_MAX_LCORE)) {
 619                 rte_errno = EINVAL;
 620                 return NULL;
 621         }
 622
 623         if (alg_type == RTE_DIST_ALG_SINGLE) {
 624                 d = malloc(sizeof(struct rte_distributor));
 625                 if (d == NULL) {
 626                         rte_errno = ENOMEM;
 627                         return NULL;
 628                 }
 629                 d->d_single = rte_distributor_create_single(name,
 630                                 socket_id, num_workers);
 631                 if (d->d_single == NULL) {
 632                         free(d);
 633                         /* rte_errno will have been set */
 634                         return NULL;
 635                 }
 636                 d->alg_type = alg_type;
 637                 return d;
 638         }
 639
 640         snprintf(mz_name, sizeof(mz_name), RTE_DISTRIB_PREFIX"%s", name);
 641         mz = rte_memzone_reserve(mz_name, sizeof(*d), socket_id, NO_FLAGS);
 642         if (mz == NULL) {
 643                 rte_errno = ENOMEM;
 644                 return NULL;
 645         }
 646
 647         d = mz->addr;
 648         strlcpy(d->name, name, sizeof(d->name));
 649         d->num_workers = num_workers;
 650         d->alg_type = alg_type;
 651
 652         d->dist_match_fn = RTE_DIST_MATCH_SCALAR;
 653 #if defined(RTE_ARCH_X86)
 654         d->dist_match_fn = RTE_DIST_MATCH_VECTOR;
 655 #endif
 656
 657         /*
 658          * Set up the backlog tags so they're pointing at the second cache
 659          * line for performance during flow matching
 660          */
 661         for (i = 0 ; i < num_workers ; i++)
 662                 d->backlog[i].tags = &d->in_flight_tags[i][RTE_DIST_BURST_SIZE];
 663
 664         dist_burst_list = RTE_TAILQ_CAST(rte_dist_burst_tailq.head,
 665                                           rte_dist_burst_list);
 666
 667
 668         rte_mcfg_tailq_write_lock();
 669         TAILQ_INSERT_TAIL(dist_burst_list, d, next);
 670         rte_mcfg_tailq_write_unlock();
 671
 672         return d;
 673 }