FS-Cache: Fix operation state management and accounting

Fix the state management of internal fscache operations and the accounting of
what operations are in what states.

This is done by:

 (1) Give struct fscache_operation a enum variable that directly represents the
     state it's currently in, rather than spreading this knowledge over a bunch
     of flags, who's processing the operation at the moment and whether it is
     queued or not.

     This makes it easier to write assertions to check the state at various
     points and to prevent invalid state transitions.

 (2) Add an 'operation complete' state and supply a function to indicate the
     completion of an operation (fscache_op_complete()) and make things call
     it.  The final call to fscache_put_operation() can then check that an op
     in the appropriate state (complete or cancelled).

 (3) Adjust the use of object->n_ops, ->n_in_progress, ->n_exclusive to better
     govern the state of an object:

	(a) The ->n_ops is now the number of extant operations on the object
	    and is now decremented by fscache_put_operation() only.

	(b) The ->n_in_progress is simply the number of objects that have been
	    taken off of the object's pending queue for the purposes of being
	    run.  This is decremented by fscache_op_complete() only.

	(c) The ->n_exclusive is the number of exclusive ops that have been
	    submitted and queued or are in progress.  It is decremented by
	    fscache_op_complete() and by fscache_cancel_op().

     fscache_put_operation() and fscache_operation_gc() now no longer try to
     clean up ->n_exclusive and ->n_in_progress.  That was leading to double
     decrements against fscache_cancel_op().

     fscache_cancel_op() now no longer decrements ->n_ops.  That was leading to
     double decrements against fscache_put_operation().

     fscache_submit_exclusive_op() now decides whether it has to queue an op
     based on ->n_in_progress being > 0 rather than ->n_ops > 0 as the latter
     will persist in being true even after all preceding operations have been
     cancelled or completed.  Furthermore, if an object is active and there are
     runnable ops against it, there must be at least one op running.

 (4) Add a remaining-pages counter (n_pages) to struct fscache_retrieval and
     provide a function to record completion of the pages as they complete.

     When n_pages reaches 0, the operation is deemed to be complete and
     fscache_op_complete() is called.

     Add calls to fscache_retrieval_complete() anywhere we've finished with a
     page we've been given to read or allocate for.  This includes places where
     we just return pages to the netfs for reading from the server and where
     accessing the cache fails and we discard the proposed netfs page.

The bugs in the unfixed state management manifest themselves as oopses like the
following where the operation completion gets out of sync with return of the
cookie by the netfs.  This is possible because the cache unlocks and returns
all the netfs pages before recording its completion - which means that there's
nothing to stop the netfs discarding them and returning the cookie.


FS-Cache: Cookie 'NFS.fh' still has outstanding reads
------------[ cut here ]------------
kernel BUG at fs/fscache/cookie.c:519!
invalid opcode: 0000 [#1] SMP
CPU 1
Modules linked in: cachefiles nfs fscache auth_rpcgss nfs_acl lockd sunrpc

Pid: 400, comm: kswapd0 Not tainted 3.1.0-rc7-fsdevel+ #1090                  /DG965RY
RIP: 0010:[<ffffffffa007050a>]  [<ffffffffa007050a>] __fscache_relinquish_cookie+0x170/0x343 [fscache]
RSP: 0018:ffff8800368cfb00  EFLAGS: 00010282
RAX: 000000000000003c RBX: ffff880023cc8790 RCX: 0000000000000000
RDX: 0000000000002f2e RSI: 0000000000000001 RDI: ffffffff813ab86c
RBP: ffff8800368cfb50 R08: 0000000000000002 R09: 0000000000000000
R10: ffff88003a1b7890 R11: ffff88001df6e488 R12: ffff880023d8ed98
R13: ffff880023cc8798 R14: 0000000000000004 R15: ffff88003b8bf370
FS:  0000000000000000(0000) GS:ffff88003bd00000(0000) knlGS:0000000000000000
CS:  0010 DS: 0000 ES: 0000 CR0: 000000008005003b
CR2: 00000000008ba008 CR3: 0000000023d93000 CR4: 00000000000006e0
DR0: 0000000000000000 DR1: 0000000000000000 DR2: 0000000000000000
DR3: 0000000000000000 DR6: 00000000ffff0ff0 DR7: 0000000000000400
Process kswapd0 (pid: 400, threadinfo ffff8800368ce000, task ffff88003b8bf040)
Stack:
 ffff88003b8bf040 ffff88001df6e528 ffff88001df6e528 ffffffffa00b46b0
 ffff88003b8bf040 ffff88001df6e488 ffff88001df6e620 ffffffffa00b46b0
 ffff88001ebd04c8 0000000000000004 ffff8800368cfb70 ffffffffa00b2c91
Call Trace:
 [<ffffffffa00b2c91>] nfs_fscache_release_inode_cookie+0x3b/0x47 [nfs]
 [<ffffffffa008f25f>] nfs_clear_inode+0x3c/0x41 [nfs]
 [<ffffffffa0090df1>] nfs4_evict_inode+0x2f/0x33 [nfs]
 [<ffffffff810d8d47>] evict+0xa1/0x15c
 [<ffffffff810d8e2e>] dispose_list+0x2c/0x38
 [<ffffffff810d9ebd>] prune_icache_sb+0x28c/0x29b
 [<ffffffff810c56b7>] prune_super+0xd5/0x140
 [<ffffffff8109b615>] shrink_slab+0x102/0x1ab
 [<ffffffff8109d690>] balance_pgdat+0x2f2/0x595
 [<ffffffff8103e009>] ? process_timeout+0xb/0xb
 [<ffffffff8109dba3>] kswapd+0x270/0x289
 [<ffffffff8104c5ea>] ? __init_waitqueue_head+0x46/0x46
 [<ffffffff8109d933>] ? balance_pgdat+0x595/0x595
 [<ffffffff8104bf7a>] kthread+0x7f/0x87
 [<ffffffff813ad6b4>] kernel_thread_helper+0x4/0x10
 [<ffffffff81026b98>] ? finish_task_switch+0x45/0xc0
 [<ffffffff813abcdd>] ? retint_restore_args+0xe/0xe
 [<ffffffff8104befb>] ? __init_kthread_worker+0x53/0x53
 [<ffffffff813ad6b0>] ? gs_change+0xb/0xb

Signed-off-by: David Howells <dhowells@redhat.com>

3 files changed, 80 insertions, 38 deletions

diff --git a/fs/fscache/object.c b/fs/fscache/object.c
index b6b897c550ac..773bc798a416 100644
--- a/fs/fscache/object.c
+++ b/fs/fscache/object.c
@@ -587,8 +587,6 @@ static void fscache_object_available(struct fscache_object *object)
         if (object->n_in_progress == 0) {
                 if (object->n_ops > 0) {
                         ASSERTCMP(object->n_ops, >=, object->n_obj_ops);
-                        ASSERTIF(object->n_ops > object->n_obj_ops,
-                                 !list_empty(&object->pending_ops));
                         fscache_start_operations(object);
                 } else {
                         ASSERT(list_empty(&object->pending_ops));
diff --git a/fs/fscache/operation.c b/fs/fscache/operation.c
index c857ab824d6e..748f9553c2cb 100644
--- a/fs/fscache/operation.c
+++ b/fs/fscache/operation.c
@@ -37,6 +37,7 @@ void fscache_enqueue_operation(struct fscache_operation *op)
         ASSERT(op->processor != NULL);
         ASSERTCMP(op->object->state, >=, FSCACHE_OBJECT_AVAILABLE);
         ASSERTCMP(atomic_read(&op->usage), >, 0);
+        ASSERTCMP(op->state, ==, FSCACHE_OP_ST_IN_PROGRESS);
 
         fscache_stat(&fscache_n_op_enqueue);
         switch (op->flags & FSCACHE_OP_TYPE) {
@@ -64,6 +65,9 @@ EXPORT_SYMBOL(fscache_enqueue_operation);
 static void fscache_run_op(struct fscache_object *object,
                            struct fscache_operation *op)
 {
+        ASSERTCMP(op->state, ==, FSCACHE_OP_ST_PENDING);
+
+        op->state = FSCACHE_OP_ST_IN_PROGRESS;
         object->n_in_progress++;
         if (test_and_clear_bit(FSCACHE_OP_WAITING, &op->flags))
                 wake_up_bit(&op->flags, FSCACHE_OP_WAITING);
@@ -80,22 +84,23 @@ static void fscache_run_op(struct fscache_object *object,
 int fscache_submit_exclusive_op(struct fscache_object *object,
                                 struct fscache_operation *op)
 {
-        int ret;
-
         _enter("{OBJ%x OP%x},", object->debug_id, op->debug_id);
 
+        ASSERTCMP(op->state, ==, FSCACHE_OP_ST_INITIALISED);
+        ASSERTCMP(atomic_read(&op->usage), >, 0);
+
         spin_lock(&object->lock);
         ASSERTCMP(object->n_ops, >=, object->n_in_progress);
         ASSERTCMP(object->n_ops, >=, object->n_exclusive);
         ASSERT(list_empty(&op->pend_link));
 
-        ret = -ENOBUFS;
+        op->state = FSCACHE_OP_ST_PENDING;
         if (fscache_object_is_active(object)) {
                 op->object = object;
                 object->n_ops++;
                 object->n_exclusive++;  /* reads and writes must wait */
 
-                if (object->n_ops > 1) {
+                if (object->n_in_progress > 0) {
                         atomic_inc(&op->usage);
                         list_add_tail(&op->pend_link, &object->pending_ops);
                         fscache_stat(&fscache_n_op_pend);
@@ -111,7 +116,6 @@ int fscache_submit_exclusive_op(struct fscache_object *object,
 
                 /* need to issue a new write op after this */
                 clear_bit(FSCACHE_OBJECT_PENDING_WRITE, &object->flags);
-                ret = 0;
         } else if (object->state == FSCACHE_OBJECT_CREATING) {
                 op->object = object;
                 object->n_ops++;
@@ -119,14 +123,13 @@ int fscache_submit_exclusive_op(struct fscache_object *object,
                 atomic_inc(&op->usage);
                 list_add_tail(&op->pend_link, &object->pending_ops);
                 fscache_stat(&fscache_n_op_pend);
-                ret = 0;
         } else {
                 /* not allowed to submit ops in any other state */
                 BUG();
         }
 
         spin_unlock(&object->lock);
-        return ret;
+        return 0;
 }
 
 /*
@@ -186,6 +189,7 @@ int fscache_submit_op(struct fscache_object *object,
         _enter("{OBJ%x OP%x},{%u}",
                object->debug_id, op->debug_id, atomic_read(&op->usage));
 
+        ASSERTCMP(op->state, ==, FSCACHE_OP_ST_INITIALISED);
         ASSERTCMP(atomic_read(&op->usage), >, 0);
 
         spin_lock(&object->lock);
@@ -196,6 +200,7 @@ int fscache_submit_op(struct fscache_object *object,
         ostate = object->state;
         smp_rmb();
 
+        op->state = FSCACHE_OP_ST_PENDING;
         if (fscache_object_is_active(object)) {
                 op->object = object;
                 object->n_ops++;
@@ -225,12 +230,15 @@ int fscache_submit_op(struct fscache_object *object,
                    object->state == FSCACHE_OBJECT_LC_DYING ||
                    object->state == FSCACHE_OBJECT_WITHDRAWING) {
                 fscache_stat(&fscache_n_op_rejected);
+                op->state = FSCACHE_OP_ST_CANCELLED;
                 ret = -ENOBUFS;
         } else if (!test_bit(FSCACHE_IOERROR, &object->cache->flags)) {
                 fscache_report_unexpected_submission(object, op, ostate);
                 ASSERT(!fscache_object_is_active(object));
+                op->state = FSCACHE_OP_ST_CANCELLED;
                 ret = -ENOBUFS;
         } else {
+                op->state = FSCACHE_OP_ST_CANCELLED;
                 ret = -ENOBUFS;
         }
 
@@ -290,13 +298,18 @@ int fscache_cancel_op(struct fscache_operation *op)
 
         _enter("OBJ%x OP%x}", op->object->debug_id, op->debug_id);
 
+        ASSERTCMP(op->state, >=, FSCACHE_OP_ST_PENDING);
+        ASSERTCMP(op->state, !=, FSCACHE_OP_ST_CANCELLED);
+        ASSERTCMP(atomic_read(&op->usage), >, 0);
+
         spin_lock(&object->lock);
 
         ret = -EBUSY;
-        if (!list_empty(&op->pend_link)) {
+        if (op->state == FSCACHE_OP_ST_PENDING) {
+                ASSERT(!list_empty(&op->pend_link));
                 fscache_stat(&fscache_n_op_cancelled);
                 list_del_init(&op->pend_link);
-                object->n_ops--;
+                op->state = FSCACHE_OP_ST_CANCELLED;
                 if (test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags))
                         object->n_exclusive--;
                 if (test_and_clear_bit(FSCACHE_OP_WAITING, &op->flags))
@@ -311,6 +324,37 @@ int fscache_cancel_op(struct fscache_operation *op)
 }
 
 /*
+ * Record the completion of an in-progress operation.
+ */
+void fscache_op_complete(struct fscache_operation *op)
+{
+        struct fscache_object *object = op->object;
+
+        _enter("OBJ%x", object->debug_id);
+
+        ASSERTCMP(op->state, ==, FSCACHE_OP_ST_IN_PROGRESS);
+        ASSERTCMP(object->n_in_progress, >, 0);
+        ASSERTIFCMP(test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags),
+                    object->n_exclusive, >, 0);
+        ASSERTIFCMP(test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags),
+                    object->n_in_progress, ==, 1);
+
+        spin_lock(&object->lock);
+
+        op->state = FSCACHE_OP_ST_COMPLETE;
+
+        if (test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags))
+                object->n_exclusive--;
+        object->n_in_progress--;
+        if (object->n_in_progress == 0)
+                fscache_start_operations(object);
+
+        spin_unlock(&object->lock);
+        _leave("");
+}
+EXPORT_SYMBOL(fscache_op_complete);
+
+/*
  * release an operation
  * - queues pending ops if this is the last in-progress op
  */
@@ -328,8 +372,9 @@ void fscache_put_operation(struct fscache_operation *op)
                 return;
 
         _debug("PUT OP");
-        if (test_and_set_bit(FSCACHE_OP_DEAD, &op->flags))
-                BUG();
+        ASSERTIFCMP(op->state != FSCACHE_OP_ST_COMPLETE,
+                    op->state, ==, FSCACHE_OP_ST_CANCELLED);
+        op->state = FSCACHE_OP_ST_DEAD;
 
         fscache_stat(&fscache_n_op_release);
 
@@ -365,16 +410,6 @@ void fscache_put_operation(struct fscache_operation *op)
                 return;
         }
 
-        if (test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags)) {
-                ASSERTCMP(object->n_exclusive, >, 0);
-                object->n_exclusive--;
-        }
-
-        ASSERTCMP(object->n_in_progress, >, 0);
-        object->n_in_progress--;
-        if (object->n_in_progress == 0)
-                fscache_start_operations(object);
-
         ASSERTCMP(object->n_ops, >, 0);
         object->n_ops--;
         if (object->n_ops == 0)
@@ -413,23 +448,14 @@ void fscache_operation_gc(struct work_struct *work)
                 spin_unlock(&cache->op_gc_list_lock);
 
                 object = op->object;
+                spin_lock(&object->lock);
 
                 _debug("GC DEFERRED REL OBJ%x OP%x",
                        object->debug_id, op->debug_id);
                 fscache_stat(&fscache_n_op_gc);
 
                 ASSERTCMP(atomic_read(&op->usage), ==, 0);
-
-                spin_lock(&object->lock);
-                if (test_bit(FSCACHE_OP_EXCLUSIVE, &op->flags)) {
-                        ASSERTCMP(object->n_exclusive, >, 0);
-                        object->n_exclusive--;
-                }
-
-                ASSERTCMP(object->n_in_progress, >, 0);
-                object->n_in_progress--;
-                if (object->n_in_progress == 0)
-                        fscache_start_operations(object);
+                ASSERTCMP(op->state, ==, FSCACHE_OP_ST_DEAD);
 
                 ASSERTCMP(object->n_ops, >, 0);
                 object->n_ops--;
@@ -437,6 +463,7 @@ void fscache_operation_gc(struct work_struct *work)
                         fscache_raise_event(object, FSCACHE_OBJECT_EV_CLEARED);
 
                 spin_unlock(&object->lock);
+                kfree(op);
 
         } while (count++ < 20);
 
diff --git a/fs/fscache/page.c b/fs/fscache/page.c
index 248a12e22532..b38b13d2a555 100644
--- a/fs/fscache/page.c
+++ b/fs/fscache/page.c
@@ -162,6 +162,7 @@ static void fscache_attr_changed_op(struct fscache_operation *op)
                         fscache_abort_object(object);
         }
 
+        fscache_op_complete(op);
         _leave("");
 }
 
@@ -223,6 +224,8 @@ static void fscache_release_retrieval_op(struct fscache_operation *_op)
 
         _enter("{OP%x}", op->op.debug_id);
 
+        ASSERTCMP(op->n_pages, ==, 0);
+
         fscache_hist(fscache_retrieval_histogram, op->start_time);
         if (op->context)
                 fscache_put_context(op->op.object->cookie, op->context);
@@ -320,6 +323,11 @@ static int fscache_wait_for_retrieval_activation(struct fscache_object *object,
         _debug("<<< GO");
 
 check_if_dead:
+        if (op->op.state == FSCACHE_OP_ST_CANCELLED) {
+                fscache_stat(stat_object_dead);
+                _leave(" = -ENOBUFS [cancelled]");
+                return -ENOBUFS;
+        }
         if (unlikely(fscache_object_is_dead(object))) {
                 fscache_stat(stat_object_dead);
                 return -ENOBUFS;
@@ -364,6 +372,7 @@ int __fscache_read_or_alloc_page(struct fscache_cookie *cookie,
                 _leave(" = -ENOMEM");
                 return -ENOMEM;
         }
+        op->n_pages = 1;
 
         spin_lock(&cookie->lock);
 
@@ -375,10 +384,10 @@ int __fscache_read_or_alloc_page(struct fscache_cookie *cookie,
         ASSERTCMP(object->state, >, FSCACHE_OBJECT_LOOKING_UP);
 
         atomic_inc(&object->n_reads);
-        set_bit(FSCACHE_OP_DEC_READ_CNT, &op->op.flags);
+        __set_bit(FSCACHE_OP_DEC_READ_CNT, &op->op.flags);
 
         if (fscache_submit_op(object, &op->op) < 0)
-                goto nobufs_unlock;
+                goto nobufs_unlock_dec;
         spin_unlock(&cookie->lock);
 
         fscache_stat(&fscache_n_retrieval_ops);
@@ -425,6 +434,8 @@ error:
         _leave(" = %d", ret);
         return ret;
 
+nobufs_unlock_dec:
+        atomic_dec(&object->n_reads);
 nobufs_unlock:
         spin_unlock(&cookie->lock);
         kfree(op);
@@ -482,6 +493,7 @@ int __fscache_read_or_alloc_pages(struct fscache_cookie *cookie,
         op = fscache_alloc_retrieval(mapping, end_io_func, context);
         if (!op)
                 return -ENOMEM;
+        op->n_pages = *nr_pages;
 
         spin_lock(&cookie->lock);
 
@@ -491,10 +503,10 @@ int __fscache_read_or_alloc_pages(struct fscache_cookie *cookie,
                              struct fscache_object, cookie_link);
 
         atomic_inc(&object->n_reads);
-        set_bit(FSCACHE_OP_DEC_READ_CNT, &op->op.flags);
+        __set_bit(FSCACHE_OP_DEC_READ_CNT, &op->op.flags);
 
         if (fscache_submit_op(object, &op->op) < 0)
-                goto nobufs_unlock;
+                goto nobufs_unlock_dec;
         spin_unlock(&cookie->lock);
 
         fscache_stat(&fscache_n_retrieval_ops);
@@ -541,6 +553,8 @@ error:
         _leave(" = %d", ret);
         return ret;
 
+nobufs_unlock_dec:
+        atomic_dec(&object->n_reads);
 nobufs_unlock:
         spin_unlock(&cookie->lock);
         kfree(op);
@@ -583,6 +597,7 @@ int __fscache_alloc_page(struct fscache_cookie *cookie,
         op = fscache_alloc_retrieval(page->mapping, NULL, NULL);
         if (!op)
                 return -ENOMEM;
+        op->n_pages = 1;
 
         spin_lock(&cookie->lock);
 
@@ -696,6 +711,7 @@ static void fscache_write_op(struct fscache_operation *_op)
         fscache_end_page_write(object, page);
         if (ret < 0) {
                 fscache_abort_object(object);
+                fscache_op_complete(&op->op);
         } else {
                 fscache_enqueue_operation(&op->op);
         }
@@ -710,6 +726,7 @@ superseded:
         spin_unlock(&cookie->stores_lock);
         clear_bit(FSCACHE_OBJECT_PENDING_WRITE, &object->flags);
         spin_unlock(&object->lock);
+        fscache_op_complete(&op->op);
         _leave("");
 }