drivers/event/sw/sw_evdev_scheduler.c

   1 /*-
   2  *   BSD LICENSE
   3  *
   4  *   Copyright(c) 2016-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 <rte_ring.h>
  34 #include <rte_hash_crc.h>
  35 #include "sw_evdev.h"
  36 #include "iq_ring.h"
  37 #include "event_ring.h"
  38
  39 #define SW_IQS_MASK (SW_IQS_MAX-1)
  40
  41 /* Retrieve the highest priority IQ or -1 if no pkts available. Doing the
  42  * CLZ twice is faster than caching the value due to data dependencies
  43  */
  44 #define PKT_MASK_TO_IQ(pkts) \
  45         (__builtin_ctz(pkts | (1 << SW_IQS_MAX)))
  46
  47 #if SW_IQS_MAX != 4
  48 #error Misconfigured PRIO_TO_IQ caused by SW_IQS_MAX value change
  49 #endif
  50 #define PRIO_TO_IQ(prio) (prio >> 6)
  51
  52 #define MAX_PER_IQ_DEQUEUE 48
  53 #define FLOWID_MASK (SW_QID_NUM_FIDS-1)
  54 /* use cheap bit mixing, we only need to lose a few bits */
  55 #define SW_HASH_FLOWID(f) (((f) ^ (f >> 10)) & FLOWID_MASK)
  56
  57 static inline uint32_t
  58 sw_schedule_atomic_to_cq(struct sw_evdev *sw, struct sw_qid * const qid,
  59                 uint32_t iq_num, unsigned int count)
  60 {
  61         struct rte_event qes[MAX_PER_IQ_DEQUEUE]; /* count <= MAX */
  62         struct rte_event blocked_qes[MAX_PER_IQ_DEQUEUE];
  63         uint32_t nb_blocked = 0;
  64         uint32_t i;
  65
  66         if (count > MAX_PER_IQ_DEQUEUE)
  67                 count = MAX_PER_IQ_DEQUEUE;
  68
  69         /* This is the QID ID. The QID ID is static, hence it can be
  70          * used to identify the stage of processing in history lists etc
  71          */
  72         uint32_t qid_id = qid->id;
  73
  74         iq_ring_dequeue_burst(qid->iq[iq_num], qes, count);
  75         for (i = 0; i < count; i++) {
  76                 const struct rte_event *qe = &qes[i];
  77                 const uint16_t flow_id = SW_HASH_FLOWID(qes[i].flow_id);
  78                 struct sw_fid_t *fid = &qid->fids[flow_id];
  79                 int cq = fid->cq;
  80
  81                 if (cq < 0) {
  82                         uint32_t cq_idx = qid->cq_next_tx++;
  83                         if (qid->cq_next_tx == qid->cq_num_mapped_cqs)
  84                                 qid->cq_next_tx = 0;
  85                         cq = qid->cq_map[cq_idx];
  86
  87                         /* find least used */
  88                         int cq_free_cnt = sw->cq_ring_space[cq];
  89                         for (cq_idx = 0; cq_idx < qid->cq_num_mapped_cqs;
  90                                         cq_idx++) {
  91                                 int test_cq = qid->cq_map[cq_idx];
  92                                 int test_cq_free = sw->cq_ring_space[test_cq];
  93                                 if (test_cq_free > cq_free_cnt) {
  94                                         cq = test_cq;
  95                                         cq_free_cnt = test_cq_free;
  96                                 }
  97                         }
  98
  99                         fid->cq = cq; /* this pins early */
 100                 }
 101
 102                 if (sw->cq_ring_space[cq] == 0 ||
 103                                 sw->ports[cq].inflights == SW_PORT_HIST_LIST) {
 104                         blocked_qes[nb_blocked++] = *qe;
 105                         continue;
 106                 }
 107
 108                 struct sw_port *p = &sw->ports[cq];
 109
 110                 /* at this point we can queue up the packet on the cq_buf */
 111                 fid->pcount++;
 112                 p->cq_buf[p->cq_buf_count++] = *qe;
 113                 p->inflights++;
 114                 sw->cq_ring_space[cq]--;
 115
 116                 int head = (p->hist_head++ & (SW_PORT_HIST_LIST-1));
 117                 p->hist_list[head].fid = flow_id;
 118                 p->hist_list[head].qid = qid_id;
 119
 120                 p->stats.tx_pkts++;
 121                 qid->stats.tx_pkts++;
 122
 123                 /* if we just filled in the last slot, flush the buffer */
 124                 if (sw->cq_ring_space[cq] == 0) {
 125                         struct qe_ring *worker = p->cq_worker_ring;
 126                         qe_ring_enqueue_burst(worker, p->cq_buf,
 127                                         p->cq_buf_count,
 128                                         &sw->cq_ring_space[cq]);
 129                         p->cq_buf_count = 0;
 130                 }
 131         }
 132         iq_ring_put_back(qid->iq[iq_num], blocked_qes, nb_blocked);
 133
 134         return count - nb_blocked;
 135 }
 136
 137 static inline uint32_t
 138 sw_schedule_parallel_to_cq(struct sw_evdev *sw, struct sw_qid * const qid,
 139                 uint32_t iq_num, unsigned int count, int keep_order)
 140 {
 141         uint32_t i;
 142         uint32_t cq_idx = qid->cq_next_tx;
 143
 144         /* This is the QID ID. The QID ID is static, hence it can be
 145          * used to identify the stage of processing in history lists etc
 146          */
 147         uint32_t qid_id = qid->id;
 148
 149         if (count > MAX_PER_IQ_DEQUEUE)
 150                 count = MAX_PER_IQ_DEQUEUE;
 151
 152         if (keep_order)
 153                 /* only schedule as many as we have reorder buffer entries */
 154                 count = RTE_MIN(count,
 155                                 rte_ring_count(qid->reorder_buffer_freelist));
 156
 157         for (i = 0; i < count; i++) {
 158                 const struct rte_event *qe = iq_ring_peek(qid->iq[iq_num]);
 159                 uint32_t cq_check_count = 0;
 160                 uint32_t cq;
 161
 162                 /*
 163                  *  for parallel, just send to next available CQ in round-robin
 164                  * fashion. So scan for an available CQ. If all CQs are full
 165                  * just return and move on to next QID
 166                  */
 167                 do {
 168                         if (++cq_check_count > qid->cq_num_mapped_cqs)
 169                                 goto exit;
 170                         cq = qid->cq_map[cq_idx];
 171                         if (++cq_idx == qid->cq_num_mapped_cqs)
 172                                 cq_idx = 0;
 173                 } while (qe_ring_free_count(sw->ports[cq].cq_worker_ring) == 0 ||
 174                                 sw->ports[cq].inflights == SW_PORT_HIST_LIST);
 175
 176                 struct sw_port *p = &sw->ports[cq];
 177                 if (sw->cq_ring_space[cq] == 0 ||
 178                                 p->inflights == SW_PORT_HIST_LIST)
 179                         break;
 180
 181                 sw->cq_ring_space[cq]--;
 182
 183                 qid->stats.tx_pkts++;
 184
 185                 const int head = (p->hist_head & (SW_PORT_HIST_LIST-1));
 186                 p->hist_list[head].fid = SW_HASH_FLOWID(qe->flow_id);
 187                 p->hist_list[head].qid = qid_id;
 188
 189                 if (keep_order)
 190                         rte_ring_sc_dequeue(qid->reorder_buffer_freelist,
 191                                         (void *)&p->hist_list[head].rob_entry);
 192
 193                 sw->ports[cq].cq_buf[sw->ports[cq].cq_buf_count++] = *qe;
 194                 iq_ring_pop(qid->iq[iq_num]);
 195
 196                 rte_compiler_barrier();
 197                 p->inflights++;
 198                 p->stats.tx_pkts++;
 199                 p->hist_head++;
 200         }
 201 exit:
 202         qid->cq_next_tx = cq_idx;
 203         return i;
 204 }
 205
 206 static uint32_t
 207 sw_schedule_dir_to_cq(struct sw_evdev *sw, struct sw_qid * const qid,
 208                 uint32_t iq_num, unsigned int count __rte_unused)
 209 {
 210         uint32_t cq_id = qid->cq_map[0];
 211         struct sw_port *port = &sw->ports[cq_id];
 212
 213         /* get max burst enq size for cq_ring */
 214         uint32_t count_free = sw->cq_ring_space[cq_id];
 215         if (count_free == 0)
 216                 return 0;
 217
 218         /* burst dequeue from the QID IQ ring */
 219         struct iq_ring *ring = qid->iq[iq_num];
 220         uint32_t ret = iq_ring_dequeue_burst(ring,
 221                         &port->cq_buf[port->cq_buf_count], count_free);
 222         port->cq_buf_count += ret;
 223
 224         /* Update QID, Port and Total TX stats */
 225         qid->stats.tx_pkts += ret;
 226         port->stats.tx_pkts += ret;
 227
 228         /* Subtract credits from cached value */
 229         sw->cq_ring_space[cq_id] -= ret;
 230
 231         return ret;
 232 }
 233
 234 static uint32_t
 235 sw_schedule_qid_to_cq(struct sw_evdev *sw)
 236 {
 237         uint32_t pkts = 0;
 238         uint32_t qid_idx;
 239
 240         sw->sched_cq_qid_called++;
 241
 242         for (qid_idx = 0; qid_idx < sw->qid_count; qid_idx++) {
 243                 struct sw_qid *qid = sw->qids_prioritized[qid_idx];
 244
 245                 int type = qid->type;
 246                 int iq_num = PKT_MASK_TO_IQ(qid->iq_pkt_mask);
 247
 248                 /* zero mapped CQs indicates directed */
 249                 if (iq_num >= SW_IQS_MAX)
 250                         continue;
 251
 252                 uint32_t pkts_done = 0;
 253                 uint32_t count = iq_ring_count(qid->iq[iq_num]);
 254
 255                 if (count > 0) {
 256                         if (type == SW_SCHED_TYPE_DIRECT)
 257                                 pkts_done += sw_schedule_dir_to_cq(sw, qid,
 258                                                 iq_num, count);
 259                         else if (type == RTE_SCHED_TYPE_ATOMIC)
 260                                 pkts_done += sw_schedule_atomic_to_cq(sw, qid,
 261                                                 iq_num, count);
 262                         else
 263                                 pkts_done += sw_schedule_parallel_to_cq(sw, qid,
 264                                                 iq_num, count,
 265                                                 type == RTE_SCHED_TYPE_ORDERED);
 266                 }
 267
 268                 /* Check if the IQ that was polled is now empty, and unset it
 269                  * in the IQ mask if its empty.
 270                  */
 271                 int all_done = (pkts_done == count);
 272
 273                 qid->iq_pkt_mask &= ~(all_done << (iq_num));
 274                 pkts += pkts_done;
 275         }
 276
 277         return pkts;
 278 }
 279
 280 /* This function will perform re-ordering of packets, and injecting into
 281  * the appropriate QID IQ. As LB and DIR QIDs are in the same array, but *NOT*
 282  * contiguous in that array, this function accepts a "range" of QIDs to scan.
 283  */
 284 static uint16_t
 285 sw_schedule_reorder(struct sw_evdev *sw, int qid_start, int qid_end)
 286 {
 287         /* Perform egress reordering */
 288         struct rte_event *qe;
 289         uint32_t pkts_iter = 0;
 290
 291         for (; qid_start < qid_end; qid_start++) {
 292                 struct sw_qid *qid = &sw->qids[qid_start];
 293                 int i, num_entries_in_use;
 294
 295                 if (qid->type != RTE_SCHED_TYPE_ORDERED)
 296                         continue;
 297
 298                 num_entries_in_use = rte_ring_free_count(
 299                                         qid->reorder_buffer_freelist);
 300
 301                 for (i = 0; i < num_entries_in_use; i++) {
 302                         struct reorder_buffer_entry *entry;
 303                         int j;
 304
 305                         entry = &qid->reorder_buffer[qid->reorder_buffer_index];
 306
 307                         if (!entry->ready)
 308                                 break;
 309
 310                         for (j = 0; j < entry->num_fragments; j++) {
 311                                 uint16_t dest_qid;
 312                                 uint16_t dest_iq;
 313
 314                                 int idx = entry->fragment_index + j;
 315                                 qe = &entry->fragments[idx];
 316
 317                                 dest_qid = qe->queue_id;
 318                                 dest_iq  = PRIO_TO_IQ(qe->priority);
 319
 320                                 if (dest_qid >= sw->qid_count) {
 321                                         sw->stats.rx_dropped++;
 322                                         continue;
 323                                 }
 324
 325                                 struct sw_qid *dest_qid_ptr =
 326                                         &sw->qids[dest_qid];
 327                                 const struct iq_ring *dest_iq_ptr =
 328                                         dest_qid_ptr->iq[dest_iq];
 329                                 if (iq_ring_free_count(dest_iq_ptr) == 0)
 330                                         break;
 331
 332                                 pkts_iter++;
 333
 334                                 struct sw_qid *q = &sw->qids[dest_qid];
 335                                 struct iq_ring *r = q->iq[dest_iq];
 336
 337                                 /* we checked for space above, so enqueue must
 338                                  * succeed
 339                                  */
 340                                 iq_ring_enqueue(r, qe);
 341                                 q->iq_pkt_mask |= (1 << (dest_iq));
 342                                 q->iq_pkt_count[dest_iq]++;
 343                                 q->stats.rx_pkts++;
 344                         }
 345
 346                         entry->ready = (j != entry->num_fragments);
 347                         entry->num_fragments -= j;
 348                         entry->fragment_index += j;
 349
 350                         if (!entry->ready) {
 351                                 entry->fragment_index = 0;
 352
 353                                 rte_ring_sp_enqueue(
 354                                                 qid->reorder_buffer_freelist,
 355                                                 entry);
 356
 357                                 qid->reorder_buffer_index++;
 358                                 qid->reorder_buffer_index %= qid->window_size;
 359                         }
 360                 }
 361         }
 362         return pkts_iter;
 363 }
 364
 365 static inline void __attribute__((always_inline))
 366 sw_refill_pp_buf(struct sw_evdev *sw, struct sw_port *port)
 367 {
 368         RTE_SET_USED(sw);
 369         struct qe_ring *worker = port->rx_worker_ring;
 370         port->pp_buf_start = 0;
 371         port->pp_buf_count = qe_ring_dequeue_burst(worker, port->pp_buf,
 372                         RTE_DIM(port->pp_buf));
 373 }
 374
 375 static inline uint32_t __attribute__((always_inline))
 376 __pull_port_lb(struct sw_evdev *sw, uint32_t port_id, int allow_reorder)
 377 {
 378         static struct reorder_buffer_entry dummy_rob;
 379         uint32_t pkts_iter = 0;
 380         struct sw_port *port = &sw->ports[port_id];
 381
 382         /* If shadow ring has 0 pkts, pull from worker ring */
 383         if (port->pp_buf_count == 0)
 384                 sw_refill_pp_buf(sw, port);
 385
 386         while (port->pp_buf_count) {
 387                 const struct rte_event *qe = &port->pp_buf[port->pp_buf_start];
 388                 struct sw_hist_list_entry *hist_entry = NULL;
 389                 uint8_t flags = qe->op;
 390                 const uint16_t eop = !(flags & QE_FLAG_NOT_EOP);
 391                 int needs_reorder = 0;
 392                 /* if no-reordering, having PARTIAL == NEW */
 393                 if (!allow_reorder && !eop)
 394                         flags = QE_FLAG_VALID;
 395
 396                 /*
 397                  * if we don't have space for this packet in an IQ,
 398                  * then move on to next queue. Technically, for a
 399                  * packet that needs reordering, we don't need to check
 400                  * here, but it simplifies things not to special-case
 401                  */
 402                 uint32_t iq_num = PRIO_TO_IQ(qe->priority);
 403                 struct sw_qid *qid = &sw->qids[qe->queue_id];
 404
 405                 if ((flags & QE_FLAG_VALID) &&
 406                                 iq_ring_free_count(qid->iq[iq_num]) == 0)
 407                         break;
 408
 409                 /* now process based on flags. Note that for directed
 410                  * queues, the enqueue_flush masks off all but the
 411                  * valid flag. This makes FWD and PARTIAL enqueues just
 412                  * NEW type, and makes DROPS no-op calls.
 413                  */
 414                 if ((flags & QE_FLAG_COMPLETE) && port->inflights > 0) {
 415                         const uint32_t hist_tail = port->hist_tail &
 416                                         (SW_PORT_HIST_LIST - 1);
 417
 418                         hist_entry = &port->hist_list[hist_tail];
 419                         const uint32_t hist_qid = hist_entry->qid;
 420                         const uint32_t hist_fid = hist_entry->fid;
 421
 422                         struct sw_fid_t *fid =
 423                                 &sw->qids[hist_qid].fids[hist_fid];
 424                         fid->pcount -= eop;
 425                         if (fid->pcount == 0)
 426                                 fid->cq = -1;
 427
 428                         if (allow_reorder) {
 429                                 /* set reorder ready if an ordered QID */
 430                                 uintptr_t rob_ptr =
 431                                         (uintptr_t)hist_entry->rob_entry;
 432                                 const uintptr_t valid = (rob_ptr != 0);
 433                                 needs_reorder = valid;
 434                                 rob_ptr |=
 435                                         ((valid - 1) & (uintptr_t)&dummy_rob);
 436                                 struct reorder_buffer_entry *tmp_rob_ptr =
 437                                         (struct reorder_buffer_entry *)rob_ptr;
 438                                 tmp_rob_ptr->ready = eop * needs_reorder;
 439                         }
 440
 441                         port->inflights -= eop;
 442                         port->hist_tail += eop;
 443                 }
 444                 if (flags & QE_FLAG_VALID) {
 445                         port->stats.rx_pkts++;
 446
 447                         if (allow_reorder && needs_reorder) {
 448                                 struct reorder_buffer_entry *rob_entry =
 449                                                 hist_entry->rob_entry;
 450
 451                                 hist_entry->rob_entry = NULL;
 452                                 /* Although fragmentation not currently
 453                                  * supported by eventdev API, we support it
 454                                  * here. Open: How do we alert the user that
 455                                  * they've exceeded max frags?
 456                                  */
 457                                 int num_frag = rob_entry->num_fragments;
 458                                 if (num_frag == SW_FRAGMENTS_MAX)
 459                                         sw->stats.rx_dropped++;
 460                                 else {
 461                                         int idx = rob_entry->num_fragments++;
 462                                         rob_entry->fragments[idx] = *qe;
 463                                 }
 464                                 goto end_qe;
 465                         }
 466
 467                         /* Use the iq_num from above to push the QE
 468                          * into the qid at the right priority
 469                          */
 470
 471                         qid->iq_pkt_mask |= (1 << (iq_num));
 472                         iq_ring_enqueue(qid->iq[iq_num], qe);
 473                         qid->iq_pkt_count[iq_num]++;
 474                         qid->stats.rx_pkts++;
 475                         pkts_iter++;
 476                 }
 477
 478 end_qe:
 479                 port->pp_buf_start++;
 480                 port->pp_buf_count--;
 481         } /* while (avail_qes) */
 482
 483         return pkts_iter;
 484 }
 485
 486 static uint32_t
 487 sw_schedule_pull_port_lb(struct sw_evdev *sw, uint32_t port_id)
 488 {
 489         return __pull_port_lb(sw, port_id, 1);
 490 }
 491
 492 static uint32_t
 493 sw_schedule_pull_port_no_reorder(struct sw_evdev *sw, uint32_t port_id)
 494 {
 495         return __pull_port_lb(sw, port_id, 0);
 496 }
 497
 498 static uint32_t
 499 sw_schedule_pull_port_dir(struct sw_evdev *sw, uint32_t port_id)
 500 {
 501         uint32_t pkts_iter = 0;
 502         struct sw_port *port = &sw->ports[port_id];
 503
 504         /* If shadow ring has 0 pkts, pull from worker ring */
 505         if (port->pp_buf_count == 0)
 506                 sw_refill_pp_buf(sw, port);
 507
 508         while (port->pp_buf_count) {
 509                 const struct rte_event *qe = &port->pp_buf[port->pp_buf_start];
 510                 uint8_t flags = qe->op;
 511
 512                 if ((flags & QE_FLAG_VALID) == 0)
 513                         goto end_qe;
 514
 515                 uint32_t iq_num = PRIO_TO_IQ(qe->priority);
 516                 struct sw_qid *qid = &sw->qids[qe->queue_id];
 517                 struct iq_ring *iq_ring = qid->iq[iq_num];
 518
 519                 if (iq_ring_free_count(iq_ring) == 0)
 520                         break; /* move to next port */
 521
 522                 port->stats.rx_pkts++;
 523
 524                 /* Use the iq_num from above to push the QE
 525                  * into the qid at the right priority
 526                  */
 527                 qid->iq_pkt_mask |= (1 << (iq_num));
 528                 iq_ring_enqueue(iq_ring, qe);
 529                 qid->iq_pkt_count[iq_num]++;
 530                 qid->stats.rx_pkts++;
 531                 pkts_iter++;
 532
 533 end_qe:
 534                 port->pp_buf_start++;
 535                 port->pp_buf_count--;
 536         } /* while port->pp_buf_count */
 537
 538         return pkts_iter;
 539 }
 540
 541 void
 542 sw_event_schedule(struct rte_eventdev *dev)
 543 {
 544         struct sw_evdev *sw = sw_pmd_priv(dev);
 545         uint32_t in_pkts, out_pkts;
 546         uint32_t out_pkts_total = 0, in_pkts_total = 0;
 547         int32_t sched_quanta = sw->sched_quanta;
 548         uint32_t i;
 549
 550         sw->sched_called++;
 551         if (!sw->started)
 552                 return;
 553
 554         do {
 555                 uint32_t in_pkts_this_iteration = 0;
 556
 557                 /* Pull from rx_ring for ports */
 558                 do {
 559                         in_pkts = 0;
 560                         for (i = 0; i < sw->port_count; i++)
 561                                 if (sw->ports[i].is_directed)
 562                                         in_pkts += sw_schedule_pull_port_dir(sw, i);
 563                                 else if (sw->ports[i].num_ordered_qids > 0)
 564                                         in_pkts += sw_schedule_pull_port_lb(sw, i);
 565                                 else
 566                                         in_pkts += sw_schedule_pull_port_no_reorder(sw, i);
 567
 568                         /* QID scan for re-ordered */
 569                         in_pkts += sw_schedule_reorder(sw, 0,
 570                                         sw->qid_count);
 571                         in_pkts_this_iteration += in_pkts;
 572                 } while (in_pkts > 4 &&
 573                                 (int)in_pkts_this_iteration < sched_quanta);
 574
 575                 out_pkts = 0;
 576                 out_pkts += sw_schedule_qid_to_cq(sw);
 577                 out_pkts_total += out_pkts;
 578                 in_pkts_total += in_pkts_this_iteration;
 579
 580                 if (in_pkts == 0 && out_pkts == 0)
 581                         break;
 582         } while ((int)out_pkts_total < sched_quanta);
 583
 584         /* push all the internal buffered QEs in port->cq_ring to the
 585          * worker cores: aka, do the ring transfers batched.
 586          */
 587         for (i = 0; i < sw->port_count; i++) {
 588                 struct qe_ring *worker = sw->ports[i].cq_worker_ring;
 589                 qe_ring_enqueue_burst(worker, sw->ports[i].cq_buf,
 590                                 sw->ports[i].cq_buf_count,
 591                                 &sw->cq_ring_space[i]);
 592                 sw->ports[i].cq_buf_count = 0;
 593         }
 594
 595         sw->stats.tx_pkts += out_pkts_total;
 596         sw->stats.rx_pkts += in_pkts_total;
 597
 598         sw->sched_no_iq_enqueues += (in_pkts_total == 0);
 599         sw->sched_no_cq_enqueues += (out_pkts_total == 0);
 600
 601 }