kclient: fix cap writeback vs revocation

We need to reply to a revocation via check_caps and __send_cap if we are
writing back dirty metadata.

We also need to take care to look at issued|implemented (not just issued)
when deciding what to write back (since it may be being revoked).

diff --git a/src/kernel/caps.c b/src/kernel/caps.c
index b0bc9c75741167cd4f0d62fdb38d61a90ff76147..8f9ab258e514514143f43e52ab525eb0475ab3a5 100644 (file)
--- a/src/kernel/caps.c
+++ b/src/kernel/caps.c
@@ -499,6 +499,7 @@ static void __send_cap(struct ceph_mds_client *mdsc,
 {
        struct ceph_inode_info *ci = cap->ci;
        struct inode *inode = &ci->vfs_inode;
+       int held = cap->issued | cap->implemented;
        int revoking = cap->implemented & ~cap->issued;
        int dropping = cap->issued & ~retain;
        int keep;
@@ -517,11 +518,13 @@ static void __send_cap(struct ceph_mds_client *mdsc,
        if (retain == 0)
                op = CEPH_CAP_OP_RELEASE;
 
-       dout(10, "__send_cap cap %p session %p %d -> %d\n", cap, cap->session,
-            cap->issued, cap->issued & retain);
+       dout(10, "__send_cap cap %p session %p %s -> %s (revoking %s)\n",
+            cap, cap->session,
+            ceph_cap_string(held), ceph_cap_string(held & retain),
+            ceph_cap_string(revoking));
 
        dirty = __ceph_caps_dirty(ci);
-       cap->flushing |= dirty & cap->issued;
+       cap->flushing |= dirty & held;
        if (cap->flushing) {
                ci->i_dirty_caps &= ~cap->flushing;
                dout(10, "__send_cap flushing %s, dirty_caps now %s\n",
@@ -705,14 +708,14 @@ static void ceph_flush_snaps(struct ceph_inode_info *ci)
  * @is_delayed indicates caller is delayed work and we should not
  * delay further.
  */
-void ceph_check_caps(struct ceph_inode_info *ci, int is_delayed, int drop)
+void ceph_check_caps(struct ceph_inode_info *ci, int is_delayed, int drop,
+                    struct ceph_mds_session *session)
 {
        struct ceph_client *client = ceph_inode_to_client(&ci->vfs_inode);
        struct ceph_mds_client *mdsc = &client->mdsc;
        struct inode *inode = &ci->vfs_inode;
        struct ceph_cap *cap;
        int file_wanted, used;
-       struct ceph_mds_session *session = NULL;    /* if set, i hold s_mutex */
        int took_snap_rwsem = 0;             /* true if mdsc->snap_rwsem held */
        int want, retain, revoking;
        int mds = -1;   /* keep track of how far we've gone through i_caps list
@@ -1025,7 +1028,7 @@ void ceph_put_cap_refs(struct ceph_inode_info *ci, int had)
             last ? "last" : "");
 
        if (last && !flushsnaps)
-               ceph_check_caps(ci, 0, 0);
+               ceph_check_caps(ci, 0, 0, NULL);
        else if (flushsnaps)
                ceph_flush_snaps(ci);
        if (wake)
@@ -1087,7 +1090,7 @@ void ceph_put_wrbuffer_cap_refs(struct ceph_inode_info *ci, int nr,
        spin_unlock(&inode->i_lock);
 
        if (last) {
-               ceph_check_caps(ci, 0, 0);
+               ceph_check_caps(ci, 0, 0, NULL);
        } else if (last_snap) {
                ceph_flush_snaps(ci);
                wake_up(&ci->i_cap_wq);
@@ -1104,6 +1107,7 @@ void ceph_put_wrbuffer_cap_refs(struct ceph_inode_info *ci, int nr,
  * return value:
  *  0 - ok
  *  1 - send the msg back to mds
+ *  2 - check_caps
  */
 static int handle_cap_grant(struct inode *inode, struct ceph_mds_caps *grant,
                            struct ceph_mds_session *session,
@@ -1116,13 +1120,11 @@ static int handle_cap_grant(struct inode *inode, struct ceph_mds_caps *grant,
        int mds = session->s_mds;
        int seq = le32_to_cpu(grant->seq);
        int newcaps = le32_to_cpu(grant->caps);
-       int used;
-       int issued; /* to me, before */
-       int wanted;
-       int reply = 0;
+       int issued, used, wanted, dirty;
        u64 size = le64_to_cpu(grant->size);
        u64 max_size = le64_to_cpu(grant->max_size);
        struct timespec mtime, atime, ctime;
+       int reply = 0;
        int wake = 0;
        int writeback = 0;
        int revoked_rdcache = 0;
@@ -1246,8 +1248,11 @@ start:
        /* check cap bits */
        wanted = __ceph_caps_wanted(ci);
        used = __ceph_caps_used(ci);
-       dout(10, " my wanted = %s, used = %s\n", ceph_cap_string(wanted),
-            ceph_cap_string(used));
+       dirty = __ceph_caps_dirty(ci);
+       dout(10, " my wanted = %s, used = %s, dirty %s\n",
+            ceph_cap_string(wanted),
+            ceph_cap_string(used),
+            ceph_cap_string(dirty));
        if (wanted != le32_to_cpu(grant->wanted)) {
                dout(10, "mds wanted %s -> %s\n",
                     ceph_cap_string(le32_to_cpu(grant->wanted)),
@@ -1266,7 +1271,9 @@ start:
                     ceph_cap_string(newcaps));
                if ((used & ~newcaps) & CEPH_CAP_FILE_WRBUFFER) {
                        writeback = 1; /* will delay ack */
-               } else if (((used & ~newcaps) & CEPH_CAP_FILE_RDCACHE) == 0 ||
+               } else if (dirty & ~newcaps)
+                       reply = 2;     /* initiate writeback in check_caps */
+               else if (((used & ~newcaps) & CEPH_CAP_FILE_RDCACHE) == 0 ||
                           revoked_rdcache) {
                        /*
                         * we're not using revoked caps.. ack now.
@@ -1573,6 +1580,7 @@ void ceph_handle_caps(struct ceph_mds_client *mdsc,
        u64 size, max_size;
        int check_caps = 0;
        void *xattr_data = NULL;
+       int r;
 
        dout(10, "handle_caps from mds%d\n", mds);
 
@@ -1652,10 +1660,14 @@ void ceph_handle_caps(struct ceph_mds_client *mdsc,
        switch (op) {
        case CEPH_CAP_OP_GRANT:
                up_write(&mdsc->snap_rwsem);
-               if (handle_cap_grant(inode, h, session, cap,&xattr_data) == 1) {
+               r = handle_cap_grant(inode, h, session, cap,&xattr_data);
+               if (r == 1) {
                        dout(10, " sending reply back to mds%d\n", mds);
                        ceph_msg_get(msg);
                        ceph_send_msg_mds(mdsc, msg, mds);
+               } else if (r == 2) {
+                       ceph_check_caps(ceph_inode(inode), 1, 0, session);
+                       goto done_unlocked;
                }
                break;
 
@@ -1679,11 +1691,12 @@ void ceph_handle_caps(struct ceph_mds_client *mdsc,
 
 done:
        mutex_unlock(&session->s_mutex);
+done_unlocked:
        ceph_put_mds_session(session);
 
        kfree(xattr_data);
        if (check_caps)
-               ceph_check_caps(ceph_inode(inode), 1, 0);
+               ceph_check_caps(ceph_inode(inode), 1, 0, NULL);
        if (inode)
                iput(inode);
        return;
@@ -1724,7 +1737,7 @@ void ceph_check_delayed_caps(struct ceph_mds_client *mdsc)
                list_del_init(&ci->i_cap_delay_list);
                spin_unlock(&mdsc->cap_delay_lock);
                dout(10, "check_delayed_caps on %p\n", &ci->vfs_inode);
-               ceph_check_caps(ci, 1, 0);
+               ceph_check_caps(ci, 1, 0, NULL);
                iput(&ci->vfs_inode);
        }
        spin_unlock(&mdsc->cap_delay_lock);

diff --git a/src/kernel/file.c b/src/kernel/file.c
index 4574e4c20f9f426eaa24852b69cf124740f774e9..018618da8e4f63ebb202c94a62e573d887becb6f 100644 (file)
--- a/src/kernel/file.c
+++ b/src/kernel/file.c
@@ -346,7 +346,7 @@ static void check_max_size(struct inode *inode, loff_t endoff)
        }
        spin_unlock(&inode->i_lock);
        if (check)
-               ceph_check_caps(ci, 0, 0);
+               ceph_check_caps(ci, 0, 0, NULL);
 }
 
 /*

diff --git a/src/kernel/inode.c b/src/kernel/inode.c
index 2bf2bc6da400d1c8ceab622cf8e626630866e35b..646d7ba5d9b1d058a0c19951ea7c8fd4956972f5 100644 (file)
--- a/src/kernel/inode.c
+++ b/src/kernel/inode.c
@@ -1237,7 +1237,7 @@ void ceph_inode_set_size(struct inode *inode, loff_t size)
        if ((size << 1) >= ci->i_max_size &&
            (ci->i_reported_size << 1) < ci->i_max_size) {
                spin_unlock(&inode->i_lock);
-               ceph_check_caps(ci, 0, 0);
+               ceph_check_caps(ci, 0, 0, NULL);
        } else {
                spin_unlock(&inode->i_lock);
        }
@@ -1261,7 +1261,7 @@ void ceph_put_fmode(struct ceph_inode_info *ci, int fmode)
        spin_unlock(&ci->vfs_inode.i_lock);
 
        if (last && ci->i_vino.snap == CEPH_NOSNAP)
-               ceph_check_caps(ci, 0, 0);
+               ceph_check_caps(ci, 0, 0, NULL);
 }
 
 
@@ -1322,7 +1322,7 @@ static void ceph_inode_invalidate_pages(struct work_struct *work)
        spin_unlock(&inode->i_lock);
 
        if (check)
-               ceph_check_caps(ci, 0, 0);
+               ceph_check_caps(ci, 0, 0, NULL);
 out:
        iput(inode);
 }
@@ -1368,7 +1368,7 @@ void __ceph_do_pending_vmtruncate(struct inode *inode)
                dout(10, "__do_pending_vmtruncate %p to %lld\n", inode, to);
                truncate_inode_pages(inode->i_mapping, to);
                if (wrbuffer_refs == 0)
-                       ceph_check_caps(ci, 0, 0);
+                       ceph_check_caps(ci, 0, 0, NULL);
        } else {
                dout(10, "__do_pending_vmtruncate %p nothing to do\n", inode);
        }
@@ -1643,7 +1643,7 @@ int ceph_setattr(struct dentry *dentry, struct iattr *attr)
                     inode->i_uid, attr->ia_uid);
        if (ia_valid & ATTR_GID)
                dout(10, "setattr: %p gid %d -> %d\n", inode,
-                    inode->i_uid, attr->ia_uid);
+                    inode->i_gid, attr->ia_gid);
        if (ia_valid & ATTR_MODE)
                dout(10, "setattr: %p mode 0%o -> 0%o\n", inode, inode->i_mode,
                     attr->ia_mode);

diff --git a/src/kernel/mds_client.c b/src/kernel/mds_client.c
index 205aba072bd5bf1fa98370a857d74b141c0783ca..cc39d30fa384d09a7cc545f3811987966c5b64a2 100644 (file)
--- a/src/kernel/mds_client.c
+++ b/src/kernel/mds_client.c
@@ -881,7 +881,7 @@ static void check_all_caps(struct ceph_mds_client *mdsc,
 
                igrab(inode);
                mutex_unlock(&session->s_mutex);
-               ceph_check_caps(ceph_inode(inode), 1, 0);
+               ceph_check_caps(ceph_inode(inode), 1, 0, NULL);
                mutex_lock(&session->s_mutex);
                iput(inode);
        }

diff --git a/src/kernel/super.h b/src/kernel/super.h
index e049f9b6d9069f76ec8a237afcd57910f270bbb9..fe7f39f90d29cfc303ceac5ef62ddcef255d634f 100644 (file)
--- a/src/kernel/super.h
+++ b/src/kernel/super.h
@@ -713,13 +713,14 @@ extern void ceph_put_wrbuffer_cap_refs(struct ceph_inode_info *ci, int nr,
                                       struct ceph_snap_context *snapc);
 extern void __ceph_flush_snaps(struct ceph_inode_info *ci,
                               struct ceph_mds_session **psession);
-extern void ceph_check_caps(struct ceph_inode_info *ci, int delayed, int drop);
+extern void ceph_check_caps(struct ceph_inode_info *ci, int delayed, int drop,
+                           struct ceph_mds_session *session);
 extern void ceph_check_delayed_caps(struct ceph_mds_client *mdsc);
 void ceph_trim_session_rdcaps(struct ceph_mds_session *session);
 
 static inline void ceph_release_caps(struct inode *inode, int mask)
 {
-       ceph_check_caps(ceph_inode(inode), 1, mask);
+       ceph_check_caps(ceph_inode(inode), 1, mask, NULL);
 }
 
 /* addr.c */