lib/librte_distributor/rte_distributor.c

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