lib/librte_vhost/vhost_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 <stdbool.h>
  36 #include <linux/virtio_net.h>
  37
  38 #include <rte_mbuf.h>
  39 #include <rte_memcpy.h>
  40 #include <rte_virtio_net.h>
  41
  42 #include "vhost-net.h"
  43
  44 #define MAX_PKT_BURST 32
  45
  46 static bool
  47 is_valid_virt_queue_idx(uint32_t idx, int is_tx, uint32_t qp_nb)
  48 {
  49         return (is_tx ^ (idx & 1)) == 0 && idx < qp_nb * VIRTIO_QNUM;
  50 }
  51
  52 /**
  53  * This function adds buffers to the virtio devices RX virtqueue. Buffers can
  54  * be received from the physical port or from another virtio device. A packet
  55  * count is returned to indicate the number of packets that are succesfully
  56  * added to the RX queue. This function works when the mbuf is scattered, but
  57  * it doesn't support the mergeable feature.
  58  */
  59 static inline uint32_t __attribute__((always_inline))
  60 virtio_dev_rx(struct virtio_net *dev, uint16_t queue_id,
  61         struct rte_mbuf **pkts, uint32_t count)
  62 {
  63         struct vhost_virtqueue *vq;
  64         struct vring_desc *desc;
  65         struct rte_mbuf *buff;
  66         /* The virtio_hdr is initialised to 0. */
  67         struct virtio_net_hdr_mrg_rxbuf virtio_hdr = {{0, 0, 0, 0, 0, 0}, 0};
  68         uint64_t buff_addr = 0;
  69         uint64_t buff_hdr_addr = 0;
  70         uint32_t head[MAX_PKT_BURST];
  71         uint32_t head_idx, packet_success = 0;
  72         uint16_t avail_idx, res_cur_idx;
  73         uint16_t res_base_idx, res_end_idx;
  74         uint16_t free_entries;
  75         uint8_t success = 0;
  76
  77         LOG_DEBUG(VHOST_DATA, "(%"PRIu64") virtio_dev_rx()\n", dev->device_fh);
  78         if (unlikely(!is_valid_virt_queue_idx(queue_id, 0, dev->virt_qp_nb))) {
  79                 RTE_LOG(ERR, VHOST_DATA,
  80                         "%s (%"PRIu64"): virtqueue idx:%d invalid.\n",
  81                         __func__, dev->device_fh, queue_id);
  82                 return 0;
  83         }
  84
  85         vq = dev->virtqueue[queue_id];
  86         if (unlikely(vq->enabled == 0))
  87                 return 0;
  88
  89         count = (count > MAX_PKT_BURST) ? MAX_PKT_BURST : count;
  90
  91         /*
  92          * As many data cores may want access to available buffers,
  93          * they need to be reserved.
  94          */
  95         do {
  96                 res_base_idx = vq->last_used_idx_res;
  97                 avail_idx = *((volatile uint16_t *)&vq->avail->idx);
  98
  99                 free_entries = (avail_idx - res_base_idx);
 100                 /*check that we have enough buffers*/
 101                 if (unlikely(count > free_entries))
 102                         count = free_entries;
 103
 104                 if (count == 0)
 105                         return 0;
 106
 107                 res_end_idx = res_base_idx + count;
 108                 /* vq->last_used_idx_res is atomically updated. */
 109                 /* TODO: Allow to disable cmpset if no concurrency in application. */
 110                 success = rte_atomic16_cmpset(&vq->last_used_idx_res,
 111                                 res_base_idx, res_end_idx);
 112         } while (unlikely(success == 0));
 113         res_cur_idx = res_base_idx;
 114         LOG_DEBUG(VHOST_DATA, "(%"PRIu64") Current Index %d| End Index %d\n",
 115                         dev->device_fh, res_cur_idx, res_end_idx);
 116
 117         /* Prefetch available ring to retrieve indexes. */
 118         rte_prefetch0(&vq->avail->ring[res_cur_idx & (vq->size - 1)]);
 119
 120         /* Retrieve all of the head indexes first to avoid caching issues. */
 121         for (head_idx = 0; head_idx < count; head_idx++)
 122                 head[head_idx] = vq->avail->ring[(res_cur_idx + head_idx) &
 123                                         (vq->size - 1)];
 124
 125         /*Prefetch descriptor index. */
 126         rte_prefetch0(&vq->desc[head[packet_success]]);
 127
 128         while (res_cur_idx != res_end_idx) {
 129                 uint32_t offset = 0, vb_offset = 0;
 130                 uint32_t pkt_len, len_to_cpy, data_len, total_copied = 0;
 131                 uint8_t hdr = 0, uncompleted_pkt = 0;
 132
 133                 /* Get descriptor from available ring */
 134                 desc = &vq->desc[head[packet_success]];
 135
 136                 buff = pkts[packet_success];
 137
 138                 /* Convert from gpa to vva (guest physical addr -> vhost virtual addr) */
 139                 buff_addr = gpa_to_vva(dev, desc->addr);
 140                 /* Prefetch buffer address. */
 141                 rte_prefetch0((void *)(uintptr_t)buff_addr);
 142
 143                 /* Copy virtio_hdr to packet and increment buffer address */
 144                 buff_hdr_addr = buff_addr;
 145
 146                 /*
 147                  * If the descriptors are chained the header and data are
 148                  * placed in separate buffers.
 149                  */
 150                 if ((desc->flags & VRING_DESC_F_NEXT) &&
 151                         (desc->len == vq->vhost_hlen)) {
 152                         desc = &vq->desc[desc->next];
 153                         /* Buffer address translation. */
 154                         buff_addr = gpa_to_vva(dev, desc->addr);
 155                 } else {
 156                         vb_offset += vq->vhost_hlen;
 157                         hdr = 1;
 158                 }
 159
 160                 pkt_len = rte_pktmbuf_pkt_len(buff);
 161                 data_len = rte_pktmbuf_data_len(buff);
 162                 len_to_cpy = RTE_MIN(data_len,
 163                         hdr ? desc->len - vq->vhost_hlen : desc->len);
 164                 while (total_copied < pkt_len) {
 165                         /* Copy mbuf data to buffer */
 166                         rte_memcpy((void *)(uintptr_t)(buff_addr + vb_offset),
 167                                 rte_pktmbuf_mtod_offset(buff, const void *, offset),
 168                                 len_to_cpy);
 169                         PRINT_PACKET(dev, (uintptr_t)(buff_addr + vb_offset),
 170                                 len_to_cpy, 0);
 171
 172                         offset += len_to_cpy;
 173                         vb_offset += len_to_cpy;
 174                         total_copied += len_to_cpy;
 175
 176                         /* The whole packet completes */
 177                         if (total_copied == pkt_len)
 178                                 break;
 179
 180                         /* The current segment completes */
 181                         if (offset == data_len) {
 182                                 buff = buff->next;
 183                                 offset = 0;
 184                                 data_len = rte_pktmbuf_data_len(buff);
 185                         }
 186
 187                         /* The current vring descriptor done */
 188                         if (vb_offset == desc->len) {
 189                                 if (desc->flags & VRING_DESC_F_NEXT) {
 190                                         desc = &vq->desc[desc->next];
 191                                         buff_addr = gpa_to_vva(dev, desc->addr);
 192                                         vb_offset = 0;
 193                                 } else {
 194                                         /* Room in vring buffer is not enough */
 195                                         uncompleted_pkt = 1;
 196                                         break;
 197                                 }
 198                         }
 199                         len_to_cpy = RTE_MIN(data_len - offset, desc->len - vb_offset);
 200                 }
 201
 202                 /* Update used ring with desc information */
 203                 vq->used->ring[res_cur_idx & (vq->size - 1)].id =
 204                                                         head[packet_success];
 205
 206                 /* Drop the packet if it is uncompleted */
 207                 if (unlikely(uncompleted_pkt == 1))
 208                         vq->used->ring[res_cur_idx & (vq->size - 1)].len =
 209                                                         vq->vhost_hlen;
 210                 else
 211                         vq->used->ring[res_cur_idx & (vq->size - 1)].len =
 212                                                         pkt_len + vq->vhost_hlen;
 213
 214                 res_cur_idx++;
 215                 packet_success++;
 216
 217                 if (unlikely(uncompleted_pkt == 1))
 218                         continue;
 219
 220                 rte_memcpy((void *)(uintptr_t)buff_hdr_addr,
 221                         (const void *)&virtio_hdr, vq->vhost_hlen);
 222
 223                 PRINT_PACKET(dev, (uintptr_t)buff_hdr_addr, vq->vhost_hlen, 1);
 224
 225                 if (res_cur_idx < res_end_idx) {
 226                         /* Prefetch descriptor index. */
 227                         rte_prefetch0(&vq->desc[head[packet_success]]);
 228                 }
 229         }
 230
 231         rte_compiler_barrier();
 232
 233         /* Wait until it's our turn to add our buffer to the used ring. */
 234         while (unlikely(vq->last_used_idx != res_base_idx))
 235                 rte_pause();
 236
 237         *(volatile uint16_t *)&vq->used->idx += count;
 238         vq->last_used_idx = res_end_idx;
 239
 240         /* flush used->idx update before we read avail->flags. */
 241         rte_mb();
 242
 243         /* Kick the guest if necessary. */
 244         if (!(vq->avail->flags & VRING_AVAIL_F_NO_INTERRUPT))
 245                 eventfd_write(vq->callfd, (eventfd_t)1);
 246         return count;
 247 }
 248
 249 static inline uint32_t __attribute__((always_inline))
 250 copy_from_mbuf_to_vring(struct virtio_net *dev, uint32_t queue_id,
 251                         uint16_t res_base_idx, uint16_t res_end_idx,
 252                         struct rte_mbuf *pkt)
 253 {
 254         uint32_t vec_idx = 0;
 255         uint32_t entry_success = 0;
 256         struct vhost_virtqueue *vq;
 257         /* The virtio_hdr is initialised to 0. */
 258         struct virtio_net_hdr_mrg_rxbuf virtio_hdr = {
 259                 {0, 0, 0, 0, 0, 0}, 0};
 260         uint16_t cur_idx = res_base_idx;
 261         uint64_t vb_addr = 0;
 262         uint64_t vb_hdr_addr = 0;
 263         uint32_t seg_offset = 0;
 264         uint32_t vb_offset = 0;
 265         uint32_t seg_avail;
 266         uint32_t vb_avail;
 267         uint32_t cpy_len, entry_len;
 268
 269         if (pkt == NULL)
 270                 return 0;
 271
 272         LOG_DEBUG(VHOST_DATA, "(%"PRIu64") Current Index %d| "
 273                 "End Index %d\n",
 274                 dev->device_fh, cur_idx, res_end_idx);
 275
 276         /*
 277          * Convert from gpa to vva
 278          * (guest physical addr -> vhost virtual addr)
 279          */
 280         vq = dev->virtqueue[queue_id];
 281
 282         vb_addr = gpa_to_vva(dev, vq->buf_vec[vec_idx].buf_addr);
 283         vb_hdr_addr = vb_addr;
 284
 285         /* Prefetch buffer address. */
 286         rte_prefetch0((void *)(uintptr_t)vb_addr);
 287
 288         virtio_hdr.num_buffers = res_end_idx - res_base_idx;
 289
 290         LOG_DEBUG(VHOST_DATA, "(%"PRIu64") RX: Num merge buffers %d\n",
 291                 dev->device_fh, virtio_hdr.num_buffers);
 292
 293         rte_memcpy((void *)(uintptr_t)vb_hdr_addr,
 294                 (const void *)&virtio_hdr, vq->vhost_hlen);
 295
 296         PRINT_PACKET(dev, (uintptr_t)vb_hdr_addr, vq->vhost_hlen, 1);
 297
 298         seg_avail = rte_pktmbuf_data_len(pkt);
 299         vb_offset = vq->vhost_hlen;
 300         vb_avail = vq->buf_vec[vec_idx].buf_len - vq->vhost_hlen;
 301
 302         entry_len = vq->vhost_hlen;
 303
 304         if (vb_avail == 0) {
 305                 uint32_t desc_idx =
 306                         vq->buf_vec[vec_idx].desc_idx;
 307
 308                 if ((vq->desc[desc_idx].flags
 309                         & VRING_DESC_F_NEXT) == 0) {
 310                         /* Update used ring with desc information */
 311                         vq->used->ring[cur_idx & (vq->size - 1)].id
 312                                 = vq->buf_vec[vec_idx].desc_idx;
 313                         vq->used->ring[cur_idx & (vq->size - 1)].len
 314                                 = entry_len;
 315
 316                         entry_len = 0;
 317                         cur_idx++;
 318                         entry_success++;
 319                 }
 320
 321                 vec_idx++;
 322                 vb_addr = gpa_to_vva(dev, vq->buf_vec[vec_idx].buf_addr);
 323
 324                 /* Prefetch buffer address. */
 325                 rte_prefetch0((void *)(uintptr_t)vb_addr);
 326                 vb_offset = 0;
 327                 vb_avail = vq->buf_vec[vec_idx].buf_len;
 328         }
 329
 330         cpy_len = RTE_MIN(vb_avail, seg_avail);
 331
 332         while (cpy_len > 0) {
 333                 /* Copy mbuf data to vring buffer */
 334                 rte_memcpy((void *)(uintptr_t)(vb_addr + vb_offset),
 335                         rte_pktmbuf_mtod_offset(pkt, const void *, seg_offset),
 336                         cpy_len);
 337
 338                 PRINT_PACKET(dev,
 339                         (uintptr_t)(vb_addr + vb_offset),
 340                         cpy_len, 0);
 341
 342                 seg_offset += cpy_len;
 343                 vb_offset += cpy_len;
 344                 seg_avail -= cpy_len;
 345                 vb_avail -= cpy_len;
 346                 entry_len += cpy_len;
 347
 348                 if (seg_avail != 0) {
 349                         /*
 350                          * The virtio buffer in this vring
 351                          * entry reach to its end.
 352                          * But the segment doesn't complete.
 353                          */
 354                         if ((vq->desc[vq->buf_vec[vec_idx].desc_idx].flags &
 355                                 VRING_DESC_F_NEXT) == 0) {
 356                                 /* Update used ring with desc information */
 357                                 vq->used->ring[cur_idx & (vq->size - 1)].id
 358                                         = vq->buf_vec[vec_idx].desc_idx;
 359                                 vq->used->ring[cur_idx & (vq->size - 1)].len
 360                                         = entry_len;
 361                                 entry_len = 0;
 362                                 cur_idx++;
 363                                 entry_success++;
 364                         }
 365
 366                         vec_idx++;
 367                         vb_addr = gpa_to_vva(dev,
 368                                 vq->buf_vec[vec_idx].buf_addr);
 369                         vb_offset = 0;
 370                         vb_avail = vq->buf_vec[vec_idx].buf_len;
 371                         cpy_len = RTE_MIN(vb_avail, seg_avail);
 372                 } else {
 373                         /*
 374                          * This current segment complete, need continue to
 375                          * check if the whole packet complete or not.
 376                          */
 377                         pkt = pkt->next;
 378                         if (pkt != NULL) {
 379                                 /*
 380                                  * There are more segments.
 381                                  */
 382                                 if (vb_avail == 0) {
 383                                         /*
 384                                          * This current buffer from vring is
 385                                          * used up, need fetch next buffer
 386                                          * from buf_vec.
 387                                          */
 388                                         uint32_t desc_idx =
 389                                                 vq->buf_vec[vec_idx].desc_idx;
 390
 391                                         if ((vq->desc[desc_idx].flags &
 392                                                 VRING_DESC_F_NEXT) == 0) {
 393                                                 uint16_t wrapped_idx =
 394                                                         cur_idx & (vq->size - 1);
 395                                                 /*
 396                                                  * Update used ring with the
 397                                                  * descriptor information
 398                                                  */
 399                                                 vq->used->ring[wrapped_idx].id
 400                                                         = desc_idx;
 401                                                 vq->used->ring[wrapped_idx].len
 402                                                         = entry_len;
 403                                                 entry_success++;
 404                                                 entry_len = 0;
 405                                                 cur_idx++;
 406                                         }
 407
 408                                         /* Get next buffer from buf_vec. */
 409                                         vec_idx++;
 410                                         vb_addr = gpa_to_vva(dev,
 411                                                 vq->buf_vec[vec_idx].buf_addr);
 412                                         vb_avail =
 413                                                 vq->buf_vec[vec_idx].buf_len;
 414                                         vb_offset = 0;
 415                                 }
 416
 417                                 seg_offset = 0;
 418                                 seg_avail = rte_pktmbuf_data_len(pkt);
 419                                 cpy_len = RTE_MIN(vb_avail, seg_avail);
 420                         } else {
 421                                 /*
 422                                  * This whole packet completes.
 423                                  */
 424                                 /* Update used ring with desc information */
 425                                 vq->used->ring[cur_idx & (vq->size - 1)].id
 426                                         = vq->buf_vec[vec_idx].desc_idx;
 427                                 vq->used->ring[cur_idx & (vq->size - 1)].len
 428                                         = entry_len;
 429                                 entry_success++;
 430                                 break;
 431                         }
 432                 }
 433         }
 434
 435         return entry_success;
 436 }
 437
 438 static inline void __attribute__((always_inline))
 439 update_secure_len(struct vhost_virtqueue *vq, uint32_t id,
 440         uint32_t *secure_len, uint32_t *vec_idx)
 441 {
 442         uint16_t wrapped_idx = id & (vq->size - 1);
 443         uint32_t idx = vq->avail->ring[wrapped_idx];
 444         uint8_t next_desc;
 445         uint32_t len = *secure_len;
 446         uint32_t vec_id = *vec_idx;
 447
 448         do {
 449                 next_desc = 0;
 450                 len += vq->desc[idx].len;
 451                 vq->buf_vec[vec_id].buf_addr = vq->desc[idx].addr;
 452                 vq->buf_vec[vec_id].buf_len = vq->desc[idx].len;
 453                 vq->buf_vec[vec_id].desc_idx = idx;
 454                 vec_id++;
 455
 456                 if (vq->desc[idx].flags & VRING_DESC_F_NEXT) {
 457                         idx = vq->desc[idx].next;
 458                         next_desc = 1;
 459                 }
 460         } while (next_desc);
 461
 462         *secure_len = len;
 463         *vec_idx = vec_id;
 464 }
 465
 466 /*
 467  * This function works for mergeable RX.
 468  */
 469 static inline uint32_t __attribute__((always_inline))
 470 virtio_dev_merge_rx(struct virtio_net *dev, uint16_t queue_id,
 471         struct rte_mbuf **pkts, uint32_t count)
 472 {
 473         struct vhost_virtqueue *vq;
 474         uint32_t pkt_idx = 0, entry_success = 0;
 475         uint16_t avail_idx;
 476         uint16_t res_base_idx, res_cur_idx;
 477         uint8_t success = 0;
 478
 479         LOG_DEBUG(VHOST_DATA, "(%"PRIu64") virtio_dev_merge_rx()\n",
 480                 dev->device_fh);
 481         if (unlikely(!is_valid_virt_queue_idx(queue_id, 0, dev->virt_qp_nb))) {
 482                 RTE_LOG(ERR, VHOST_DATA,
 483                         "%s (%"PRIu64"): virtqueue idx:%d invalid.\n",
 484                         __func__, dev->device_fh, queue_id);
 485                 return 0;
 486         }
 487
 488         vq = dev->virtqueue[queue_id];
 489         if (unlikely(vq->enabled == 0))
 490                 return 0;
 491
 492         count = RTE_MIN((uint32_t)MAX_PKT_BURST, count);
 493
 494         if (count == 0)
 495                 return 0;
 496
 497         for (pkt_idx = 0; pkt_idx < count; pkt_idx++) {
 498                 uint32_t pkt_len = pkts[pkt_idx]->pkt_len + vq->vhost_hlen;
 499
 500                 do {
 501                         /*
 502                          * As many data cores may want access to available
 503                          * buffers, they need to be reserved.
 504                          */
 505                         uint32_t secure_len = 0;
 506                         uint32_t vec_idx = 0;
 507
 508                         res_base_idx = vq->last_used_idx_res;
 509                         res_cur_idx = res_base_idx;
 510
 511                         do {
 512                                 avail_idx = *((volatile uint16_t *)&vq->avail->idx);
 513                                 if (unlikely(res_cur_idx == avail_idx)) {
 514                                         LOG_DEBUG(VHOST_DATA,
 515                                                 "(%"PRIu64") Failed "
 516                                                 "to get enough desc from "
 517                                                 "vring\n",
 518                                                 dev->device_fh);
 519                                         goto merge_rx_exit;
 520                                 } else {
 521                                         update_secure_len(vq, res_cur_idx, &secure_len, &vec_idx);
 522                                         res_cur_idx++;
 523                                 }
 524                         } while (pkt_len > secure_len);
 525
 526                         /* vq->last_used_idx_res is atomically updated. */
 527                         success = rte_atomic16_cmpset(&vq->last_used_idx_res,
 528                                                         res_base_idx,
 529                                                         res_cur_idx);
 530                 } while (success == 0);
 531
 532                 entry_success = copy_from_mbuf_to_vring(dev, queue_id,
 533                         res_base_idx, res_cur_idx, pkts[pkt_idx]);
 534
 535                 rte_compiler_barrier();
 536
 537                 /*
 538                  * Wait until it's our turn to add our buffer
 539                  * to the used ring.
 540                  */
 541                 while (unlikely(vq->last_used_idx != res_base_idx))
 542                         rte_pause();
 543
 544                 *(volatile uint16_t *)&vq->used->idx += entry_success;
 545                 vq->last_used_idx = res_cur_idx;
 546         }
 547
 548 merge_rx_exit:
 549         if (likely(pkt_idx)) {
 550                 /* flush used->idx update before we read avail->flags. */
 551                 rte_mb();
 552
 553                 /* Kick the guest if necessary. */
 554                 if (!(vq->avail->flags & VRING_AVAIL_F_NO_INTERRUPT))
 555                         eventfd_write(vq->callfd, (eventfd_t)1);
 556         }
 557
 558         return pkt_idx;
 559 }
 560
 561 uint16_t
 562 rte_vhost_enqueue_burst(struct virtio_net *dev, uint16_t queue_id,
 563         struct rte_mbuf **pkts, uint16_t count)
 564 {
 565         if (unlikely(dev->features & (1 << VIRTIO_NET_F_MRG_RXBUF)))
 566                 return virtio_dev_merge_rx(dev, queue_id, pkts, count);
 567         else
 568                 return virtio_dev_rx(dev, queue_id, pkts, count);
 569 }
 570
 571 uint16_t
 572 rte_vhost_dequeue_burst(struct virtio_net *dev, uint16_t queue_id,
 573         struct rte_mempool *mbuf_pool, struct rte_mbuf **pkts, uint16_t count)
 574 {
 575         struct rte_mbuf *m, *prev;
 576         struct vhost_virtqueue *vq;
 577         struct vring_desc *desc;
 578         uint64_t vb_addr = 0;
 579         uint32_t head[MAX_PKT_BURST];
 580         uint32_t used_idx;
 581         uint32_t i;
 582         uint16_t free_entries, entry_success = 0;
 583         uint16_t avail_idx;
 584
 585         if (unlikely(!is_valid_virt_queue_idx(queue_id, 1, dev->virt_qp_nb))) {
 586                 RTE_LOG(ERR, VHOST_DATA,
 587                         "%s (%"PRIu64"): virtqueue idx:%d invalid.\n",
 588                         __func__, dev->device_fh, queue_id);
 589                 return 0;
 590         }
 591
 592         vq = dev->virtqueue[queue_id];
 593         if (unlikely(vq->enabled == 0))
 594                 return 0;
 595
 596         avail_idx =  *((volatile uint16_t *)&vq->avail->idx);
 597
 598         /* If there are no available buffers then return. */
 599         if (vq->last_used_idx == avail_idx)
 600                 return 0;
 601
 602         LOG_DEBUG(VHOST_DATA, "%s (%"PRIu64")\n", __func__,
 603                 dev->device_fh);
 604
 605         /* Prefetch available ring to retrieve head indexes. */
 606         rte_prefetch0(&vq->avail->ring[vq->last_used_idx & (vq->size - 1)]);
 607
 608         /*get the number of free entries in the ring*/
 609         free_entries = (avail_idx - vq->last_used_idx);
 610
 611         free_entries = RTE_MIN(free_entries, count);
 612         /* Limit to MAX_PKT_BURST. */
 613         free_entries = RTE_MIN(free_entries, MAX_PKT_BURST);
 614
 615         LOG_DEBUG(VHOST_DATA, "(%"PRIu64") Buffers available %d\n",
 616                         dev->device_fh, free_entries);
 617         /* Retrieve all of the head indexes first to avoid caching issues. */
 618         for (i = 0; i < free_entries; i++)
 619                 head[i] = vq->avail->ring[(vq->last_used_idx + i) & (vq->size - 1)];
 620
 621         /* Prefetch descriptor index. */
 622         rte_prefetch0(&vq->desc[head[entry_success]]);
 623         rte_prefetch0(&vq->used->ring[vq->last_used_idx & (vq->size - 1)]);
 624
 625         while (entry_success < free_entries) {
 626                 uint32_t vb_avail, vb_offset;
 627                 uint32_t seg_avail, seg_offset;
 628                 uint32_t cpy_len;
 629                 uint32_t seg_num = 0;
 630                 struct rte_mbuf *cur;
 631                 uint8_t alloc_err = 0;
 632
 633                 desc = &vq->desc[head[entry_success]];
 634
 635                 /* Discard first buffer as it is the virtio header */
 636                 if (desc->flags & VRING_DESC_F_NEXT) {
 637                         desc = &vq->desc[desc->next];
 638                         vb_offset = 0;
 639                         vb_avail = desc->len;
 640                 } else {
 641                         vb_offset = vq->vhost_hlen;
 642                         vb_avail = desc->len - vb_offset;
 643                 }
 644
 645                 /* Buffer address translation. */
 646                 vb_addr = gpa_to_vva(dev, desc->addr);
 647                 /* Prefetch buffer address. */
 648                 rte_prefetch0((void *)(uintptr_t)vb_addr);
 649
 650                 used_idx = vq->last_used_idx & (vq->size - 1);
 651
 652                 if (entry_success < (free_entries - 1)) {
 653                         /* Prefetch descriptor index. */
 654                         rte_prefetch0(&vq->desc[head[entry_success+1]]);
 655                         rte_prefetch0(&vq->used->ring[(used_idx + 1) & (vq->size - 1)]);
 656                 }
 657
 658                 /* Update used index buffer information. */
 659                 vq->used->ring[used_idx].id = head[entry_success];
 660                 vq->used->ring[used_idx].len = 0;
 661
 662                 /* Allocate an mbuf and populate the structure. */
 663                 m = rte_pktmbuf_alloc(mbuf_pool);
 664                 if (unlikely(m == NULL)) {
 665                         RTE_LOG(ERR, VHOST_DATA,
 666                                 "Failed to allocate memory for mbuf.\n");
 667                         break;
 668                 }
 669                 seg_offset = 0;
 670                 seg_avail = m->buf_len - RTE_PKTMBUF_HEADROOM;
 671                 cpy_len = RTE_MIN(vb_avail, seg_avail);
 672
 673                 PRINT_PACKET(dev, (uintptr_t)vb_addr, desc->len, 0);
 674
 675                 seg_num++;
 676                 cur = m;
 677                 prev = m;
 678                 while (cpy_len != 0) {
 679                         rte_memcpy(rte_pktmbuf_mtod_offset(cur, void *, seg_offset),
 680                                 (void *)((uintptr_t)(vb_addr + vb_offset)),
 681                                 cpy_len);
 682
 683                         seg_offset += cpy_len;
 684                         vb_offset += cpy_len;
 685                         vb_avail -= cpy_len;
 686                         seg_avail -= cpy_len;
 687
 688                         if (vb_avail != 0) {
 689                                 /*
 690                                  * The segment reachs to its end,
 691                                  * while the virtio buffer in TX vring has
 692                                  * more data to be copied.
 693                                  */
 694                                 cur->data_len = seg_offset;
 695                                 m->pkt_len += seg_offset;
 696                                 /* Allocate mbuf and populate the structure. */
 697                                 cur = rte_pktmbuf_alloc(mbuf_pool);
 698                                 if (unlikely(cur == NULL)) {
 699                                         RTE_LOG(ERR, VHOST_DATA, "Failed to "
 700                                                 "allocate memory for mbuf.\n");
 701                                         rte_pktmbuf_free(m);
 702                                         alloc_err = 1;
 703                                         break;
 704                                 }
 705
 706                                 seg_num++;
 707                                 prev->next = cur;
 708                                 prev = cur;
 709                                 seg_offset = 0;
 710                                 seg_avail = cur->buf_len - RTE_PKTMBUF_HEADROOM;
 711                         } else {
 712                                 if (desc->flags & VRING_DESC_F_NEXT) {
 713                                         /*
 714                                          * There are more virtio buffers in
 715                                          * same vring entry need to be copied.
 716                                          */
 717                                         if (seg_avail == 0) {
 718                                                 /*
 719                                                  * The current segment hasn't
 720                                                  * room to accomodate more
 721                                                  * data.
 722                                                  */
 723                                                 cur->data_len = seg_offset;
 724                                                 m->pkt_len += seg_offset;
 725                                                 /*
 726                                                  * Allocate an mbuf and
 727                                                  * populate the structure.
 728                                                  */
 729                                                 cur = rte_pktmbuf_alloc(mbuf_pool);
 730                                                 if (unlikely(cur == NULL)) {
 731                                                         RTE_LOG(ERR,
 732                                                                 VHOST_DATA,
 733                                                                 "Failed to "
 734                                                                 "allocate memory "
 735                                                                 "for mbuf\n");
 736                                                         rte_pktmbuf_free(m);
 737                                                         alloc_err = 1;
 738                                                         break;
 739                                                 }
 740                                                 seg_num++;
 741                                                 prev->next = cur;
 742                                                 prev = cur;
 743                                                 seg_offset = 0;
 744                                                 seg_avail = cur->buf_len - RTE_PKTMBUF_HEADROOM;
 745                                         }
 746
 747                                         desc = &vq->desc[desc->next];
 748
 749                                         /* Buffer address translation. */
 750                                         vb_addr = gpa_to_vva(dev, desc->addr);
 751                                         /* Prefetch buffer address. */
 752                                         rte_prefetch0((void *)(uintptr_t)vb_addr);
 753                                         vb_offset = 0;
 754                                         vb_avail = desc->len;
 755
 756                                         PRINT_PACKET(dev, (uintptr_t)vb_addr,
 757                                                 desc->len, 0);
 758                                 } else {
 759                                         /* The whole packet completes. */
 760                                         cur->data_len = seg_offset;
 761                                         m->pkt_len += seg_offset;
 762                                         vb_avail = 0;
 763                                 }
 764                         }
 765
 766                         cpy_len = RTE_MIN(vb_avail, seg_avail);
 767                 }
 768
 769                 if (unlikely(alloc_err == 1))
 770                         break;
 771
 772                 m->nb_segs = seg_num;
 773
 774                 pkts[entry_success] = m;
 775                 vq->last_used_idx++;
 776                 entry_success++;
 777         }
 778
 779         rte_compiler_barrier();
 780         vq->used->idx += entry_success;
 781         /* Kick guest if required. */
 782         if (!(vq->avail->flags & VRING_AVAIL_F_NO_INTERRUPT))
 783                 eventfd_write(vq->callfd, (eventfd_t)1);
 784         return entry_success;
 785 }