[PATCH net v2 09/16] afs: Simplify call refcounting
From: David Howells <dhowells@redhat.com>
Date: 2026-07-10 19:23:17
Also in:
lkml
Subsystem:
afs filesystem, filesystems (vfs and infrastructure), the rest, tracing · Maintainers:
David Howells, Marc Dionne, Alexander Viro, Christian Brauner, Linus Torvalds, Steven Rostedt, Masami Hiramatsu
Simplify afs_call refcounting so that a queued work item doesn't hold a ref on the call. Rather, for async calls, put the retaining ref when the call completes. For synchronous calls, the caller of afs_make_call() holds its own ref. This means that queuing a call's async work doesn't require a ref to be taken first on a call - and then there's no need to try and revert the ref taken from the context of the rxrpc I/O thread if the call is already queued. Further, the AFS cache manager server RPC handler functions (SRXAFSCB_*) are then called directly from afs_deliver_to_call() rather then being dispatched to a different workqueue. With that, call->work is changed to a function pointer. Also, there is no longer a need to keep an extra ref on an async call and call->drop_ref can be removed. Signed-off-by: David Howells <dhowells@redhat.com> cc: Marc Dionne <marc.dionne@auristor.com> cc: Jeffrey Altman <redacted> 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 --- fs/afs/cmservice.c | 33 ++++++------------ fs/afs/file.c | 10 +++--- fs/afs/internal.h | 18 ++-------- fs/afs/rxrpc.c | 69 +++++++++++--------------------------- include/trace/events/afs.h | 1 + 5 files changed, 39 insertions(+), 92 deletions(-)
diff --git a/fs/afs/cmservice.c b/fs/afs/cmservice.c
index 3c12a0ecee8e..f92bddd4f1e9 100644
--- a/fs/afs/cmservice.c
+++ b/fs/afs/cmservice.c@@ -23,11 +23,11 @@ static int afs_deliver_cb_callback(struct afs_call *); static int afs_deliver_cb_probe_uuid(struct afs_call *); static int afs_deliver_cb_tell_me_about_yourself(struct afs_call *); static void afs_cm_destructor(struct afs_call *); -static void SRXAFSCB_CallBack(struct work_struct *); -static void SRXAFSCB_InitCallBackState(struct work_struct *); -static void SRXAFSCB_Probe(struct work_struct *); -static void SRXAFSCB_ProbeUuid(struct work_struct *); -static void SRXAFSCB_TellMeAboutYourself(struct work_struct *); +static void SRXAFSCB_CallBack(struct afs_call *call); +static void SRXAFSCB_InitCallBackState(struct afs_call *call); +static void SRXAFSCB_Probe(struct afs_call *call); +static void SRXAFSCB_ProbeUuid(struct afs_call *call); +static void SRXAFSCB_TellMeAboutYourself(struct afs_call *call); static int afs_deliver_yfs_cb_callback(struct afs_call *);
@@ -161,10 +161,8 @@ static void afs_abort_service_call(struct afs_call *call, u32 abort_code, int er /* * The server supplied a list of callbacks that it wanted to break. */ -static void SRXAFSCB_CallBack(struct work_struct *work) +static void SRXAFSCB_CallBack(struct afs_call *call) { - struct afs_call *call = container_of(work, struct afs_call, work); - _enter(""); /* We need to break the callbacks before sending the reply as the
@@ -180,7 +178,6 @@ static void SRXAFSCB_CallBack(struct work_struct *work) } afs_send_empty_reply(call); - afs_put_call(call); _leave(""); }
@@ -284,16 +281,13 @@ static int afs_deliver_cb_callback(struct afs_call *call) /* * allow the fileserver to request callback state (re-)initialisation */ -static void SRXAFSCB_InitCallBackState(struct work_struct *work) +static void SRXAFSCB_InitCallBackState(struct afs_call *call) { - struct afs_call *call = container_of(work, struct afs_call, work); - _enter("{%p}", call->server); if (call->server) afs_init_callback_state(call->server); afs_send_empty_reply(call); - afs_put_call(call); _leave(""); }
@@ -374,13 +368,10 @@ static int afs_deliver_cb_init_call_back_state3(struct afs_call *call) /* * allow the fileserver to see if the cache manager is still alive */ -static void SRXAFSCB_Probe(struct work_struct *work) +static void SRXAFSCB_Probe(struct afs_call *call) { - struct afs_call *call = container_of(work, struct afs_call, work); - _enter(""); afs_send_empty_reply(call); - afs_put_call(call); _leave(""); }
@@ -407,9 +398,8 @@ static int afs_deliver_cb_probe(struct afs_call *call) * Allow the fileserver to quickly find out if the cache manager has been * rebooted. */ -static void SRXAFSCB_ProbeUuid(struct work_struct *work) +static void SRXAFSCB_ProbeUuid(struct afs_call *call) { - struct afs_call *call = container_of(work, struct afs_call, work); struct afs_uuid *r = call->request; _enter("");
@@ -419,7 +409,6 @@ static void SRXAFSCB_ProbeUuid(struct work_struct *work) else afs_abort_service_call(call, 1, 1, afs_abort_probeuuid_negative); - afs_put_call(call); _leave(""); }
@@ -481,9 +470,8 @@ static int afs_deliver_cb_probe_uuid(struct afs_call *call) /* * allow the fileserver to ask about the cache manager's capabilities */ -static void SRXAFSCB_TellMeAboutYourself(struct work_struct *work) +static void SRXAFSCB_TellMeAboutYourself(struct afs_call *call) { - struct afs_call *call = container_of(work, struct afs_call, work); int loop; struct {
@@ -515,7 +503,6 @@ static void SRXAFSCB_TellMeAboutYourself(struct work_struct *work) reply.cap.capcount = htonl(1); reply.cap.caps[0] = htonl(AFS_CAP_ERROR_TRANSLATION); afs_send_simple_reply(call, &reply, sizeof(reply)); - afs_put_call(call); _leave(""); }
diff --git a/fs/afs/file.c b/fs/afs/file.c
index 0467742bfeee..35d68f7f498d 100644
--- a/fs/afs/file.c
+++ b/fs/afs/file.c@@ -316,15 +316,17 @@ void afs_fetch_data_async_rx(struct work_struct *work) struct afs_call *call = container_of(work, struct afs_call, async_work); afs_read_receive(call); - afs_put_call(call); + + if (call->state == AFS_CALL_COMPLETE) { + cancel_work(&call->async_work); + afs_put_call(call); + } } void afs_fetch_data_immediate_cancel(struct afs_call *call) { if (call->async) { - afs_get_call(call, afs_call_trace_wake); - if (!queue_work(afs_async_calls, &call->async_work)) - afs_deferred_put_call(call); + queue_work(afs_async_calls, &call->async_work); flush_work(&call->async_work); } }
diff --git a/fs/afs/internal.h b/fs/afs/internal.h
index 0b72a8566299..cd00956bc4cf 100644
--- a/fs/afs/internal.h
+++ b/fs/afs/internal.h@@ -129,7 +129,7 @@ struct afs_call { const struct afs_call_type *type; /* type of call */ wait_queue_head_t waitq; /* processes awaiting completion */ struct work_struct async_work; /* async I/O processor */ - struct work_struct work; /* actual work processor */ + void (*work)(struct afs_call *call); /* Worker function */ struct work_struct free_work; /* Deferred free processor */ struct rxrpc_call *rxcall; /* RxRPC call handle */ struct rxrpc_peer *peer; /* Remote endpoint */
@@ -169,7 +169,6 @@ struct afs_call { unsigned reply_max; /* maximum size of reply */ unsigned count2; /* count used in unmarshalling */ unsigned char unmarshall; /* unmarshalling phase */ - bool drop_ref; /* T if need to drop ref for incoming call */ bool need_attention; /* T if RxRPC poked us */ bool async; /* T if asynchronous */ bool upgrade; /* T to request service upgrade */
@@ -208,7 +207,7 @@ struct afs_call_type { void (*async_rx)(struct work_struct *work); /* Work function */ - void (*work)(struct work_struct *work); + void (*work)(struct afs_call *call); /* Call done function (gets called immediately on success or failure) */ void (*done)(struct afs_call *call);
@@ -1382,7 +1381,6 @@ extern int __net_init afs_open_socket(struct afs_net *); extern void __net_exit afs_close_socket(struct afs_net *); extern void afs_charge_preallocation(struct work_struct *); extern void afs_put_call(struct afs_call *); -void afs_deferred_put_call(struct afs_call *call); void afs_make_call(struct afs_call *call, gfp_t gfp); void afs_deliver_to_call(struct afs_call *call); void afs_wait_for_call_to_complete(struct afs_call *call);
@@ -1495,7 +1493,6 @@ static inline void afs_set_call_complete(struct afs_call *call, int error, u32 remote_abort) { enum afs_call_state state; - bool ok = false; spin_lock_bh(&call->state_lock); state = call->state;
@@ -1505,19 +1502,8 @@ static inline void afs_set_call_complete(struct afs_call *call, call->state = AFS_CALL_COMPLETE; trace_afs_call_state(call, state, AFS_CALL_COMPLETE, error, remote_abort); - ok = true; } spin_unlock_bh(&call->state_lock); - if (ok) { - trace_afs_call_done(call); - - /* Asynchronous calls have two refs to release - one from the alloc and - * one queued with the work item - and we can't just deallocate the - * call because the work item may be queued again. - */ - if (call->drop_ref) - afs_put_call(call); - } } /*
diff --git a/fs/afs/rxrpc.c b/fs/afs/rxrpc.c
index 06c711c75f55..a404b6f0cc7c 100644
--- a/fs/afs/rxrpc.c
+++ b/fs/afs/rxrpc.c@@ -177,7 +177,6 @@ static struct afs_call *afs_alloc_call(struct afs_net *net, call->debug_id = atomic_inc_return(&rxrpc_debug_id); refcount_set(&call->ref, 1); INIT_WORK(&call->async_work, type->async_rx ?: afs_process_async_call); - INIT_WORK(&call->work, call->type->work); INIT_WORK(&call->free_work, afs_deferred_free_worker); init_waitqueue_head(&call->waitq); spin_lock_init(&call->state_lock);
@@ -244,37 +243,6 @@ static void afs_deferred_free_worker(struct work_struct *work) afs_free_call(call); } -/* - * Dispose of a reference on a call, deferring the cleanup to a workqueue - * to avoid lock recursion. - */ -void afs_deferred_put_call(struct afs_call *call) -{ - struct afs_net *net = call->net; - unsigned int debug_id = call->debug_id; - bool zero; - int r, o; - - zero = __refcount_dec_and_test(&call->ref, &r); - o = atomic_read(&net->nr_outstanding_calls); - trace_afs_call(debug_id, afs_call_trace_put, r - 1, o, - __builtin_return_address(0)); - if (zero) - schedule_work(&call->free_work); -} - -/* - * Queue the call for actual work. - */ -static void afs_queue_call_work(struct afs_call *call) -{ - if (call->type->work) { - afs_get_call(call, afs_call_trace_work); - if (!queue_work(afs_wq, &call->work)) - afs_put_call(call); - } -} - /* * allocate a call with flat request and reply buffers */
@@ -375,10 +343,8 @@ void afs_make_call(struct afs_call *call, gfp_t gfp) /* If the call is going to be asynchronous, we need an extra ref for * the call to hold itself so the caller need not hang on to its ref. */ - if (call->async) { + if (call->async) afs_get_call(call, afs_call_trace_get); - call->drop_ref = true; - } /* create a call */ rxcall = rxrpc_kernel_begin_call(call->net->socket, call->peer, call->key,
@@ -479,8 +445,7 @@ void afs_make_call(struct afs_call *call, gfp_t gfp) if (call->rxcall) rxrpc_kernel_shutdown_call(call->net->socket, call->rxcall); if (call->async) { - if (cancel_work_sync(&call->async_work)) - afs_put_call(call); + cancel_work_sync(&call->async_work); afs_set_call_complete(call, ret, 0); }
@@ -566,7 +531,8 @@ void afs_deliver_to_call(struct afs_call *call) switch (ret) { case 0: call->responded = true; - afs_queue_call_work(call); + if (call->work) + call->work(call); if (state == AFS_CALL_CL_PROC_REPLY) { if (call->op) set_bit(AFS_SERVER_FL_MAY_HAVE_CB,
@@ -616,6 +582,7 @@ void afs_deliver_to_call(struct afs_call *call) } done: + trace_afs_call_done(call); if (call->type->done) call->type->done(call); out:
@@ -704,19 +671,16 @@ static void afs_wake_up_async_call(struct sock *sk, struct rxrpc_call *rxcall, unsigned long call_user_ID) { struct afs_call *call = (struct afs_call *)call_user_ID; - int r; trace_afs_notify_call(rxcall, call); call->need_attention = true; - if (__refcount_inc_not_zero(&call->ref, &r)) { - trace_afs_call(call->debug_id, afs_call_trace_wake, r + 1, - atomic_read(&call->net->nr_outstanding_calls), - __builtin_return_address(0)); + trace_afs_call(call->debug_id, afs_call_trace_wake, + refcount_read(&call->ref), + atomic_read(&call->net->nr_outstanding_calls), + __builtin_return_address(0)); - if (!queue_work(afs_async_calls, &call->async_work)) - afs_deferred_put_call(call); - } + queue_work(afs_async_calls, &call->async_work); } /*
@@ -729,12 +693,20 @@ static void afs_process_async_call(struct work_struct *work) _enter(""); + trace_afs_call(call->debug_id, afs_call_trace_async_process, + refcount_read(&call->ref), + atomic_read(&call->net->nr_outstanding_calls), + __builtin_return_address(0)); + if (call->state < AFS_CALL_COMPLETE && call->need_attention) { call->need_attention = false; afs_deliver_to_call(call); } - afs_put_call(call); + if (call->state == AFS_CALL_COMPLETE) { + cancel_work(&call->async_work); + afs_put_call(call); + } _leave(""); }
@@ -760,7 +732,6 @@ void afs_charge_preallocation(struct work_struct *work) if (!call) break; - call->drop_ref = true; call->async = true; call->state = AFS_CALL_SV_AWAIT_OP_ID; init_waitqueue_head(&call->waitq);
@@ -836,7 +807,7 @@ static int afs_deliver_cm_op_id(struct afs_call *call) &call->enctype); trace_afs_cb_call(call); - call->work.func = call->type->work; + call->work = call->type->work; /* pass responsibility for the remainder of this message off to the * cache manager op */
diff --git a/include/trace/events/afs.h b/include/trace/events/afs.h
index cf7218efb861..4f18a2a5b9f6 100644
--- a/include/trace/events/afs.h
+++ b/include/trace/events/afs.h@@ -123,6 +123,7 @@ enum yfs_cm_operation { EM(afs_call_trace_alloc, "ALLOC") \ EM(afs_call_trace_async_abort, "ASYAB") \ EM(afs_call_trace_async_kill, "ASYKL") \ + EM(afs_call_trace_async_process, "ASYPR") \ EM(afs_call_trace_free, "FREE ") \ EM(afs_call_trace_get, "GET ") \ EM(afs_call_trace_put, "PUT ") \