drivers/net/liquidio/lio_rxtx.c

   1 /*
   2  *   BSD LICENSE
   3  *
   4  *   Copyright(c) 2017 Cavium, Inc.. All rights reserved.
   5  *   All rights reserved.
   6  *
   7  *   Redistribution and use in source and binary forms, with or without
   8  *   modification, are permitted provided that the following conditions
   9  *   are met:
  10  *
  11  *     * Redistributions of source code must retain the above copyright
  12  *       notice, this list of conditions and the following disclaimer.
  13  *     * Redistributions in binary form must reproduce the above copyright
  14  *       notice, this list of conditions and the following disclaimer in
  15  *       the documentation and/or other materials provided with the
  16  *       distribution.
  17  *     * Neither the name of Cavium, Inc. nor the names of its
  18  *       contributors may be used to endorse or promote products derived
  19  *       from this software without specific prior written permission.
  20  *
  21  *   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  22  *   "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  23  *   LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  24  *   A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  25  *   OWNER(S) OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  26  *   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  27  *   LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  28  *   DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  29  *   THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  30  *   (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  31  *   OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  32  */
  33
  34 #include <rte_ethdev.h>
  35 #include <rte_cycles.h>
  36 #include <rte_malloc.h>
  37
  38 #include "lio_logs.h"
  39 #include "lio_struct.h"
  40 #include "lio_ethdev.h"
  41 #include "lio_rxtx.h"
  42
  43 static void
  44 lio_droq_compute_max_packet_bufs(struct lio_droq *droq)
  45 {
  46         uint32_t count = 0;
  47
  48         do {
  49                 count += droq->buffer_size;
  50         } while (count < LIO_MAX_RX_PKTLEN);
  51 }
  52
  53 static void
  54 lio_droq_reset_indices(struct lio_droq *droq)
  55 {
  56         droq->read_idx  = 0;
  57         droq->write_idx = 0;
  58         droq->refill_idx = 0;
  59         droq->refill_count = 0;
  60         rte_atomic64_set(&droq->pkts_pending, 0);
  61 }
  62
  63 static void
  64 lio_droq_destroy_ring_buffers(struct lio_droq *droq)
  65 {
  66         uint32_t i;
  67
  68         for (i = 0; i < droq->max_count; i++) {
  69                 if (droq->recv_buf_list[i].buffer) {
  70                         rte_pktmbuf_free((struct rte_mbuf *)
  71                                          droq->recv_buf_list[i].buffer);
  72                         droq->recv_buf_list[i].buffer = NULL;
  73                 }
  74         }
  75
  76         lio_droq_reset_indices(droq);
  77 }
  78
  79 static void *
  80 lio_recv_buffer_alloc(struct lio_device *lio_dev, int q_no)
  81 {
  82         struct lio_droq *droq = lio_dev->droq[q_no];
  83         struct rte_mempool *mpool = droq->mpool;
  84         struct rte_mbuf *m;
  85
  86         m = rte_pktmbuf_alloc(mpool);
  87         if (m == NULL) {
  88                 lio_dev_err(lio_dev, "Cannot allocate\n");
  89                 return NULL;
  90         }
  91
  92         rte_mbuf_refcnt_set(m, 1);
  93         m->next = NULL;
  94         m->data_off = RTE_PKTMBUF_HEADROOM;
  95         m->nb_segs = 1;
  96         m->pool = mpool;
  97
  98         return m;
  99 }
 100
 101 static int
 102 lio_droq_setup_ring_buffers(struct lio_device *lio_dev,
 103                             struct lio_droq *droq)
 104 {
 105         struct lio_droq_desc *desc_ring = droq->desc_ring;
 106         uint32_t i;
 107         void *buf;
 108
 109         for (i = 0; i < droq->max_count; i++) {
 110                 buf = lio_recv_buffer_alloc(lio_dev, droq->q_no);
 111                 if (buf == NULL) {
 112                         lio_dev_err(lio_dev, "buffer alloc failed\n");
 113                         lio_droq_destroy_ring_buffers(droq);
 114                         return -ENOMEM;
 115                 }
 116
 117                 droq->recv_buf_list[i].buffer = buf;
 118                 droq->info_list[i].length = 0;
 119
 120                 /* map ring buffers into memory */
 121                 desc_ring[i].info_ptr = lio_map_ring_info(droq, i);
 122                 desc_ring[i].buffer_ptr =
 123                         lio_map_ring(droq->recv_buf_list[i].buffer);
 124         }
 125
 126         lio_droq_reset_indices(droq);
 127
 128         lio_droq_compute_max_packet_bufs(droq);
 129
 130         return 0;
 131 }
 132
 133 static void
 134 lio_dma_zone_free(struct lio_device *lio_dev, const struct rte_memzone *mz)
 135 {
 136         const struct rte_memzone *mz_tmp;
 137         int ret = 0;
 138
 139         if (mz == NULL) {
 140                 lio_dev_err(lio_dev, "Memzone NULL\n");
 141                 return;
 142         }
 143
 144         mz_tmp = rte_memzone_lookup(mz->name);
 145         if (mz_tmp == NULL) {
 146                 lio_dev_err(lio_dev, "Memzone %s Not Found\n", mz->name);
 147                 return;
 148         }
 149
 150         ret = rte_memzone_free(mz);
 151         if (ret)
 152                 lio_dev_err(lio_dev, "Memzone free Failed ret %d\n", ret);
 153 }
 154
 155 /**
 156  *  Frees the space for descriptor ring for the droq.
 157  *
 158  *  @param lio_dev      - pointer to the lio device structure
 159  *  @param q_no         - droq no.
 160  */
 161 static void
 162 lio_delete_droq(struct lio_device *lio_dev, uint32_t q_no)
 163 {
 164         struct lio_droq *droq = lio_dev->droq[q_no];
 165
 166         lio_dev_dbg(lio_dev, "OQ[%d]\n", q_no);
 167
 168         lio_droq_destroy_ring_buffers(droq);
 169         rte_free(droq->recv_buf_list);
 170         droq->recv_buf_list = NULL;
 171         lio_dma_zone_free(lio_dev, droq->info_mz);
 172         lio_dma_zone_free(lio_dev, droq->desc_ring_mz);
 173
 174         memset(droq, 0, LIO_DROQ_SIZE);
 175 }
 176
 177 static void *
 178 lio_alloc_info_buffer(struct lio_device *lio_dev,
 179                       struct lio_droq *droq, unsigned int socket_id)
 180 {
 181         droq->info_mz = rte_eth_dma_zone_reserve(lio_dev->eth_dev,
 182                                                  "info_list", droq->q_no,
 183                                                  (droq->max_count *
 184                                                         LIO_DROQ_INFO_SIZE),
 185                                                  RTE_CACHE_LINE_SIZE,
 186                                                  socket_id);
 187
 188         if (droq->info_mz == NULL)
 189                 return NULL;
 190
 191         droq->info_list_dma = droq->info_mz->phys_addr;
 192         droq->info_alloc_size = droq->info_mz->len;
 193         droq->info_base_addr = (size_t)droq->info_mz->addr;
 194
 195         return droq->info_mz->addr;
 196 }
 197
 198 /**
 199  *  Allocates space for the descriptor ring for the droq and
 200  *  sets the base addr, num desc etc in Octeon registers.
 201  *
 202  * @param lio_dev       - pointer to the lio device structure
 203  * @param q_no          - droq no.
 204  * @param app_ctx       - pointer to application context
 205  * @return Success: 0   Failure: -1
 206  */
 207 static int
 208 lio_init_droq(struct lio_device *lio_dev, uint32_t q_no,
 209               uint32_t num_descs, uint32_t desc_size,
 210               struct rte_mempool *mpool, unsigned int socket_id)
 211 {
 212         uint32_t c_refill_threshold;
 213         uint32_t desc_ring_size;
 214         struct lio_droq *droq;
 215
 216         lio_dev_dbg(lio_dev, "OQ[%d]\n", q_no);
 217
 218         droq = lio_dev->droq[q_no];
 219         droq->lio_dev = lio_dev;
 220         droq->q_no = q_no;
 221         droq->mpool = mpool;
 222
 223         c_refill_threshold = LIO_OQ_REFILL_THRESHOLD_CFG(lio_dev);
 224
 225         droq->max_count = num_descs;
 226         droq->buffer_size = desc_size;
 227
 228         desc_ring_size = droq->max_count * LIO_DROQ_DESC_SIZE;
 229         droq->desc_ring_mz = rte_eth_dma_zone_reserve(lio_dev->eth_dev,
 230                                                       "droq", q_no,
 231                                                       desc_ring_size,
 232                                                       RTE_CACHE_LINE_SIZE,
 233                                                       socket_id);
 234
 235         if (droq->desc_ring_mz == NULL) {
 236                 lio_dev_err(lio_dev,
 237                             "Output queue %d ring alloc failed\n", q_no);
 238                 return -1;
 239         }
 240
 241         droq->desc_ring_dma = droq->desc_ring_mz->phys_addr;
 242         droq->desc_ring = (struct lio_droq_desc *)droq->desc_ring_mz->addr;
 243
 244         lio_dev_dbg(lio_dev, "droq[%d]: desc_ring: virt: 0x%p, dma: %lx\n",
 245                     q_no, droq->desc_ring, (unsigned long)droq->desc_ring_dma);
 246         lio_dev_dbg(lio_dev, "droq[%d]: num_desc: %d\n", q_no,
 247                     droq->max_count);
 248
 249         droq->info_list = lio_alloc_info_buffer(lio_dev, droq, socket_id);
 250         if (droq->info_list == NULL) {
 251                 lio_dev_err(lio_dev, "Cannot allocate memory for info list.\n");
 252                 goto init_droq_fail;
 253         }
 254
 255         droq->recv_buf_list = rte_zmalloc_socket("recv_buf_list",
 256                                                  (droq->max_count *
 257                                                         LIO_DROQ_RECVBUF_SIZE),
 258                                                  RTE_CACHE_LINE_SIZE,
 259                                                  socket_id);
 260         if (droq->recv_buf_list == NULL) {
 261                 lio_dev_err(lio_dev,
 262                             "Output queue recv buf list alloc failed\n");
 263                 goto init_droq_fail;
 264         }
 265
 266         if (lio_droq_setup_ring_buffers(lio_dev, droq))
 267                 goto init_droq_fail;
 268
 269         droq->refill_threshold = c_refill_threshold;
 270
 271         rte_spinlock_init(&droq->lock);
 272
 273         lio_dev->fn_list.setup_oq_regs(lio_dev, q_no);
 274
 275         lio_dev->io_qmask.oq |= (1ULL << q_no);
 276
 277         return 0;
 278
 279 init_droq_fail:
 280         lio_delete_droq(lio_dev, q_no);
 281
 282         return -1;
 283 }
 284
 285 int
 286 lio_setup_droq(struct lio_device *lio_dev, int oq_no, int num_descs,
 287                int desc_size, struct rte_mempool *mpool, unsigned int socket_id)
 288 {
 289         struct lio_droq *droq;
 290
 291         PMD_INIT_FUNC_TRACE();
 292
 293         if (lio_dev->droq[oq_no]) {
 294                 lio_dev_dbg(lio_dev, "Droq %d in use\n", oq_no);
 295                 return 0;
 296         }
 297
 298         /* Allocate the DS for the new droq. */
 299         droq = rte_zmalloc_socket("ethdev RX queue", sizeof(*droq),
 300                                   RTE_CACHE_LINE_SIZE, socket_id);
 301         if (droq == NULL)
 302                 return -ENOMEM;
 303
 304         lio_dev->droq[oq_no] = droq;
 305
 306         /* Initialize the Droq */
 307         if (lio_init_droq(lio_dev, oq_no, num_descs, desc_size, mpool,
 308                           socket_id)) {
 309                 lio_dev_err(lio_dev, "Droq[%u] Initialization Failed\n", oq_no);
 310                 rte_free(lio_dev->droq[oq_no]);
 311                 lio_dev->droq[oq_no] = NULL;
 312                 return -ENOMEM;
 313         }
 314
 315         lio_dev->num_oqs++;
 316
 317         lio_dev_dbg(lio_dev, "Total number of OQ: %d\n", lio_dev->num_oqs);
 318
 319         /* Send credit for octeon output queues. credits are always
 320          * sent after the output queue is enabled.
 321          */
 322         rte_write32(lio_dev->droq[oq_no]->max_count,
 323                     lio_dev->droq[oq_no]->pkts_credit_reg);
 324         rte_wmb();
 325
 326         return 0;
 327 }
 328
 329 static inline uint32_t
 330 lio_droq_get_bufcount(uint32_t buf_size, uint32_t total_len)
 331 {
 332         uint32_t buf_cnt = 0;
 333
 334         while (total_len > (buf_size * buf_cnt))
 335                 buf_cnt++;
 336
 337         return buf_cnt;
 338 }
 339
 340 /* If we were not able to refill all buffers, try to move around
 341  * the buffers that were not dispatched.
 342  */
 343 static inline uint32_t
 344 lio_droq_refill_pullup_descs(struct lio_droq *droq,
 345                              struct lio_droq_desc *desc_ring)
 346 {
 347         uint32_t refill_index = droq->refill_idx;
 348         uint32_t desc_refilled = 0;
 349
 350         while (refill_index != droq->read_idx) {
 351                 if (droq->recv_buf_list[refill_index].buffer) {
 352                         droq->recv_buf_list[droq->refill_idx].buffer =
 353                                 droq->recv_buf_list[refill_index].buffer;
 354                         desc_ring[droq->refill_idx].buffer_ptr =
 355                                 desc_ring[refill_index].buffer_ptr;
 356                         droq->recv_buf_list[refill_index].buffer = NULL;
 357                         desc_ring[refill_index].buffer_ptr = 0;
 358                         do {
 359                                 droq->refill_idx = lio_incr_index(
 360                                                         droq->refill_idx, 1,
 361                                                         droq->max_count);
 362                                 desc_refilled++;
 363                                 droq->refill_count--;
 364                         } while (droq->recv_buf_list[droq->refill_idx].buffer);
 365                 }
 366                 refill_index = lio_incr_index(refill_index, 1,
 367                                               droq->max_count);
 368         }       /* while */
 369
 370         return desc_refilled;
 371 }
 372
 373 /* lio_droq_refill
 374  *
 375  * @param lio_dev       - pointer to the lio device structure
 376  * @param droq          - droq in which descriptors require new buffers.
 377  *
 378  * Description:
 379  *  Called during normal DROQ processing in interrupt mode or by the poll
 380  *  thread to refill the descriptors from which buffers were dispatched
 381  *  to upper layers. Attempts to allocate new buffers. If that fails, moves
 382  *  up buffers (that were not dispatched) to form a contiguous ring.
 383  *
 384  * Returns:
 385  *  No of descriptors refilled.
 386  *
 387  * Locks:
 388  * This routine is called with droq->lock held.
 389  */
 390 static uint32_t
 391 lio_droq_refill(struct lio_device *lio_dev, struct lio_droq *droq)
 392 {
 393         struct lio_droq_desc *desc_ring;
 394         uint32_t desc_refilled = 0;
 395         void *buf = NULL;
 396
 397         desc_ring = droq->desc_ring;
 398
 399         while (droq->refill_count && (desc_refilled < droq->max_count)) {
 400                 /* If a valid buffer exists (happens if there is no dispatch),
 401                  * reuse the buffer, else allocate.
 402                  */
 403                 if (droq->recv_buf_list[droq->refill_idx].buffer == NULL) {
 404                         buf = lio_recv_buffer_alloc(lio_dev, droq->q_no);
 405                         /* If a buffer could not be allocated, no point in
 406                          * continuing
 407                          */
 408                         if (buf == NULL)
 409                                 break;
 410
 411                         droq->recv_buf_list[droq->refill_idx].buffer = buf;
 412                 }
 413
 414                 desc_ring[droq->refill_idx].buffer_ptr =
 415                     lio_map_ring(droq->recv_buf_list[droq->refill_idx].buffer);
 416                 /* Reset any previous values in the length field. */
 417                 droq->info_list[droq->refill_idx].length = 0;
 418
 419                 droq->refill_idx = lio_incr_index(droq->refill_idx, 1,
 420                                                   droq->max_count);
 421                 desc_refilled++;
 422                 droq->refill_count--;
 423         }
 424
 425         if (droq->refill_count)
 426                 desc_refilled += lio_droq_refill_pullup_descs(droq, desc_ring);
 427
 428         /* if droq->refill_count
 429          * The refill count would not change in pass two. We only moved buffers
 430          * to close the gap in the ring, but we would still have the same no. of
 431          * buffers to refill.
 432          */
 433         return desc_refilled;
 434 }
 435
 436 static int
 437 lio_droq_fast_process_packet(struct lio_device *lio_dev,
 438                              struct lio_droq *droq,
 439                              struct rte_mbuf **rx_pkts)
 440 {
 441         struct rte_mbuf *nicbuf = NULL;
 442         struct lio_droq_info *info;
 443         uint32_t total_len = 0;
 444         int data_total_len = 0;
 445         uint32_t pkt_len = 0;
 446         union octeon_rh *rh;
 447         int data_pkts = 0;
 448
 449         info = &droq->info_list[droq->read_idx];
 450         lio_swap_8B_data((uint64_t *)info, 2);
 451
 452         if (!info->length)
 453                 return -1;
 454
 455         /* Len of resp hdr in included in the received data len. */
 456         info->length -= OCTEON_RH_SIZE;
 457         rh = &info->rh;
 458
 459         total_len += (uint32_t)info->length;
 460
 461         if (lio_opcode_slow_path(rh)) {
 462                 uint32_t buf_cnt;
 463
 464                 buf_cnt = lio_droq_get_bufcount(droq->buffer_size,
 465                                                 (uint32_t)info->length);
 466                 droq->read_idx = lio_incr_index(droq->read_idx, buf_cnt,
 467                                                 droq->max_count);
 468                 droq->refill_count += buf_cnt;
 469         } else {
 470                 if (info->length <= droq->buffer_size) {
 471                         if (rh->r_dh.has_hash)
 472                                 pkt_len = (uint32_t)(info->length - 8);
 473                         else
 474                                 pkt_len = (uint32_t)info->length;
 475
 476                         nicbuf = droq->recv_buf_list[droq->read_idx].buffer;
 477                         droq->recv_buf_list[droq->read_idx].buffer = NULL;
 478                         droq->read_idx = lio_incr_index(
 479                                                 droq->read_idx, 1,
 480                                                 droq->max_count);
 481                         droq->refill_count++;
 482
 483                         if (likely(nicbuf != NULL)) {
 484                                 nicbuf->data_off = RTE_PKTMBUF_HEADROOM;
 485                                 nicbuf->nb_segs = 1;
 486                                 nicbuf->next = NULL;
 487                                 /* We don't have a way to pass flags yet */
 488                                 nicbuf->ol_flags = 0;
 489                                 if (rh->r_dh.has_hash) {
 490                                         uint64_t *hash_ptr;
 491
 492                                         nicbuf->ol_flags |= PKT_RX_RSS_HASH;
 493                                         hash_ptr = rte_pktmbuf_mtod(nicbuf,
 494                                                                     uint64_t *);
 495                                         lio_swap_8B_data(hash_ptr, 1);
 496                                         nicbuf->hash.rss = (uint32_t)*hash_ptr;
 497                                         nicbuf->data_off += 8;
 498                                 }
 499
 500                                 nicbuf->pkt_len = pkt_len;
 501                                 nicbuf->data_len = pkt_len;
 502                                 nicbuf->port = lio_dev->port_id;
 503                                 /* Store the mbuf */
 504                                 rx_pkts[data_pkts++] = nicbuf;
 505                                 data_total_len += pkt_len;
 506                         }
 507
 508                         /* Prefetch buffer pointers when on a cache line
 509                          * boundary
 510                          */
 511                         if ((droq->read_idx & 3) == 0) {
 512                                 rte_prefetch0(
 513                                     &droq->recv_buf_list[droq->read_idx]);
 514                                 rte_prefetch0(
 515                                     &droq->info_list[droq->read_idx]);
 516                         }
 517                 } else {
 518                         struct rte_mbuf *first_buf = NULL;
 519                         struct rte_mbuf *last_buf = NULL;
 520
 521                         while (pkt_len < info->length) {
 522                                 int cpy_len = 0;
 523
 524                                 cpy_len = ((pkt_len + droq->buffer_size) >
 525                                                 info->length)
 526                                                 ? ((uint32_t)info->length -
 527                                                         pkt_len)
 528                                                 : droq->buffer_size;
 529
 530                                 nicbuf =
 531                                     droq->recv_buf_list[droq->read_idx].buffer;
 532                                 droq->recv_buf_list[droq->read_idx].buffer =
 533                                     NULL;
 534
 535                                 if (likely(nicbuf != NULL)) {
 536                                         /* Note the first seg */
 537                                         if (!pkt_len)
 538                                                 first_buf = nicbuf;
 539
 540                                         nicbuf->data_off = RTE_PKTMBUF_HEADROOM;
 541                                         nicbuf->nb_segs = 1;
 542                                         nicbuf->next = NULL;
 543                                         nicbuf->port = lio_dev->port_id;
 544                                         /* We don't have a way to pass
 545                                          * flags yet
 546                                          */
 547                                         nicbuf->ol_flags = 0;
 548                                         if ((!pkt_len) && (rh->r_dh.has_hash)) {
 549                                                 uint64_t *hash_ptr;
 550
 551                                                 nicbuf->ol_flags |=
 552                                                     PKT_RX_RSS_HASH;
 553                                                 hash_ptr = rte_pktmbuf_mtod(
 554                                                     nicbuf, uint64_t *);
 555                                                 lio_swap_8B_data(hash_ptr, 1);
 556                                                 nicbuf->hash.rss =
 557                                                     (uint32_t)*hash_ptr;
 558                                                 nicbuf->data_off += 8;
 559                                                 nicbuf->pkt_len = cpy_len - 8;
 560                                                 nicbuf->data_len = cpy_len - 8;
 561                                         } else {
 562                                                 nicbuf->pkt_len = cpy_len;
 563                                                 nicbuf->data_len = cpy_len;
 564                                         }
 565
 566                                         if (pkt_len)
 567                                                 first_buf->nb_segs++;
 568
 569                                         if (last_buf)
 570                                                 last_buf->next = nicbuf;
 571
 572                                         last_buf = nicbuf;
 573                                 } else {
 574                                         PMD_RX_LOG(lio_dev, ERR, "no buf\n");
 575                                 }
 576
 577                                 pkt_len += cpy_len;
 578                                 droq->read_idx = lio_incr_index(
 579                                                         droq->read_idx,
 580                                                         1, droq->max_count);
 581                                 droq->refill_count++;
 582
 583                                 /* Prefetch buffer pointers when on a
 584                                  * cache line boundary
 585                                  */
 586                                 if ((droq->read_idx & 3) == 0) {
 587                                         rte_prefetch0(&droq->recv_buf_list
 588                                                               [droq->read_idx]);
 589
 590                                         rte_prefetch0(
 591                                             &droq->info_list[droq->read_idx]);
 592                                 }
 593                         }
 594                         rx_pkts[data_pkts++] = first_buf;
 595                         if (rh->r_dh.has_hash)
 596                                 data_total_len += (pkt_len - 8);
 597                         else
 598                                 data_total_len += pkt_len;
 599                 }
 600
 601                 /* Inform upper layer about packet checksum verification */
 602                 struct rte_mbuf *m = rx_pkts[data_pkts - 1];
 603
 604                 if (rh->r_dh.csum_verified & LIO_IP_CSUM_VERIFIED)
 605                         m->ol_flags |= PKT_RX_IP_CKSUM_GOOD;
 606
 607                 if (rh->r_dh.csum_verified & LIO_L4_CSUM_VERIFIED)
 608                         m->ol_flags |= PKT_RX_L4_CKSUM_GOOD;
 609         }
 610
 611         if (droq->refill_count >= droq->refill_threshold) {
 612                 int desc_refilled = lio_droq_refill(lio_dev, droq);
 613
 614                 /* Flush the droq descriptor data to memory to be sure
 615                  * that when we update the credits the data in memory is
 616                  * accurate.
 617                  */
 618                 rte_wmb();
 619                 rte_write32(desc_refilled, droq->pkts_credit_reg);
 620                 /* make sure mmio write completes */
 621                 rte_wmb();
 622         }
 623
 624         info->length = 0;
 625         info->rh.rh64 = 0;
 626
 627         return data_pkts;
 628 }
 629
 630 static uint32_t
 631 lio_droq_fast_process_packets(struct lio_device *lio_dev,
 632                               struct lio_droq *droq,
 633                               struct rte_mbuf **rx_pkts,
 634                               uint32_t pkts_to_process)
 635 {
 636         int ret, data_pkts = 0;
 637         uint32_t pkt;
 638
 639         for (pkt = 0; pkt < pkts_to_process; pkt++) {
 640                 ret = lio_droq_fast_process_packet(lio_dev, droq,
 641                                                    &rx_pkts[data_pkts]);
 642                 if (ret < 0) {
 643                         lio_dev_err(lio_dev, "Port[%d] DROQ[%d] idx: %d len:0, pkt_cnt: %d\n",
 644                                     lio_dev->port_id, droq->q_no,
 645                                     droq->read_idx, pkts_to_process);
 646                         break;
 647                 }
 648                 data_pkts += ret;
 649         }
 650
 651         rte_atomic64_sub(&droq->pkts_pending, pkt);
 652
 653         return data_pkts;
 654 }
 655
 656 static inline uint32_t
 657 lio_droq_check_hw_for_pkts(struct lio_droq *droq)
 658 {
 659         uint32_t last_count;
 660         uint32_t pkt_count;
 661
 662         pkt_count = rte_read32(droq->pkts_sent_reg);
 663
 664         last_count = pkt_count - droq->pkt_count;
 665         droq->pkt_count = pkt_count;
 666
 667         if (last_count)
 668                 rte_atomic64_add(&droq->pkts_pending, last_count);
 669
 670         return last_count;
 671 }
 672
 673 uint16_t
 674 lio_dev_recv_pkts(void *rx_queue,
 675                   struct rte_mbuf **rx_pkts,
 676                   uint16_t budget)
 677 {
 678         struct lio_droq *droq = rx_queue;
 679         struct lio_device *lio_dev = droq->lio_dev;
 680         uint32_t pkts_processed = 0;
 681         uint32_t pkt_count = 0;
 682
 683         lio_droq_check_hw_for_pkts(droq);
 684
 685         pkt_count = rte_atomic64_read(&droq->pkts_pending);
 686         if (!pkt_count)
 687                 return 0;
 688
 689         if (pkt_count > budget)
 690                 pkt_count = budget;
 691
 692         /* Grab the lock */
 693         rte_spinlock_lock(&droq->lock);
 694         pkts_processed = lio_droq_fast_process_packets(lio_dev,
 695                                                        droq, rx_pkts,
 696                                                        pkt_count);
 697
 698         if (droq->pkt_count) {
 699                 rte_write32(droq->pkt_count, droq->pkts_sent_reg);
 700                 droq->pkt_count = 0;
 701         }
 702
 703         /* Release the spin lock */
 704         rte_spinlock_unlock(&droq->lock);
 705
 706         return pkts_processed;
 707 }
 708
 709 void
 710 lio_delete_droq_queue(struct lio_device *lio_dev,
 711                       int oq_no)
 712 {
 713         lio_delete_droq(lio_dev, oq_no);
 714         lio_dev->num_oqs--;
 715         rte_free(lio_dev->droq[oq_no]);
 716         lio_dev->droq[oq_no] = NULL;
 717 }
 718
 719 /**
 720  *  lio_init_instr_queue()
 721  *  @param lio_dev      - pointer to the lio device structure.
 722  *  @param txpciq       - queue to be initialized.
 723  *
 724  *  Called at driver init time for each input queue. iq_conf has the
 725  *  configuration parameters for the queue.
 726  *
 727  *  @return  Success: 0 Failure: -1
 728  */
 729 static int
 730 lio_init_instr_queue(struct lio_device *lio_dev,
 731                      union octeon_txpciq txpciq,
 732                      uint32_t num_descs, unsigned int socket_id)
 733 {
 734         uint32_t iq_no = (uint32_t)txpciq.s.q_no;
 735         struct lio_instr_queue *iq;
 736         uint32_t instr_type;
 737         uint32_t q_size;
 738
 739         instr_type = LIO_IQ_INSTR_TYPE(lio_dev);
 740
 741         q_size = instr_type * num_descs;
 742         iq = lio_dev->instr_queue[iq_no];
 743         iq->iq_mz = rte_eth_dma_zone_reserve(lio_dev->eth_dev,
 744                                              "instr_queue", iq_no, q_size,
 745                                              RTE_CACHE_LINE_SIZE,
 746                                              socket_id);
 747         if (iq->iq_mz == NULL) {
 748                 lio_dev_err(lio_dev, "Cannot allocate memory for instr queue %d\n",
 749                             iq_no);
 750                 return -1;
 751         }
 752
 753         iq->base_addr_dma = iq->iq_mz->phys_addr;
 754         iq->base_addr = (uint8_t *)iq->iq_mz->addr;
 755
 756         iq->max_count = num_descs;
 757
 758         /* Initialize a list to holds requests that have been posted to Octeon
 759          * but has yet to be fetched by octeon
 760          */
 761         iq->request_list = rte_zmalloc_socket("request_list",
 762                                               sizeof(*iq->request_list) *
 763                                                         num_descs,
 764                                               RTE_CACHE_LINE_SIZE,
 765                                               socket_id);
 766         if (iq->request_list == NULL) {
 767                 lio_dev_err(lio_dev, "Alloc failed for IQ[%d] nr free list\n",
 768                             iq_no);
 769                 lio_dma_zone_free(lio_dev, iq->iq_mz);
 770                 return -1;
 771         }
 772
 773         lio_dev_dbg(lio_dev, "IQ[%d]: base: %p basedma: %lx count: %d\n",
 774                     iq_no, iq->base_addr, (unsigned long)iq->base_addr_dma,
 775                     iq->max_count);
 776
 777         iq->lio_dev = lio_dev;
 778         iq->txpciq.txpciq64 = txpciq.txpciq64;
 779         iq->fill_cnt = 0;
 780         iq->host_write_index = 0;
 781         iq->lio_read_index = 0;
 782         iq->flush_index = 0;
 783
 784         rte_atomic64_set(&iq->instr_pending, 0);
 785
 786         /* Initialize the spinlock for this instruction queue */
 787         rte_spinlock_init(&iq->lock);
 788         rte_spinlock_init(&iq->post_lock);
 789
 790         rte_atomic64_clear(&iq->iq_flush_running);
 791
 792         lio_dev->io_qmask.iq |= (1ULL << iq_no);
 793
 794         /* Set the 32B/64B mode for each input queue */
 795         lio_dev->io_qmask.iq64B |= ((instr_type == 64) << iq_no);
 796         iq->iqcmd_64B = (instr_type == 64);
 797
 798         lio_dev->fn_list.setup_iq_regs(lio_dev, iq_no);
 799
 800         return 0;
 801 }
 802
 803 int
 804 lio_setup_instr_queue0(struct lio_device *lio_dev)
 805 {
 806         union octeon_txpciq txpciq;
 807         uint32_t num_descs = 0;
 808         uint32_t iq_no = 0;
 809
 810         num_descs = LIO_NUM_DEF_TX_DESCS_CFG(lio_dev);
 811
 812         lio_dev->num_iqs = 0;
 813
 814         lio_dev->instr_queue[0] = rte_zmalloc(NULL,
 815                                         sizeof(struct lio_instr_queue), 0);
 816         if (lio_dev->instr_queue[0] == NULL)
 817                 return -ENOMEM;
 818
 819         lio_dev->instr_queue[0]->q_index = 0;
 820         lio_dev->instr_queue[0]->app_ctx = (void *)(size_t)0;
 821         txpciq.txpciq64 = 0;
 822         txpciq.s.q_no = iq_no;
 823         txpciq.s.pkind = lio_dev->pfvf_hsword.pkind;
 824         txpciq.s.use_qpg = 0;
 825         txpciq.s.qpg = 0;
 826         if (lio_init_instr_queue(lio_dev, txpciq, num_descs, SOCKET_ID_ANY)) {
 827                 rte_free(lio_dev->instr_queue[0]);
 828                 lio_dev->instr_queue[0] = NULL;
 829                 return -1;
 830         }
 831
 832         lio_dev->num_iqs++;
 833
 834         return 0;
 835 }
 836
 837 /**
 838  *  lio_delete_instr_queue()
 839  *  @param lio_dev      - pointer to the lio device structure.
 840  *  @param iq_no        - queue to be deleted.
 841  *
 842  *  Called at driver unload time for each input queue. Deletes all
 843  *  allocated resources for the input queue.
 844  */
 845 static void
 846 lio_delete_instr_queue(struct lio_device *lio_dev, uint32_t iq_no)
 847 {
 848         struct lio_instr_queue *iq = lio_dev->instr_queue[iq_no];
 849
 850         rte_free(iq->request_list);
 851         iq->request_list = NULL;
 852         lio_dma_zone_free(lio_dev, iq->iq_mz);
 853 }
 854
 855 void
 856 lio_free_instr_queue0(struct lio_device *lio_dev)
 857 {
 858         lio_delete_instr_queue(lio_dev, 0);
 859         rte_free(lio_dev->instr_queue[0]);
 860         lio_dev->instr_queue[0] = NULL;
 861         lio_dev->num_iqs--;
 862 }
 863
 864 static inline void
 865 lio_ring_doorbell(struct lio_device *lio_dev,
 866                   struct lio_instr_queue *iq)
 867 {
 868         if (rte_atomic64_read(&lio_dev->status) == LIO_DEV_RUNNING) {
 869                 rte_write32(iq->fill_cnt, iq->doorbell_reg);
 870                 /* make sure doorbell write goes through */
 871                 rte_wmb();
 872                 iq->fill_cnt = 0;
 873         }
 874 }
 875
 876 static inline void
 877 copy_cmd_into_iq(struct lio_instr_queue *iq, uint8_t *cmd)
 878 {
 879         uint8_t *iqptr, cmdsize;
 880
 881         cmdsize = ((iq->iqcmd_64B) ? 64 : 32);
 882         iqptr = iq->base_addr + (cmdsize * iq->host_write_index);
 883
 884         rte_memcpy(iqptr, cmd, cmdsize);
 885 }
 886
 887 static inline struct lio_iq_post_status
 888 post_command2(struct lio_instr_queue *iq, uint8_t *cmd)
 889 {
 890         struct lio_iq_post_status st;
 891
 892         st.status = LIO_IQ_SEND_OK;
 893
 894         /* This ensures that the read index does not wrap around to the same
 895          * position if queue gets full before Octeon could fetch any instr.
 896          */
 897         if (rte_atomic64_read(&iq->instr_pending) >=
 898                         (int32_t)(iq->max_count - 1)) {
 899                 st.status = LIO_IQ_SEND_FAILED;
 900                 st.index = -1;
 901                 return st;
 902         }
 903
 904         if (rte_atomic64_read(&iq->instr_pending) >=
 905                         (int32_t)(iq->max_count - 2))
 906                 st.status = LIO_IQ_SEND_STOP;
 907
 908         copy_cmd_into_iq(iq, cmd);
 909
 910         /* "index" is returned, host_write_index is modified. */
 911         st.index = iq->host_write_index;
 912         iq->host_write_index = lio_incr_index(iq->host_write_index, 1,
 913                                               iq->max_count);
 914         iq->fill_cnt++;
 915
 916         /* Flush the command into memory. We need to be sure the data is in
 917          * memory before indicating that the instruction is pending.
 918          */
 919         rte_wmb();
 920
 921         rte_atomic64_inc(&iq->instr_pending);
 922
 923         return st;
 924 }
 925
 926 static inline void
 927 lio_add_to_request_list(struct lio_instr_queue *iq,
 928                         int idx, void *buf, int reqtype)
 929 {
 930         iq->request_list[idx].buf = buf;
 931         iq->request_list[idx].reqtype = reqtype;
 932 }
 933
 934 static int
 935 lio_send_command(struct lio_device *lio_dev, uint32_t iq_no, void *cmd,
 936                  void *buf, uint32_t datasize __rte_unused, uint32_t reqtype)
 937 {
 938         struct lio_instr_queue *iq = lio_dev->instr_queue[iq_no];
 939         struct lio_iq_post_status st;
 940
 941         rte_spinlock_lock(&iq->post_lock);
 942
 943         st = post_command2(iq, cmd);
 944
 945         if (st.status != LIO_IQ_SEND_FAILED) {
 946                 lio_add_to_request_list(iq, st.index, buf, reqtype);
 947                 lio_ring_doorbell(lio_dev, iq);
 948         }
 949
 950         rte_spinlock_unlock(&iq->post_lock);
 951
 952         return st.status;
 953 }
 954
 955 void
 956 lio_prepare_soft_command(struct lio_device *lio_dev,
 957                          struct lio_soft_command *sc, uint8_t opcode,
 958                          uint8_t subcode, uint32_t irh_ossp, uint64_t ossp0,
 959                          uint64_t ossp1)
 960 {
 961         struct octeon_instr_pki_ih3 *pki_ih3;
 962         struct octeon_instr_ih3 *ih3;
 963         struct octeon_instr_irh *irh;
 964         struct octeon_instr_rdp *rdp;
 965
 966         RTE_ASSERT(opcode <= 15);
 967         RTE_ASSERT(subcode <= 127);
 968
 969         ih3       = (struct octeon_instr_ih3 *)&sc->cmd.cmd3.ih3;
 970
 971         ih3->pkind = lio_dev->instr_queue[sc->iq_no]->txpciq.s.pkind;
 972
 973         pki_ih3 = (struct octeon_instr_pki_ih3 *)&sc->cmd.cmd3.pki_ih3;
 974
 975         pki_ih3->w      = 1;
 976         pki_ih3->raw    = 1;
 977         pki_ih3->utag   = 1;
 978         pki_ih3->uqpg   = lio_dev->instr_queue[sc->iq_no]->txpciq.s.use_qpg;
 979         pki_ih3->utt    = 1;
 980
 981         pki_ih3->tag    = LIO_CONTROL;
 982         pki_ih3->tagtype = OCTEON_ATOMIC_TAG;
 983         pki_ih3->qpg    = lio_dev->instr_queue[sc->iq_no]->txpciq.s.qpg;
 984         pki_ih3->pm     = 0x7;
 985         pki_ih3->sl     = 8;
 986
 987         if (sc->datasize)
 988                 ih3->dlengsz = sc->datasize;
 989
 990         irh             = (struct octeon_instr_irh *)&sc->cmd.cmd3.irh;
 991         irh->opcode     = opcode;
 992         irh->subcode    = subcode;
 993
 994         /* opcode/subcode specific parameters (ossp) */
 995         irh->ossp = irh_ossp;
 996         sc->cmd.cmd3.ossp[0] = ossp0;
 997         sc->cmd.cmd3.ossp[1] = ossp1;
 998
 999         if (sc->rdatasize) {
1000                 rdp = (struct octeon_instr_rdp *)&sc->cmd.cmd3.rdp;
1001                 rdp->pcie_port = lio_dev->pcie_port;
1002                 rdp->rlen      = sc->rdatasize;
1003                 irh->rflag = 1;
1004                 /* PKI IH3 */
1005                 ih3->fsz    = OCTEON_SOFT_CMD_RESP_IH3;
1006         } else {
1007                 irh->rflag = 0;
1008                 /* PKI IH3 */
1009                 ih3->fsz    = OCTEON_PCI_CMD_O3;
1010         }
1011 }
1012
1013 int
1014 lio_send_soft_command(struct lio_device *lio_dev,
1015                       struct lio_soft_command *sc)
1016 {
1017         struct octeon_instr_ih3 *ih3;
1018         struct octeon_instr_irh *irh;
1019         uint32_t len = 0;
1020
1021         ih3 = (struct octeon_instr_ih3 *)&sc->cmd.cmd3.ih3;
1022         if (ih3->dlengsz) {
1023                 RTE_ASSERT(sc->dmadptr);
1024                 sc->cmd.cmd3.dptr = sc->dmadptr;
1025         }
1026
1027         irh = (struct octeon_instr_irh *)&sc->cmd.cmd3.irh;
1028         if (irh->rflag) {
1029                 RTE_ASSERT(sc->dmarptr);
1030                 RTE_ASSERT(sc->status_word != NULL);
1031                 *sc->status_word = LIO_COMPLETION_WORD_INIT;
1032                 sc->cmd.cmd3.rptr = sc->dmarptr;
1033         }
1034
1035         len = (uint32_t)ih3->dlengsz;
1036
1037         if (sc->wait_time)
1038                 sc->timeout = lio_uptime + sc->wait_time;
1039
1040         return lio_send_command(lio_dev, sc->iq_no, &sc->cmd, sc, len,
1041                                 LIO_REQTYPE_SOFT_COMMAND);
1042 }
1043
1044 int
1045 lio_setup_sc_buffer_pool(struct lio_device *lio_dev)
1046 {
1047         char sc_pool_name[RTE_MEMPOOL_NAMESIZE];
1048         uint16_t buf_size;
1049
1050         buf_size = LIO_SOFT_COMMAND_BUFFER_SIZE + RTE_PKTMBUF_HEADROOM;
1051         snprintf(sc_pool_name, sizeof(sc_pool_name),
1052                  "lio_sc_pool_%u", lio_dev->port_id);
1053         lio_dev->sc_buf_pool = rte_pktmbuf_pool_create(sc_pool_name,
1054                                                 LIO_MAX_SOFT_COMMAND_BUFFERS,
1055                                                 0, 0, buf_size, SOCKET_ID_ANY);
1056         return 0;
1057 }
1058
1059 void
1060 lio_free_sc_buffer_pool(struct lio_device *lio_dev)
1061 {
1062         rte_mempool_free(lio_dev->sc_buf_pool);
1063 }
1064
1065 struct lio_soft_command *
1066 lio_alloc_soft_command(struct lio_device *lio_dev, uint32_t datasize,
1067                        uint32_t rdatasize, uint32_t ctxsize)
1068 {
1069         uint32_t offset = sizeof(struct lio_soft_command);
1070         struct lio_soft_command *sc;
1071         struct rte_mbuf *m;
1072         uint64_t dma_addr;
1073
1074         RTE_ASSERT((offset + datasize + rdatasize + ctxsize) <=
1075                    LIO_SOFT_COMMAND_BUFFER_SIZE);
1076
1077         m = rte_pktmbuf_alloc(lio_dev->sc_buf_pool);
1078         if (m == NULL) {
1079                 lio_dev_err(lio_dev, "Cannot allocate mbuf for sc\n");
1080                 return NULL;
1081         }
1082
1083         /* set rte_mbuf data size and there is only 1 segment */
1084         m->pkt_len = LIO_SOFT_COMMAND_BUFFER_SIZE;
1085         m->data_len = LIO_SOFT_COMMAND_BUFFER_SIZE;
1086
1087         /* use rte_mbuf buffer for soft command */
1088         sc = rte_pktmbuf_mtod(m, struct lio_soft_command *);
1089         memset(sc, 0, LIO_SOFT_COMMAND_BUFFER_SIZE);
1090         sc->size = LIO_SOFT_COMMAND_BUFFER_SIZE;
1091         sc->dma_addr = rte_mbuf_data_dma_addr(m);
1092         sc->mbuf = m;
1093
1094         dma_addr = sc->dma_addr;
1095
1096         if (ctxsize) {
1097                 sc->ctxptr = (uint8_t *)sc + offset;
1098                 sc->ctxsize = ctxsize;
1099         }
1100
1101         /* Start data at 128 byte boundary */
1102         offset = (offset + ctxsize + 127) & 0xffffff80;
1103
1104         if (datasize) {
1105                 sc->virtdptr = (uint8_t *)sc + offset;
1106                 sc->dmadptr = dma_addr + offset;
1107                 sc->datasize = datasize;
1108         }
1109
1110         /* Start rdata at 128 byte boundary */
1111         offset = (offset + datasize + 127) & 0xffffff80;
1112
1113         if (rdatasize) {
1114                 RTE_ASSERT(rdatasize >= 16);
1115                 sc->virtrptr = (uint8_t *)sc + offset;
1116                 sc->dmarptr = dma_addr + offset;
1117                 sc->rdatasize = rdatasize;
1118                 sc->status_word = (uint64_t *)((uint8_t *)(sc->virtrptr) +
1119                                                rdatasize - 8);
1120         }
1121
1122         return sc;
1123 }
1124
1125 void
1126 lio_free_soft_command(struct lio_soft_command *sc)
1127 {
1128         rte_pktmbuf_free(sc->mbuf);
1129 }
1130
1131 void
1132 lio_setup_response_list(struct lio_device *lio_dev)
1133 {
1134         STAILQ_INIT(&lio_dev->response_list.head);
1135         rte_spinlock_init(&lio_dev->response_list.lock);
1136         rte_atomic64_set(&lio_dev->response_list.pending_req_count, 0);
1137 }
1138
1139 int
1140 lio_process_ordered_list(struct lio_device *lio_dev)
1141 {
1142         int resp_to_process = LIO_MAX_ORD_REQS_TO_PROCESS;
1143         struct lio_response_list *ordered_sc_list;
1144         struct lio_soft_command *sc;
1145         int request_complete = 0;
1146         uint64_t status64;
1147         uint32_t status;
1148
1149         ordered_sc_list = &lio_dev->response_list;
1150
1151         do {
1152                 rte_spinlock_lock(&ordered_sc_list->lock);
1153
1154                 if (STAILQ_EMPTY(&ordered_sc_list->head)) {
1155                         /* ordered_sc_list is empty; there is
1156                          * nothing to process
1157                          */
1158                         rte_spinlock_unlock(&ordered_sc_list->lock);
1159                         return -1;
1160                 }
1161
1162                 sc = LIO_STQUEUE_FIRST_ENTRY(&ordered_sc_list->head,
1163                                              struct lio_soft_command, node);
1164
1165                 status = LIO_REQUEST_PENDING;
1166
1167                 /* check if octeon has finished DMA'ing a response
1168                  * to where rptr is pointing to
1169                  */
1170                 status64 = *sc->status_word;
1171
1172                 if (status64 != LIO_COMPLETION_WORD_INIT) {
1173                         /* This logic ensures that all 64b have been written.
1174                          * 1. check byte 0 for non-FF
1175                          * 2. if non-FF, then swap result from BE to host order
1176                          * 3. check byte 7 (swapped to 0) for non-FF
1177                          * 4. if non-FF, use the low 32-bit status code
1178                          * 5. if either byte 0 or byte 7 is FF, don't use status
1179                          */
1180                         if ((status64 & 0xff) != 0xff) {
1181                                 lio_swap_8B_data(&status64, 1);
1182                                 if (((status64 & 0xff) != 0xff)) {
1183                                         /* retrieve 16-bit firmware status */
1184                                         status = (uint32_t)(status64 &
1185                                                             0xffffULL);
1186                                         if (status) {
1187                                                 status =
1188                                                 LIO_FIRMWARE_STATUS_CODE(
1189                                                                         status);
1190                                         } else {
1191                                                 /* i.e. no error */
1192                                                 status = LIO_REQUEST_DONE;
1193                                         }
1194                                 }
1195                         }
1196                 } else if ((sc->timeout && lio_check_timeout(lio_uptime,
1197                                                              sc->timeout))) {
1198                         lio_dev_err(lio_dev,
1199                                     "cmd failed, timeout (%ld, %ld)\n",
1200                                     (long)lio_uptime, (long)sc->timeout);
1201                         status = LIO_REQUEST_TIMEOUT;
1202                 }
1203
1204                 if (status != LIO_REQUEST_PENDING) {
1205                         /* we have received a response or we have timed out.
1206                          * remove node from linked list
1207                          */
1208                         STAILQ_REMOVE(&ordered_sc_list->head,
1209                                       &sc->node, lio_stailq_node, entries);
1210                         rte_atomic64_dec(
1211                             &lio_dev->response_list.pending_req_count);
1212                         rte_spinlock_unlock(&ordered_sc_list->lock);
1213
1214                         if (sc->callback)
1215                                 sc->callback(status, sc->callback_arg);
1216
1217                         request_complete++;
1218                 } else {
1219                         /* no response yet */
1220                         request_complete = 0;
1221                         rte_spinlock_unlock(&ordered_sc_list->lock);
1222                 }
1223
1224                 /* If we hit the Max Ordered requests to process every loop,
1225                  * we quit and let this function be invoked the next time
1226                  * the poll thread runs to process the remaining requests.
1227                  * This function can take up the entire CPU if there is
1228                  * no upper limit to the requests processed.
1229                  */
1230                 if (request_complete >= resp_to_process)
1231                         break;
1232         } while (request_complete);
1233
1234         return 0;
1235 }