Re: Kernel WARNING: at net/core/dev.c:1330 __netif_schedule+0x2c/0x98()

Peter Zijlstra
2008-07-24T09:11:04Z

On Wed, 2008-07-23 at 13:16 -0700, David Miller wrote:
> From: Jarek Poplawski <jarkao2@gmail.com>
> Date: Wed, 23 Jul 2008 11:49:14 +0000
>
> > On Wed, Jul 23, 2008 at 11:35:19AM +0000, Jarek Poplawski wrote:
> > > On Wed, Jul 23, 2008 at 12:58:16PM +0200, Peter Zijlstra wrote:
> > ...
> > > > When I look at the mac802.11 code in ieee80211_tx_pending() it looks
> > > > like it can do with just one lock at a time, instead of all - but I
> > > > might be missing some obvious details.
> > > >
> > > > So I guess my question is, is netif_tx_lock() here to stay, or is the
> > > > right fix to convert all those drivers to use __netif_tx_lock() which
> > > > locks only a single queue?
> > > >
> > >
> > > It's a new thing mainly for new hardware/drivers, and just after
> > > conversion (older drivers effectively use __netif_tx_lock()), so it'll
> > > probably stay for some time until something better is found. David,
> > > will tell the rest, I hope.
> >
> > ...And, of course, these new drivers should also lock a single queue
> > where possible.
>
> It isn't going away.
>
> There will always be a need for a "stop all the TX queues" operation.

Ok, then how about something like this, the idea is to wrap the per tx
lock with a read lock of the device and let the netif_tx_lock() be the
write side, therefore excluding all device locks, but not incure the
cacheline bouncing on the read side by using per-cpu counters like rcu
does.

This of course requires that netif_tx_lock() is rare, otherwise stuff
will go bounce anyway...

Probably missed a few details,.. but I think the below ought to show the
idea...

struct tx_lock {
int busy;
spinlock_t lock;
unsigned long *counters;
};

int tx_lock_init(struct tx_lock *txl)
{
txl->busy = 0;
spin_lock_init(&txl->lock);
txl->counters = alloc_percpu(unsigned long);

if (!txl->counters)
return -ENOMEM;

return 0;
}

void __netif_tx_lock(struct netdev_queue *txq, cpu)
{
struct net_device *dev = txq->dev;

if (rcu_dereference(dev->tx_lock.busy)) {
spin_lock(&dev->tx_lock.lock);
(*percpu_ptr(dev->tx_lock.counters, cpu))++;
spin_unlock(&dev->tx_lock.lock);
} else
(*percpu_ptr(dev->tx_lock.counters, cpu))++;

spin_lock(&txq->_xmit_lock);
txq->xmit_lock_owner = cpu;
}

void __netif_tx_unlock(struct netdev_queue *txq)
{
struct net_device *dev = txq->dev;

(*percpu_ptr(dev->tx_lock.counters, txq->xmit_lock_owner))--;
txq->xmit_lock_owner = -1;
spin_unlock(&txq->xmit_lock);
}

unsigned long tx_lock_read_counters(struct tx_lock *txl)
{
int i;
unsigned long counter = 0;

/* can use online - the inc/dec are matched per cpu */
for_each_online_cpu(i)
counter += *percpu_ptr(txl->counters, i);

return counter;
}

void netif_tx_lock(struct net_device *dev)
{
spin_lock(&dev->tx_lock.lock);
rcu_assign_pointer(dev->tx_lock.busy, 1);

while (tx_lock_read_counters(&dev->tx_lock)
cpu_relax();
}

void netif_tx_unlock(struct net_device *dev)
{
rcu_assign_pointer(dev->tx_lock.busy, 0);
smp_wmb(); /* because rcu_assign_pointer is broken */
spin_unlock(&dev->tx_lock.lock);
}