lib/librte_pmd_virtio/virtio_rxtx.c

   1 /*-
   2  *   BSD LICENSE
   3  *
   4  *   Copyright(c) 2010-2014 Intel Corporation. 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 Intel Corporation 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 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 <stdint.h>
  35 #include <stdio.h>
  36 #include <stdlib.h>
  37 #include <string.h>
  38 #include <errno.h>
  39
  40 #include <rte_cycles.h>
  41 #include <rte_memory.h>
  42 #include <rte_memzone.h>
  43 #include <rte_branch_prediction.h>
  44 #include <rte_mempool.h>
  45 #include <rte_malloc.h>
  46 #include <rte_mbuf.h>
  47 #include <rte_ether.h>
  48 #include <rte_ethdev.h>
  49 #include <rte_prefetch.h>
  50 #include <rte_string_fns.h>
  51 #include <rte_errno.h>
  52
  53 #include "virtio_logs.h"
  54 #include "virtio_ethdev.h"
  55 #include "virtqueue.h"
  56
  57 #ifdef RTE_LIBRTE_VIRTIO_DEBUG_DUMP
  58 #define VIRTIO_DUMP_PACKET(m, len) rte_pktmbuf_dump(stdout, m, len)
  59 #else
  60 #define  VIRTIO_DUMP_PACKET(m, len) do { } while (0)
  61 #endif
  62
  63 static void
  64 vq_ring_free_chain(struct virtqueue *vq, uint16_t desc_idx)
  65 {
  66         struct vring_desc *dp, *dp_tail;
  67         struct vq_desc_extra *dxp;
  68         uint16_t desc_idx_last = desc_idx;
  69
  70         dp  = &vq->vq_ring.desc[desc_idx];
  71         dxp = &vq->vq_descx[desc_idx];
  72         vq->vq_free_cnt = (uint16_t)(vq->vq_free_cnt + dxp->ndescs);
  73         if ((dp->flags & VRING_DESC_F_INDIRECT) == 0) {
  74                 while (dp->flags & VRING_DESC_F_NEXT) {
  75                         desc_idx_last = dp->next;
  76                         dp = &vq->vq_ring.desc[dp->next];
  77                 }
  78         }
  79         dxp->ndescs = 0;
  80
  81         /*
  82          * We must append the existing free chain, if any, to the end of
  83          * newly freed chain. If the virtqueue was completely used, then
  84          * head would be VQ_RING_DESC_CHAIN_END (ASSERTed above).
  85          */
  86         if (vq->vq_desc_tail_idx == VQ_RING_DESC_CHAIN_END) {
  87                 vq->vq_desc_head_idx = desc_idx;
  88         } else {
  89                 dp_tail = &vq->vq_ring.desc[vq->vq_desc_tail_idx];
  90                 dp_tail->next = desc_idx;
  91         }
  92
  93         vq->vq_desc_tail_idx = desc_idx_last;
  94         dp->next = VQ_RING_DESC_CHAIN_END;
  95 }
  96
  97 static uint16_t
  98 virtqueue_dequeue_burst_rx(struct virtqueue *vq, struct rte_mbuf **rx_pkts,
  99                            uint32_t *len, uint16_t num)
 100 {
 101         struct vring_used_elem *uep;
 102         struct rte_mbuf *cookie;
 103         uint16_t used_idx, desc_idx;
 104         uint16_t i;
 105
 106         /*  Caller does the check */
 107         for (i = 0; i < num ; i++) {
 108                 used_idx = (uint16_t)(vq->vq_used_cons_idx & (vq->vq_nentries - 1));
 109                 uep = &vq->vq_ring.used->ring[used_idx];
 110                 desc_idx = (uint16_t) uep->id;
 111                 len[i] = uep->len;
 112                 cookie = (struct rte_mbuf *)vq->vq_descx[desc_idx].cookie;
 113
 114                 if (unlikely(cookie == NULL)) {
 115                         PMD_DRV_LOG(ERR, "vring descriptor with no mbuf cookie at %u\n",
 116                                 vq->vq_used_cons_idx);
 117                         break;
 118                 }
 119
 120                 rte_prefetch0(cookie);
 121                 rte_packet_prefetch(cookie->pkt.data);
 122                 rx_pkts[i]  = cookie;
 123                 vq->vq_used_cons_idx++;
 124                 vq_ring_free_chain(vq, desc_idx);
 125                 vq->vq_descx[desc_idx].cookie = NULL;
 126         }
 127
 128         return i;
 129 }
 130
 131 static void
 132 virtqueue_dequeue_pkt_tx(struct virtqueue *vq)
 133 {
 134         struct vring_used_elem *uep;
 135         uint16_t used_idx, desc_idx;
 136
 137         used_idx = (uint16_t)(vq->vq_used_cons_idx & (vq->vq_nentries - 1));
 138         uep = &vq->vq_ring.used->ring[used_idx];
 139         desc_idx = (uint16_t) uep->id;
 140         vq->vq_used_cons_idx++;
 141         vq_ring_free_chain(vq, desc_idx);
 142 }
 143
 144
 145 static inline int
 146 virtqueue_enqueue_recv_refill(struct virtqueue *vq, struct rte_mbuf *cookie)
 147 {
 148         struct vq_desc_extra *dxp;
 149         struct vring_desc *start_dp;
 150         uint16_t needed = 1;
 151         uint16_t head_idx, idx;
 152
 153         if (unlikely(vq->vq_free_cnt == 0))
 154                 return -ENOSPC;
 155         if (unlikely(vq->vq_free_cnt < needed))
 156                 return -EMSGSIZE;
 157
 158         head_idx = vq->vq_desc_head_idx;
 159         if (unlikely(head_idx >= vq->vq_nentries))
 160                 return -EFAULT;
 161
 162         idx = head_idx;
 163         dxp = &vq->vq_descx[idx];
 164         dxp->cookie = (void *)cookie;
 165         dxp->ndescs = needed;
 166
 167         start_dp = vq->vq_ring.desc;
 168         start_dp[idx].addr  =
 169                 (uint64_t) (cookie->buf_physaddr + RTE_PKTMBUF_HEADROOM - sizeof(struct virtio_net_hdr));
 170         start_dp[idx].len   = cookie->buf_len - RTE_PKTMBUF_HEADROOM + sizeof(struct virtio_net_hdr);
 171         start_dp[idx].flags =  VRING_DESC_F_WRITE;
 172         idx = start_dp[idx].next;
 173         vq->vq_desc_head_idx = idx;
 174         if (vq->vq_desc_head_idx == VQ_RING_DESC_CHAIN_END)
 175                 vq->vq_desc_tail_idx = idx;
 176         vq->vq_free_cnt = (uint16_t)(vq->vq_free_cnt - needed);
 177         vq_update_avail_ring(vq, head_idx);
 178
 179         return 0;
 180 }
 181
 182 static int
 183 virtqueue_enqueue_xmit(struct virtqueue *txvq, struct rte_mbuf *cookie)
 184 {
 185         struct vq_desc_extra *dxp;
 186         struct vring_desc *start_dp;
 187         uint16_t needed = 2;
 188         uint16_t head_idx, idx;
 189
 190         if (unlikely(txvq->vq_free_cnt == 0))
 191                 return -ENOSPC;
 192         if (unlikely(txvq->vq_free_cnt < needed))
 193                 return -EMSGSIZE;
 194         head_idx = txvq->vq_desc_head_idx;
 195         if (unlikely(head_idx >= txvq->vq_nentries))
 196                 return -EFAULT;
 197
 198         idx = head_idx;
 199         dxp = &txvq->vq_descx[idx];
 200         if (dxp->cookie != NULL)
 201                 rte_pktmbuf_free_seg(dxp->cookie);
 202         dxp->cookie = (void *)cookie;
 203         dxp->ndescs = needed;
 204
 205         start_dp = txvq->vq_ring.desc;
 206         start_dp[idx].addr  =
 207                 txvq->virtio_net_hdr_mem + idx * sizeof(struct virtio_net_hdr);
 208         start_dp[idx].len   = sizeof(struct virtio_net_hdr);
 209         start_dp[idx].flags = VRING_DESC_F_NEXT;
 210         idx = start_dp[idx].next;
 211         start_dp[idx].addr  = RTE_MBUF_DATA_DMA_ADDR(cookie);
 212         start_dp[idx].len   = cookie->pkt.data_len;
 213         start_dp[idx].flags = 0;
 214         idx = start_dp[idx].next;
 215         txvq->vq_desc_head_idx = idx;
 216         if (txvq->vq_desc_head_idx == VQ_RING_DESC_CHAIN_END)
 217                 txvq->vq_desc_tail_idx = idx;
 218         txvq->vq_free_cnt = (uint16_t)(txvq->vq_free_cnt - needed);
 219         vq_update_avail_ring(txvq, head_idx);
 220
 221         return 0;
 222 }
 223
 224 static inline struct rte_mbuf *
 225 rte_rxmbuf_alloc(struct rte_mempool *mp)
 226 {
 227         struct rte_mbuf *m;
 228
 229         m = __rte_mbuf_raw_alloc(mp);
 230         __rte_mbuf_sanity_check_raw(m, RTE_MBUF_PKT, 0);
 231
 232         return m;
 233 }
 234
 235 static void
 236 virtio_dev_vring_start(struct rte_eth_dev *dev, struct virtqueue *vq, int queue_type)
 237 {
 238         struct rte_mbuf *m;
 239         int i, nbufs, error, size = vq->vq_nentries;
 240         struct vring *vr = &vq->vq_ring;
 241         uint8_t *ring_mem = vq->vq_ring_virt_mem;
 242         char vq_name[VIRTQUEUE_MAX_NAME_SZ];
 243         PMD_INIT_FUNC_TRACE();
 244
 245         /*
 246          * Reinitialise since virtio port might have been stopped and restarted
 247          */
 248         memset(vq->vq_ring_virt_mem, 0, vq->vq_ring_size);
 249         vring_init(vr, size, ring_mem, vq->vq_alignment);
 250         vq->vq_used_cons_idx = 0;
 251         vq->vq_desc_head_idx = 0;
 252         vq->vq_avail_idx = 0;
 253         vq->vq_desc_tail_idx = (uint16_t)(vq->vq_nentries - 1);
 254         vq->vq_free_cnt = vq->vq_nentries;
 255         memset(vq->vq_descx, 0, sizeof(struct vq_desc_extra) * vq->vq_nentries);
 256
 257         /* Chain all the descriptors in the ring with an END */
 258         for (i = 0; i < size - 1; i++)
 259                 vr->desc[i].next = (uint16_t)(i + 1);
 260         vr->desc[i].next = VQ_RING_DESC_CHAIN_END;
 261
 262         /*
 263          * Disable device(host) interrupting guest
 264          */
 265         virtqueue_disable_intr(vq);
 266
 267         snprintf(vq_name, sizeof(vq_name), "port_%d_rx_vq",
 268                                         dev->data->port_id);
 269         PMD_INIT_LOG(DEBUG, "vq name: %s", vq->vq_name);
 270
 271         /* Only rx virtqueue needs mbufs to be allocated at initialization */
 272         if (queue_type == VTNET_RQ) {
 273                 if (vq->mpool == NULL)
 274                         rte_exit(EXIT_FAILURE,
 275                         "Cannot allocate initial mbufs for rx virtqueue");
 276
 277                 /* Allocate blank mbufs for the each rx descriptor */
 278                 nbufs = 0;
 279                 error = ENOSPC;
 280                 while (!virtqueue_full(vq)) {
 281                         m = rte_rxmbuf_alloc(vq->mpool);
 282                         if (m == NULL)
 283                                 break;
 284
 285                         /******************************************
 286                         *         Enqueue allocated buffers        *
 287                         *******************************************/
 288                         error = virtqueue_enqueue_recv_refill(vq, m);
 289
 290                         if (error) {
 291                                 rte_pktmbuf_free_seg(m);
 292                                 break;
 293                         }
 294                         nbufs++;
 295                 }
 296
 297                 vq_update_avail_idx(vq);
 298
 299                 PMD_INIT_LOG(DEBUG, "Allocated %d bufs", nbufs);
 300
 301                 VIRTIO_WRITE_REG_2(vq->hw, VIRTIO_PCI_QUEUE_SEL,
 302                         vq->vq_queue_index);
 303                 VIRTIO_WRITE_REG_4(vq->hw, VIRTIO_PCI_QUEUE_PFN,
 304                         vq->mz->phys_addr >> VIRTIO_PCI_QUEUE_ADDR_SHIFT);
 305         } else if (queue_type == VTNET_TQ) {
 306                 VIRTIO_WRITE_REG_2(vq->hw, VIRTIO_PCI_QUEUE_SEL,
 307                         vq->vq_queue_index);
 308                 VIRTIO_WRITE_REG_4(vq->hw, VIRTIO_PCI_QUEUE_PFN,
 309                         vq->mz->phys_addr >> VIRTIO_PCI_QUEUE_ADDR_SHIFT);
 310         } else {
 311                 VIRTIO_WRITE_REG_2(vq->hw, VIRTIO_PCI_QUEUE_SEL,
 312                         vq->vq_queue_index);
 313                 VIRTIO_WRITE_REG_4(vq->hw, VIRTIO_PCI_QUEUE_PFN,
 314                         vq->mz->phys_addr >> VIRTIO_PCI_QUEUE_ADDR_SHIFT);
 315         }
 316 }
 317
 318 void
 319 virtio_dev_cq_start(struct rte_eth_dev *dev)
 320 {
 321         struct virtio_hw *hw
 322                 = VIRTIO_DEV_PRIVATE_TO_HW(dev->data->dev_private);
 323
 324         virtio_dev_vring_start(dev, hw->cvq, VTNET_CQ);
 325         VIRTQUEUE_DUMP((struct virtqueue *)hw->cvq);
 326 }
 327
 328 void
 329 virtio_dev_rxtx_start(struct rte_eth_dev *dev)
 330 {
 331         /*
 332          * Start receive and transmit vrings
 333          * -    Setup vring structure for all queues
 334          * -    Initialize descriptor for the rx vring
 335          * -    Allocate blank mbufs for the each rx descriptor
 336          *
 337          */
 338         int i;
 339
 340         PMD_INIT_FUNC_TRACE();
 341
 342         /* Start rx vring. */
 343         for (i = 0; i < dev->data->nb_rx_queues; i++) {
 344                 virtio_dev_vring_start(dev, dev->data->rx_queues[i], VTNET_RQ);
 345                 VIRTQUEUE_DUMP((struct virtqueue *)dev->data->rx_queues[i]);
 346         }
 347
 348         /* Start tx vring. */
 349         for (i = 0; i < dev->data->nb_tx_queues; i++) {
 350                 virtio_dev_vring_start(dev, dev->data->tx_queues[i], VTNET_TQ);
 351                 VIRTQUEUE_DUMP((struct virtqueue *)dev->data->tx_queues[i]);
 352         }
 353 }
 354
 355 int
 356 virtio_dev_rx_queue_setup(struct rte_eth_dev *dev,
 357                         uint16_t queue_idx,
 358                         uint16_t nb_desc,
 359                         unsigned int socket_id,
 360                         __rte_unused const struct rte_eth_rxconf *rx_conf,
 361                         struct rte_mempool *mp)
 362 {
 363         uint8_t vtpci_queue_idx = 2 * queue_idx + VTNET_SQ_RQ_QUEUE_IDX;
 364         struct virtqueue *vq;
 365         int ret;
 366
 367         PMD_INIT_FUNC_TRACE();
 368         ret = virtio_dev_queue_setup(dev, VTNET_RQ, queue_idx, vtpci_queue_idx,
 369                         nb_desc, socket_id, &vq);
 370         if (ret < 0) {
 371                 PMD_INIT_LOG(ERR, "tvq initialization failed");
 372                 return ret;
 373         }
 374
 375         /* Create mempool for rx mbuf allocation */
 376         vq->mpool = mp;
 377
 378         dev->data->rx_queues[queue_idx] = vq;
 379         return 0;
 380 }
 381
 382 /*
 383  * struct rte_eth_dev *dev: Used to update dev
 384  * uint16_t nb_desc: Defaults to values read from config space
 385  * unsigned int socket_id: Used to allocate memzone
 386  * const struct rte_eth_txconf *tx_conf: Used to setup tx engine
 387  * uint16_t queue_idx: Just used as an index in dev txq list
 388  */
 389 int
 390 virtio_dev_tx_queue_setup(struct rte_eth_dev *dev,
 391                         uint16_t queue_idx,
 392                         uint16_t nb_desc,
 393                         unsigned int socket_id,
 394                         const struct rte_eth_txconf *tx_conf)
 395 {
 396         uint8_t vtpci_queue_idx = 2 * queue_idx + VTNET_SQ_TQ_QUEUE_IDX;
 397         struct virtqueue *vq;
 398         int ret;
 399
 400         PMD_INIT_FUNC_TRACE();
 401
 402         if ((tx_conf->txq_flags & ETH_TXQ_FLAGS_NOOFFLOADS)
 403             != ETH_TXQ_FLAGS_NOOFFLOADS) {
 404                 PMD_INIT_LOG(ERR, "TX checksum offload not supported\n");
 405                 return -EINVAL;
 406         }
 407
 408         ret = virtio_dev_queue_setup(dev, VTNET_TQ, queue_idx, vtpci_queue_idx,
 409                         nb_desc, socket_id, &vq);
 410         if (ret < 0) {
 411                 PMD_INIT_LOG(ERR, "rvq initialization failed");
 412                 return ret;
 413         }
 414
 415         dev->data->tx_queues[queue_idx] = vq;
 416         return 0;
 417 }
 418
 419 static void
 420 virtio_discard_rxbuf(struct virtqueue *vq, struct rte_mbuf *m)
 421 {
 422         int error;
 423         /*
 424          * Requeue the discarded mbuf. This should always be
 425          * successful since it was just dequeued.
 426          */
 427         error = virtqueue_enqueue_recv_refill(vq, m);
 428         if (unlikely(error)) {
 429                 RTE_LOG(ERR, PMD, "cannot requeue discarded mbuf");
 430                 rte_pktmbuf_free_seg(m);
 431         }
 432 }
 433
 434 #define VIRTIO_MBUF_BURST_SZ 64
 435 #define DESC_PER_CACHELINE (CACHE_LINE_SIZE / sizeof(struct vring_desc))
 436 uint16_t
 437 virtio_recv_pkts(void *rx_queue, struct rte_mbuf **rx_pkts, uint16_t nb_pkts)
 438 {
 439         struct virtqueue *rxvq = rx_queue;
 440         struct virtio_hw *hw = rxvq->hw;
 441         struct rte_mbuf *rxm, *new_mbuf;
 442         uint16_t nb_used, num, nb_rx = 0;
 443         uint32_t len[VIRTIO_MBUF_BURST_SZ];
 444         struct rte_mbuf *rcv_pkts[VIRTIO_MBUF_BURST_SZ];
 445         int error;
 446         uint32_t i, nb_enqueued = 0;
 447
 448         nb_used = VIRTQUEUE_NUSED(rxvq);
 449
 450         rmb();
 451
 452         num = (uint16_t)(likely(nb_used <= nb_pkts) ? nb_used : nb_pkts);
 453         num = (uint16_t)(likely(num <= VIRTIO_MBUF_BURST_SZ) ? num : VIRTIO_MBUF_BURST_SZ);
 454         if (likely(num > DESC_PER_CACHELINE))
 455                 num = num - ((rxvq->vq_used_cons_idx + num) % DESC_PER_CACHELINE);
 456
 457         if (num == 0)
 458                 return 0;
 459
 460         num = virtqueue_dequeue_burst_rx(rxvq, rcv_pkts, len, num);
 461         PMD_RX_LOG(DEBUG, "used:%d dequeue:%d", nb_used, num);
 462         for (i = 0; i < num ; i++) {
 463                 rxm = rcv_pkts[i];
 464
 465                 PMD_RX_LOG(DEBUG, "packet len:%d", len[i]);
 466
 467                 if (unlikely(len[i]
 468                              < (uint32_t)hw->vtnet_hdr_size + ETHER_HDR_LEN)) {
 469                         PMD_RX_LOG(ERR, "Packet drop");
 470                         nb_enqueued++;
 471                         virtio_discard_rxbuf(rxvq, rxm);
 472                         rxvq->errors++;
 473                         continue;
 474                 }
 475
 476                 rxm->pkt.in_port = rxvq->port_id;
 477                 rxm->pkt.data = (char *)rxm->buf_addr + RTE_PKTMBUF_HEADROOM;
 478                 rxm->pkt.nb_segs = 1;
 479                 rxm->pkt.next = NULL;
 480                 rxm->pkt.pkt_len  = (uint32_t)(len[i]
 481                                                - sizeof(struct virtio_net_hdr));
 482                 rxm->pkt.data_len = (uint16_t)(len[i]
 483                                                - sizeof(struct virtio_net_hdr));
 484
 485                 VIRTIO_DUMP_PACKET(rxm, rxm->pkt.data_len);
 486
 487                 rx_pkts[nb_rx++] = rxm;
 488                 rxvq->bytes += len[i] - sizeof(struct virtio_net_hdr);
 489         }
 490
 491         rxvq->packets += nb_rx;
 492
 493         /* Allocate new mbuf for the used descriptor */
 494         error = ENOSPC;
 495         while (likely(!virtqueue_full(rxvq))) {
 496                 new_mbuf = rte_rxmbuf_alloc(rxvq->mpool);
 497                 if (unlikely(new_mbuf == NULL)) {
 498                         struct rte_eth_dev *dev
 499                                 = &rte_eth_devices[rxvq->port_id];
 500                         dev->data->rx_mbuf_alloc_failed++;
 501                         break;
 502                 }
 503                 error = virtqueue_enqueue_recv_refill(rxvq, new_mbuf);
 504                 if (unlikely(error)) {
 505                         rte_pktmbuf_free_seg(new_mbuf);
 506                         break;
 507                 }
 508                 nb_enqueued++;
 509         }
 510         if (likely(nb_enqueued)) {
 511                 if (unlikely(virtqueue_kick_prepare(rxvq))) {
 512                         virtqueue_notify(rxvq);
 513                         PMD_RX_LOG(DEBUG, "Notified");
 514                 }
 515         }
 516
 517         vq_update_avail_idx(rxvq);
 518
 519         return nb_rx;
 520 }
 521
 522 uint16_t
 523 virtio_xmit_pkts(void *tx_queue, struct rte_mbuf **tx_pkts, uint16_t nb_pkts)
 524 {
 525         struct virtqueue *txvq = tx_queue;
 526         struct rte_mbuf *txm;
 527         uint16_t nb_used, nb_tx, num;
 528         int error;
 529
 530         nb_tx = 0;
 531
 532         if (unlikely(nb_pkts < 1))
 533                 return nb_pkts;
 534
 535         PMD_TX_LOG(DEBUG, "%d packets to xmit", nb_pkts);
 536         nb_used = VIRTQUEUE_NUSED(txvq);
 537
 538         rmb();
 539
 540         num = (uint16_t)(likely(nb_used < VIRTIO_MBUF_BURST_SZ) ? nb_used : VIRTIO_MBUF_BURST_SZ);
 541
 542         while (nb_tx < nb_pkts) {
 543                 if (virtqueue_full(txvq) && num) {
 544                         virtqueue_dequeue_pkt_tx(txvq);
 545                         num--;
 546                 }
 547
 548                 if (!virtqueue_full(txvq)) {
 549                         txm = tx_pkts[nb_tx];
 550                         /* Enqueue Packet buffers */
 551                         error = virtqueue_enqueue_xmit(txvq, txm);
 552                         if (unlikely(error)) {
 553                                 if (error == ENOSPC)
 554                                         PMD_TX_LOG(ERR, "virtqueue_enqueue Free count = 0");
 555                                 else if (error == EMSGSIZE)
 556                                         PMD_TX_LOG(ERR, "virtqueue_enqueue Free count < 1");
 557                                 else
 558                                         PMD_TX_LOG(ERR, "virtqueue_enqueue error: %d", error);
 559                                 break;
 560                         }
 561                         nb_tx++;
 562                         txvq->bytes += txm->pkt.data_len;
 563                 } else {
 564                         PMD_TX_LOG(ERR, "No free tx descriptors to transmit");
 565                         break;
 566                 }
 567         }
 568         vq_update_avail_idx(txvq);
 569
 570         txvq->packets += nb_tx;
 571
 572         if (unlikely(virtqueue_kick_prepare(txvq))) {
 573                 virtqueue_notify(txvq);
 574                 PMD_TX_LOG(DEBUG, "Notified backend after xmit");
 575         }
 576
 577         return nb_tx;
 578 }