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