Commit f9b2ee71 authored by Trond Myklebust's avatar Trond Myklebust

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: default avatarTrond Myklebust <trond.myklebust@primarydata.com>
parent edc1b01c
...@@ -972,42 +972,36 @@ static void xs_local_data_ready(struct sock *sk) ...@@ -972,42 +972,36 @@ static void xs_local_data_ready(struct sock *sk)
} }
/** /**
* xs_udp_data_ready - "data ready" callback for UDP sockets * xs_udp_data_read_skb - receive callback for UDP sockets
* @sk: socket with data to read * @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_task *task;
struct rpc_xprt *xprt;
struct rpc_rqst *rovr; struct rpc_rqst *rovr;
struct sk_buff *skb; int repsize, copied;
int err, repsize, copied;
u32 _xid; u32 _xid;
__be32 *xp; __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); repsize = skb->len - sizeof(struct udphdr);
if (repsize < 4) { if (repsize < 4) {
dprintk("RPC: impossible RPC reply size %d!\n", repsize); dprintk("RPC: impossible RPC reply size %d!\n", repsize);
goto dropit; return;
} }
/* Copy the XID from the skb... */ /* Copy the XID from the skb... */
xp = skb_header_pointer(skb, sizeof(struct udphdr), xp = skb_header_pointer(skb, sizeof(struct udphdr),
sizeof(_xid), &_xid); sizeof(_xid), &_xid);
if (xp == NULL) if (xp == NULL)
goto dropit; return;
/* Look up and lock the request corresponding to the given XID */ /* 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); rovr = xprt_lookup_rqst(xprt, *xp);
if (!rovr) if (!rovr)
goto out_unlock; goto out_unlock;
...@@ -1028,10 +1022,54 @@ static void xs_udp_data_ready(struct sock *sk) ...@@ -1028,10 +1022,54 @@ static void xs_udp_data_ready(struct sock *sk)
xprt_complete_rqst(task, copied); xprt_complete_rqst(task, copied);
out_unlock: out_unlock:
spin_unlock(&xprt->transport_lock); spin_unlock_bh(&xprt->transport_lock);
dropit: }
skb_free_datagram(sk, skb);
out: 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); read_unlock_bh(&sk->sk_callback_lock);
} }
...@@ -2094,7 +2132,7 @@ static void xs_udp_finish_connecting(struct rpc_xprt *xprt, struct socket *sock) ...@@ -2094,7 +2132,7 @@ static void xs_udp_finish_connecting(struct rpc_xprt *xprt, struct socket *sock)
xs_save_old_callbacks(transport, sk); xs_save_old_callbacks(transport, sk);
sk->sk_user_data = xprt; 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_write_space = xs_udp_write_space;
sk->sk_allocation = GFP_NOIO; sk->sk_allocation = GFP_NOIO;
...@@ -2811,7 +2849,7 @@ static struct rpc_xprt *xs_setup_udp(struct xprt_create *args) ...@@ -2811,7 +2849,7 @@ static struct rpc_xprt *xs_setup_udp(struct xprt_create *args)
xprt->timeout = &xs_udp_default_timeout; 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); INIT_DELAYED_WORK(&transport->connect_worker, xs_udp_setup_socket);
switch (addr->sa_family) { switch (addr->sa_family) {
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment