Ethernet receive threads and loopback sends can enqueue packets at the
same time: they hold different locks, and the skb queue spinlocks are
no-ops on Hurd. Concurrent updates can corrupt the receive backlog,
causing packet loss and TCP timeouts.
Add backlog_lock to protect the queue and netdev_dropping in netif_rx(),
net_bh() and dev_clear_backlog(). Acquire it after the caller's locks and
release it before freeing packets or waking the worker. Check for an
empty queue and dequeue under the same lock.
* pfinet/pfinet.h (backlog_lock): Declare.
* pfinet/sched.c (backlog_lock): Define.
(net_bh_worker): Correct the locking comment.
* pfinet/linux-src/net/core/dev.c (netif_rx, net_bh, dev_clear_backlog):
Protect the receive backlog and drop state with backlog_lock.
* pfinet/loopback.c (loopback_xmit): Correct the locking comment.
---
pfinet/linux-src/net/core/dev.c | 52 +++++++++++++++++++++------------
pfinet/loopback.c | 5 ++--
pfinet/pfinet.h | 1 +
pfinet/sched.c | 19 ++++--------
4 files changed, 43 insertions(+), 34 deletions(-)
diff --git a/pfinet/linux-src/net/core/dev.c b/pfinet/linux-src/net/core/dev.c
index b47c502..ee59d92 100644
--- a/pfinet/linux-src/net/core/dev.c
+++ b/pfinet/linux-src/net/core/dev.c
@@ -718,7 +718,6 @@ static void netdev_wakeup(void)
static void dev_clear_backlog(struct device *dev)
{
struct sk_buff *curr;
- unsigned long flags;
/*
*
@@ -727,30 +726,35 @@ static void dev_clear_backlog(struct device *dev)
* We are competing here both with netif_rx() and net_bh().
* We don't want either of those to mess with skb ptrs
* while we work on them, thus we must grab the
- * skb_queue_lock.
+ * backlog_lock.
*/
+ pthread_mutex_lock(&backlog_lock);
if (backlog.qlen) {
repeat:
- spin_lock_irqsave(&skb_queue_lock, flags);
for (curr = backlog.next;
curr != (struct sk_buff *)(&backlog);
curr = curr->next)
if (curr->dev == dev)
{
__skb_unlink(curr, &backlog);
- spin_unlock_irqrestore(&skb_queue_lock, flags);
+ pthread_mutex_unlock(&backlog_lock);
kfree_skb(curr);
+ pthread_mutex_lock(&backlog_lock);
goto repeat;
}
- spin_unlock_irqrestore(&skb_queue_lock, flags);
#ifdef CONFIG_NET_HW_FLOWCONTROL
if (netdev_dropping)
+ {
+ pthread_mutex_unlock(&backlog_lock);
netdev_wakeup();
+ return;
+ }
#else
netdev_dropping = 0;
#endif
}
+ pthread_mutex_unlock(&backlog_lock);
}
/*
@@ -771,13 +775,17 @@ void netif_rx(struct sk_buff *skb)
short when CPU is congested, but is still operating.
*/
+ pthread_mutex_lock (&backlog_lock);
+
if (backlog.qlen <= netdev_max_backlog) {
if (backlog.qlen) {
if (netdev_dropping == 0) {
skb_queue_tail(&backlog,skb);
+ pthread_mutex_unlock (&backlog_lock);
mark_bh(NET_BH);
return;
}
+ pthread_mutex_unlock (&backlog_lock);
atomic_inc(&netdev_rx_dropped);
kfree_skb(skb);
return;
@@ -789,10 +797,12 @@ void netif_rx(struct sk_buff *skb)
netdev_dropping = 0;
#endif
skb_queue_tail(&backlog,skb);
+ pthread_mutex_unlock (&backlog_lock);
mark_bh(NET_BH);
return;
}
netdev_dropping = 1;
+ pthread_mutex_unlock (&backlog_lock);
atomic_inc(&netdev_rx_dropped);
kfree_skb(skb);
}
@@ -871,14 +881,10 @@ void net_bh(void)
*/
/*
- * While the queue is not empty..
- *
- * Note that the queue never shrinks due to
- * an interrupt, so we can do this test without
- * disabling interrupts.
+ * Drain the queue under backlog_lock.
*/
- while (!skb_queue_empty(&backlog))
+ for (;;)
{
struct sk_buff * skb;
@@ -889,9 +895,25 @@ void net_bh(void)
#endif
/*
- * We have a packet. Therefore the queue has shrunk
+ * Dequeue and reset the drop state under backlog_lock.
*/
+ pthread_mutex_lock (&backlog_lock);
skb = skb_dequeue(&backlog);
+ if (skb == NULL)
+ {
+#ifdef CONFIG_NET_HW_FLOWCONTROL
+ int dropping = netdev_dropping;
+#else
+ netdev_dropping = 0;
+#endif
+ pthread_mutex_unlock (&backlog_lock);
+#ifdef CONFIG_NET_HW_FLOWCONTROL
+ if (dropping)
+ netdev_wakeup();
+#endif
+ break;
+ }
+ pthread_mutex_unlock (&backlog_lock);
#ifndef _HURD_
#ifdef CONFIG_CPU_IS_SLOW
@@ -1027,12 +1049,6 @@ void net_bh(void)
start_busy = 0;
}
#endif
-#endif
-#ifdef CONFIG_NET_HW_FLOWCONTROL
- if (netdev_dropping)
- netdev_wakeup();
-#else
- netdev_dropping = 0;
#endif
NET_PROFILE_LEAVE(net_bh);
return;
diff --git a/pfinet/loopback.c b/pfinet/loopback.c
index e15e426..66c4b4f 100644
--- a/pfinet/loopback.c
+++ b/pfinet/loopback.c
@@ -73,9 +73,8 @@ static int loopback_xmit(struct sk_buff *skb, struct device
*dev)
#endif
/*
- * Calling netif_rx() requires locking net_bh_lock, which
- * has already been done since this function is called by
- * the net_bh worker thread.
+ * netif_rx() protects the backlog with its own lock. Callers
+ * may hold either global_lock or net_bh_lock.
*/
netif_rx(skb);
diff --git a/pfinet/pfinet.h b/pfinet/pfinet.h
index df8da52..2e9ce0e 100644
--- a/pfinet/pfinet.h
+++ b/pfinet/pfinet.h
@@ -38,6 +38,7 @@
extern pthread_mutex_t global_lock;
extern pthread_mutex_t net_bh_lock;
+extern pthread_mutex_t backlog_lock;
extern struct port_bucket *pfinet_bucket;
extern struct port_class *addrport_class;
diff --git a/pfinet/sched.c b/pfinet/sched.c
index 5a3dd3e..9e9f00c 100644
--- a/pfinet/sched.c
+++ b/pfinet/sched.c
@@ -26,6 +26,7 @@
pthread_mutex_t global_lock = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t net_bh_lock = PTHREAD_MUTEX_INITIALIZER;
+pthread_mutex_t backlog_lock = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t net_bh_wakeup = PTHREAD_COND_INITIALIZER;
int net_bh_raised = 0;
@@ -44,19 +45,11 @@ sock_wake_async (struct socket *sock, int how)
}
-/* This function is the "net_bh worker thread".
- The packet receiver thread calls net/core/dev.c::netif_rx with a packet;
- netif_rx either drops the packet, or enqueues it and wakes us up
- via mark_bh which is really condition_broadcast on net_bh_wakeup.
- The packet receiver thread holds net_bh_lock while calling netif_rx.
- We wake up and take global_lock, which locks out RPC service threads.
- We then also take net_bh_lock running net_bh.
- Thus, only this thread running net_bh locks out the packet receiver
- thread (which takes only net_bh_lock while calling netif_rx), so packets
- are quickly moved from the Mach port's message queue to the `backlog'
- queue, or dropped, without synchronizing with RPC service threads.
- (The RPC service threads lock out the running of net_bh, but not
- the queuing/dropping of packets in netif_rx.) */
+/* This is the net_bh worker thread. Packet receivers hold net_bh_lock;
+ RPC service threads hold global_lock. netif_rx protects the shared
+ backlog with backlog_lock, acquired after either caller's lock.
+ mark_bh wakes this thread, which waits with net_bh_lock and takes
+ global_lock before running net_bh. */
void *
net_bh_worker (void *arg)
{
