1. Rework the Tx completion flush to walk descriptors one at a time and
free each completed segment with rte_pktmbuf_free_seg(), add prefetch
hints on the next descriptor cache line, switch the ring-held mbufs
in zxdh_dev_free_mbufs() to rte_pktmbuf_free_seg().
2. The flush relies on `desc[k].id == k` already being set by the
enqueue paths (`zxdh_xmit_enqueue_push` writes `dp->id = id` and
`zxdh_xmit_enqueue_append` writes `start_dp[idx].id = idx`) and
preserved by the device.
To match the commercial branch the flush also rewrites
`desc[used_idx].id = used_idx` and keeps the inner
`do { ... } while (curr_id != id)`; under the invariant the inner
loop runs once, and the curr_id-anchored walk backstops the case
where the device overwrites the id.
3. Guard the walk with a `budget` counter initialised to `vq_nentries`,
break on `id >= size`, and mask the prefetch index with `(size - 1)`
so the walk and `free_cnt` stay bounded even if the device writes
an out-of-range or stale id.
4. Document the Tx xmit changes in the 26.11 release notes.
Signed-off-by: Junlong Wang <wang.junlong1@zte.com.cn>
---
doc/guides/rel_notes/release_26_11.rst | 2 +
drivers/net/zxdh/zxdh_ethdev.c | 4 +-
drivers/net/zxdh/zxdh_queue.h | 1 -
drivers/net/zxdh/zxdh_rxtx.c | 136 ++++++++++---------------
4 files changed, 57 insertions(+), 86 deletions(-)
diff --git a/doc/guides/rel_notes/release_26_11.rst b/doc/guides/rel_notes/release_26_11.rst
index bf1f8d7334..2d6d06d228 100644
--- a/doc/guides/rel_notes/release_26_11.rst
+++ b/doc/guides/rel_notes/release_26_11.rst
@@ -71,6 +71,8 @@ New Features
is selected when the MTU fits in a single buffer.
* Optimized the packed-ring Rx recv path.
* Changed the set of per-queue xstats counters.
+ * Optimized the packed-ring Tx xmit path with per-descriptor mbuf
+ free (``rte_pktmbuf_free_seg``) and prefetch hints.
Removed Items
-------------
diff --git a/drivers/net/zxdh/zxdh_ethdev.c b/drivers/net/zxdh/zxdh_ethdev.c
index 493d25a766..7c2f894642 100644
--- a/drivers/net/zxdh/zxdh_ethdev.c
+++ b/drivers/net/zxdh/zxdh_ethdev.c
@@ -489,7 +489,7 @@ zxdh_dev_free_mbufs(struct rte_eth_dev *dev)
if (!vq)
continue;
while ((buf = zxdh_queue_detach_unused(vq)) != NULL)
- rte_pktmbuf_free(buf);
+ rte_pktmbuf_free_seg(buf);
PMD_DRV_LOG(DEBUG, "freeing %s[%d] used and unused buf",
"rxq", i * 2);
}
@@ -498,7 +498,7 @@ zxdh_dev_free_mbufs(struct rte_eth_dev *dev)
if (!vq)
continue;
while ((buf = zxdh_queue_detach_unused(vq)) != NULL)
- rte_pktmbuf_free(buf);
+ rte_pktmbuf_free_seg(buf);
PMD_DRV_LOG(DEBUG, "freeing %s[%d] used and unused buf",
"txq", i * 2 + 1);
}
diff --git a/drivers/net/zxdh/zxdh_queue.h b/drivers/net/zxdh/zxdh_queue.h
index 49970fbef2..47048f6926 100644
--- a/drivers/net/zxdh/zxdh_queue.h
+++ b/drivers/net/zxdh/zxdh_queue.h
@@ -124,7 +124,6 @@ struct zxdh_vring_packed {
struct zxdh_vq_desc_extra {
void *cookie;
- uint16_t ndescs;
uint16_t next;
};
diff --git a/drivers/net/zxdh/zxdh_rxtx.c b/drivers/net/zxdh/zxdh_rxtx.c
index 004e528adc..8e2fae7c6c 100644
--- a/drivers/net/zxdh/zxdh_rxtx.c
+++ b/drivers/net/zxdh/zxdh_rxtx.c
@@ -114,6 +114,8 @@
RTE_MBUF_F_TX_SEC_OFFLOAD | \
RTE_MBUF_F_TX_UDP_SEG)
+#define ZXDH_NEXT_CACHELINE_OFF_16B (RTE_CACHE_LINE_SIZE / 16)
+
uint32_t zxdh_outer_l2_type[16] = {
0,
RTE_PTYPE_L2_ETHER,
@@ -201,43 +203,6 @@ uint32_t zxdh_inner_l4_type[16] = {
0,
};
-static void
-zxdh_xmit_cleanup_inorder_packed(struct zxdh_virtqueue *vq, int32_t num)
-{
- uint16_t used_idx = 0;
- uint16_t id = 0;
- uint16_t curr_id = 0;
- uint16_t free_cnt = 0;
- uint16_t size = vq->vq_nentries;
- struct zxdh_vring_packed_desc *desc = vq->vq_packed.ring.desc;
- struct zxdh_vq_desc_extra *dxp = NULL;
-
- used_idx = vq->vq_used_cons_idx;
- /* desc_is_used has a load-acquire or rte_io_rmb inside
- * and wait for used desc in virtqueue.
- */
- while (num > 0 && desc_is_used(&desc[used_idx], vq)) {
- id = desc[used_idx].id;
- do {
- curr_id = used_idx;
- dxp = &vq->vq_descx[used_idx];
- used_idx += dxp->ndescs;
- free_cnt += dxp->ndescs;
- num -= dxp->ndescs;
- if (used_idx >= size) {
- used_idx -= size;
- vq->used_wrap_counter ^= 1;
- }
- if (dxp->cookie != NULL) {
- rte_pktmbuf_free(dxp->cookie);
- dxp->cookie = NULL;
- }
- } while (curr_id != id);
- }
- vq->vq_used_cons_idx = used_idx;
- vq->vq_free_cnt += free_cnt;
-}
-
static inline uint16_t
zxdh_get_mtu(struct zxdh_virtqueue *vq)
{
@@ -334,7 +299,7 @@ zxdh_xmit_fill_net_hdr(struct zxdh_virtqueue *vq, struct rte_mbuf *cookie,
}
static inline void
-zxdh_enqueue_xmit_packed_fast(struct zxdh_virtnet_tx *txvq,
+zxdh_xmit_enqueue_push(struct zxdh_virtnet_tx *txvq,
struct rte_mbuf *cookie)
{
struct zxdh_virtqueue *vq = txvq->vq;
@@ -345,7 +310,6 @@ zxdh_enqueue_xmit_packed_fast(struct zxdh_virtnet_tx *txvq,
uint8_t hdr_len = vq->hw->dl_net_hdr_len;
struct zxdh_vring_packed_desc *dp = &vq->vq_packed.ring.desc[id];
- dxp->ndescs = 1;
dxp->cookie = cookie;
hdr = rte_pktmbuf_mtod_offset(cookie, struct zxdh_net_hdr_dl *, -hdr_len);
zxdh_xmit_fill_net_hdr(vq, cookie, hdr);
@@ -362,51 +326,56 @@ zxdh_enqueue_xmit_packed_fast(struct zxdh_virtnet_tx *txvq,
}
static inline void
-zxdh_enqueue_xmit_packed(struct zxdh_virtnet_tx *txvq,
+zxdh_xmit_enqueue_append(struct zxdh_virtnet_tx *txvq,
struct rte_mbuf *cookie,
uint16_t needed)
{
struct zxdh_tx_region *txr = txvq->zxdh_net_hdr_mz->addr;
struct zxdh_virtqueue *vq = txvq->vq;
- uint16_t id = vq->vq_avail_idx;
- struct zxdh_vq_desc_extra *dxp = &vq->vq_descx[id];
uint16_t head_idx = vq->vq_avail_idx;
uint16_t idx = head_idx;
struct zxdh_vring_packed_desc *start_dp = vq->vq_packed.ring.desc;
struct zxdh_vring_packed_desc *head_dp = &vq->vq_packed.ring.desc[idx];
struct zxdh_net_hdr_dl *hdr = NULL;
- uint16_t head_flags = cookie->next ? ZXDH_VRING_DESC_F_NEXT : 0;
+ uint16_t id = vq->vq_avail_idx;
+ struct zxdh_vq_desc_extra *dxp = &vq->vq_descx[id];
uint8_t hdr_len = vq->hw->dl_net_hdr_len;
+ uint16_t head_flags = 0;
- dxp->ndescs = needed;
- dxp->cookie = cookie;
- head_flags |= vq->cached_flags;
+ dxp->cookie = NULL;
+ /*
+ * Head descriptor has no mbuf cookie. Per-segment cookies are
+ * stored on the segment descs so zxdh_xmit_fast_flush() can free
+ * each via rte_pktmbuf_free_seg(). zxdh_queue_detach_unused() walks
+ * the cookies on queue teardown and is the only reader that would
+ * otherwise observe a stray head cookie.
+ */
+ /* setup first tx ring slot to point to header stored in reserved region. */
start_dp[idx].addr = txvq->zxdh_net_hdr_mem + RTE_PTR_DIFF(&txr[idx].tx_hdr, txr);
start_dp[idx].len = hdr_len;
- head_flags |= ZXDH_VRING_DESC_F_NEXT;
+ start_dp[idx].id = idx;
+ head_flags |= vq->cached_flags | ZXDH_VRING_DESC_F_NEXT;
hdr = (void *)&txr[idx].tx_hdr;
- rte_prefetch1(hdr);
+ zxdh_xmit_fill_net_hdr(vq, cookie, hdr);
+
idx++;
if (idx >= vq->vq_nentries) {
idx -= vq->vq_nentries;
vq->cached_flags ^= ZXDH_VRING_PACKED_DESC_F_AVAIL_USED;
}
- zxdh_xmit_fill_net_hdr(vq, cookie, hdr);
-
do {
start_dp[idx].addr = rte_pktmbuf_iova(cookie);
start_dp[idx].len = cookie->data_len;
- start_dp[idx].id = id;
- if (likely(idx != head_idx)) {
- uint16_t flags = cookie->next ? ZXDH_VRING_DESC_F_NEXT : 0;
+ start_dp[idx].id = idx;
- flags |= vq->cached_flags;
- start_dp[idx].flags = flags;
- }
+ vq->vq_descx[idx].cookie = cookie;
+ uint16_t flags = cookie->next ? ZXDH_VRING_DESC_F_NEXT : 0;
+ flags |= vq->cached_flags;
+ start_dp[idx].flags = flags;
idx++;
if (idx >= vq->vq_nentries) {
@@ -430,7 +399,7 @@ zxdh_update_packet_stats(struct zxdh_virtnet_stats *stats, struct rte_mbuf *mbuf
}
static void
-zxdh_xmit_flush(struct zxdh_virtqueue *vq)
+zxdh_xmit_fast_flush(struct zxdh_virtqueue *vq)
{
uint16_t id = 0;
uint16_t curr_id = 0;
@@ -439,27 +408,40 @@ zxdh_xmit_flush(struct zxdh_virtqueue *vq)
struct zxdh_vring_packed_desc *desc = vq->vq_packed.ring.desc;
struct zxdh_vq_desc_extra *dxp = NULL;
uint16_t used_idx = vq->vq_used_cons_idx;
+ uint16_t budget = size;
/*
* The function desc_is_used performs a load-acquire operation
* or calls rte_io_rmb to ensure memory consistency. It waits
* for a used descriptor in the virtqueue.
*/
- while (desc_is_used(&desc[used_idx], vq)) {
+ while (budget > 0 && desc_is_used(&desc[used_idx], vq)) {
+ /*
+ * vq_nentries is validated as power-of-two in
+ * zxdh_queue_desc_pre_setup(), so mask the prefetch index to keep
+ * it inside desc[] when used_idx is near the end of the ring.
+ */
+ rte_prefetch0(&desc[(used_idx + ZXDH_NEXT_CACHELINE_OFF_16B) & (size - 1)]);
id = desc[used_idx].id;
+ if (unlikely(id >= size))
+ break;
do {
+ desc[used_idx].id = used_idx;
curr_id = used_idx;
dxp = &vq->vq_descx[used_idx];
- used_idx += dxp->ndescs;
- free_cnt += dxp->ndescs;
- if (used_idx >= size) {
- used_idx -= size;
- vq->used_wrap_counter ^= 1;
- }
if (dxp->cookie != NULL) {
- rte_pktmbuf_free(dxp->cookie);
+ rte_pktmbuf_free_seg(dxp->cookie);
dxp->cookie = NULL;
}
+ used_idx += 1;
+ free_cnt += 1;
+ budget -= 1;
+ if (unlikely(used_idx == size)) {
+ used_idx = 0;
+ vq->used_wrap_counter ^= 1;
+ }
+ if (unlikely(budget == 0))
+ break;
} while (curr_id != id);
}
vq->vq_used_cons_idx = used_idx;
@@ -473,13 +455,12 @@ zxdh_xmit_pkts_packed(void *tx_queue, struct rte_mbuf **tx_pkts, uint16_t nb_pkt
struct zxdh_virtqueue *vq = txvq->vq;
uint16_t nb_tx = 0;
- zxdh_xmit_flush(vq);
+ zxdh_xmit_fast_flush(vq);
for (nb_tx = 0; nb_tx < nb_pkts; nb_tx++) {
struct rte_mbuf *txm = tx_pkts[nb_tx];
int32_t can_push = 0;
int32_t slots = 0;
- int32_t need = 0;
rte_prefetch0(txm);
/* optimize ring usage */
@@ -496,26 +477,15 @@ zxdh_xmit_pkts_packed(void *tx_queue, struct rte_mbuf **tx_pkts, uint16_t nb_pkt
* default => number of segments + 1
**/
slots = txm->nb_segs + !can_push;
- need = slots - vq->vq_free_cnt;
- /* Positive value indicates it need free vring descriptors */
- if (unlikely(need > 0)) {
- zxdh_xmit_cleanup_inorder_packed(vq, need);
- need = slots - vq->vq_free_cnt;
- if (unlikely(need > 0)) {
- PMD_TX_LOG(ERR,
- " No enough %d free tx descriptors to transmit."
- "freecnt %d",
- need,
- vq->vq_free_cnt);
- break;
- }
- }
+
+ if (unlikely(slots > vq->vq_free_cnt))
+ break;
/* Enqueue Packet buffers */
if (can_push)
- zxdh_enqueue_xmit_packed_fast(txvq, txm);
+ zxdh_xmit_enqueue_push(txvq, txm);
else
- zxdh_enqueue_xmit_packed(txvq, txm, slots);
+ zxdh_xmit_enqueue_append(txvq, txm, slots);
zxdh_update_packet_stats(&txvq->stats, txm);
}
txvq->stats.packets += nb_tx;
--
2.27.0