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