net: sock_def_readable() and friends RCU conversion

sk_callback_lock rwlock actually protects sk->sk_sleep pointer, so we
need two atomic operations (and associated dirtying) per incoming
packet.

RCU conversion is pretty much needed :

1) Add a new structure, called "struct socket_wq" to hold all fields
that will need rcu_read_lock() protection (currently: a
wait_queue_head_t and a struct fasync_struct pointer).

[Future patch will add a list anchor for wakeup coalescing]

2) Attach one of such structure to each "struct socket" created in
sock_alloc_inode().

3) Respect RCU grace period when freeing a "struct socket_wq"

4) Change sk_sleep pointer in "struct sock" by sk_wq, pointer to "struct
socket_wq"

5) Change sk_sleep() function to use new sk->sk_wq instead of
sk->sk_sleep

6) Change sk_has_sleeper() to wq_has_sleeper() that must be used inside
a rcu_read_lock() section.

7) Change all sk_has_sleeper() callers to :
  - Use rcu_read_lock() instead of read_lock(&sk->sk_callback_lock)
  - Use wq_has_sleeper() to eventually wakeup tasks.
  - Use rcu_read_unlock() instead of read_unlock(&sk->sk_callback_lock)

8) sock_wake_async() is modified to use rcu protection as well.

9) Exceptions :
  macvtap, drivers/net/tun.c, af_unix use integrated "struct socket_wq"
instead of dynamically allocated ones. They dont need rcu freeing.

Some cleanups or followups are probably needed, (possible
sk_callback_lock conversion to a spinlock for example...).

Signed-off-by: Eric Dumazet <eric.dumazet@gmail.com>
Signed-off-by: David S. Miller <davem@davemloft.net>

11 files changed, 121 insertions, 68 deletions

diff --git a/net/atm/common.c b/net/atm/common.c
index e3e10e6f8628..b43feb1a3995 100644
--- a/net/atm/common.c
+++ b/net/atm/common.c
@@ -90,10 +90,13 @@ static void vcc_sock_destruct(struct sock *sk)
 
 static void vcc_def_wakeup(struct sock *sk)
 {
-        read_lock(&sk->sk_callback_lock);
-        if (sk_has_sleeper(sk))
-                wake_up(sk_sleep(sk));
-        read_unlock(&sk->sk_callback_lock);
+        struct socket_wq *wq;
+
+        rcu_read_lock();
+        wq = rcu_dereference(sk->sk_wq);
+        if (wq_has_sleeper(wq))
+                wake_up(&wq->wait);
+        rcu_read_unlock();
 }
 
 static inline int vcc_writable(struct sock *sk)
@@ -106,16 +109,19 @@ static inline int vcc_writable(struct sock *sk)
 
 static void vcc_write_space(struct sock *sk)
 {
-        read_lock(&sk->sk_callback_lock);
+        struct socket_wq *wq;
+
+        rcu_read_lock();
 
         if (vcc_writable(sk)) {
-                if (sk_has_sleeper(sk))
-                        wake_up_interruptible(sk_sleep(sk));
+                wq = rcu_dereference(sk->sk_wq);
+                if (wq_has_sleeper(wq))
+                        wake_up_interruptible(&wq->wait);
 
                 sk_wake_async(sk, SOCK_WAKE_SPACE, POLL_OUT);
         }
 
-        read_unlock(&sk->sk_callback_lock);
+        rcu_read_unlock();
 }
 
 static struct proto vcc_proto = {
diff --git a/net/core/sock.c b/net/core/sock.c
index 51041759517e..94c4affdda9b 100644
--- a/net/core/sock.c
+++ b/net/core/sock.c
@@ -1211,7 +1211,7 @@ struct sock *sk_clone(const struct sock *sk, const gfp_t priority)
                  */
                 sk_refcnt_debug_inc(newsk);
                 sk_set_socket(newsk, NULL);
-                newsk->sk_sleep  = NULL;
+                newsk->sk_wq = NULL;
 
                 if (newsk->sk_prot->sockets_allocated)
                         percpu_counter_inc(newsk->sk_prot->sockets_allocated);
@@ -1800,41 +1800,53 @@ EXPORT_SYMBOL(sock_no_sendpage);
 
 static void sock_def_wakeup(struct sock *sk)
 {
-        read_lock(&sk->sk_callback_lock);
-        if (sk_has_sleeper(sk))
-                wake_up_interruptible_all(sk_sleep(sk));
-        read_unlock(&sk->sk_callback_lock);
+        struct socket_wq *wq;
+
+        rcu_read_lock();
+        wq = rcu_dereference(sk->sk_wq);
+        if (wq_has_sleeper(wq))
+                wake_up_interruptible_all(&wq->wait);
+        rcu_read_unlock();
 }
 
 static void sock_def_error_report(struct sock *sk)
 {
-        read_lock(&sk->sk_callback_lock);
-        if (sk_has_sleeper(sk))
-                wake_up_interruptible_poll(sk_sleep(sk), POLLERR);
+        struct socket_wq *wq;
+
+        rcu_read_lock();
+        wq = rcu_dereference(sk->sk_wq);
+        if (wq_has_sleeper(wq))
+                wake_up_interruptible_poll(&wq->wait, POLLERR);
         sk_wake_async(sk, SOCK_WAKE_IO, POLL_ERR);
-        read_unlock(&sk->sk_callback_lock);
+        rcu_read_unlock();
 }
 
 static void sock_def_readable(struct sock *sk, int len)
 {
-        read_lock(&sk->sk_callback_lock);
-        if (sk_has_sleeper(sk))
-                wake_up_interruptible_sync_poll(sk_sleep(sk), POLLIN |
+        struct socket_wq *wq;
+
+        rcu_read_lock();
+        wq = rcu_dereference(sk->sk_wq);
+        if (wq_has_sleeper(wq))
+                wake_up_interruptible_sync_poll(&wq->wait, POLLIN |
                                                 POLLRDNORM | POLLRDBAND);
         sk_wake_async(sk, SOCK_WAKE_WAITD, POLL_IN);
-        read_unlock(&sk->sk_callback_lock);
+        rcu_read_unlock();
 }
 
 static void sock_def_write_space(struct sock *sk)
 {
-        read_lock(&sk->sk_callback_lock);
+        struct socket_wq *wq;
+
+        rcu_read_lock();
 
         /* Do not wake up a writer until he can make "significant"
          * progress.  --DaveM
          */
         if ((atomic_read(&sk->sk_wmem_alloc) << 1) <= sk->sk_sndbuf) {
-                if (sk_has_sleeper(sk))
-                        wake_up_interruptible_sync_poll(sk_sleep(sk), POLLOUT |
+                wq = rcu_dereference(sk->sk_wq);
+                if (wq_has_sleeper(wq))
+                        wake_up_interruptible_sync_poll(&wq->wait, POLLOUT |
                                                 POLLWRNORM | POLLWRBAND);
 
                 /* Should agree with poll, otherwise some programs break */
@@ -1842,7 +1854,7 @@ static void sock_def_write_space(struct sock *sk)
                         sk_wake_async(sk, SOCK_WAKE_SPACE, POLL_OUT);
         }
 
-        read_unlock(&sk->sk_callback_lock);
+        rcu_read_unlock();
 }
 
 static void sock_def_destruct(struct sock *sk)
@@ -1896,10 +1908,10 @@ void sock_init_data(struct socket *sock, struct sock *sk)
 
         if (sock) {
                 sk->sk_type     =       sock->type;
-                sk->sk_sleep    =       &sock->wait;
+                sk->sk_wq       =       sock->wq;
                 sock->sk        =       sk;
         } else
-                sk->sk_sleep    =       NULL;
+                sk->sk_wq       =       NULL;
 
         spin_lock_init(&sk->sk_dst_lock);
         rwlock_init(&sk->sk_callback_lock);
diff --git a/net/core/stream.c b/net/core/stream.c
index 7b3c3f30b107..cc196f42b8d8 100644
--- a/net/core/stream.c
+++ b/net/core/stream.c
@@ -28,15 +28,19 @@
 void sk_stream_write_space(struct sock *sk)
 {
         struct socket *sock = sk->sk_socket;
+        struct socket_wq *wq;
 
         if (sk_stream_wspace(sk) >= sk_stream_min_wspace(sk) && sock) {
                 clear_bit(SOCK_NOSPACE, &sock->flags);
 
-                if (sk_sleep(sk) && waitqueue_active(sk_sleep(sk)))
-                        wake_up_interruptible_poll(sk_sleep(sk), POLLOUT |
+                rcu_read_lock();
+                wq = rcu_dereference(sk->sk_wq);
+                if (wq_has_sleeper(wq))
+                        wake_up_interruptible_poll(&wq->wait, POLLOUT |
                                                 POLLWRNORM | POLLWRBAND);
-                if (sock->fasync_list && !(sk->sk_shutdown & SEND_SHUTDOWN))
+                if (wq && wq->fasync_list && !(sk->sk_shutdown & SEND_SHUTDOWN))
                         sock_wake_async(sock, SOCK_WAKE_SPACE, POLL_OUT);
+                rcu_read_unlock();
         }
 }
 
diff --git a/net/dccp/output.c b/net/dccp/output.c
index 2d3dcb39851f..aadbdb58758b 100644
--- a/net/dccp/output.c
+++ b/net/dccp/output.c
@@ -195,15 +195,17 @@ EXPORT_SYMBOL_GPL(dccp_sync_mss);
 
 void dccp_write_space(struct sock *sk)
 {
-        read_lock(&sk->sk_callback_lock);
+        struct socket_wq *wq;
 
-        if (sk_has_sleeper(sk))
-                wake_up_interruptible(sk_sleep(sk));
+        rcu_read_lock();
+        wq = rcu_dereference(sk->sk_wq);
+        if (wq_has_sleeper(wq))
+                wake_up_interruptible(&wq->wait);
         /* Should agree with poll, otherwise some programs break */
         if (sock_writeable(sk))
                 sk_wake_async(sk, SOCK_WAKE_SPACE, POLL_OUT);
 
-        read_unlock(&sk->sk_callback_lock);
+        rcu_read_unlock();
 }
 
 /**
diff --git a/net/iucv/af_iucv.c b/net/iucv/af_iucv.c
index 9636b7d27b48..8be324fe08b9 100644
--- a/net/iucv/af_iucv.c
+++ b/net/iucv/af_iucv.c
@@ -305,11 +305,14 @@ static inline int iucv_below_msglim(struct sock *sk)
  */
 static void iucv_sock_wake_msglim(struct sock *sk)
 {
-        read_lock(&sk->sk_callback_lock);
-        if (sk_has_sleeper(sk))
-                wake_up_interruptible_all(sk_sleep(sk));
+        struct socket_wq *wq;
+
+        rcu_read_lock();
+        wq = rcu_dereference(sk->sk_wq);
+        if (wq_has_sleeper(wq))
+                wake_up_interruptible_all(&wq->wait);
         sk_wake_async(sk, SOCK_WAKE_SPACE, POLL_OUT);
-        read_unlock(&sk->sk_callback_lock);
+        rcu_read_unlock();
 }
 
 /* Timers */
diff --git a/net/phonet/pep.c b/net/phonet/pep.c
index e2a95762abd3..af4d38bc3b22 100644
--- a/net/phonet/pep.c
+++ b/net/phonet/pep.c
@@ -664,12 +664,12 @@ static int pep_wait_connreq(struct sock *sk, int noblock)
                 if (signal_pending(tsk))
                         return sock_intr_errno(timeo);
 
-                prepare_to_wait_exclusive(&sk->sk_socket->wait, &wait,
+                prepare_to_wait_exclusive(sk_sleep(sk), &wait,
                                                 TASK_INTERRUPTIBLE);
                 release_sock(sk);
                 timeo = schedule_timeout(timeo);
                 lock_sock(sk);
-                finish_wait(&sk->sk_socket->wait, &wait);
+                finish_wait(sk_sleep(sk), &wait);
         }
 
         return 0;
@@ -910,10 +910,10 @@ disabled:
                         goto out;
                 }
 
-                prepare_to_wait(&sk->sk_socket->wait, &wait,
+                prepare_to_wait(sk_sleep(sk), &wait,
                                 TASK_INTERRUPTIBLE);
                 done = sk_wait_event(sk, &timeo, atomic_read(&pn->tx_credits));
-                finish_wait(&sk->sk_socket->wait, &wait);
+                finish_wait(sk_sleep(sk), &wait);
 
                 if (sk->sk_state != TCP_ESTABLISHED)
                         goto disabled;
diff --git a/net/phonet/socket.c b/net/phonet/socket.c
index c785bfd0744f..6e9848bf0370 100644
--- a/net/phonet/socket.c
+++ b/net/phonet/socket.c
@@ -265,7 +265,7 @@ static unsigned int pn_socket_poll(struct file *file, struct socket *sock,
         struct pep_sock *pn = pep_sk(sk);
         unsigned int mask = 0;
 
-        poll_wait(file, &sock->wait, wait);
+        poll_wait(file, sk_sleep(sk), wait);
 
         switch (sk->sk_state) {
         case TCP_LISTEN:
diff --git a/net/rxrpc/af_rxrpc.c b/net/rxrpc/af_rxrpc.c
index c432d76f415e..0b9bb2085ce4 100644
--- a/net/rxrpc/af_rxrpc.c
+++ b/net/rxrpc/af_rxrpc.c
@@ -62,13 +62,15 @@ static inline int rxrpc_writable(struct sock *sk)
 static void rxrpc_write_space(struct sock *sk)
 {
         _enter("%p", sk);
-        read_lock(&sk->sk_callback_lock);
+        rcu_read_lock();
         if (rxrpc_writable(sk)) {
-                if (sk_has_sleeper(sk))
-                        wake_up_interruptible(sk_sleep(sk));
+                struct socket_wq *wq = rcu_dereference(sk->sk_wq);
+
+                if (wq_has_sleeper(wq))
+                        wake_up_interruptible(&wq->wait);
                 sk_wake_async(sk, SOCK_WAKE_SPACE, POLL_OUT);
         }
-        read_unlock(&sk->sk_callback_lock);
+        rcu_read_unlock();
 }
 
 /*
diff --git a/net/sctp/socket.c b/net/sctp/socket.c
index 13d8229f3a9c..d54700af927a 100644
--- a/net/sctp/socket.c
+++ b/net/sctp/socket.c
@@ -6065,7 +6065,7 @@ static void __sctp_write_space(struct sctp_association *asoc)
                          * here by modeling from the current TCP/UDP code.
                          * We have not tested with it yet.
                          */
-                        if (sock->fasync_list &&
+                        if (sock->wq->fasync_list &&
                             !(sk->sk_shutdown & SEND_SHUTDOWN))
                                 sock_wake_async(sock,
                                                 SOCK_WAKE_SPACE, POLL_OUT);
diff --git a/net/socket.c b/net/socket.c
index cb7c1f6c0d6e..dae8c6b84a09 100644
--- a/net/socket.c
+++ b/net/socket.c
@@ -252,9 +252,14 @@ static struct inode *sock_alloc_inode(struct super_block *sb)
         ei = kmem_cache_alloc(sock_inode_cachep, GFP_KERNEL);
         if (!ei)
                 return NULL;
-        init_waitqueue_head(&ei->socket.wait);
+        ei->socket.wq = kmalloc(sizeof(struct socket_wq), GFP_KERNEL);
+        if (!ei->socket.wq) {
+                kmem_cache_free(sock_inode_cachep, ei);
+                return NULL;
+        }
+        init_waitqueue_head(&ei->socket.wq->wait);
+        ei->socket.wq->fasync_list = NULL;
 
-        ei->socket.fasync_list = NULL;
         ei->socket.state = SS_UNCONNECTED;
         ei->socket.flags = 0;
         ei->socket.ops = NULL;
@@ -264,10 +269,21 @@ static struct inode *sock_alloc_inode(struct super_block *sb)
         return &ei->vfs_inode;
 }
 
+
+static void wq_free_rcu(struct rcu_head *head)
+{
+        struct socket_wq *wq = container_of(head, struct socket_wq, rcu);
+
+        kfree(wq);
+}
+
 static void sock_destroy_inode(struct inode *inode)
 {
-        kmem_cache_free(sock_inode_cachep,
-                        container_of(inode, struct socket_alloc, vfs_inode));
+        struct socket_alloc *ei;
+
+        ei = container_of(inode, struct socket_alloc, vfs_inode);
+        call_rcu(&ei->socket.wq->rcu, wq_free_rcu);
+        kmem_cache_free(sock_inode_cachep, ei);
 }
 
 static void init_once(void *foo)
@@ -513,7 +529,7 @@ void sock_release(struct socket *sock)
                 module_put(owner);
         }
 
-        if (sock->fasync_list)
+        if (sock->wq->fasync_list)
                 printk(KERN_ERR "sock_release: fasync list not empty!\n");
 
         percpu_sub(sockets_in_use, 1);
@@ -1080,9 +1096,9 @@ static int sock_fasync(int fd, struct file *filp, int on)
 
         lock_sock(sk);
 
-        fasync_helper(fd, filp, on, &sock->fasync_list);
+        fasync_helper(fd, filp, on, &sock->wq->fasync_list);
 
-        if (!sock->fasync_list)
+        if (!sock->wq->fasync_list)
                 sock_reset_flag(sk, SOCK_FASYNC);
         else
                 sock_set_flag(sk, SOCK_FASYNC);
@@ -1091,12 +1107,20 @@ static int sock_fasync(int fd, struct file *filp, int on)
         return 0;
 }
 
-/* This function may be called only under socket lock or callback_lock */
+/* This function may be called only under socket lock or callback_lock or rcu_lock */
 
 int sock_wake_async(struct socket *sock, int how, int band)
 {
-        if (!sock || !sock->fasync_list)
+        struct socket_wq *wq;
+
+        if (!sock)
                 return -1;
+        rcu_read_lock();
+        wq = rcu_dereference(sock->wq);
+        if (!wq || !wq->fasync_list) {
+                rcu_read_unlock();
+                return -1;
+        }
         switch (how) {
         case SOCK_WAKE_WAITD:
                 if (test_bit(SOCK_ASYNC_WAITDATA, &sock->flags))
@@ -1108,11 +1132,12 @@ int sock_wake_async(struct socket *sock, int how, int band)
                 /* fall through */
         case SOCK_WAKE_IO:
 call_kill:
-                kill_fasync(&sock->fasync_list, SIGIO, band);
+                kill_fasync(&wq->fasync_list, SIGIO, band);
                 break;
         case SOCK_WAKE_URG:
-                kill_fasync(&sock->fasync_list, SIGURG, band);
+                kill_fasync(&wq->fasync_list, SIGURG, band);
         }
+        rcu_read_unlock();
         return 0;
 }
 
diff --git a/net/unix/af_unix.c b/net/unix/af_unix.c
index 87c0360eaa25..fef2cc5e9d2b 100644
--- a/net/unix/af_unix.c
+++ b/net/unix/af_unix.c
@@ -313,13 +313,16 @@ static inline int unix_writable(struct sock *sk)
 
 static void unix_write_space(struct sock *sk)
 {
-        read_lock(&sk->sk_callback_lock);
+        struct socket_wq *wq;
+
+        rcu_read_lock();
         if (unix_writable(sk)) {
-                if (sk_has_sleeper(sk))
-                        wake_up_interruptible_sync(sk_sleep(sk));
+                wq = rcu_dereference(sk->sk_wq);
+                if (wq_has_sleeper(wq))
+                        wake_up_interruptible_sync(&wq->wait);
                 sk_wake_async(sk, SOCK_WAKE_SPACE, POLL_OUT);
         }
-        read_unlock(&sk->sk_callback_lock);
+        rcu_read_unlock();
 }
 
 /* When dgram socket disconnects (or changes its peer), we clear its receive
@@ -406,9 +409,7 @@ static int unix_release_sock(struct sock *sk, int embrion)
                                 skpair->sk_err = ECONNRESET;
                         unix_state_unlock(skpair);
                         skpair->sk_state_change(skpair);
-                        read_lock(&skpair->sk_callback_lock);
                         sk_wake_async(skpair, SOCK_WAKE_WAITD, POLL_HUP);
-                        read_unlock(&skpair->sk_callback_lock);
                 }
                 sock_put(skpair); /* It may now die */
                 unix_peer(sk) = NULL;
@@ -1142,7 +1143,7 @@ restart:
         newsk->sk_peercred.pid  = task_tgid_vnr(current);
         current_euid_egid(&newsk->sk_peercred.uid, &newsk->sk_peercred.gid);
         newu = unix_sk(newsk);
-        newsk->sk_sleep         = &newu->peer_wait;
+        newsk->sk_wq            = &newu->peer_wq;
         otheru = unix_sk(other);
 
         /* copy address information from listening to new sock*/
@@ -1931,12 +1932,10 @@ static int unix_shutdown(struct socket *sock, int mode)
                         other->sk_shutdown |= peer_mode;
                         unix_state_unlock(other);
                         other->sk_state_change(other);
-                        read_lock(&other->sk_callback_lock);
                         if (peer_mode == SHUTDOWN_MASK)
                                 sk_wake_async(other, SOCK_WAKE_WAITD, POLL_HUP);
                         else if (peer_mode & RCV_SHUTDOWN)
                                 sk_wake_async(other, SOCK_WAKE_WAITD, POLL_IN);
-                        read_unlock(&other->sk_callback_lock);
                 }
                 if (other)
                         sock_put(other);