Commit c0783818 authored by David Howells's avatar David Howells Committed by Paolo Abeni

rxrpc: Fix overproduction of wakeups to recvmsg()

Fix three cases of overproduction of wakeups:

 (1) rxrpc_input_split_jumbo() conditionally notifies the app that there's
     data for recvmsg() to collect if it queues some data - and then its
     only caller, rxrpc_input_data(), goes and wakes up recvmsg() anyway.

     Fix the rxrpc_input_data() to only do the wakeup in failure cases.

 (2) If a DATA packet is received for a call by the I/O thread whilst
     recvmsg() is busy draining the call's rx queue in the app thread, the
     call will left on the recvmsg() queue for recvmsg() to pick up, even
     though there isn't any data on it.

     This can cause an unexpected recvmsg() with a 0 return and no MSG_EOR
     set after the reply has been posted to a service call.

     Fix this by discarding pending calls from the recvmsg() queue that
     don't need servicing yet.

 (3) Not-yet-completed calls get requeued after having data read from them,
     even if they have no data to read.

     Fix this by only requeuing them if they have data waiting on them; if
     they don't, the I/O thread will requeue them when data arrives or they
     fail.
Signed-off-by: default avatarDavid Howells <dhowells@redhat.com>
cc: Marc Dionne <marc.dionne@auristor.com>
cc: linux-afs@lists.infradead.org
Link: https://lore.kernel.org/r/3386149.1676497685@warthog.procyon.org.ukSigned-off-by: default avatarPaolo Abeni <pabeni@redhat.com>
parent d269ac13
...@@ -318,6 +318,7 @@ ...@@ -318,6 +318,7 @@
EM(rxrpc_recvmsg_return, "RETN") \ EM(rxrpc_recvmsg_return, "RETN") \
EM(rxrpc_recvmsg_terminal, "TERM") \ EM(rxrpc_recvmsg_terminal, "TERM") \
EM(rxrpc_recvmsg_to_be_accepted, "TBAC") \ EM(rxrpc_recvmsg_to_be_accepted, "TBAC") \
EM(rxrpc_recvmsg_unqueue, "UNQU") \
E_(rxrpc_recvmsg_wait, "WAIT") E_(rxrpc_recvmsg_wait, "WAIT")
#define rxrpc_rtt_tx_traces \ #define rxrpc_rtt_tx_traces \
......
...@@ -606,7 +606,7 @@ static void rxrpc_input_data(struct rxrpc_call *call, struct sk_buff *skb) ...@@ -606,7 +606,7 @@ static void rxrpc_input_data(struct rxrpc_call *call, struct sk_buff *skb)
rxrpc_proto_abort(call, sp->hdr.seq, rxrpc_badmsg_bad_jumbo); rxrpc_proto_abort(call, sp->hdr.seq, rxrpc_badmsg_bad_jumbo);
goto out_notify; goto out_notify;
} }
skb = NULL; return;
out_notify: out_notify:
trace_rxrpc_notify_socket(call->debug_id, serial); trace_rxrpc_notify_socket(call->debug_id, serial);
......
...@@ -334,10 +334,23 @@ int rxrpc_recvmsg(struct socket *sock, struct msghdr *msg, size_t len, ...@@ -334,10 +334,23 @@ int rxrpc_recvmsg(struct socket *sock, struct msghdr *msg, size_t len,
/* Find the next call and dequeue it if we're not just peeking. If we /* Find the next call and dequeue it if we're not just peeking. If we
* do dequeue it, that comes with a ref that we will need to release. * do dequeue it, that comes with a ref that we will need to release.
* We also want to weed out calls that got requeued whilst we were
* shovelling data out.
*/ */
spin_lock(&rx->recvmsg_lock); spin_lock(&rx->recvmsg_lock);
l = rx->recvmsg_q.next; l = rx->recvmsg_q.next;
call = list_entry(l, struct rxrpc_call, recvmsg_link); call = list_entry(l, struct rxrpc_call, recvmsg_link);
if (!rxrpc_call_is_complete(call) &&
skb_queue_empty(&call->recvmsg_queue)) {
list_del_init(&call->recvmsg_link);
spin_unlock(&rx->recvmsg_lock);
release_sock(&rx->sk);
trace_rxrpc_recvmsg(call->debug_id, rxrpc_recvmsg_unqueue, 0);
rxrpc_put_call(call, rxrpc_call_put_recvmsg);
goto try_again;
}
if (!(flags & MSG_PEEK)) if (!(flags & MSG_PEEK))
list_del_init(&call->recvmsg_link); list_del_init(&call->recvmsg_link);
else else
...@@ -402,7 +415,8 @@ int rxrpc_recvmsg(struct socket *sock, struct msghdr *msg, size_t len, ...@@ -402,7 +415,8 @@ int rxrpc_recvmsg(struct socket *sock, struct msghdr *msg, size_t len,
if (rxrpc_call_has_failed(call)) if (rxrpc_call_has_failed(call))
goto call_failed; goto call_failed;
rxrpc_notify_socket(call); if (!skb_queue_empty(&call->recvmsg_queue))
rxrpc_notify_socket(call);
goto not_yet_complete; goto not_yet_complete;
call_failed: call_failed:
......
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