On 23.09.25 18:36, Michael S. Tsirkin wrote:

On Tue, Sep 23, 2025 at 12:15:49AM +0200, Simon Schippers wrote:

quoted

The new wrappers tun_ring_consume/tap_ring_consume deal with consuming an
entry of the ptr_ring and then waking the netdev queue when entries got
invalidated to be used again by the producer.
To avoid waking the netdev queue when the ptr_ring is full, it is checked
if the netdev queue is stopped before invalidating entries. Like that the
netdev queue can be safely woken after invalidating entries.

The READ_ONCE in __ptr_ring_peek, paired with the smp_wmb() in
__ptr_ring_produce within tun_net_xmit guarantees that the information
about the netdev queue being stopped is visible after __ptr_ring_peek is
called.

The netdev queue is also woken after resizing the ptr_ring.

Co-developed-by: Tim Gebauer <redacted>
Signed-off-by: Tim Gebauer <redacted>
Signed-off-by: Simon Schippers <redacted>
---
 drivers/net/tap.c | 44 +++++++++++++++++++++++++++++++++++++++++++-
 drivers/net/tun.c | 47 +++++++++++++++++++++++++++++++++++++++++++++--
 2 files changed, 88 insertions(+), 3 deletions(-)

diff --git a/drivers/net/tap.c b/drivers/net/tap.c
index 1197f245e873..f8292721a9d6 100644
--- a/drivers/net/tap.c
+++ b/drivers/net/tap.c

@@ -753,6 +753,46 @@ static ssize_t tap_put_user(struct tap_queue *q,
 	return ret ? ret : total;
 }
 
+static struct sk_buff *tap_ring_consume(struct tap_queue *q)
+{
+	struct netdev_queue *txq;
+	struct net_device *dev;
+	bool will_invalidate;
+	bool stopped;
+	void *ptr;
+
+	spin_lock(&q->ring.consumer_lock);
+	ptr = __ptr_ring_peek(&q->ring);
+	if (!ptr) {
+		spin_unlock(&q->ring.consumer_lock);
+		return ptr;
+	}
+
+	/* Check if the queue stopped before zeroing out, so no ptr get
+	 * produced in the meantime, because this could result in waking
+	 * even though the ptr_ring is full.

So what? Maybe it would be a bit suboptimal? But with your design, I do
not get what prevents this:


	stopped? -> No
		ring is stopped
	discard

and queue stays stopped forever

I totally missed this (but I am not sure why it did not happen in my 
testing with different ptr_ring sizes..).

I guess you are right, there must be some type of locking.
It probably makes sense to lock the netdev txq->_xmit_lock whenever the 
consumer invalidates old ptr_ring entries (so when r->consumer_head >= 
r->consumer_tail). The producer holds this lock with dev->lltx=false. Then 
the consumer is able to wake the queue safely.

So I would now just change the implementation to:
tun_net_xmit:
...
if ptr_ring_produce
    // Could happen because of unproduce in vhost_net..
    netif_tx_stop_queue
    ...
    goto drop

if ptr_ring_full
    netif_tx_stop_queue
...

tun_ring_recv/tap_do_read (the implementation for the batched methods 
would be done in the similar way):
...
ptr_ring_consume
if r->consumer_head >= r->consumer_tail
    __netif_tx_lock_bh
    netif_tx_wake_queue
    __netif_tx_unlock_bh

This implementation does not need any new ptr_ring helpers and no fancy 
ordering tricks.
Would this implementation be sufficient in your opinion?

quoted

The order of the operations
+	 * is ensured by barrier().
+	 */
+	will_invalidate = __ptr_ring_will_invalidate(&q->ring);
+	if (unlikely(will_invalidate)) {
+		rcu_read_lock();
+		dev = rcu_dereference(q->tap)->dev;
+		txq = netdev_get_tx_queue(dev, q->queue_index);
+		stopped = netif_tx_queue_stopped(txq);
+	}
+	barrier();
+	__ptr_ring_discard_one(&q->ring, will_invalidate);
+
+	if (unlikely(will_invalidate)) {
+		if (stopped)
+			netif_tx_wake_queue(txq);
+		rcu_read_unlock();
+	}

After an entry is consumed, you can detect this by checking

	                r->consumer_head >= r->consumer_tail


so it seems you could keep calling regular ptr_ring_consume
and check afterwards?

quoted

+	spin_unlock(&q->ring.consumer_lock);
+
+	return ptr;
+}
+
 static ssize_t tap_do_read(struct tap_queue *q,
 			   struct iov_iter *to,
 			   int noblock, struct sk_buff *skb)

@@ -774,7 +814,7 @@ static ssize_t tap_do_read(struct tap_queue *q,
 					TASK_INTERRUPTIBLE);
 
 		/* Read frames from the queue */
-		skb = ptr_ring_consume(&q->ring);
+		skb = tap_ring_consume(q);
 		if (skb)
 			break;
 		if (noblock) {

@@ -1207,6 +1247,8 @@ int tap_queue_resize(struct tap_dev *tap)
 	ret = ptr_ring_resize_multiple_bh(rings, n,
 					  dev->tx_queue_len, GFP_KERNEL,
 					  __skb_array_destroy_skb);
+	if (netif_running(dev))
+		netif_tx_wake_all_queues(dev);
 
 	kfree(rings);
 	return ret;

diff --git a/drivers/net/tun.c b/drivers/net/tun.c
index c6b22af9bae8..682df8157b55 100644
--- a/drivers/net/tun.c
+++ b/drivers/net/tun.c

@@ -2114,13 +2114,53 @@ static ssize_t tun_put_user(struct tun_struct *tun,
 	return total;
 }
 
+static void *tun_ring_consume(struct tun_file *tfile)
+{
+	struct netdev_queue *txq;
+	struct net_device *dev;
+	bool will_invalidate;
+	bool stopped;
+	void *ptr;
+
+	spin_lock(&tfile->tx_ring.consumer_lock);
+	ptr = __ptr_ring_peek(&tfile->tx_ring);
+	if (!ptr) {
+		spin_unlock(&tfile->tx_ring.consumer_lock);
+		return ptr;
+	}
+
+	/* Check if the queue stopped before zeroing out, so no ptr get
+	 * produced in the meantime, because this could result in waking
+	 * even though the ptr_ring is full. The order of the operations
+	 * is ensured by barrier().
+	 */
+	will_invalidate = __ptr_ring_will_invalidate(&tfile->tx_ring);
+	if (unlikely(will_invalidate)) {
+		rcu_read_lock();
+		dev = rcu_dereference(tfile->tun)->dev;
+		txq = netdev_get_tx_queue(dev, tfile->queue_index);
+		stopped = netif_tx_queue_stopped(txq);
+	}
+	barrier();
+	__ptr_ring_discard_one(&tfile->tx_ring, will_invalidate);
+
+	if (unlikely(will_invalidate)) {
+		if (stopped)
+			netif_tx_wake_queue(txq);
+		rcu_read_unlock();
+	}
+	spin_unlock(&tfile->tx_ring.consumer_lock);
+
+	return ptr;
+}
+
 static void *tun_ring_recv(struct tun_file *tfile, int noblock, int *err)
 {
 	DECLARE_WAITQUEUE(wait, current);
 	void *ptr = NULL;
 	int error = 0;
 
-	ptr = ptr_ring_consume(&tfile->tx_ring);
+	ptr = tun_ring_consume(tfile);
 	if (ptr)
 		goto out;
 	if (noblock) {

@@ -2132,7 +2172,7 @@ static void *tun_ring_recv(struct tun_file *tfile, int noblock, int *err)
 
 	while (1) {
 		set_current_state(TASK_INTERRUPTIBLE);
-		ptr = ptr_ring_consume(&tfile->tx_ring);
+		ptr = tun_ring_consume(tfile);
 		if (ptr)
 			break;
 		if (signal_pending(current)) {

@@ -3621,6 +3661,9 @@ static int tun_queue_resize(struct tun_struct *tun)
 					  dev->tx_queue_len, GFP_KERNEL,
 					  tun_ptr_free);
 
+	if (netif_running(dev))
+		netif_tx_wake_all_queues(dev);
+
 	kfree(rings);
 	return ret;
 }

-- 
2.43.0