Merge branch 'afs' of git://git.kernel.org/pub/scm/linux/kernel/git/dhowells/linux-fs

Pull AFS fixes and cleanups from David Howells:
 "Here are some patches to the AFS filesystem:

  1) Fix problems in the clean-up parts of the cache manager service
     handler.

  2) Split afs_end_call() introduced in (1) and replace some identical
     code elsewhere with a call to the first half of the split function.

  3) Fix an error introduced in the workqueue PREPARE_WORK() elimination
     commits.

  4) Clean up argument passing to functions called from the workqueue as
     there's now an insulating layer between them and the workqueue.
     This is possible from (3)"

* 'afs' of git://git.kernel.org/pub/scm/linux/kernel/git/dhowells/linux-fs:
  AFS: Pass an afs_call* to call->async_workfn() instead of a work_struct*
  AFS: Fix kafs module unloading
  AFS: Part of afs_end_call() is identical to code elsewhere, so split it
  AFS: Fix cache manager service handlers

3 files changed, 63 insertions, 44 deletions

diff --git a/fs/afs/cmservice.c b/fs/afs/cmservice.c
index 1c8c6cc6de30..4b0eff6da674 100644
--- a/fs/afs/cmservice.c
+++ b/fs/afs/cmservice.c
@@ -130,6 +130,15 @@ static void afs_cm_destructor(struct afs_call *call)
 {
         _enter("");
 
+        /* Break the callbacks here so that we do it after the final ACK is
+         * received.  The step number here must match the final number in
+         * afs_deliver_cb_callback().
+         */
+        if (call->unmarshall == 6) {
+                ASSERT(call->server && call->count && call->request);
+                afs_break_callbacks(call->server, call->count, call->request);
+        }
+
         afs_put_server(call->server);
         call->server = NULL;
         kfree(call->buffer);
@@ -272,6 +281,16 @@ static int afs_deliver_cb_callback(struct afs_call *call, struct sk_buff *skb,
                 _debug("trailer");
                 if (skb->len != 0)
                         return -EBADMSG;
+
+                /* Record that the message was unmarshalled successfully so
+                 * that the call destructor can know do the callback breaking
+                 * work, even if the final ACK isn't received.
+                 *
+                 * If the step number changes, then afs_cm_destructor() must be
+                 * updated also.
+                 */
+                call->unmarshall++;
+        case 6:
                 break;
         }
 
diff --git a/fs/afs/internal.h b/fs/afs/internal.h
index be75b500005d..590b55f46d61 100644
--- a/fs/afs/internal.h
+++ b/fs/afs/internal.h
@@ -75,7 +75,7 @@ struct afs_call {
         const struct afs_call_type *type;       /* type of call */
         const struct afs_wait_mode *wait_mode;  /* completion wait mode */
         wait_queue_head_t       waitq;          /* processes awaiting completion */
-        work_func_t             async_workfn;
+        void (*async_workfn)(struct afs_call *call); /* asynchronous work function */
         struct work_struct      async_work;     /* asynchronous work processor */
         struct work_struct      work;           /* actual work processor */
         struct sk_buff_head     rx_queue;       /* received packets */
diff --git a/fs/afs/rxrpc.c b/fs/afs/rxrpc.c
index ef943df73b8c..03a3beb17004 100644
--- a/fs/afs/rxrpc.c
+++ b/fs/afs/rxrpc.c
@@ -25,7 +25,7 @@ static void afs_wake_up_call_waiter(struct afs_call *);
 static int afs_wait_for_call_to_complete(struct afs_call *);
 static void afs_wake_up_async_call(struct afs_call *);
 static int afs_dont_wait_for_call_to_complete(struct afs_call *);
-static void afs_process_async_call(struct work_struct *);
+static void afs_process_async_call(struct afs_call *);
 static void afs_rx_interceptor(struct sock *, unsigned long, struct sk_buff *);
 static int afs_deliver_cm_op_id(struct afs_call *, struct sk_buff *, bool);
 
@@ -58,6 +58,13 @@ static void afs_collect_incoming_call(struct work_struct *);
 static struct sk_buff_head afs_incoming_calls;
 static DECLARE_WORK(afs_collect_incoming_call_work, afs_collect_incoming_call);
 
+static void afs_async_workfn(struct work_struct *work)
+{
+        struct afs_call *call = container_of(work, struct afs_call, async_work);
+
+        call->async_workfn(call);
+}
+
 /*
  * open an RxRPC socket and bind it to be a server for callback notifications
  * - the socket is left in blocking mode and non-blocking ops use MSG_DONTWAIT
@@ -184,6 +191,28 @@ static void afs_free_call(struct afs_call *call)
 }
 
 /*
+ * End a call but do not free it
+ */
+static void afs_end_call_nofree(struct afs_call *call)
+{
+        if (call->rxcall) {
+                rxrpc_kernel_end_call(call->rxcall);
+                call->rxcall = NULL;
+        }
+        if (call->type->destructor)
+                call->type->destructor(call);
+}
+
+/*
+ * End a call and free it
+ */
+static void afs_end_call(struct afs_call *call)
+{
+        afs_end_call_nofree(call);
+        afs_free_call(call);
+}
+
+/*
  * allocate a call with flat request and reply buffers
  */
 struct afs_call *afs_alloc_flat_call(const struct afs_call_type *type,
@@ -326,7 +355,8 @@ int afs_make_call(struct in_addr *addr, struct afs_call *call, gfp_t gfp,
                atomic_read(&afs_outstanding_calls));
 
         call->wait_mode = wait_mode;
-        INIT_WORK(&call->async_work, afs_process_async_call);
+        call->async_workfn = afs_process_async_call;
+        INIT_WORK(&call->async_work, afs_async_workfn);
 
         memset(&srx, 0, sizeof(srx));
         srx.srx_family = AF_RXRPC;
@@ -383,11 +413,8 @@ error_do_abort:
         rxrpc_kernel_abort_call(rxcall, RX_USER_ABORT);
         while ((skb = skb_dequeue(&call->rx_queue)))
                 afs_free_skb(skb);
-        rxrpc_kernel_end_call(rxcall);
-        call->rxcall = NULL;
 error_kill_call:
-        call->type->destructor(call);
-        afs_free_call(call);
+        afs_end_call(call);
         _leave(" = %d", ret);
         return ret;
 }
@@ -509,12 +536,8 @@ static void afs_deliver_to_call(struct afs_call *call)
         if (call->state >= AFS_CALL_COMPLETE) {
                 while ((skb = skb_dequeue(&call->rx_queue)))
                         afs_free_skb(skb);
-                if (call->incoming) {
-                        rxrpc_kernel_end_call(call->rxcall);
-                        call->rxcall = NULL;
-                        call->type->destructor(call);
-                        afs_free_call(call);
-                }
+                if (call->incoming)
+                        afs_end_call(call);
         }
 
         _leave("");
@@ -564,10 +587,7 @@ static int afs_wait_for_call_to_complete(struct afs_call *call)
         }
 
         _debug("call complete");
-        rxrpc_kernel_end_call(call->rxcall);
-        call->rxcall = NULL;
-        call->type->destructor(call);
-        afs_free_call(call);
+        afs_end_call(call);
         _leave(" = %d", ret);
         return ret;
 }
@@ -603,11 +623,8 @@ static int afs_dont_wait_for_call_to_complete(struct afs_call *call)
 /*
  * delete an asynchronous call
  */
-static void afs_delete_async_call(struct work_struct *work)
+static void afs_delete_async_call(struct afs_call *call)
 {
-        struct afs_call *call =
-                container_of(work, struct afs_call, async_work);
-
         _enter("");
 
         afs_free_call(call);
@@ -620,11 +637,8 @@ static void afs_delete_async_call(struct work_struct *work)
  * - on a multiple-thread workqueue this work item may try to run on several
  *   CPUs at the same time
  */
-static void afs_process_async_call(struct work_struct *work)
+static void afs_process_async_call(struct afs_call *call)
 {
-        struct afs_call *call =
-                container_of(work, struct afs_call, async_work);
-
         _enter("");
 
         if (!skb_queue_empty(&call->rx_queue))
@@ -637,10 +651,7 @@ static void afs_process_async_call(struct work_struct *work)
                 call->reply = NULL;
 
                 /* kill the call */
-                rxrpc_kernel_end_call(call->rxcall);
-                call->rxcall = NULL;
-                if (call->type->destructor)
-                        call->type->destructor(call);
+                afs_end_call_nofree(call);
 
                 /* we can't just delete the call because the work item may be
                  * queued */
@@ -663,13 +674,6 @@ void afs_transfer_reply(struct afs_call *call, struct sk_buff *skb)
         call->reply_size += len;
 }
 
-static void afs_async_workfn(struct work_struct *work)
-{
-        struct afs_call *call = container_of(work, struct afs_call, async_work);
-
-        call->async_workfn(work);
-}
-
 /*
  * accept the backlog of incoming calls
  */
@@ -790,10 +794,7 @@ void afs_send_empty_reply(struct afs_call *call)
                 _debug("oom");
                 rxrpc_kernel_abort_call(call->rxcall, RX_USER_ABORT);
         default:
-                rxrpc_kernel_end_call(call->rxcall);
-                call->rxcall = NULL;
-                call->type->destructor(call);
-                afs_free_call(call);
+                afs_end_call(call);
                 _leave(" [error]");
                 return;
         }
@@ -823,17 +824,16 @@ void afs_send_simple_reply(struct afs_call *call, const void *buf, size_t len)
         call->state = AFS_CALL_AWAIT_ACK;
         n = rxrpc_kernel_send_data(call->rxcall, &msg, len);
         if (n >= 0) {
+                /* Success */
                 _leave(" [replied]");
                 return;
         }
+
         if (n == -ENOMEM) {
                 _debug("oom");
                 rxrpc_kernel_abort_call(call->rxcall, RX_USER_ABORT);
         }
-        rxrpc_kernel_end_call(call->rxcall);
-        call->rxcall = NULL;
-        call->type->destructor(call);
-        afs_free_call(call);
+        afs_end_call(call);
         _leave(" [error]");
 }