SUNRPC: Move UDP receive data path into a workqueue context

Now that we've done it for TCP, let's convert UDP as well.

Signed-off-by: Trond Myklebust <trond.myklebust@primarydata.com>

diff --git a/net/sunrpc/xprtsock.c b/net/sunrpc/xprtsock.c
index 58dc90c..df8bdcc 100644 (file)
--- a/net/sunrpc/xprtsock.c
+++ b/net/sunrpc/xprtsock.c
@@ -972,42 +972,36 @@ static void xs_local_data_ready(struct sock *sk)
 }
 
 /**
- * xs_udp_data_ready - "data ready" callback for UDP sockets
- * @sk: socket with data to read
+ * xs_udp_data_read_skb - receive callback for UDP sockets
+ * @xprt: transport
+ * @sk: socket
+ * @skb: skbuff
  *
  */
-static void xs_udp_data_ready(struct sock *sk)
+static void xs_udp_data_read_skb(struct rpc_xprt *xprt,
+               struct sock *sk,
+               struct sk_buff *skb)
 {
        struct rpc_task *task;
-       struct rpc_xprt *xprt;
        struct rpc_rqst *rovr;
-       struct sk_buff *skb;
-       int err, repsize, copied;
+       int repsize, copied;
        u32 _xid;
        __be32 *xp;
 
-       read_lock_bh(&sk->sk_callback_lock);
-       dprintk("RPC:       xs_udp_data_ready...\n");
-       if (!(xprt = xprt_from_sock(sk)))
-               goto out;
-
-       if ((skb = skb_recv_datagram(sk, 0, 1, &err)) == NULL)
-               goto out;
-
        repsize = skb->len - sizeof(struct udphdr);
        if (repsize < 4) {
                dprintk("RPC:       impossible RPC reply size %d!\n", repsize);
-               goto dropit;
+               return;
        }
 
        /* Copy the XID from the skb... */
        xp = skb_header_pointer(skb, sizeof(struct udphdr),
                                sizeof(_xid), &_xid);
        if (xp == NULL)
-               goto dropit;
+               return;
 
        /* Look up and lock the request corresponding to the given XID */
-       spin_lock(&xprt->transport_lock);
+       spin_lock_bh(&xprt->transport_lock);
        rovr = xprt_lookup_rqst(xprt, *xp);
        if (!rovr)
                goto out_unlock;
@@ -1028,10 +1022,54 @@ static void xs_udp_data_ready(struct sock *sk)
        xprt_complete_rqst(task, copied);
 
  out_unlock:
-       spin_unlock(&xprt->transport_lock);
- dropit:
-       skb_free_datagram(sk, skb);
- out:
+       spin_unlock_bh(&xprt->transport_lock);
+}
+
+static void xs_udp_data_receive(struct sock_xprt *transport)
+{
+       struct sk_buff *skb;
+       struct sock *sk;
+       int err;
+
+       mutex_lock(&transport->recv_mutex);
+       sk = transport->inet;
+       if (sk == NULL)
+               goto out;
+       for (;;) {
+               skb = skb_recv_datagram(sk, 0, 1, &err);
+               if (skb == NULL)
+                       break;
+               xs_udp_data_read_skb(&transport->xprt, sk, skb);
+               skb_free_datagram(sk, skb);
+       }
+out:
+       mutex_unlock(&transport->recv_mutex);
+}
+
+static void xs_udp_data_receive_workfn(struct work_struct *work)
+{
+       struct sock_xprt *transport =
+               container_of(work, struct sock_xprt, recv_worker);
+       xs_udp_data_receive(transport);
+}
+
+/**
+ * xs_data_ready - "data ready" callback for UDP sockets
+ * @sk: socket with data to read
+ *
+ */
+static void xs_data_ready(struct sock *sk)
+{
+       struct rpc_xprt *xprt;
+
+       read_lock_bh(&sk->sk_callback_lock);
+       dprintk("RPC:       xs_data_ready...\n");
+       xprt = xprt_from_sock(sk);
+       if (xprt != NULL) {
+               struct sock_xprt *transport = container_of(xprt,
+                               struct sock_xprt, xprt);
+               queue_work(rpciod_workqueue, &transport->recv_worker);
+       }
        read_unlock_bh(&sk->sk_callback_lock);
 }
 
@@ -2094,7 +2132,7 @@ static void xs_udp_finish_connecting(struct rpc_xprt *xprt, struct socket *sock)
                xs_save_old_callbacks(transport, sk);
 
                sk->sk_user_data = xprt;
-               sk->sk_data_ready = xs_udp_data_ready;
+               sk->sk_data_ready = xs_data_ready;
                sk->sk_write_space = xs_udp_write_space;
                sk->sk_allocation = GFP_NOIO;
 
@@ -2811,7 +2849,7 @@ static struct rpc_xprt *xs_setup_udp(struct xprt_create *args)
 
        xprt->timeout = &xs_udp_default_timeout;
 
-       INIT_WORK(&transport->recv_worker, xs_dummy_data_receive_workfn);
+       INIT_WORK(&transport->recv_worker, xs_udp_data_receive_workfn);
        INIT_DELAYED_WORK(&transport->connect_worker, xs_udp_setup_socket);
 
        switch (addr->sa_family) {