[PATCH net-next 0/4] kcm: Fix two locking issues

[PATCH net-next 0/4] kcm: Fix two locking issues

[Tom Herbert]

One issue is lockdep warnings when sock_owned_by_user returns true
in strparser. Fix is to add and call sock_owned_by_user_nocheck since
the check for owned by user is not an error condition in this case.

The other issue is a potential deadlock between TX and RX paths

KCM socket lock and the psock socket lock are acquired in both
the RX and TX path, however they take the locks in opposite order
which can lead to deadlock. The fix is to add try_sock_lock to see
if psock socket lock can get acquired in the TX path with KCM lock
held. If not, then KCM socket is released and the psock socket lock
and KCM socket lock are acquired in the same order as the RX path.

Tested:

Ran KCM traffic without incident.

Tom Herbert (4):
  sock: Add sock_owned_by_user_nocheck
  strparser: Call sock_owned_by_user_nocheck
  sock_lock: Add try_sock_lock
  kcm: Address deadlock between TX and RX paths

 include/net/kcm.h         |  1 +
 include/net/sock.h        | 12 +++++++++
 net/core/sock.c           | 20 +++++++++++++++
 net/kcm/kcmsock.c         | 64 ++++++++++++++++++++++++++++++++++-------------
 net/strparser/strparser.c |  2 +-
 5 files changed, 81 insertions(+), 18 deletions(-)

-- 
2.11.0

[PATCH net-next 1/4] sock: Add sock_owned_by_user_nocheck

[Tom Herbert]

This allows checking socket lock ownership with producing lockdep
warnings.

Signed-off-by: Tom Herbert <tom@quantonium.net>
---
 include/net/sock.h | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/include/net/sock.h b/include/net/sock.h
index 0a32f3ce381c..3b4ca2046f8c 100644
--- a/include/net/sock.h
+++ b/include/net/sock.h
@@ -1515,6 +1515,11 @@ static inline bool sock_owned_by_user(const struct sock *sk)
 	return sk->sk_lock.owned;
 }
 
+static inline bool sock_owned_by_user_nocheck(const struct sock *sk)
+{
+	return sk->sk_lock.owned;
+}
+
 /* no reclassification while locks are held */
 static inline bool sock_allow_reclassification(const struct sock *csk)
 {
-- 
2.11.0

[PATCH net-next 2/4] strparser: Call sock_owned_by_user_nocheck

[Tom Herbert]

strparser wants to check socket ownership without producing any
warnings. As indicated by the comment in the code, it is permissible
for owned_by_user to return true.

Fixes: 43a0c6751a322847 ("strparser: Stream parser for messages")
Reported-by: syzbot <syzkaller@googlegroups.com>
Signed-off-by: Tom Herbert <tom@quantonium.net>
---
 net/strparser/strparser.c | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/net/strparser/strparser.c b/net/strparser/strparser.c
index c5fda15ba319..1fdab5c4eda8 100644
--- a/net/strparser/strparser.c
+++ b/net/strparser/strparser.c
@@ -401,7 +401,7 @@ void strp_data_ready(struct strparser *strp)
 	 * allows a thread in BH context to safely check if the process
 	 * lock is held. In this case, if the lock is held, queue work.
 	 */
-	if (sock_owned_by_user(strp->sk)) {
+	if (sock_owned_by_user_nocheck(strp->sk)) {
 		queue_work(strp_wq, &strp->work);
 		return;
 	}
-- 
2.11.0

[PATCH net-next 3/4] sock_lock: Add try_sock_lock

[Tom Herbert]

try_sock lock is an opportunistic attempt to acquire a socket lock
without blocking or sleeping. If the socket lock is acquired then
true is returned, else false is returned.

Signed-off-by: Tom Herbert <tom@quantonium.net>
---
 include/net/sock.h |  7 +++++++
 net/core/sock.c    | 20 ++++++++++++++++++++
 2 files changed, 27 insertions(+)

diff --git a/include/net/sock.h b/include/net/sock.h
index 3b4ca2046f8c..69fdd1a89591 100644
--- a/include/net/sock.h
+++ b/include/net/sock.h
@@ -1462,6 +1462,13 @@ static inline void lock_sock(struct sock *sk)
 	lock_sock_nested(sk, 0);
 }
 
+bool try_lock_sock_nested(struct sock *sk, int subclass);
+
+static inline bool try_lock_sock(struct sock *sk)
+{
+	return try_lock_sock_nested(sk, 0);
+}
+
 void release_sock(struct sock *sk);
 
 /* BH context may only use the following locking interface. */
diff --git a/net/core/sock.c b/net/core/sock.c
index 72d14b221784..40fb772e2d52 100644
--- a/net/core/sock.c
+++ b/net/core/sock.c
@@ -2782,6 +2782,26 @@ void lock_sock_nested(struct sock *sk, int subclass)
 }
 EXPORT_SYMBOL(lock_sock_nested);
 
+bool try_lock_sock_nested(struct sock *sk, int subclass)
+{
+	spin_lock_bh(&sk->sk_lock.slock);
+	if (sk->sk_lock.owned) {
+		spin_unlock_bh(&sk->sk_lock.slock);
+		return false;
+	}
+
+	sk->sk_lock.owned = 1;
+	spin_unlock(&sk->sk_lock.slock);
+
+	/* The sk_lock has mutex_lock() semantics here: */
+
+	mutex_acquire(&sk->sk_lock.dep_map, subclass, 0, _RET_IP_);
+	local_bh_enable();
+
+	return true;
+}
+EXPORT_SYMBOL(try_lock_sock_nested);
+
 void release_sock(struct sock *sk)
 {
 	spin_lock_bh(&sk->sk_lock.slock);
-- 
2.11.0

[PATCH net-next 4/4] kcm: Address deadlock between TX and RX paths

[Tom Herbert]

Both the transmit and receive paths of KCM need to take both the
KCM socket lock and the psock socket lock, however they take the
locks in opposite order which can lead to deadlock.

This patch changes the transmit path (kcm_write_msgs to be specific)
so the locks are taken in the proper order. try_sock_lock is first used
to get the lower socket lock, if that is successful then sending data
can proceed with dropping KCM lock. If try_sock_lock fails then the KCM
lock is released and lock_sock is done on the lower socket followed by
the lock_sock on the KCM sock.

A doing_write_msgs flag has been added to kcm structure to prevent
multiple threads doing write_msgs when the KCM lock is dropped.
kernel_sendpage_locked is now called to do the send data with lock
already held.

Signed-off-by: Tom Herbert <tom@quantonium.net>
---
 include/net/kcm.h |  1 +
 net/kcm/kcmsock.c | 64 ++++++++++++++++++++++++++++++++++++++++---------------
 2 files changed, 48 insertions(+), 17 deletions(-)

diff --git a/include/net/kcm.h b/include/net/kcm.h
index 2a8965819db0..22bd7dd3eedb 100644
--- a/include/net/kcm.h
+++ b/include/net/kcm.h
@@ -78,6 +78,7 @@ struct kcm_sock {
 	/* Don't use bit fields here, these are set under different locks */
 	bool tx_wait;
 	bool tx_wait_more;
+	bool doing_write_msgs;
 
 	/* Receive */
 	struct kcm_psock *rx_psock;
diff --git a/net/kcm/kcmsock.c b/net/kcm/kcmsock.c
index d4e98f20fc2a..3eb3179b96b3 100644
--- a/net/kcm/kcmsock.c
+++ b/net/kcm/kcmsock.c
@@ -574,13 +574,19 @@ static void kcm_report_tx_retry(struct kcm_sock *kcm)
 static int kcm_write_msgs(struct kcm_sock *kcm)
 {
 	struct sock *sk = &kcm->sk;
-	struct kcm_psock *psock;
-	struct sk_buff *skb, *head;
-	struct kcm_tx_msg *txm;
+	struct sk_buff *head = skb_peek(&sk->sk_write_queue);
 	unsigned short fragidx, frag_offset;
 	unsigned int sent, total_sent = 0;
+	struct kcm_psock *psock;
+	struct kcm_tx_msg *txm;
+	struct sk_buff *skb;
 	int ret = 0;
 
+	if (unlikely(kcm->doing_write_msgs))
+		return 0;
+
+	kcm->doing_write_msgs = true;
+
 	kcm->tx_wait_more = false;
 	psock = kcm->tx_psock;
 	if (unlikely(psock && psock->tx_stopped)) {
@@ -598,15 +604,36 @@ static int kcm_write_msgs(struct kcm_sock *kcm)
 		return 0;
 	}
 
+try_again:
+	psock = reserve_psock(kcm);
+	if (!psock)
+		goto out_no_release_psock;
+
+	/* Get lock for lower sock */
+	if (!try_lock_sock(psock->sk)) {
+		/* Someone  else is holding the lower sock lock. We need to
+		 * release the KCM lock and get the psock lock first. This is
+		 * needed since the receive path obtains the locks in reverse
+		 * order and we want to avoid deadlock. Note that
+		 * write_msgs can't be reentered when we drop the KCM lock
+		 * since doing_write_msgs is set.
+		 */
+		release_sock(&kcm->sk);
+
+		/* Take locks in order that receive path does */
+		lock_sock(psock->sk);
+		lock_sock(&kcm->sk);
+	}
+
+	/* At this point we have a reserved psock and its lower socket is
+	 * locked.
+	 */
+
 	head = skb_peek(&sk->sk_write_queue);
 	txm = kcm_tx_msg(head);
 
 	if (txm->sent) {
 		/* Send of first skbuff in queue already in progress */
-		if (WARN_ON(!psock)) {
-			ret = -EINVAL;
-			goto out;
-		}
 		sent = txm->sent;
 		frag_offset = txm->frag_offset;
 		fragidx = txm->fragidx;
@@ -615,11 +642,6 @@ static int kcm_write_msgs(struct kcm_sock *kcm)
 		goto do_frag;
 	}
 
-try_again:
-	psock = reserve_psock(kcm);
-	if (!psock)
-		goto out;
-
 	do {
 		skb = head;
 		txm = kcm_tx_msg(head);
@@ -643,11 +665,12 @@ static int kcm_write_msgs(struct kcm_sock *kcm)
 				goto out;
 			}
 
-			ret = kernel_sendpage(psock->sk->sk_socket,
-					      frag->page.p,
-					      frag->page_offset + frag_offset,
-					      frag->size - frag_offset,
-					      MSG_DONTWAIT);
+			ret = kernel_sendpage_locked(psock->sk, frag->page.p,
+						     frag->page_offset +
+							frag_offset,
+						     frag->size -
+							frag_offset,
+						     MSG_DONTWAIT);
 			if (ret <= 0) {
 				if (ret == -EAGAIN) {
 					/* Save state to try again when there's
@@ -669,6 +692,7 @@ static int kcm_write_msgs(struct kcm_sock *kcm)
 				 */
 				kcm_abort_tx_psock(psock, ret ? -ret : EPIPE,
 						   true);
+				release_sock(psock->sk);
 				unreserve_psock(kcm);
 
 				txm->sent = 0;
@@ -704,7 +728,13 @@ static int kcm_write_msgs(struct kcm_sock *kcm)
 		total_sent += sent;
 		KCM_STATS_INCR(psock->stats.tx_msgs);
 	} while ((head = skb_peek(&sk->sk_write_queue)));
+
 out:
+	release_sock(psock->sk);
+
+out_no_release_psock:
+	kcm->doing_write_msgs = false;
+
 	if (!head) {
 		/* Done with all queued messages. */
 		WARN_ON(!skb_queue_empty(&sk->sk_write_queue));
-- 
2.11.0