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 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
 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         /* Only rx virtqueue needs mbufs to be allocated at initialization */
 268         if (queue_type == VTNET_RQ) {
 269                 if (vq->mpool == NULL)
 270                         rte_exit(EXIT_FAILURE,
 271                         "Cannot allocate initial mbufs for rx virtqueue");
 272
 273                 /* Allocate blank mbufs for the each rx descriptor */
 274                 nbufs = 0;
 275                 error = ENOSPC;
 276                 while (!virtqueue_full(vq)) {
 277                         m = rte_rxmbuf_alloc(vq->mpool);
 278                         if (m == NULL)
 279                                 break;
 280
 281                         /******************************************
 282                         *         Enqueue allocated buffers        *
 283                         *******************************************/
 284                         error = virtqueue_enqueue_recv_refill(vq, m);
 285
 286                         if (error) {
 287                                 rte_pktmbuf_free_seg(m);
 288                                 break;
 289                         }
 290                         nbufs++;
 291                 }
 292
 293                 vq_update_avail_idx(vq);
 294
 295                 PMD_INIT_LOG(DEBUG, "Allocated %d bufs", nbufs);
 296
 297                 VIRTIO_WRITE_REG_2(vq->hw, VIRTIO_PCI_QUEUE_SEL,
 298                         vq->vq_queue_index);
 299                 VIRTIO_WRITE_REG_4(vq->hw, VIRTIO_PCI_QUEUE_PFN,
 300                         vq->mz->phys_addr >> VIRTIO_PCI_QUEUE_ADDR_SHIFT);
 301         } else if (queue_type == VTNET_TQ) {
 302                 VIRTIO_WRITE_REG_2(vq->hw, VIRTIO_PCI_QUEUE_SEL,
 303                         vq->vq_queue_index);
 304                 VIRTIO_WRITE_REG_4(vq->hw, VIRTIO_PCI_QUEUE_PFN,
 305                         vq->mz->phys_addr >> VIRTIO_PCI_QUEUE_ADDR_SHIFT);
 306         } else {
 307                 VIRTIO_WRITE_REG_2(vq->hw, VIRTIO_PCI_QUEUE_SEL,
 308                         vq->vq_queue_index);
 309                 VIRTIO_WRITE_REG_4(vq->hw, VIRTIO_PCI_QUEUE_PFN,
 310                         vq->mz->phys_addr >> VIRTIO_PCI_QUEUE_ADDR_SHIFT);
 311         }
 312 }
 313
 314 void
 315 virtio_dev_cq_start(struct rte_eth_dev *dev)
 316 {
 317         struct virtio_hw *hw
 318                 = VIRTIO_DEV_PRIVATE_TO_HW(dev->data->dev_private);
 319
 320         virtio_dev_vring_start(hw->cvq, VTNET_CQ);
 321         VIRTQUEUE_DUMP((struct virtqueue *)hw->cvq);
 322 }
 323
 324 void
 325 virtio_dev_rxtx_start(struct rte_eth_dev *dev)
 326 {
 327         /*
 328          * Start receive and transmit vrings
 329          * -    Setup vring structure for all queues
 330          * -    Initialize descriptor for the rx vring
 331          * -    Allocate blank mbufs for the each rx descriptor
 332          *
 333          */
 334         int i;
 335
 336         PMD_INIT_FUNC_TRACE();
 337
 338         /* Start rx vring. */
 339         for (i = 0; i < dev->data->nb_rx_queues; i++) {
 340                 virtio_dev_vring_start(dev->data->rx_queues[i], VTNET_RQ);
 341                 VIRTQUEUE_DUMP((struct virtqueue *)dev->data->rx_queues[i]);
 342         }
 343
 344         /* Start tx vring. */
 345         for (i = 0; i < dev->data->nb_tx_queues; i++) {
 346                 virtio_dev_vring_start(dev->data->tx_queues[i], VTNET_TQ);
 347                 VIRTQUEUE_DUMP((struct virtqueue *)dev->data->tx_queues[i]);
 348         }
 349 }
 350
 351 int
 352 virtio_dev_rx_queue_setup(struct rte_eth_dev *dev,
 353                         uint16_t queue_idx,
 354                         uint16_t nb_desc,
 355                         unsigned int socket_id,
 356                         __rte_unused const struct rte_eth_rxconf *rx_conf,
 357                         struct rte_mempool *mp)
 358 {
 359         uint8_t vtpci_queue_idx = 2 * queue_idx + VTNET_SQ_RQ_QUEUE_IDX;
 360         struct virtqueue *vq;
 361         int ret;
 362
 363         PMD_INIT_FUNC_TRACE();
 364         ret = virtio_dev_queue_setup(dev, VTNET_RQ, queue_idx, vtpci_queue_idx,
 365                         nb_desc, socket_id, &vq);
 366         if (ret < 0) {
 367                 PMD_INIT_LOG(ERR, "tvq initialization failed");
 368                 return ret;
 369         }
 370
 371         /* Create mempool for rx mbuf allocation */
 372         vq->mpool = mp;
 373
 374         dev->data->rx_queues[queue_idx] = vq;
 375         return 0;
 376 }
 377
 378 /*
 379  * struct rte_eth_dev *dev: Used to update dev
 380  * uint16_t nb_desc: Defaults to values read from config space
 381  * unsigned int socket_id: Used to allocate memzone
 382  * const struct rte_eth_txconf *tx_conf: Used to setup tx engine
 383  * uint16_t queue_idx: Just used as an index in dev txq list
 384  */
 385 int
 386 virtio_dev_tx_queue_setup(struct rte_eth_dev *dev,
 387                         uint16_t queue_idx,
 388                         uint16_t nb_desc,
 389                         unsigned int socket_id,
 390                         const struct rte_eth_txconf *tx_conf)
 391 {
 392         uint8_t vtpci_queue_idx = 2 * queue_idx + VTNET_SQ_TQ_QUEUE_IDX;
 393         struct virtqueue *vq;
 394         int ret;
 395
 396         PMD_INIT_FUNC_TRACE();
 397
 398         if ((tx_conf->txq_flags & ETH_TXQ_FLAGS_NOOFFLOADS)
 399             != ETH_TXQ_FLAGS_NOOFFLOADS) {
 400                 PMD_INIT_LOG(ERR, "TX checksum offload not supported\n");
 401                 return -EINVAL;
 402         }
 403
 404         ret = virtio_dev_queue_setup(dev, VTNET_TQ, queue_idx, vtpci_queue_idx,
 405                         nb_desc, socket_id, &vq);
 406         if (ret < 0) {
 407                 PMD_INIT_LOG(ERR, "rvq initialization failed");
 408                 return ret;
 409         }
 410
 411         dev->data->tx_queues[queue_idx] = vq;
 412         return 0;
 413 }
 414
 415 static void
 416 virtio_discard_rxbuf(struct virtqueue *vq, struct rte_mbuf *m)
 417 {
 418         int error;
 419         /*
 420          * Requeue the discarded mbuf. This should always be
 421          * successful since it was just dequeued.
 422          */
 423         error = virtqueue_enqueue_recv_refill(vq, m);
 424         if (unlikely(error)) {
 425                 RTE_LOG(ERR, PMD, "cannot requeue discarded mbuf");
 426                 rte_pktmbuf_free_seg(m);
 427         }
 428 }
 429
 430 #define VIRTIO_MBUF_BURST_SZ 64
 431 #define DESC_PER_CACHELINE (CACHE_LINE_SIZE / sizeof(struct vring_desc))
 432 uint16_t
 433 virtio_recv_pkts(void *rx_queue, struct rte_mbuf **rx_pkts, uint16_t nb_pkts)
 434 {
 435         struct virtqueue *rxvq = rx_queue;
 436         struct virtio_hw *hw = rxvq->hw;
 437         struct rte_mbuf *rxm, *new_mbuf;
 438         uint16_t nb_used, num, nb_rx = 0;
 439         uint32_t len[VIRTIO_MBUF_BURST_SZ];
 440         struct rte_mbuf *rcv_pkts[VIRTIO_MBUF_BURST_SZ];
 441         int error;
 442         uint32_t i, nb_enqueued = 0;
 443
 444         nb_used = VIRTQUEUE_NUSED(rxvq);
 445
 446         rmb();
 447
 448         num = (uint16_t)(likely(nb_used <= nb_pkts) ? nb_used : nb_pkts);
 449         num = (uint16_t)(likely(num <= VIRTIO_MBUF_BURST_SZ) ? num : VIRTIO_MBUF_BURST_SZ);
 450         if (likely(num > DESC_PER_CACHELINE))
 451                 num = num - ((rxvq->vq_used_cons_idx + num) % DESC_PER_CACHELINE);
 452
 453         if (num == 0)
 454                 return 0;
 455
 456         num = virtqueue_dequeue_burst_rx(rxvq, rcv_pkts, len, num);
 457         PMD_RX_LOG(DEBUG, "used:%d dequeue:%d", nb_used, num);
 458         for (i = 0; i < num ; i++) {
 459                 rxm = rcv_pkts[i];
 460
 461                 PMD_RX_LOG(DEBUG, "packet len:%d", len[i]);
 462
 463                 if (unlikely(len[i]
 464                              < (uint32_t)hw->vtnet_hdr_size + ETHER_HDR_LEN)) {
 465                         PMD_RX_LOG(ERR, "Packet drop");
 466                         nb_enqueued++;
 467                         virtio_discard_rxbuf(rxvq, rxm);
 468                         rxvq->errors++;
 469                         continue;
 470                 }
 471
 472                 rxm->pkt.in_port = rxvq->port_id;
 473                 rxm->pkt.data = (char *)rxm->buf_addr + RTE_PKTMBUF_HEADROOM;
 474                 rxm->pkt.nb_segs = 1;
 475                 rxm->pkt.next = NULL;
 476                 rxm->pkt.pkt_len  = (uint32_t)(len[i]
 477                                                - sizeof(struct virtio_net_hdr));
 478                 rxm->pkt.data_len = (uint16_t)(len[i]
 479                                                - sizeof(struct virtio_net_hdr));
 480
 481                 VIRTIO_DUMP_PACKET(rxm, rxm->pkt.data_len);
 482
 483                 rx_pkts[nb_rx++] = rxm;
 484                 rxvq->bytes += len[i] - sizeof(struct virtio_net_hdr);
 485         }
 486
 487         rxvq->packets += nb_rx;
 488
 489         /* Allocate new mbuf for the used descriptor */
 490         error = ENOSPC;
 491         while (likely(!virtqueue_full(rxvq))) {
 492                 new_mbuf = rte_rxmbuf_alloc(rxvq->mpool);
 493                 if (unlikely(new_mbuf == NULL)) {
 494                         struct rte_eth_dev *dev
 495                                 = &rte_eth_devices[rxvq->port_id];
 496                         dev->data->rx_mbuf_alloc_failed++;
 497                         break;
 498                 }
 499                 error = virtqueue_enqueue_recv_refill(rxvq, new_mbuf);
 500                 if (unlikely(error)) {
 501                         rte_pktmbuf_free_seg(new_mbuf);
 502                         break;
 503                 }
 504                 nb_enqueued++;
 505         }
 506         if (likely(nb_enqueued)) {
 507                 if (unlikely(virtqueue_kick_prepare(rxvq))) {
 508                         virtqueue_notify(rxvq);
 509                         PMD_RX_LOG(DEBUG, "Notified");
 510                 }
 511         }
 512
 513         vq_update_avail_idx(rxvq);
 514
 515         return nb_rx;
 516 }
 517
 518 uint16_t
 519 virtio_xmit_pkts(void *tx_queue, struct rte_mbuf **tx_pkts, uint16_t nb_pkts)
 520 {
 521         struct virtqueue *txvq = tx_queue;
 522         struct rte_mbuf *txm;
 523         uint16_t nb_used, nb_tx, num;
 524         int error;
 525
 526         nb_tx = 0;
 527
 528         if (unlikely(nb_pkts < 1))
 529                 return nb_pkts;
 530
 531         PMD_TX_LOG(DEBUG, "%d packets to xmit", nb_pkts);
 532         nb_used = VIRTQUEUE_NUSED(txvq);
 533
 534         rmb();
 535
 536         num = (uint16_t)(likely(nb_used < VIRTIO_MBUF_BURST_SZ) ? nb_used : VIRTIO_MBUF_BURST_SZ);
 537
 538         while (nb_tx < nb_pkts) {
 539                 if (virtqueue_full(txvq) && num) {
 540                         virtqueue_dequeue_pkt_tx(txvq);
 541                         num--;
 542                 }
 543
 544                 if (!virtqueue_full(txvq)) {
 545                         txm = tx_pkts[nb_tx];
 546                         /* Enqueue Packet buffers */
 547                         error = virtqueue_enqueue_xmit(txvq, txm);
 548                         if (unlikely(error)) {
 549                                 if (error == ENOSPC)
 550                                         PMD_TX_LOG(ERR, "virtqueue_enqueue Free count = 0");
 551                                 else if (error == EMSGSIZE)
 552                                         PMD_TX_LOG(ERR, "virtqueue_enqueue Free count < 1");
 553                                 else
 554                                         PMD_TX_LOG(ERR, "virtqueue_enqueue error: %d", error);
 555                                 break;
 556                         }
 557                         nb_tx++;
 558                         txvq->bytes += txm->pkt.data_len;
 559                 } else {
 560                         PMD_TX_LOG(ERR, "No free tx descriptors to transmit");
 561                         break;
 562                 }
 563         }
 564         vq_update_avail_idx(txvq);
 565
 566         txvq->packets += nb_tx;
 567
 568         if (unlikely(virtqueue_kick_prepare(txvq))) {
 569                 virtqueue_notify(txvq);
 570                 PMD_TX_LOG(DEBUG, "Notified backend after xmit");
 571         }
 572
 573         return nb_tx;
 574 }