ceph: properly handle aborted mds requests

Previously, if the MDS request was interrupted, we would unregister the
request and ignore any reply.  This could cause the caps or other cache
state to become out of sync.  (For instance, aborting dbench and doing
rm -r on clients would complain about a non-empty directory because the
client didn't realize it's aborted file create request completed.)

Even we don't unregister, we still can't process the reply normally because
we are no longer holding the caller's locks (like the dir i_mutex).

So, mark aborted operations with r_aborted, and in the reply handler, be
sure to process all the caps.  Do not process the namespace changes,
though, since we no longer will hold the dir i_mutex.  The dentry lease
state can also be ignored as it's more forgiving.

Signed-off-by: Sage Weil <sage@newdream.net>

1 files changed, 23 insertions, 5 deletions

diff --git a/fs/ceph/mds_client.c b/fs/ceph/mds_client.c
index 6e08f488a30f..623c67cd484b 100644
--- a/fs/ceph/mds_client.c
+++ b/fs/ceph/mds_client.c
@@ -1624,11 +1624,29 @@ int ceph_mdsc_do_request(struct ceph_mds_client *mdsc,
                 err = PTR_ERR(req->r_reply);
                 req->r_reply = NULL;
 
-                /* clean up */
-                __unregister_request(mdsc, req);
-                if (!list_empty(&req->r_unsafe_item))
-                        list_del_init(&req->r_unsafe_item);
-                complete(&req->r_safe_completion);
+                if (err == -ERESTARTSYS) {
+                        /* aborted */
+                        req->r_aborted = true;
+
+                        if (req->r_locked_dir &&
+                            (req->r_op & CEPH_MDS_OP_WRITE)) {
+                                struct ceph_inode_info *ci =
+                                        ceph_inode(req->r_locked_dir);
+
+                                dout("aborted, clearing I_COMPLETE on %p\n", 
+                                     req->r_locked_dir);
+                                spin_lock(&req->r_locked_dir->i_lock);
+                                ci->i_ceph_flags &= ~CEPH_I_COMPLETE;
+                                ci->i_release_count++;
+                                spin_unlock(&req->r_locked_dir->i_lock);
+                        }
+                } else {
+                        /* clean up this request */
+                        __unregister_request(mdsc, req);
+                        if (!list_empty(&req->r_unsafe_item))
+                                list_del_init(&req->r_unsafe_item);
+                        complete(&req->r_safe_completion);
+                }
         } else if (req->r_err) {
                 err = req->r_err;
         } else {