Commit 7e4f3dbf authored by Dariusz Stojaczyk's avatar Dariusz Stojaczyk Committed by Daniel Verkamp
Browse files

rte_virtio: remove virtio_queue.h/.c 



Moved their contents to virtio_dev
files. virtio_queue was already deeply
connected with virtio_dev. There's
no need to keep them separate.

Change-Id: I540c04eb954fe0618817c020f9453aebf3d98ba9
Signed-off-by: default avatarDariusz Stojaczyk <dariuszx.stojaczyk@intel.com>
Reviewed-on: https://review.gerrithub.io/382619


Reviewed-by: default avatarJim Harris <james.r.harris@intel.com>
Reviewed-by: default avatarDaniel Verkamp <daniel.verkamp@intel.com>
Tested-by: default avatarSPDK Automated Test System <sys_sgsw@intel.com>
parent 66d101ff

lib/bdev/virtio/Makefile

+1 −1
Original line number Diff line number Diff line
@@ -38,7 +38,7 @@ CFLAGS += $(ENV_CFLAGS) -I$(SPDK_ROOT_DIR)/lib/bdev/ -Irte_virtio 
CFLAGS += -I$(SPDK_ROOT_DIR)/lib/vhost/linux



C_SRCS = bdev_virtio.c

C_SRCS += rte_virtio/virtio_dev.c rte_virtio/virtio_pci.c rte_virtio/virtio_queue.c

C_SRCS += rte_virtio/virtio_dev.c rte_virtio/virtio_pci.c

C_SRCS += rte_virtio/virtio_user.c

C_SRCS += rte_virtio/virtio_user/vhost_user.c rte_virtio/virtio_user/virtio_user_dev.c

lib/bdev/virtio/rte_virtio/virtio_dev.c

+203 −1
Original line number Diff line number Diff line
@@ -52,12 +52,12 @@ 
#include <rte_memory.h>

#include <rte_eal.h>

#include <rte_dev.h>

#include <rte_prefetch.h>



#include "virtio_user/vhost.h"

#include "virtio_dev.h"

#include "virtio_pci.h"

#include "virtio_logs.h"

#include "virtio_queue.h"



static void

virtio_init_vring(struct virtqueue *vq)
@@ -330,3 +330,205 @@ virtio_dev_start(struct virtio_dev *vdev) 


	return 0;

}



static void

vq_ring_free_chain(struct virtqueue *vq, uint16_t desc_idx)

{

	struct vring_desc *dp, *dp_tail;

	struct vq_desc_extra *dxp;

	uint16_t desc_idx_last = desc_idx;



	dp  = &vq->vq_ring.desc[desc_idx];

	dxp = &vq->vq_descx[desc_idx];

	vq->vq_free_cnt = (uint16_t)(vq->vq_free_cnt + dxp->ndescs);

	if ((dp->flags & VRING_DESC_F_INDIRECT) == 0) {

		while (dp->flags & VRING_DESC_F_NEXT) {

			desc_idx_last = dp->next;

			dp = &vq->vq_ring.desc[dp->next];

		}

	}

	dxp->ndescs = 0;



	/*

	 * We must append the existing free chain, if any, to the end of

	 * newly freed chain. If the virtqueue was completely used, then

	 * head would be VQ_RING_DESC_CHAIN_END (ASSERTed above).

	 */

	if (vq->vq_desc_tail_idx == VQ_RING_DESC_CHAIN_END) {

		vq->vq_desc_head_idx = desc_idx;

	} else {

		dp_tail = &vq->vq_ring.desc[vq->vq_desc_tail_idx];

		dp_tail->next = desc_idx;

	}



	vq->vq_desc_tail_idx = desc_idx_last;

	dp->next = VQ_RING_DESC_CHAIN_END;

}



static uint16_t

virtqueue_dequeue_burst_rx(struct virtqueue *vq, struct virtio_req **rx_pkts,

			   uint32_t *len, uint16_t num)

{

	struct vring_used_elem *uep;

	struct virtio_req *cookie;

	uint16_t used_idx, desc_idx;

	uint16_t i;



	/*  Caller does the check */

	for (i = 0; i < num ; i++) {

		used_idx = (uint16_t)(vq->vq_used_cons_idx & (vq->vq_nentries - 1));

		uep = &vq->vq_ring.used->ring[used_idx];

		desc_idx = (uint16_t) uep->id;

		len[i] = uep->len;

		cookie = (struct virtio_req *)vq->vq_descx[desc_idx].cookie;



		if (spdk_unlikely(cookie == NULL)) {

			PMD_DRV_LOG(ERR, "vring descriptor with no mbuf cookie at %u",

				vq->vq_used_cons_idx);

			break;

		}



		rte_prefetch0(cookie);

		rx_pkts[i]  = cookie;

		vq->vq_used_cons_idx++;

		vq_ring_free_chain(vq, desc_idx);

		vq->vq_descx[desc_idx].cookie = NULL;

	}



	return i;

}



static inline void

virtqueue_iov_to_desc(struct virtqueue *vq, uint16_t desc_idx, struct iovec *iov)

{

	if (!vq->vdev->is_hw) {

		vq->vq_ring.desc[desc_idx].addr  = (uintptr_t)iov->iov_base;

	} else {

		vq->vq_ring.desc[desc_idx].addr = spdk_vtophys(iov->iov_base);

	}



	vq->vq_ring.desc[desc_idx].len = iov->iov_len;

}



static inline void

virtqueue_enqueue_xmit(struct virtqueue *vq, struct virtio_req *req)

{

	struct vq_desc_extra *dxp;

	struct vring_desc *descs;

	uint32_t i;

	uint16_t head_idx, idx;

	uint32_t total_iovs = req->iovcnt + 2;

	struct iovec *iov = req->iov;



	if (total_iovs > vq->vq_free_cnt) {

		PMD_DRV_LOG(ERR, "not enough free descriptors. requested %"PRIu32", got %"PRIu32"\n",

			total_iovs, vq->vq_free_cnt);

		return;

	}



	head_idx = vq->vq_desc_head_idx;

	idx = head_idx;

	dxp = &vq->vq_descx[idx];

	dxp->cookie = (void *)req;

	dxp->ndescs = total_iovs;



	descs = vq->vq_ring.desc;



	virtqueue_iov_to_desc(vq, idx, &req->iov_req);

	descs[idx].flags = VRING_DESC_F_NEXT;

	idx = descs[idx].next;



	if (req->is_write) {

		for (i = 0; i < req->iovcnt; i++) {

			virtqueue_iov_to_desc(vq, idx, &iov[i]);

			descs[idx].flags = VRING_DESC_F_NEXT;

			idx = descs[idx].next;

		}



		virtqueue_iov_to_desc(vq, idx, &req->iov_resp);

		descs[idx].flags = VRING_DESC_F_WRITE;

		idx = descs[idx].next;

	} else {

		virtqueue_iov_to_desc(vq, idx, &req->iov_resp);

		descs[idx].flags = VRING_DESC_F_WRITE | VRING_DESC_F_NEXT;

		idx = descs[idx].next;



		for (i = 0; i < req->iovcnt; i++) {

			virtqueue_iov_to_desc(vq, idx, &iov[i]);

			descs[idx].flags = VRING_DESC_F_WRITE;

			descs[idx].flags |= (i + 1) != req->iovcnt ? VRING_DESC_F_NEXT : 0;

			idx = descs[idx].next;

		}

	}



	vq->vq_desc_head_idx = idx;

	if (vq->vq_desc_head_idx == VQ_RING_DESC_CHAIN_END) {

		assert(vq->vq_free_cnt == 0);

		vq->vq_desc_tail_idx = VQ_RING_DESC_CHAIN_END;

	}

	vq->vq_free_cnt = (uint16_t)(vq->vq_free_cnt - total_iovs);

	vq_update_avail_ring(vq, head_idx);

}



#define VIRTIO_MBUF_BURST_SZ 64

#define DESC_PER_CACHELINE (RTE_CACHE_LINE_SIZE / sizeof(struct vring_desc))

uint16_t

virtio_recv_pkts(struct virtqueue *vq, struct virtio_req **reqs, uint16_t nb_pkts)

{

	struct virtio_dev *vdev = vq->vdev;

	struct virtio_req *rxm;

	uint16_t nb_used, num, nb_rx;

	uint32_t len[VIRTIO_MBUF_BURST_SZ];

	struct virtio_req *rcv_pkts[VIRTIO_MBUF_BURST_SZ];

	uint32_t i;



	nb_rx = 0;

	if (spdk_unlikely(vdev->started == 0))

		return nb_rx;



	nb_used = VIRTQUEUE_NUSED(vq);



	virtio_rmb();



	num = (uint16_t)(spdk_likely(nb_used <= nb_pkts) ? nb_used : nb_pkts);

	num = (uint16_t)(spdk_likely(num <= VIRTIO_MBUF_BURST_SZ) ? num : VIRTIO_MBUF_BURST_SZ);

	if (spdk_likely(num > DESC_PER_CACHELINE))

		num = num - ((vq->vq_used_cons_idx + num) % DESC_PER_CACHELINE);



	num = virtqueue_dequeue_burst_rx(vq, rcv_pkts, len, num);

	PMD_RX_LOG(DEBUG, "used:%d dequeue:%d", nb_used, num);



	for (i = 0; i < num ; i++) {

		rxm = rcv_pkts[i];



		PMD_RX_LOG(DEBUG, "packet len:%d", len[i]);



		rxm->data_transferred = (uint16_t)(len[i]);



		reqs[nb_rx++] = rxm;

	}



	return nb_rx;

}



uint16_t

virtio_xmit_pkts(struct virtqueue *vq, struct virtio_req *req)

{

	struct virtio_dev *vdev = vq->vdev;



	if (spdk_unlikely(vdev->started == 0))

		return 0;



	virtio_rmb();



	virtqueue_enqueue_xmit(vq, req);



	vq_update_avail_idx(vq);



	if (spdk_unlikely(virtqueue_kick_prepare(vq))) {

		vtpci_ops(vdev)->notify_queue(vdev, vq);

		PMD_TX_LOG(DEBUG, "Notified backend after xmit");

	}



	return 1;

}

lib/bdev/virtio/rte_virtio/virtio_dev.h

+131 −0
Original line number Diff line number Diff line
@@ -38,6 +38,36 @@ 
#include <sys/uio.h>

#include <sys/queue.h>



#include <linux/virtio_ring.h>



#include <rte_memory.h>

#include <rte_mempool.h>



#include "virtio_logs.h"



#include "spdk/likely.h"



/*

 * Per virtio_config.h in Linux.

 *     For virtio_pci on SMP, we don't need to order with respect to MMIO

 *     accesses through relaxed memory I/O windows, so smp_mb() et al are

 *     sufficient.

 *

 */

#define virtio_mb()	rte_smp_mb()

#define virtio_rmb()	rte_smp_rmb()

#define virtio_wmb()	rte_smp_wmb()



#define VIRTQUEUE_MAX_NAME_SZ 32



/**

 * The maximum virtqueue size is 2^15. Use that value as the end of

 * descriptor chain terminator since it will never be a valid index

 * in the descriptor table. This is used to verify we are correctly

 * handling vq_free_cnt.

 */

#define VQ_RING_DESC_CHAIN_END 32768



struct virtio_dev {

	struct virtqueue **vqs;

	uint16_t	started;
@@ -59,6 +89,49 @@ struct virtio_dev { 
	TAILQ_ENTRY(virtio_dev) tailq;

};



struct vq_desc_extra {

	void *cookie;

	uint16_t ndescs;

};



struct virtqueue {

	struct virtio_dev *vdev; /**< owner of this virtqueue */

	struct vring vq_ring;  /**< vring keeping desc, used and avail */

	/**

	 * Last consumed descriptor in the used table,

	 * trails vq_ring.used->idx.

	 */

	uint16_t vq_used_cons_idx;

	uint16_t vq_nentries;  /**< vring desc numbers */

	uint16_t vq_free_cnt;  /**< num of desc available */

	uint16_t vq_avail_idx; /**< sync until needed */



	void *vq_ring_virt_mem;  /**< virtual address of vring */

	unsigned int vq_ring_size;



	const struct rte_memzone *mz;    /**< mem zone to populate TX ring. */



	phys_addr_t vq_ring_mem; /**< physical address of vring */



	/**

	 * Head of the free chain in the descriptor table. If

	 * there are no free descriptors, this will be set to

	 * VQ_RING_DESC_CHAIN_END.

	 */

	uint16_t  vq_desc_head_idx;



	/**

	 * Tail of the free chain in desc table. If

	 * there are no free descriptors, this will be set to

	 * VQ_RING_DESC_CHAIN_END.

	 */

	uint16_t  vq_desc_tail_idx;

	uint16_t  vq_queue_index;   /**< PCI queue index */

	uint16_t  *notify_addr;



	struct vq_desc_extra vq_descx[0];

};



struct virtio_req {

	struct iovec	*iov;

	struct iovec	iov_req;
@@ -82,4 +155,62 @@ int virtio_dev_init(struct virtio_dev *hw, uint64_t req_features); 
void virtio_dev_free(struct virtio_dev *dev);

int virtio_dev_start(struct virtio_dev *hw);



/* Chain all the descriptors in the ring with an END */

static inline void

vring_desc_init(struct vring_desc *dp, uint16_t n)

{

	uint16_t i;



	for (i = 0; i < n - 1; i++)

		dp[i].next = (uint16_t)(i + 1);

	dp[i].next = VQ_RING_DESC_CHAIN_END;

}



/**

 * Tell the backend not to interrupt us.

 */

static inline void

virtqueue_disable_intr(struct virtqueue *vq)

{

	vq->vq_ring.avail->flags |= VRING_AVAIL_F_NO_INTERRUPT;

}



static inline int

virtqueue_full(const struct virtqueue *vq)

{

	return vq->vq_free_cnt == 0;

}



#define VIRTQUEUE_NUSED(vq) ((uint16_t)((vq)->vq_ring.used->idx - (vq)->vq_used_cons_idx))



static inline void

vq_update_avail_idx(struct virtqueue *vq)

{

	virtio_wmb();

	vq->vq_ring.avail->idx = vq->vq_avail_idx;

}



static inline void

vq_update_avail_ring(struct virtqueue *vq, uint16_t desc_idx)

{

	uint16_t avail_idx;

	/*

	 * Place the head of the descriptor chain into the next slot and make

	 * it usable to the host. The chain is made available now rather than

	 * deferring to virtqueue_notify() in the hopes that if the host is

	 * currently running on another CPU, we can keep it processing the new

	 * descriptor.

	 */

	avail_idx = (uint16_t)(vq->vq_avail_idx & (vq->vq_nentries - 1));

	if (spdk_unlikely(vq->vq_ring.avail->ring[avail_idx] != desc_idx))

		vq->vq_ring.avail->ring[avail_idx] = desc_idx;

	vq->vq_avail_idx++;

}



static inline int

virtqueue_kick_prepare(struct virtqueue *vq)

{

	return !(vq->vq_ring.used->flags & VRING_USED_F_NO_NOTIFY);

}



#endif /* _VIRTIO_DEV_H_ */

lib/bdev/virtio/rte_virtio/virtio_pci.c

+0 −1
Original line number Diff line number Diff line
@@ -38,7 +38,6 @@ 


#include "virtio_pci.h"

#include "virtio_logs.h"

#include "virtio_queue.h"



struct virtio_driver g_virtio_driver = {

	.init_ctrlrs = TAILQ_HEAD_INITIALIZER(g_virtio_driver.init_ctrlrs),

lib/bdev/virtio/rte_virtio/virtio_queue.c

deleted100644 → 0
+0 −251
Original line number Diff line number Diff line 
/*-

 *   BSD LICENSE

 *

 *   Copyright(c) 2010-2014 Intel Corporation. All rights reserved.

 *   All rights reserved.

 *

 *   Redistribution and use in source and binary forms, with or without

 *   modification, are permitted provided that the following conditions

 *   are met:

 *

 *     * Redistributions of source code must retain the above copyright

 *       notice, this list of conditions and the following disclaimer.

 *     * Redistributions in binary form must reproduce the above copyright

 *       notice, this list of conditions and the following disclaimer in

 *       the documentation and/or other materials provided with the

 *       distribution.

 *     * Neither the name of Intel Corporation nor the names of its

 *       contributors may be used to endorse or promote products derived

 *       from this software without specific prior written permission.

 *

 *   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS

 *   "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT

 *   LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR

 *   A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT

 *   OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,

 *   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT

 *   LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,

 *   DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY

 *   THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT

 *   (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE

 *   OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

 */



#include <stdint.h>

#include <stdio.h>

#include <stdlib.h>

#include <string.h>

#include <errno.h>



#include <rte_cycles.h>

#include <rte_memory.h>

#include <rte_memzone.h>

#include <rte_prefetch.h>



#include "virtio_logs.h"

#include "virtio_dev.h"

#include "virtio_pci.h"

#include "spdk/env.h"

#include "virtio_queue.h"



static void

vq_ring_free_chain(struct virtqueue *vq, uint16_t desc_idx)

{

	struct vring_desc *dp, *dp_tail;

	struct vq_desc_extra *dxp;

	uint16_t desc_idx_last = desc_idx;



	dp  = &vq->vq_ring.desc[desc_idx];

	dxp = &vq->vq_descx[desc_idx];

	vq->vq_free_cnt = (uint16_t)(vq->vq_free_cnt + dxp->ndescs);

	if ((dp->flags & VRING_DESC_F_INDIRECT) == 0) {

		while (dp->flags & VRING_DESC_F_NEXT) {

			desc_idx_last = dp->next;

			dp = &vq->vq_ring.desc[dp->next];

		}

	}

	dxp->ndescs = 0;



	/*

	 * We must append the existing free chain, if any, to the end of

	 * newly freed chain. If the virtqueue was completely used, then

	 * head would be VQ_RING_DESC_CHAIN_END (ASSERTed above).

	 */

	if (vq->vq_desc_tail_idx == VQ_RING_DESC_CHAIN_END) {

		vq->vq_desc_head_idx = desc_idx;

	} else {

		dp_tail = &vq->vq_ring.desc[vq->vq_desc_tail_idx];

		dp_tail->next = desc_idx;

	}



	vq->vq_desc_tail_idx = desc_idx_last;

	dp->next = VQ_RING_DESC_CHAIN_END;

}



static uint16_t

virtqueue_dequeue_burst_rx(struct virtqueue *vq, struct virtio_req **rx_pkts,

			   uint32_t *len, uint16_t num)

{

	struct vring_used_elem *uep;

	struct virtio_req *cookie;

	uint16_t used_idx, desc_idx;

	uint16_t i;



	/*  Caller does the check */

	for (i = 0; i < num ; i++) {

		used_idx = (uint16_t)(vq->vq_used_cons_idx & (vq->vq_nentries - 1));

		uep = &vq->vq_ring.used->ring[used_idx];

		desc_idx = (uint16_t) uep->id;

		len[i] = uep->len;

		cookie = (struct virtio_req *)vq->vq_descx[desc_idx].cookie;



		if (spdk_unlikely(cookie == NULL)) {

			PMD_DRV_LOG(ERR, "vring descriptor with no mbuf cookie at %u",

				vq->vq_used_cons_idx);

			break;

		}



		rte_prefetch0(cookie);

		rx_pkts[i]  = cookie;

		vq->vq_used_cons_idx++;

		vq_ring_free_chain(vq, desc_idx);

		vq->vq_descx[desc_idx].cookie = NULL;

	}



	return i;

}



static inline void

virtqueue_iov_to_desc(struct virtqueue *vq, uint16_t desc_idx, struct iovec *iov)

{

	if (!vq->vdev->is_hw) {

		vq->vq_ring.desc[desc_idx].addr  = (uintptr_t)iov->iov_base;

	} else {

		vq->vq_ring.desc[desc_idx].addr = spdk_vtophys(iov->iov_base);

	}



	vq->vq_ring.desc[desc_idx].len = iov->iov_len;

}



static inline void

virtqueue_enqueue_xmit(struct virtqueue *vq, struct virtio_req *req)

{

	struct vq_desc_extra *dxp;

	struct vring_desc *descs;

	uint32_t i;

	uint16_t head_idx, idx;

	uint32_t total_iovs = req->iovcnt + 2;

	struct iovec *iov = req->iov;



	if (total_iovs > vq->vq_free_cnt) {

		PMD_DRV_LOG(ERR, "not enough free descriptors. requested %"PRIu32", got %"PRIu32"\n",

			total_iovs, vq->vq_free_cnt);

		return;

	}



	head_idx = vq->vq_desc_head_idx;

	idx = head_idx;

	dxp = &vq->vq_descx[idx];

	dxp->cookie = (void *)req;

	dxp->ndescs = total_iovs;



	descs = vq->vq_ring.desc;



	virtqueue_iov_to_desc(vq, idx, &req->iov_req);

	descs[idx].flags = VRING_DESC_F_NEXT;

	idx = descs[idx].next;



	if (req->is_write) {

		for (i = 0; i < req->iovcnt; i++) {

			virtqueue_iov_to_desc(vq, idx, &iov[i]);

			descs[idx].flags = VRING_DESC_F_NEXT;

			idx = descs[idx].next;

		}



		virtqueue_iov_to_desc(vq, idx, &req->iov_resp);

		descs[idx].flags = VRING_DESC_F_WRITE;

		idx = descs[idx].next;

	} else {

		virtqueue_iov_to_desc(vq, idx, &req->iov_resp);

		descs[idx].flags = VRING_DESC_F_WRITE | VRING_DESC_F_NEXT;

		idx = descs[idx].next;



		for (i = 0; i < req->iovcnt; i++) {

			virtqueue_iov_to_desc(vq, idx, &iov[i]);

			descs[idx].flags = VRING_DESC_F_WRITE;

			descs[idx].flags |= (i + 1) != req->iovcnt ? VRING_DESC_F_NEXT : 0;

			idx = descs[idx].next;

		}

	}



	vq->vq_desc_head_idx = idx;

	if (vq->vq_desc_head_idx == VQ_RING_DESC_CHAIN_END) {

		assert(vq->vq_free_cnt == 0);

		vq->vq_desc_tail_idx = VQ_RING_DESC_CHAIN_END;

	}

	vq->vq_free_cnt = (uint16_t)(vq->vq_free_cnt - total_iovs);

	vq_update_avail_ring(vq, head_idx);

}



#define VIRTIO_MBUF_BURST_SZ 64

#define DESC_PER_CACHELINE (RTE_CACHE_LINE_SIZE / sizeof(struct vring_desc))

uint16_t

virtio_recv_pkts(struct virtqueue *vq, struct virtio_req **reqs, uint16_t nb_pkts)

{

	struct virtio_dev *vdev = vq->vdev;

	struct virtio_req *rxm;

	uint16_t nb_used, num, nb_rx;

	uint32_t len[VIRTIO_MBUF_BURST_SZ];

	struct virtio_req *rcv_pkts[VIRTIO_MBUF_BURST_SZ];

	uint32_t i;



	nb_rx = 0;

	if (spdk_unlikely(vdev->started == 0))

		return nb_rx;



	nb_used = VIRTQUEUE_NUSED(vq);



	virtio_rmb();



	num = (uint16_t)(spdk_likely(nb_used <= nb_pkts) ? nb_used : nb_pkts);

	num = (uint16_t)(spdk_likely(num <= VIRTIO_MBUF_BURST_SZ) ? num : VIRTIO_MBUF_BURST_SZ);

	if (spdk_likely(num > DESC_PER_CACHELINE))

		num = num - ((vq->vq_used_cons_idx + num) % DESC_PER_CACHELINE);



	num = virtqueue_dequeue_burst_rx(vq, rcv_pkts, len, num);

	PMD_RX_LOG(DEBUG, "used:%d dequeue:%d", nb_used, num);



	for (i = 0; i < num ; i++) {

		rxm = rcv_pkts[i];



		PMD_RX_LOG(DEBUG, "packet len:%d", len[i]);



		rxm->data_transferred = (uint16_t)(len[i]);



		reqs[nb_rx++] = rxm;

	}



	return nb_rx;

}



uint16_t

virtio_xmit_pkts(struct virtqueue *vq, struct virtio_req *req)

{

	struct virtio_dev *vdev = vq->vdev;



	if (spdk_unlikely(vdev->started == 0))

		return 0;



	virtio_rmb();



	virtqueue_enqueue_xmit(vq, req);



	vq_update_avail_idx(vq);



	if (spdk_unlikely(virtqueue_kick_prepare(vq))) {

		vtpci_ops(vdev)->notify_queue(vdev, vq);

		PMD_TX_LOG(DEBUG, "Notified backend after xmit");

	}



	return 1;

}