Re: [PATCH net-next v5 4/8] TUN & TAP: Wake netdev queue after consuming an entry

["Michael S. Tsirkin" <mst@redhat.com>]

On Wed, Sep 24, 2025 at 10:40:04AM +0200, Simon Schippers wrote:
> On 24.09.25 09:49, Michael S. Tsirkin wrote:
> > On Wed, Sep 24, 2025 at 09:42:45AM +0200, Simon Schippers wrote:
> >> On 24.09.25 08:55, Michael S. Tsirkin wrote:
> >>> On Wed, Sep 24, 2025 at 07:56:33AM +0200, Simon Schippers wrote:
> >>>> On 23.09.25 18:36, Michael S. Tsirkin wrote:
> >>>>> On Tue, Sep 23, 2025 at 12:15:49AM +0200, Simon Schippers wrote:
> >>>>>> 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 <tim.gebauer@tu-dortmund.de>
> >>>>>> Signed-off-by: Tim Gebauer <tim.gebauer@tu-dortmund.de>
> >>>>>> Signed-off-by: Simon Schippers <simon.schippers@tu-dortmund.de>
> >>>>>> ---
> >>>>>>  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?
> >>>
> >>>
> >>> Maybe you mean == ? Pls don't poke at ptr ring internals though.
> >>> What are we testing for here?
> >>> I think the point is that a batch of entries was consumed?
> >>> Maybe __ptr_ring_consumed_batch ? and a comment explaining
> >>> this returns true when last successful call to consume
> >>> freed up a batch of space in the ring for producer to make
> >>> progress.
> >>>
> >>
> >> Yes, I mean ==.
> >>
> >> Having a dedicated helper for this purpose makes sense. I just find
> >> the name __ptr_ring_consumed_batch a bit confusing next to
> >> __ptr_ring_consume_batched, since they both refer to different kinds of
> >> batches.
> > 
> > __ptr_ring_consume_created_space ?
> > 
> > /* Previous call to ptr_ring_consume created some space.
> >  *
> >  * NB: only refers to the last call to __ptr_ring_consume,
> >  * if you are calling ptr_ring_consume multiple times, you
> >  * have to check this multiple times.
> >  * Accordingly, do not use this after __ptr_ring_consume_batched.
> >  */
> >
> 
> Sounds good.
> 
> Regarding __ptr_ring_consume_batched:
> Theoretically the consumer_tail before and after calling the method could
> be compared to avoid calling __ptr_ring_consume_created_space at each
> iteration. But I guess it is also fine calling it at each iteration.

Hmm good point, though I worry about wrap-around a bit.


> >>>
> >>> consumer_head == consumer_tail also happens rather a lot,
> >>> though thankfully not on every entry.
> >>> So taking tx lock each time this happens, even if queue
> >>> is not stopped, seems heavyweight.
> >>>
> >>>
> >>
> >> Yes, I agree — but avoiding locking probably requires some fancy
> >> ordering tricks again..
> >>
> >>
> >>>
> >>>
> >>>
> >>>>>> 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?
> >>>>>
> >>>>>
> >>>>>
> >>>>>
> >>>>>> +	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
> >>>>>
> >>>
> >