Re: [PATCH net] rxrpc: Take write lock when publishing the initial RxGK key

[David Howells <dhowells@redhat.com>]

Chengfeng Ye <nicoyip.dev@gmail.com> wrote:

> On a client connection, a second sendmsg can observe
> RXRPC_CONN_CLIENT through a lockless load of conn->state,

I wonder if I need something like the attached also...  Or if it might do
instead, though I think there's no harm in doing your writelock suggestion
anyway.

David
---
commit 21a0c0765f5d2654379f3a0065b059e676d7c761
Author: David Howells <dhowells@redhat.com>
Date:   Mon Sep 14 14:40:13 2026 +0100

    rxrpc: Fix lack of conn->state barriering
    
    Because rxrpc can manipulate the connection state in one thread and then
    read it in another, but it also guards access to some members in the
    struct, it needs to have release/acquire barriers.  Fix this by using
    helpers to read/set the connection state.
    
    Fixes: 9d35d880e0e4 ("rxrpc: Move client call connection to the I/O thread")
    Link: https://netdev-ai.bots.linux.dev/sashiko/#/patchset/20260907113743.1453210-1-dhowells%40redhat.com
    Signed-off-by: David Howells <dhowells@redhat.com>
    cc: Marc Dionne <marc.dionne@auristor.com>
    cc: Eric Dumazet <edumazet@google.com>
    cc: "David S. Miller" <davem@davemloft.net>
    cc: Jakub Kicinski <kuba@kernel.org>
    cc: Paolo Abeni <pabeni@redhat.com>
    cc: Simon Horman <horms@kernel.org>
    cc: linux-afs@lists.infradead.org

diff --git a/net/rxrpc/ar-internal.h b/net/rxrpc/ar-internal.h
index cb36a709f540..dde262ffc7fe 100644
--- a/net/rxrpc/ar-internal.h
+++ b/net/rxrpc/ar-internal.h
@@ -601,7 +601,7 @@ struct rxrpc_connection {
 	unsigned long		events;
 	unsigned long		idle_timestamp;	/* Time at which last became idle */
 	spinlock_t		state_lock;	/* state-change lock */
-	enum rxrpc_conn_proto_state state;	/* current state of connection */
+	enum rxrpc_conn_proto_state _state;	/* current state of connection */
 	enum rxrpc_call_completion completion;	/* Completion condition */
 	s32			abort_code;	/* Abort code of connection abort */
 	int			debug_id;	/* debug ID for printks */
@@ -1185,10 +1185,23 @@ void rxrpc_process_delayed_final_acks(struct rxrpc_connection *, bool);
 bool rxrpc_input_conn_packet(struct rxrpc_connection *conn, struct sk_buff *skb);
 void rxrpc_input_conn_event(struct rxrpc_connection *conn, struct sk_buff *skb);
 
+static inline void rxrpc_set_conn_state(struct rxrpc_connection *conn,
+					enum rxrpc_conn_proto_state state)
+{
+	/* Order write of conn info before write of state. */
+	smp_store_release(&conn->_state, state);
+}
+
+static inline
+enum rxrpc_conn_proto_state rxrpc_conn_state(const struct rxrpc_connection *conn)
+{
+	/* Order read of state before read of conn info. */
+	return smp_load_acquire(&conn->_state);
+}
+
 static inline bool rxrpc_is_conn_aborted(const struct rxrpc_connection *conn)
 {
-	/* Order reading the abort info after the state check. */
-	return smp_load_acquire(&conn->state) == RXRPC_CONN_ABORTED;
+	return rxrpc_conn_state(conn) == RXRPC_CONN_ABORTED;
 }
 
 /*
diff --git a/net/rxrpc/call_accept.c b/net/rxrpc/call_accept.c
index 47824120f1da..1dcfd9e5fca4 100644
--- a/net/rxrpc/call_accept.c
+++ b/net/rxrpc/call_accept.c
@@ -398,8 +398,8 @@ bool rxrpc_new_incoming_call(struct rxrpc_local *local,
 		rx->app_ops->notify_new_call(&rx->sk, call, call->user_call_ID);
 
 	spin_lock(&conn->state_lock);
-	if (conn->state == RXRPC_CONN_SERVICE_UNSECURED) {
-		conn->state = RXRPC_CONN_SERVICE_CHALLENGING;
+	if (rxrpc_conn_state(conn) == RXRPC_CONN_SERVICE_UNSECURED) {
+		rxrpc_set_conn_state(conn, RXRPC_CONN_SERVICE_CHALLENGING);
 		set_bit(RXRPC_CONN_EV_CHALLENGE, &call->conn->events);
 		rxrpc_queue_conn(call->conn, rxrpc_conn_queue_challenge);
 	}
diff --git a/net/rxrpc/call_object.c b/net/rxrpc/call_object.c
index 817ed9acb91e..29f9a01394c1 100644
--- a/net/rxrpc/call_object.c
+++ b/net/rxrpc/call_object.c
@@ -459,7 +459,7 @@ void rxrpc_incoming_call(struct rxrpc_sock *rx,
 
 	spin_lock(&conn->state_lock);
 
-	switch (conn->state) {
+	switch (rxrpc_conn_state(conn)) {
 	case RXRPC_CONN_SERVICE_UNSECURED:
 	case RXRPC_CONN_SERVICE_CHALLENGING:
 		__set_bit(RXRPC_CALL_CONN_CHALLENGING, &call->flags);
diff --git a/net/rxrpc/conn_client.c b/net/rxrpc/conn_client.c
index 48519f0de185..3055ef6fd11d 100644
--- a/net/rxrpc/conn_client.c
+++ b/net/rxrpc/conn_client.c
@@ -182,11 +182,12 @@ rxrpc_alloc_client_connection(struct rxrpc_bundle *bundle)
 	conn->upgrade		= bundle->upgrade;
 	conn->orig_service_id	= bundle->service_id;
 	conn->security_level	= bundle->security_level;
-	conn->state		= RXRPC_CONN_CLIENT_UNSECURED;
 	conn->service_id	= conn->orig_service_id;
 
 	if (conn->security == &rxrpc_no_security)
-		conn->state	= RXRPC_CONN_CLIENT;
+		rxrpc_set_conn_state(conn, RXRPC_CONN_CLIENT);
+	else
+		rxrpc_set_conn_state(conn, RXRPC_CONN_CLIENT_UNSECURED);
 
 	atomic_inc(&rxnet->nr_conns);
 	write_lock(&rxnet->conn_lock);
@@ -206,6 +207,7 @@ rxrpc_alloc_client_connection(struct rxrpc_bundle *bundle)
 static bool rxrpc_may_reuse_conn(struct rxrpc_connection *conn)
 {
 	struct rxrpc_net *rxnet;
+	enum rxrpc_conn_proto_state state;
 	int id_cursor, id, distance, limit;
 
 	if (!conn)
@@ -215,8 +217,9 @@ static bool rxrpc_may_reuse_conn(struct rxrpc_connection *conn)
 	if (test_bit(RXRPC_CONN_DONT_REUSE, &conn->flags))
 		goto dont_reuse;
 
-	if ((conn->state != RXRPC_CONN_CLIENT_UNSECURED &&
-	     conn->state != RXRPC_CONN_CLIENT) ||
+	state = rxrpc_conn_state(conn);
+	if ((state != RXRPC_CONN_CLIENT_UNSECURED &&
+	     state != RXRPC_CONN_CLIENT) ||
 	    conn->proto.epoch != rxnet->epoch)
 		goto mark_dont_reuse;
 
diff --git a/net/rxrpc/conn_event.c b/net/rxrpc/conn_event.c
index 611c790bc6d0..f60bceff0bad 100644
--- a/net/rxrpc/conn_event.c
+++ b/net/rxrpc/conn_event.c
@@ -25,14 +25,13 @@ static bool rxrpc_set_conn_aborted(struct rxrpc_connection *conn,
 {
 	bool aborted = false;
 
-	if (conn->state != RXRPC_CONN_ABORTED) {
+	if (rxrpc_conn_state(conn) != RXRPC_CONN_ABORTED) {
 		spin_lock_irq(&conn->state_lock);
-		if (conn->state != RXRPC_CONN_ABORTED) {
+		if (rxrpc_conn_state(conn) != RXRPC_CONN_ABORTED) {
 			conn->abort_code = abort_code;
 			conn->error	 = err;
 			conn->completion = compl;
-			/* Order the abort info before the state change. */
-			smp_store_release(&conn->state, RXRPC_CONN_ABORTED);
+			rxrpc_set_conn_state(conn, RXRPC_CONN_ABORTED);
 			set_bit(RXRPC_CONN_DONT_REUSE, &conn->flags);
 			set_bit(RXRPC_CONN_EV_ABORT_CALLS, &conn->events);
 			aborted = true;
@@ -272,7 +271,7 @@ static int rxrpc_process_event(struct rxrpc_connection *conn,
 	bool secured = false;
 	int ret;
 
-	if (conn->state == RXRPC_CONN_ABORTED)
+	if (rxrpc_conn_state(conn) == RXRPC_CONN_ABORTED)
 		return -ECONNABORTED;
 
 	_enter("{%d},{%u,%%%u},", conn->debug_id, sp->hdr.type, sp->hdr.serial);
@@ -286,7 +285,7 @@ static int rxrpc_process_event(struct rxrpc_connection *conn,
 
 	case RXRPC_PACKET_TYPE_RESPONSE:
 		spin_lock_irq(&conn->state_lock);
-		if (conn->state != RXRPC_CONN_SERVICE_CHALLENGING) {
+		if (rxrpc_conn_state(conn) != RXRPC_CONN_SERVICE_CHALLENGING) {
 			spin_unlock_irq(&conn->state_lock);
 			return 0;
 		}
@@ -302,8 +301,8 @@ static int rxrpc_process_event(struct rxrpc_connection *conn,
 			return ret;
 
 		spin_lock_irq(&conn->state_lock);
-		if (conn->state == RXRPC_CONN_SERVICE_CHALLENGING) {
-			conn->state = RXRPC_CONN_SERVICE;
+		if (rxrpc_conn_state(conn) == RXRPC_CONN_SERVICE_CHALLENGING) {
+			rxrpc_set_conn_state(conn, RXRPC_CONN_SERVICE);
 			secured = true;
 		}
 		spin_unlock_irq(&conn->state_lock);
@@ -548,7 +547,7 @@ void rxrpc_input_conn_event(struct rxrpc_connection *conn, struct sk_buff *skb)
 		conn->tx_response = NULL;
 		spin_unlock_irq(&conn->local->lock);
 
-		if (conn->state != RXRPC_CONN_ABORTED)
+		if (rxrpc_conn_state(conn) != RXRPC_CONN_ABORTED)
 			rxrpc_send_response(conn, skb);
 		rxrpc_free_skb(skb, rxrpc_skb_put_response);
 	}
@@ -556,7 +555,7 @@ void rxrpc_input_conn_event(struct rxrpc_connection *conn, struct sk_buff *skb)
 	if (skb) {
 		switch (skb->mark) {
 		case RXRPC_SKB_MARK_SERVICE_CONN_SECURED:
-			if (conn->state != RXRPC_CONN_SERVICE)
+			if (rxrpc_conn_state(conn) != RXRPC_CONN_SERVICE)
 				break;
 
 			for (loop = 0; loop < RXRPC_MAXCALLS; loop++)
diff --git a/net/rxrpc/conn_object.c b/net/rxrpc/conn_object.c
index 1be50e0c9cee..12914fb72345 100644
--- a/net/rxrpc/conn_object.c
+++ b/net/rxrpc/conn_object.c
@@ -407,7 +407,7 @@ void rxrpc_service_connection_reaper(struct work_struct *work)
 		ASSERTCMP(atomic_read(&conn->active), >=, 0);
 		if (likely(atomic_read(&conn->active) > 0))
 			continue;
-		if (conn->state == RXRPC_CONN_SERVICE_PREALLOC)
+		if (rxrpc_conn_state(conn) == RXRPC_CONN_SERVICE_PREALLOC)
 			continue;
 
 		if (rxnet->live && !conn->local->dead) {
diff --git a/net/rxrpc/conn_service.c b/net/rxrpc/conn_service.c
index 39c908a3ca6e..04a8206f3a44 100644
--- a/net/rxrpc/conn_service.c
+++ b/net/rxrpc/conn_service.c
@@ -126,7 +126,7 @@ struct rxrpc_connection *rxrpc_prealloc_service_connection(struct rxrpc_net *rxn
 		/* We maintain an extra ref on the connection whilst it is on
 		 * the rxrpc_connections list.
 		 */
-		conn->state = RXRPC_CONN_SERVICE_PREALLOC;
+		rxrpc_set_conn_state(conn, RXRPC_CONN_SERVICE_PREALLOC);
 		refcount_set(&conn->ref, 2);
 
 		atomic_inc(&rxnet->nr_conns);
@@ -161,10 +161,6 @@ void rxrpc_new_incoming_connection(struct rxrpc_sock *rx,
 	conn->security_ix	= sp->hdr.securityIndex;
 	conn->out_clientflag	= 0;
 	conn->security		= sec;
-	if (conn->security_ix)
-		conn->state	= RXRPC_CONN_SERVICE_UNSECURED;
-	else
-		conn->state	= RXRPC_CONN_SERVICE;
 
 	/* See if we should upgrade the service.  This can only happen on the
 	 * first packet on a new connection.  Once done, it applies to all
@@ -174,6 +170,11 @@ void rxrpc_new_incoming_connection(struct rxrpc_sock *rx,
 	    conn->service_id == rx->service_upgrade.from)
 		conn->service_id = rx->service_upgrade.to;
 
+	if (conn->security_ix)
+		rxrpc_set_conn_state(conn, RXRPC_CONN_SERVICE_UNSECURED);
+	else
+		rxrpc_set_conn_state(conn, RXRPC_CONN_SERVICE);
+
 	atomic_set(&conn->active, 1);
 
 	/* Make the connection a target for incoming packets. */
diff --git a/net/rxrpc/proc.c b/net/rxrpc/proc.c
index e9a27fa7b25d..99d2c850b94e 100644
--- a/net/rxrpc/proc.c
+++ b/net/rxrpc/proc.c
@@ -146,6 +146,7 @@ static int rxrpc_connection_seq_show(struct seq_file *seq, void *v)
 	struct rxrpc_connection *conn;
 	struct rxrpc_net *rxnet = rxrpc_net(seq_file_net(seq));
 	const char *state;
+	enum rxrpc_conn_proto_state cstate;
 	char lbuff[RXRPC_PROC_ADDRBUF_SIZE], rbuff[RXRPC_PROC_ADDRBUF_SIZE];
 
 	if (v == &rxnet->conn_proc_list) {
@@ -159,7 +160,8 @@ static int rxrpc_connection_seq_show(struct seq_file *seq, void *v)
 	}
 
 	conn = list_entry(v, struct rxrpc_connection, proc_link);
-	if (conn->state == RXRPC_CONN_SERVICE_PREALLOC) {
+	cstate = rxrpc_conn_state(conn);
+	if (cstate == RXRPC_CONN_SERVICE_PREALLOC) {
 		strcpy(lbuff, "no_local");
 		strcpy(rbuff, "no_connection");
 		goto print;
@@ -168,9 +170,9 @@ static int rxrpc_connection_seq_show(struct seq_file *seq, void *v)
 	scnprintf(lbuff, sizeof(lbuff), "%pISpc", &conn->local->srx.transport);
 	scnprintf(rbuff, sizeof(rbuff), "%pISpc", &conn->peer->srx.transport);
 print:
-	state = rxrpc_is_conn_aborted(conn) ?
+	state = (cstate == RXRPC_CONN_ABORTED) ?
 		rxrpc_call_completions[conn->completion] :
-		rxrpc_conn_states[conn->state];
+		rxrpc_conn_states[cstate];
 	seq_printf(seq,
 		   "UDP   %-47.47s %-47.47s %4x %08x %s %3u %3d"
 		   " %s %08x %08x %08x %08x %08x %08x %08x\n",
diff --git a/net/rxrpc/security.c b/net/rxrpc/security.c
index 2bfbf2b2bb37..f64acaa6cf53 100644
--- a/net/rxrpc/security.c
+++ b/net/rxrpc/security.c
@@ -114,12 +114,12 @@ int rxrpc_init_client_conn_security(struct rxrpc_connection *conn)
 
 found:
 	mutex_lock(&conn->security_lock);
-	if (conn->state == RXRPC_CONN_CLIENT_UNSECURED) {
+	if (rxrpc_conn_state(conn) == RXRPC_CONN_CLIENT_UNSECURED) {
 		ret = conn->security->init_connection_security(conn, token);
 		if (ret == 0) {
 			spin_lock_irq(&conn->state_lock);
-			if (conn->state == RXRPC_CONN_CLIENT_UNSECURED)
-				conn->state = RXRPC_CONN_CLIENT;
+			if (rxrpc_conn_state(conn) == RXRPC_CONN_CLIENT_UNSECURED)
+				rxrpc_set_conn_state(conn, RXRPC_CONN_CLIENT);
 			spin_unlock_irq(&conn->state_lock);
 		}
 	}
diff --git a/net/rxrpc/sendmsg.c b/net/rxrpc/sendmsg.c
index ed7ff32da184..dcbd2033ca01 100644
--- a/net/rxrpc/sendmsg.c
+++ b/net/rxrpc/sendmsg.c
@@ -336,7 +336,7 @@ static int rxrpc_send_data(struct rxrpc_sock *rx,
 	if (ret < 0)
 		goto out_unlock;
 
-	if (call->conn->state == RXRPC_CONN_CLIENT_UNSECURED) {
+	if (rxrpc_conn_state(call->conn) == RXRPC_CONN_CLIENT_UNSECURED) {
 		ret = rxrpc_init_client_conn_security(call->conn);
 		if (ret < 0)
 			goto out_unlock;