On 29.09.25 11:43, Simon Schippers wrote:
> On 29.09.25 00:33, Michael S. Tsirkin wrote:
>> On Sun, Sep 28, 2025 at 11:27:25PM +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 <[email protected]>
>>>>> Signed-off-by: Tim Gebauer <[email protected]>
>>>>> Signed-off-by: Simon Schippers <[email protected]>
>>>>> ---
>>>>> 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 think I found a solution to this problem, see below:
>>>
>>>>
>>>>> 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)) {
>>>
>>> Here I just check for
>>>
>>> if (will_invalidate || __ptr_ring_empty(&q->ring)) {
>>>
>>> instead because, if the ptr_ring is empty and the netdev queue stopped,
>>> the race must have occurred. Then it is safe to wake the netdev queue,
>>> because it is known that space in the ptr_ring was freed when the race
>>> occurred. Also, it is guaranteed that tap_ring_consume is called at least
>>> once after the race, because a new entry is generated by the producer at
>>> the race.
>>> In my adjusted implementation, it tests fine with pktgen without any lost
>>> packets.
>>
>>
>> what if it is not empty and ring is stopped?
>>
>
> Then it can not be assumed that there is free space in the ptr_ring,
> because __ptr_ring_discard_one may only create space after one of the
> upcoming entries that it will consume. Only if the ptr_ring is empty
> (which will obviously happen after some time) it is guaranteed that there
> is free space in the ptr_ring, either because the race occurred
> previously or __ptr_ring_discard_one freed entries right before.
>
>>>
>>> Generally now I think that the whole implementation can be fine without
>>> using spinlocks at all. I am currently adjusting the implementation
>>> regarding SMP memory barrier pairings, and I have a question:
>>> In the v4 you mentioned "the stop -> wake bounce involves enough barriers
>>> already". Does it, for instance, mean that netif_tx_wake_queue already
>>> ensures memory ordering, and I do not have to use an smp_wmb() in front of
>>> netif_tx_wake_queue() and smp_rmb() in front of the ptr_ring operations
>>> in tun_net_xmit?
>>> I dug through net/core/netdevice.h and dev.c but could not really
>>> answer this question by myself...
>>> Thanks :)
>>
>> Only if it wakes up something, I think.
>>
>> Read:
>>
>> SLEEP AND WAKE-UP FUNCTIONS
>>
>>
>> in Documentation/memory-barriers.txt
>>
>>
>> IIUC this is the same.
>>
>>
>
> Thanks, I will look into it! :)
>
I do not see how the netdev queue flow control follows a setup with a
sleeper and a waker as described in SLEEP AND WAKE-UP FUNCTIONS.
Yes, there is netif_tx_wake_queue, but it does not call a waker function
like complete() or wake_up(). And I do not see how there is a sleeper
that calls schedule() somewhere.
Do I misunderstand something?
For now I would instead use an additional smp_wmb() in front of
netif_tx_wake_queue for the consumer and an smp_rmb() in front of all
ptr_ring operations for the producer. This ensures that space, which is
freed by the consumer that woke up the netdev queue, is visible for the
producer.
Thanks! :)
>>>
>>>>> + 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
>>>>
>>
