The enetc driver uses a struct enetc_int_vector which holds a fixed mapping between the interrupt vector and the RX ring it services. While this mapping is fixed, it has been pointed out that the IRQ affinity to a CPU can be changed by user space, so there is no fixed mapping between that RX ring and a CPU. That means that on LS1028A, NAPI on CPU 0 can service RX ring 1, and if a BPF program requests the XDP_TX verdict, received traffic will be sent to the TX ring reserved for XDP_TX associated with that RX ring. At the same time, XDP_REDIRECT on CPU 1 can also call ndo_xdp_xmit(), and this selects for transmission the XDP TX ring of the current CPU, aka the same as of RX ring 1. This is bad, because the transmission code is not written with concurrency in mind, and access to the TX BD ring pointers as well as the statistics counters is unsynchronized. Create a spin lock in struct enetc_bdr (common between RX and TX rings), initialize the lock for TX rings and use it to serialize these two problematic code paths. Since the calling contexts are always softirq, we can use the plain spin_lock() over spin_lock_bh() which is also more expensive. Specifically, the spin lock only synchronizes the stats counters relevant to XDP (xdp_tx/xdp_tx_drops) and the TX BDR producer state (next_to_use, tx_swbd[] and hardware counterparts). The rest should be not necessary. The lock placement is inside a 52 byte hole in cacheline 1 of struct enetc_bdr. The hole is now 48 bytes in size. Testing shows for a single-CPU XDP_TX flow on 1 interface, 64B packets: Unmitigated MPPS Naive mitigation MPPS Batched mitigation MPPS 2.32 1.9 2.2 where 'naive' means locking per packet in enetc_clean_rx_ring_xdp(), and 'batched' means what this patch is doing, i.e. acquire the TX ring lock at the beginning of the poll cycle, irrespective of XDP program verdict (necessary only for XDP_TX), and release it around xdp_do_redirect() and napi_gro_receive(). We then reacquire it around enetc_update_tx_ring_tail(), so that no matter which CPU rings the doorbell, the correct producer index (tx_ring->next_to_use) is committed to hardware. Note that in principle we could drop the rx_ring from struct enetc_int_vector, and as in the ndo_xdp_xmit() case, dynamically pick up in the NET_RX softirq that RX ring which corresponds to the current CPU. Because of the softirq context, ndo_xdp_xmit() would be excluded from running in a lockless way. However, doing this seems counterproductive to the user space intention of changing the IRQ affinity. It opens the door for user space to move all interrupt vectors to the same CPU, which will leave the other RX ring unprocessed. This seems like a bad idea considering RSS/RFS. Fixes: 9d2b68cc108d ("net: enetc: add support for XDP_REDIRECT") Signed-off-by: Vladimir Oltean --- v3->v4: patch is new --- drivers/net/ethernet/freescale/enetc/enetc.c | 21 ++++++++++++++++++-- drivers/net/ethernet/freescale/enetc/enetc.h | 4 ++++ 2 files changed, 23 insertions(+), 2 deletions(-) diff --git a/drivers/net/ethernet/freescale/enetc/enetc.c b/drivers/net/ethernet/freescale/enetc/enetc.c index 103335f3d418..bd824b55a916 100644 --- a/drivers/net/ethernet/freescale/enetc/enetc.c +++ b/drivers/net/ethernet/freescale/enetc/enetc.c @@ -1814,6 +1814,7 @@ int enetc_xdp_xmit(struct net_device *ndev, int num_frames, int err = 0; tx_ring = priv->xdp_tx_ring[smp_processor_id()]; + spin_lock(&tx_ring->xdp_lock); if (unlikely(test_bit(ENETC_TX_DOWN, &priv->flags) || !netif_carrier_ok(ndev))) { @@ -1858,6 +1859,8 @@ int enetc_xdp_xmit(struct net_device *ndev, int num_frames, tx_ring->stats.xdp_tx += xdp_tx_frm_cnt; tx_ring->stats.xdp_tx_drops += num_frames - xdp_tx_frm_cnt; + spin_unlock(&tx_ring->xdp_lock); + return err ?: xdp_tx_frm_cnt; } EXPORT_SYMBOL_GPL(enetc_xdp_xmit); @@ -1998,6 +2001,9 @@ static int enetc_clean_rx_ring_xdp(struct enetc_bdr *rx_ring, /* next descriptor to process */ i = rx_ring->next_to_clean; + tx_ring = priv->xdp_tx_ring[rx_ring->index]; + spin_lock(&tx_ring->xdp_lock); + enetc_lock_mdio(); while (likely(rx_frm_cnt < work_limit)) { @@ -2064,11 +2070,12 @@ static int enetc_clean_rx_ring_xdp(struct enetc_bdr *rx_ring, enetc_bulk_flip_buff(rx_ring, orig_i, i); enetc_unlock_mdio(); + spin_unlock(&tx_ring->xdp_lock); napi_gro_receive(napi, skb); + spin_lock(&tx_ring->xdp_lock); enetc_lock_mdio(); break; case XDP_TX: - tx_ring = priv->xdp_tx_ring[rx_ring->index]; if (unlikely(test_bit(ENETC_TX_DOWN, &priv->flags))) { enetc_xdp_drop(rx_ring, orig_i, i); tx_ring->stats.xdp_tx_drops++; @@ -2101,7 +2108,11 @@ static int enetc_clean_rx_ring_xdp(struct enetc_bdr *rx_ring, break; case XDP_REDIRECT: enetc_unlock_mdio(); + spin_unlock(&tx_ring->xdp_lock); + err = xdp_do_redirect(rx_ring->ndev, &xdp_buff, prog); + + spin_lock(&tx_ring->xdp_lock); enetc_lock_mdio(); if (unlikely(err)) { enetc_xdp_drop(rx_ring, orig_i, i); @@ -2117,6 +2128,8 @@ static int enetc_clean_rx_ring_xdp(struct enetc_bdr *rx_ring, } out: + spin_unlock(&tx_ring->xdp_lock); + rx_ring->next_to_clean = i; rx_ring->stats.packets += rx_frm_cnt; @@ -2128,8 +2141,11 @@ static int enetc_clean_rx_ring_xdp(struct enetc_bdr *rx_ring, enetc_lock_mdio(); } - if (xdp_tx_frm_cnt) + if (xdp_tx_frm_cnt) { + spin_lock(&tx_ring->xdp_lock); enetc_update_tx_ring_tail(tx_ring); + spin_unlock(&tx_ring->xdp_lock); + } if (cleaned_cnt > rx_ring->xdp.xdp_tx_in_flight) enetc_refill_rx_ring(rx_ring, enetc_bd_unused(rx_ring) - @@ -3591,6 +3607,7 @@ static int enetc_int_vector_init(struct enetc_ndev_priv *priv, int i, idx = priv->bdr_int_num * j + i; __set_bit(idx, &v->tx_rings_map); bdr = &v->tx_ring[j]; + spin_lock_init(&bdr->xdp_lock); bdr->index = idx; bdr->ndev = priv->ndev; bdr->dev = priv->dev; diff --git a/drivers/net/ethernet/freescale/enetc/enetc.h b/drivers/net/ethernet/freescale/enetc/enetc.h index d9e91832a9c1..bc7c0740498d 100644 --- a/drivers/net/ethernet/freescale/enetc/enetc.h +++ b/drivers/net/ethernet/freescale/enetc/enetc.h @@ -171,6 +171,10 @@ struct enetc_bdr { void __iomem *idr; /* Interrupt Detect Register pointer */ int buffer_offset; + + /* Serialize NAPI and ndo_xdp_xmit() targeting the same TX ring */ + spinlock_t xdp_lock; + struct enetc_xdp_data xdp; struct enetc_ring_stats stats; -- 2.43.0