[PATCH net-next 22/32] rxrpc: Split out the call state changing functions into their own file

From: David Howells
Date: Tue Dec 06 2022 - 11:04:58 EST


Split out the functions that change the state of an rxrpc call into their
own file. The idea being to remove anything to do with changing the state
of a call directly from the rxrpc sendmsg() and recvmsg() paths and have
all that done in the I/O thread only, with the ultimate aim of removing the
state lock entirely. Moving the code out of sendmsg.c and recvmsg.c makes
that easier to manage.

Signed-off-by: David Howells <dhowells@xxxxxxxxxx>
cc: Marc Dionne <marc.dionne@xxxxxxxxxxxx>
cc: linux-afs@xxxxxxxxxxxxxxxxxxx
---

net/rxrpc/Makefile | 1 +
net/rxrpc/ar-internal.h | 16 +++++---
net/rxrpc/call_state.c | 92 +++++++++++++++++++++++++++++++++++++++++++++++
net/rxrpc/recvmsg.c | 84 -------------------------------------------
4 files changed, 103 insertions(+), 90 deletions(-)
create mode 100644 net/rxrpc/call_state.c

diff --git a/net/rxrpc/Makefile b/net/rxrpc/Makefile
index e76d3459d78e..ac5caf5a48e1 100644
--- a/net/rxrpc/Makefile
+++ b/net/rxrpc/Makefile
@@ -10,6 +10,7 @@ rxrpc-y := \
call_accept.o \
call_event.o \
call_object.o \
+ call_state.o \
conn_client.o \
conn_event.o \
conn_object.o \
diff --git a/net/rxrpc/ar-internal.h b/net/rxrpc/ar-internal.h
index c0704a983055..a95e161bd980 100644
--- a/net/rxrpc/ar-internal.h
+++ b/net/rxrpc/ar-internal.h
@@ -886,6 +886,16 @@ static inline bool rxrpc_is_client_call(const struct rxrpc_call *call)
return !rxrpc_is_service_call(call);
}

+/*
+ * call_state.c
+ */
+bool __rxrpc_set_call_completion(struct rxrpc_call *, enum rxrpc_call_completion, u32, int);
+bool rxrpc_set_call_completion(struct rxrpc_call *, enum rxrpc_call_completion, u32, int);
+bool __rxrpc_call_completed(struct rxrpc_call *);
+bool rxrpc_call_completed(struct rxrpc_call *);
+bool __rxrpc_abort_call(struct rxrpc_call *, rxrpc_seq_t, u32, int, enum rxrpc_abort_reason);
+bool rxrpc_abort_call(struct rxrpc_call *, rxrpc_seq_t, u32, int, enum rxrpc_abort_reason);
+
/*
* conn_client.c
*/
@@ -1115,12 +1125,6 @@ extern const struct seq_operations rxrpc_local_seq_ops;
* recvmsg.c
*/
void rxrpc_notify_socket(struct rxrpc_call *);
-bool __rxrpc_set_call_completion(struct rxrpc_call *, enum rxrpc_call_completion, u32, int);
-bool rxrpc_set_call_completion(struct rxrpc_call *, enum rxrpc_call_completion, u32, int);
-bool __rxrpc_call_completed(struct rxrpc_call *);
-bool rxrpc_call_completed(struct rxrpc_call *);
-bool __rxrpc_abort_call(struct rxrpc_call *, rxrpc_seq_t, u32, int, enum rxrpc_abort_reason);
-bool rxrpc_abort_call(struct rxrpc_call *, rxrpc_seq_t, u32, int, enum rxrpc_abort_reason);
int rxrpc_recvmsg(struct socket *, struct msghdr *, size_t, int);

/*
diff --git a/net/rxrpc/call_state.c b/net/rxrpc/call_state.c
new file mode 100644
index 000000000000..24d240773a6f
--- /dev/null
+++ b/net/rxrpc/call_state.c
@@ -0,0 +1,92 @@
+// SPDX-License-Identifier: GPL-2.0-or-later
+/* Call state changing functions.
+ *
+ * Copyright (C) 2022 Red Hat, Inc. All Rights Reserved.
+ * Written by David Howells (dhowells@xxxxxxxxxx)
+ */
+
+#include "ar-internal.h"
+
+/*
+ * Transition a call to the complete state.
+ */
+bool __rxrpc_set_call_completion(struct rxrpc_call *call,
+ enum rxrpc_call_completion compl,
+ u32 abort_code,
+ int error)
+{
+ if (call->state < RXRPC_CALL_COMPLETE) {
+ call->abort_code = abort_code;
+ call->error = error;
+ call->completion = compl;
+ /* Allow reader of completion state to operate locklessly */
+ smp_store_release(&call->state, RXRPC_CALL_COMPLETE);
+ trace_rxrpc_call_complete(call);
+ wake_up(&call->waitq);
+ rxrpc_notify_socket(call);
+ return true;
+ }
+ return false;
+}
+
+bool rxrpc_set_call_completion(struct rxrpc_call *call,
+ enum rxrpc_call_completion compl,
+ u32 abort_code,
+ int error)
+{
+ bool ret = false;
+
+ if (call->state < RXRPC_CALL_COMPLETE) {
+ spin_lock(&call->state_lock);
+ ret = __rxrpc_set_call_completion(call, compl, abort_code, error);
+ spin_unlock(&call->state_lock);
+ }
+ return ret;
+}
+
+/*
+ * Record that a call successfully completed.
+ */
+bool __rxrpc_call_completed(struct rxrpc_call *call)
+{
+ return __rxrpc_set_call_completion(call, RXRPC_CALL_SUCCEEDED, 0, 0);
+}
+
+bool rxrpc_call_completed(struct rxrpc_call *call)
+{
+ bool ret = false;
+
+ if (call->state < RXRPC_CALL_COMPLETE) {
+ spin_lock(&call->state_lock);
+ ret = __rxrpc_call_completed(call);
+ spin_unlock(&call->state_lock);
+ }
+ return ret;
+}
+
+/*
+ * Record that a call is locally aborted.
+ */
+bool __rxrpc_abort_call(struct rxrpc_call *call, rxrpc_seq_t seq,
+ u32 abort_code, int error,
+ enum rxrpc_abort_reason why)
+{
+ trace_rxrpc_abort(call->debug_id, why, call->cid, call->call_id, seq,
+ abort_code, error);
+ return __rxrpc_set_call_completion(call, RXRPC_CALL_LOCALLY_ABORTED,
+ abort_code, error);
+}
+
+bool rxrpc_abort_call(struct rxrpc_call *call, rxrpc_seq_t seq,
+ u32 abort_code, int error,
+ enum rxrpc_abort_reason why)
+{
+ bool ret;
+
+ spin_lock(&call->state_lock);
+ ret = __rxrpc_abort_call(call, seq, abort_code, error, why);
+ spin_unlock(&call->state_lock);
+ if (ret)
+ rxrpc_send_abort_packet(call);
+ return ret;
+}
diff --git a/net/rxrpc/recvmsg.c b/net/rxrpc/recvmsg.c
index b7178f2150c0..c163c103aa1d 100644
--- a/net/rxrpc/recvmsg.c
+++ b/net/rxrpc/recvmsg.c
@@ -58,90 +58,6 @@ void rxrpc_notify_socket(struct rxrpc_call *call)
_leave("");
}

-/*
- * Transition a call to the complete state.
- */
-bool __rxrpc_set_call_completion(struct rxrpc_call *call,
- enum rxrpc_call_completion compl,
- u32 abort_code,
- int error)
-{
- if (call->state < RXRPC_CALL_COMPLETE) {
- call->abort_code = abort_code;
- call->error = error;
- call->completion = compl;
- /* Allow reader of completion state to operate locklessly */
- smp_store_release(&call->state, RXRPC_CALL_COMPLETE);
- trace_rxrpc_call_complete(call);
- wake_up(&call->waitq);
- rxrpc_notify_socket(call);
- return true;
- }
- return false;
-}
-
-bool rxrpc_set_call_completion(struct rxrpc_call *call,
- enum rxrpc_call_completion compl,
- u32 abort_code,
- int error)
-{
- bool ret = false;
-
- if (call->state < RXRPC_CALL_COMPLETE) {
- spin_lock(&call->state_lock);
- ret = __rxrpc_set_call_completion(call, compl, abort_code, error);
- spin_unlock(&call->state_lock);
- }
- return ret;
-}
-
-/*
- * Record that a call successfully completed.
- */
-bool __rxrpc_call_completed(struct rxrpc_call *call)
-{
- return __rxrpc_set_call_completion(call, RXRPC_CALL_SUCCEEDED, 0, 0);
-}
-
-bool rxrpc_call_completed(struct rxrpc_call *call)
-{
- bool ret = false;
-
- if (call->state < RXRPC_CALL_COMPLETE) {
- spin_lock(&call->state_lock);
- ret = __rxrpc_call_completed(call);
- spin_unlock(&call->state_lock);
- }
- return ret;
-}
-
-/*
- * Record that a call is locally aborted.
- */
-bool __rxrpc_abort_call(struct rxrpc_call *call, rxrpc_seq_t seq,
- u32 abort_code, int error,
- enum rxrpc_abort_reason why)
-{
- trace_rxrpc_abort(call->debug_id, why, call->cid, call->call_id, seq,
- abort_code, error);
- return __rxrpc_set_call_completion(call, RXRPC_CALL_LOCALLY_ABORTED,
- abort_code, error);
-}
-
-bool rxrpc_abort_call(struct rxrpc_call *call, rxrpc_seq_t seq,
- u32 abort_code, int error,
- enum rxrpc_abort_reason why)
-{
- bool ret;
-
- spin_lock(&call->state_lock);
- ret = __rxrpc_abort_call(call, seq, abort_code, error, why);
- spin_unlock(&call->state_lock);
- if (ret)
- rxrpc_send_abort_packet(call);
- return ret;
-}
-
/*
* Pass a call terminating message to userspace.
*/