qrtr_recvmsg() takes lock_sock() before skb_recv_datagram() and keeps it while waiting for a datagram. Any other call on the same socket, such as sendmsg(), bind() or ioctl(), blocks until a packet arrives. Take the lock only once a datagram has been dequeued. The receive queue has its own lock and the enqueue path never takes the socket lock. The lock is still needed around the copy and skb_free_datagram(), since TIOCINQ peeks the queue under lock_sock() only. Testing SOCK_ZAPPED without the lock means a recv() racing with bind() may return -EADDRNOTAVAIL for a port that was just published. There is no ordering between the two calls anyway. Signed-off-by: Chunkai Deng --- net/qrtr/af_qrtr.c | 12 ++++-------- 1 file changed, 4 insertions(+), 8 deletions(-) diff --git a/net/qrtr/af_qrtr.c b/net/qrtr/af_qrtr.c index d02ef9a74c3ca..2d7339338e460 100644 --- a/net/qrtr/af_qrtr.c +++ b/net/qrtr/af_qrtr.c @@ -1034,18 +1034,14 @@ static int qrtr_recvmsg(struct socket *sock, struct msghdr *msg, struct qrtr_cb *cb; int copied, rc; - lock_sock(sk); - - if (sock_flag(sk, SOCK_ZAPPED)) { - release_sock(sk); + if (sock_flag(sk, SOCK_ZAPPED)) return -EADDRNOTAVAIL; - } skb = skb_recv_datagram(sk, flags, &rc); - if (!skb) { - release_sock(sk); + if (!skb) return rc; - } + + lock_sock(sk); cb = (struct qrtr_cb *)skb->cb; copied = skb->len; -- 2.43.0 Since commit 179d0ba0c454 ("net: qrtr: fix error return code of qrtr_sendmsg()"), qrtr_sendmsg() returns -ENOMEM whenever sock_alloc_send_skb() fails. The helper already sets rc on every failure path, so this overwrites the real error. In particular, sock_alloc_send_skb() consumes a pending sk_err, so the ENETRESET set by qrtr_reset_ports() is lost and userspace never learns that a new controller came up. -EAGAIN, -EPIPE and signal errors are hidden as well. Return the error from sock_alloc_send_skb() as is. Signed-off-by: Chunkai Deng --- net/qrtr/af_qrtr.c | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/net/qrtr/af_qrtr.c b/net/qrtr/af_qrtr.c index 2d7339338e460..09c1cb443f5cb 100644 --- a/net/qrtr/af_qrtr.c +++ b/net/qrtr/af_qrtr.c @@ -959,10 +959,8 @@ static int qrtr_sendmsg(struct socket *sock, struct msghdr *msg, size_t len) plen = (len + 3) & ~3; skb = sock_alloc_send_skb(sk, plen + QRTR_HDR_MAX_SIZE, msg->msg_flags & MSG_DONTWAIT, &rc); - if (!skb) { - rc = -ENOMEM; + if (!skb) goto out_node; - } skb_reserve(skb, QRTR_HDR_MAX_SIZE); -- 2.43.0 QRTR uses sock_no_shutdown(), so shutdown() fails with -EOPNOTSUPP and there is no way to wake up a thread blocked in recv() on the socket. Implement it the same way as unix_shutdown(): map SHUT_* to the *_SHUTDOWN bits, set them in sk_shutdown and call sk_state_change(). Do not use lock_sock() here. qrtr_sendmsg() holds it while sleeping in qrtr_tx_wait() until the remote returns flow control credit, which would leave shutdown() stuck in an uninterruptible wait. Take sk_lock.slock to serialize concurrent callers instead, and use WRITE_ONCE() for the lockless readers. Do the same for the store in qrtr_release(). A sender already sleeping in qrtr_tx_wait() is not woken by SHUT_WR; later sendmsg() calls fail with -EPIPE. Signed-off-by: Chunkai Deng --- net/qrtr/af_qrtr.c | 35 +++++++++++++++++++++++++++++++++-- 1 file changed, 33 insertions(+), 2 deletions(-) diff --git a/net/qrtr/af_qrtr.c b/net/qrtr/af_qrtr.c index 09c1cb443f5c..c955c05e71bd 100644 --- a/net/qrtr/af_qrtr.c +++ b/net/qrtr/af_qrtr.c @@ -4,6 +4,7 @@ * Copyright (c) 2013, The Linux Foundation. All rights reserved. */ #include +#include #include #include #include /* For TIOCINQ/OUTQ */ @@ -1193,6 +1194,36 @@ static int qrtr_ioctl(struct socket *sock, unsigned int cmd, unsigned long arg) return rc; } +static int qrtr_shutdown(struct socket *sock, int mode) +{ + struct sock *sk; + + if (mode < SHUT_RD || mode > SHUT_RDWR) + return -EINVAL; + + /* + * This maps: + * SHUT_RD (0) -> RCV_SHUTDOWN (1) + * SHUT_WR (1) -> SEND_SHUTDOWN (2) + * SHUT_RDWR (2) -> SHUTDOWN_MASK (3) + */ + mode++; + + sk = sock->sk; + + /* + * Do not use lock_sock() here: sendmsg() holds it while sleeping on + * remote flow control, which would block shutdown() indefinitely. + * The spinlock only serializes concurrent shutdown() callers. + */ + spin_lock_bh(&sk->sk_lock.slock); + WRITE_ONCE(sk->sk_shutdown, sk->sk_shutdown | mode); + spin_unlock_bh(&sk->sk_lock.slock); + sk->sk_state_change(sk); + + return 0; +} + static int qrtr_release(struct socket *sock) { struct sock *sk = sock->sk; @@ -1204,7 +1235,7 @@ static int qrtr_release(struct socket *sock) lock_sock(sk); ipc = qrtr_sk(sk); - sk->sk_shutdown = SHUTDOWN_MASK; + WRITE_ONCE(sk->sk_shutdown, SHUTDOWN_MASK); if (!sock_flag(sk, SOCK_DEAD)) sk->sk_state_change(sk); @@ -1237,7 +1268,7 @@ static const struct proto_ops qrtr_proto_ops = { .ioctl = qrtr_ioctl, .gettstamp = sock_gettstamp, .poll = datagram_poll, - .shutdown = sock_no_shutdown, + .shutdown = qrtr_shutdown, .release = qrtr_release, .mmap = sock_no_mmap, }; -- 2.43.0