[PATCH net v2 09/16] afs: Simplify call refcounting

[David Howells <dhowells@redhat.com>]

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 <jaltman@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
---
 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  ") \