From patchwork Thu Jul 8 10:25:28 2021 Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit X-Patchwork-Submitter: "Jiang, Cheng1" X-Patchwork-Id: 95550 X-Patchwork-Delegate: maxime.coquelin@redhat.com Return-Path: X-Original-To: patchwork@inbox.dpdk.org Delivered-To: patchwork@inbox.dpdk.org Received: from mails.dpdk.org (mails.dpdk.org [217.70.189.124]) by inbox.dpdk.org (Postfix) with ESMTP id 8DAC8A0C4A; Thu, 8 Jul 2021 12:41:04 +0200 (CEST) Received: from [217.70.189.124] (localhost [127.0.0.1]) by mails.dpdk.org (Postfix) with ESMTP id 414E040696; Thu, 8 Jul 2021 12:41:04 +0200 (CEST) Received: from mga06.intel.com (mga06.intel.com [134.134.136.31]) by mails.dpdk.org (Postfix) with ESMTP id B604A4014F; Thu, 8 Jul 2021 12:41:02 +0200 (CEST) X-IronPort-AV: E=McAfee;i="6200,9189,10038"; a="270597860" X-IronPort-AV: E=Sophos;i="5.84,222,1620716400"; d="scan'208";a="270597860" Received: from fmsmga003.fm.intel.com ([10.253.24.29]) by orsmga104.jf.intel.com with ESMTP/TLS/ECDHE-RSA-AES256-GCM-SHA384; 08 Jul 2021 03:41:01 -0700 X-ExtLoop1: 1 X-IronPort-AV: E=Sophos;i="5.84,222,1620716400"; d="scan'208";a="487504093" Received: from dpdk_jiangcheng.sh.intel.com ([10.67.119.149]) by FMSMGA003.fm.intel.com with ESMTP; 08 Jul 2021 03:40:59 -0700 From: Cheng Jiang To: maxime.coquelin@redhat.com, Chenbo.Xia@intel.com Cc: dev@dpdk.org, jiayu.hu@intel.com, yvonnex.yang@intel.com, Cheng Jiang , stable@dpdk.org Date: Thu, 8 Jul 2021 10:25:28 +0000 Message-Id: <20210708102528.36581-1-cheng1.jiang@intel.com> X-Mailer: git-send-email 2.29.2 MIME-Version: 1.0 Subject: [dpdk-dev] [PATCH] vhost: fix async packed ring batch datapath X-BeenThere: dev@dpdk.org X-Mailman-Version: 2.1.29 Precedence: list List-Id: DPDK patches and discussions List-Unsubscribe: , List-Archive: List-Post: List-Help: List-Subscribe: , Errors-To: dev-bounces@dpdk.org Sender: "dev" We assume that in the sync path, if there is no buffer wrap in the avail descriptors fetched in a batch, there is no buffer wrap in the used descriptors which need to be written back in this batch, but this assumption is wrong in the async path since there are inflight descriptors which are processed by the DMA device. This patch refactors the batch copy code and adds used ring buffer wrap check as a batch copy condition to fix this issue. Fixes: 873e8dad6f49 ("vhost: support packed ring in async datapath") Cc: stable@dpdk.org Signed-off-by: Cheng Jiang Reviewed-by: Maxime Coquelin --- lib/vhost/virtio_net.c | 163 ++++++++++++++++++++++++++++++++--------- 1 file changed, 128 insertions(+), 35 deletions(-) diff --git a/lib/vhost/virtio_net.c b/lib/vhost/virtio_net.c index 6bd00b746b..f4a2c88d8b 100644 --- a/lib/vhost/virtio_net.c +++ b/lib/vhost/virtio_net.c @@ -221,11 +221,6 @@ vhost_flush_enqueue_batch_packed(struct virtio_net *dev, uint16_t last_used_idx; struct vring_packed_desc *desc_base; - if (vq->shadow_used_idx) { - do_data_copy_enqueue(dev, vq); - vhost_flush_enqueue_shadow_packed(dev, vq); - } - last_used_idx = vq->last_used_idx; desc_base = &vq->desc_packed[last_used_idx]; @@ -1258,18 +1253,16 @@ virtio_dev_rx_split(struct virtio_net *dev, struct vhost_virtqueue *vq, } static __rte_always_inline int -virtio_dev_rx_batch_packed(struct virtio_net *dev, +virtio_dev_rx_sync_batch_check(struct virtio_net *dev, struct vhost_virtqueue *vq, - struct rte_mbuf **pkts) + struct rte_mbuf **pkts, + uint64_t *desc_addrs, + uint64_t *lens) { bool wrap_counter = vq->avail_wrap_counter; struct vring_packed_desc *descs = vq->desc_packed; uint16_t avail_idx = vq->last_avail_idx; - uint64_t desc_addrs[PACKED_BATCH_SIZE]; - struct virtio_net_hdr_mrg_rxbuf *hdrs[PACKED_BATCH_SIZE]; uint32_t buf_offset = sizeof(struct virtio_net_hdr_mrg_rxbuf); - uint64_t lens[PACKED_BATCH_SIZE]; - uint16_t ids[PACKED_BATCH_SIZE]; uint16_t i; if (unlikely(avail_idx & PACKED_BATCH_MASK)) @@ -1307,6 +1300,84 @@ virtio_dev_rx_batch_packed(struct virtio_net *dev, return -1; } + return 0; +} + +static __rte_always_inline int +virtio_dev_rx_async_batch_check(struct virtio_net *dev, + struct vhost_virtqueue *vq, + struct rte_mbuf **pkts, + uint64_t *desc_addrs, + uint64_t *lens) +{ + bool wrap_counter = vq->avail_wrap_counter; + struct vring_packed_desc *descs = vq->desc_packed; + uint16_t avail_idx = vq->last_avail_idx; + uint16_t used_idx = vq->last_used_idx; + uint32_t buf_offset = sizeof(struct virtio_net_hdr_mrg_rxbuf); + uint32_t cpy_threshold = vq->async_threshold; + uint16_t i; + + vhost_for_each_try_unroll(i, 0, PACKED_BATCH_SIZE) { + if (unlikely(pkts[i]->data_len >= cpy_threshold)) + return -1; + } + + if (unlikely(avail_idx & PACKED_BATCH_MASK)) + return -1; + + if (unlikely((avail_idx + PACKED_BATCH_SIZE) > vq->size)) + return -1; + + if (unlikely((used_idx + PACKED_BATCH_SIZE) > vq->size)) + return -1; + + vhost_for_each_try_unroll(i, 0, PACKED_BATCH_SIZE) { + if (unlikely(pkts[i]->next != NULL)) + return -1; + if (unlikely(!desc_is_avail(&descs[avail_idx + i], + wrap_counter))) + return -1; + } + + vhost_for_each_try_unroll(i, 0, PACKED_BATCH_SIZE) + lens[i] = descs[avail_idx + i].len; + + vhost_for_each_try_unroll(i, 0, PACKED_BATCH_SIZE) { + if (unlikely(pkts[i]->pkt_len > (lens[i] - buf_offset))) + return -1; + } + + vhost_for_each_try_unroll(i, 0, PACKED_BATCH_SIZE) + desc_addrs[i] = vhost_iova_to_vva(dev, vq, + descs[avail_idx + i].addr, + &lens[i], + VHOST_ACCESS_RW); + + vhost_for_each_try_unroll(i, 0, PACKED_BATCH_SIZE) { + if (unlikely(!desc_addrs[i])) + return -1; + if (unlikely(lens[i] != descs[avail_idx + i].len)) + return -1; + } + + return 0; +} + +static __rte_always_inline void +virtio_dev_rx_batch_packed_copy(struct virtio_net *dev, + struct vhost_virtqueue *vq, + struct rte_mbuf **pkts, + uint64_t *desc_addrs, + uint64_t *lens) +{ + uint32_t buf_offset = sizeof(struct virtio_net_hdr_mrg_rxbuf); + struct virtio_net_hdr_mrg_rxbuf *hdrs[PACKED_BATCH_SIZE]; + struct vring_packed_desc *descs = vq->desc_packed; + uint16_t avail_idx = vq->last_avail_idx; + uint16_t ids[PACKED_BATCH_SIZE]; + uint16_t i; + vhost_for_each_try_unroll(i, 0, PACKED_BATCH_SIZE) { rte_prefetch0((void *)(uintptr_t)desc_addrs[i]); hdrs[i] = (struct virtio_net_hdr_mrg_rxbuf *) @@ -1334,6 +1405,51 @@ virtio_dev_rx_batch_packed(struct virtio_net *dev, ids[i] = descs[avail_idx + i].id; vhost_flush_enqueue_batch_packed(dev, vq, lens, ids); +} + +static __rte_always_inline int +virtio_dev_rx_sync_batch_packed(struct virtio_net *dev, + struct vhost_virtqueue *vq, + struct rte_mbuf **pkts) +{ + uint64_t desc_addrs[PACKED_BATCH_SIZE]; + uint64_t lens[PACKED_BATCH_SIZE]; + + if (virtio_dev_rx_sync_batch_check(dev, vq, pkts, desc_addrs, lens) == -1) + return -1; + + if (vq->shadow_used_idx) { + do_data_copy_enqueue(dev, vq); + vhost_flush_enqueue_shadow_packed(dev, vq); + } + + virtio_dev_rx_batch_packed_copy(dev, vq, pkts, desc_addrs, lens); + + return 0; +} + +static __rte_always_inline int +virtio_dev_rx_async_batch_packed(struct virtio_net *dev, + struct vhost_virtqueue *vq, + struct rte_mbuf **pkts, + struct rte_mbuf **comp_pkts, uint32_t *pkt_done) +{ + uint16_t i; + uint64_t desc_addrs[PACKED_BATCH_SIZE]; + uint64_t lens[PACKED_BATCH_SIZE]; + + if (virtio_dev_rx_async_batch_check(dev, vq, pkts, desc_addrs, lens) == -1) + return -1; + + virtio_dev_rx_batch_packed_copy(dev, vq, pkts, desc_addrs, lens); + + if (vq->shadow_used_idx) { + do_data_copy_enqueue(dev, vq); + vhost_flush_enqueue_shadow_packed(dev, vq); + } + + vhost_for_each_try_unroll(i, 0, PACKED_BATCH_SIZE) + comp_pkts[(*pkt_done)++] = pkts[i]; return 0; } @@ -1375,7 +1491,7 @@ virtio_dev_rx_packed(struct virtio_net *dev, rte_prefetch0(&vq->desc_packed[vq->last_avail_idx]); if (count - pkt_idx >= PACKED_BATCH_SIZE) { - if (!virtio_dev_rx_batch_packed(dev, vq, + if (!virtio_dev_rx_sync_batch_packed(dev, vq, &pkts[pkt_idx])) { pkt_idx += PACKED_BATCH_SIZE; continue; @@ -1728,29 +1844,6 @@ vhost_update_used_packed(struct vhost_virtqueue *vq, vq->desc_packed[head_idx].flags = head_flags; } -static __rte_always_inline int -virtio_dev_rx_async_batch_packed(struct virtio_net *dev, - struct vhost_virtqueue *vq, - struct rte_mbuf **pkts, - struct rte_mbuf **comp_pkts, uint32_t *pkt_done) -{ - uint16_t i; - uint32_t cpy_threshold = vq->async_threshold; - - vhost_for_each_try_unroll(i, 0, PACKED_BATCH_SIZE) { - if (unlikely(pkts[i]->pkt_len >= cpy_threshold)) - return -1; - } - if (!virtio_dev_rx_batch_packed(dev, vq, pkts)) { - vhost_for_each_try_unroll(i, 0, PACKED_BATCH_SIZE) - comp_pkts[(*pkt_done)++] = pkts[i]; - - return 0; - } - - return -1; -} - static __rte_always_inline int vhost_enqueue_async_single_packed(struct virtio_net *dev, struct vhost_virtqueue *vq,