kclient: close wanted->needed cap transition ordering hole

diff --git a/src/TODO b/src/TODO
index 496e151ad3b51924c9dc2563db2d277605d449f7..7f9dd68512880297d00ac2eea7391f7657292f6a 100644 (file)
--- a/src/TODO
+++ b/src/TODO
@@ -22,8 +22,6 @@ userspace client
 - clean up client mds session vs mdsmap behavior?
 
 kernel client
-- close WR|WRBUFFER -> WR transition overwrite hole
-  - get_cap_refs needs to block if want WR+WRBUFFER, issued WR, but still implemented WRBUFFER.
 - flush caps on sync, fsync, etc.
   - do we need to block?
 - timeout mds session close on umount

diff --git a/src/kernel/file.c b/src/kernel/file.c
index 0c4680beddcf11c785a668bba2a2fde53f8494b2..4269317e0ca0721e2ac7eb77f7a1e95b4d5e25df 100644 (file)
--- a/src/kernel/file.c
+++ b/src/kernel/file.c
@@ -97,7 +97,7 @@ int ceph_open(struct inode *inode, struct file *file)
 
        /* can we re-use existing caps? */
        spin_lock(&inode->i_lock);
-       if ((__ceph_caps_issued(ci) & wantcaps) == wantcaps) {
+       if ((__ceph_caps_issued(ci, 0) & wantcaps) == wantcaps) {
                dout(10, "open fmode %d caps %d using existing on %p\n",
                     fmode, wantcaps, inode);
                __ceph_get_fmode(ci, fmode);

diff --git a/src/kernel/inode.c b/src/kernel/inode.c
index 1b6504a2ba59215bfc8d68134c826925136f3005..5dab42a30fc37bb166dfdf0fa7a23ff77a2ed77a 100644 (file)
--- a/src/kernel/inode.c
+++ b/src/kernel/inode.c
@@ -302,7 +302,7 @@ int ceph_fill_inode(struct inode *inode,
        ceph_decode_timespec(&atime, &info->atime);
        ceph_decode_timespec(&mtime, &info->mtime);
        ceph_decode_timespec(&ctime, &info->ctime);
-       issued = __ceph_caps_issued(ci);
+       issued = __ceph_caps_issued(ci, 0);
 
        fill_file_bits(inode, issued, le64_to_cpu(info->time_warp_seq), size,
                       &ctime, &mtime, &atime);
@@ -470,7 +470,7 @@ int ceph_inode_lease_valid(struct inode *inode, int mask)
 
        /* EXCL cap counts for an ICONTENT lease... check caps? */
        if ((mask & CEPH_LOCK_ICONTENT) &&
-           __ceph_caps_issued(ci) & CEPH_CAP_EXCL) {
+           __ceph_caps_issued(ci, 0) & CEPH_CAP_EXCL) {
                dout(20, "lease_valid inode %p EXCL cap -> ICONTENT\n", inode);
                havemask |= CEPH_LOCK_ICONTENT;
        }
@@ -1015,7 +1015,7 @@ int ceph_add_cap(struct inode *inode,
        return 0;
 }
 
-int __ceph_caps_issued(struct ceph_inode_info *ci)
+int __ceph_caps_issued(struct ceph_inode_info *ci, int *implemented)
 {
        int have = 0;
        struct ceph_inode_cap *cap;
@@ -1040,6 +1040,8 @@ int __ceph_caps_issued(struct ceph_inode_info *ci)
                dout(30, "__ceph_caps_issued %p cap %p issued %d\n",
                     &ci->vfs_inode, cap, cap->issued);
                have |= cap->issued;
+               if (implemented)
+                       *implemented |= cap->implemented;
        }
        return have;
 }
@@ -1121,7 +1123,7 @@ retry:
        wanted = __ceph_caps_wanted(ci);
        used = __ceph_caps_used(ci);
        dout(10, "check_caps %p wanted %d used %d issued %d\n", inode,
-            wanted, used, __ceph_caps_issued(ci));
+            wanted, used, __ceph_caps_issued(ci, 0));
 
        if (!is_delayed)
                __ceph_cap_delay_requeue(mdsc, ci);
@@ -1276,7 +1278,7 @@ int ceph_handle_cap_grant(struct inode *inode, struct ceph_mds_file_caps *grant,
        cap->gen = session->s_cap_gen;
 
        /* size/ctime/mtime/atime? */
-       issued = __ceph_caps_issued(ci);
+       issued = __ceph_caps_issued(ci, 0);
        ceph_decode_timespec(&mtime, &grant->mtime);
        ceph_decode_timespec(&atime, &grant->atime);
        ceph_decode_timespec(&ctime, &grant->ctime);
@@ -1322,6 +1324,7 @@ int ceph_handle_cap_grant(struct inode *inode, struct ceph_mds_file_caps *grant,
                        ceph_encode_timespec(&grant->atime, &inode->i_atime);
                        grant->time_warp_seq = cpu_to_le64(ci->i_time_warp_seq);
                        reply = 1;
+                       wake = 1;
                }
                cap->issued = newcaps;
                goto out;
@@ -1553,7 +1556,7 @@ int ceph_get_cap_refs(struct ceph_inode_info *ci, int need, int want, int *got,
                      loff_t endoff)
 {
        int ret = 0;
-       int have;
+       int have, implemented;
 
        dout(30, "get_cap_refs on %p need %d want %d\n", &ci->vfs_inode,
             need, want);
@@ -1563,13 +1566,25 @@ int ceph_get_cap_refs(struct ceph_inode_info *ci, int need, int want, int *got,
                     endoff, ci->i_max_size);
                goto sorry;
        }
-       have = __ceph_caps_issued(ci);
-       dout(30, "get_cap_refs have %d\n", have);
+       have = __ceph_caps_issued(ci, &implemented);
        if ((have & need) == need) {
-               *got = need | (have & want);
-               __take_cap_refs(ci, *got);
-               ret = 1;
-       }
+               /*
+                * look at (implemented & ~have & not) so that we keep waiting
+                * on transition from wanted -> needed caps.  this is needed
+                * for WRBUFFER|WR -> WR to avoid a new WR sync write from
+                * going before a prior buffered writeback happens.
+                */
+               int not = want & ~(have & need);
+               int revoking = implemented & ~have;
+               dout(30, "get_cap_refs have %d but not %d (revoking %d)\n", 
+                    have, not, revoking);
+               if ((revoking & not) == 0) {
+                       *got = need | (have & want);
+                       __take_cap_refs(ci, *got);
+                       ret = 1;
+               }
+       } else
+               dout(30, "get_cap_refs have %d needed %d\n", have, need);
 sorry:
        spin_unlock(&ci->vfs_inode.i_lock);
        dout(30, "get_cap_refs on %p ret %d got %d\n", &ci->vfs_inode,

diff --git a/src/kernel/mds_client.c b/src/kernel/mds_client.c
index 8d11376e59fcb90d356c14dbc07d1d50551bd556..a6656790b25a88a614fe7ec3edf6761495e0ac30 100644 (file)
--- a/src/kernel/mds_client.c
+++ b/src/kernel/mds_client.c
@@ -1339,7 +1339,7 @@ retry:
                spin_lock(&ci->vfs_inode.i_lock);
                cap->seq = 0;  /* reset cap seq */
                rec->wanted = cpu_to_le32(__ceph_caps_wanted(ci));
-               rec->issued = cpu_to_le32(__ceph_caps_issued(ci));
+               rec->issued = cpu_to_le32(__ceph_caps_issued(ci, 0));
                rec->size = cpu_to_le64(ci->vfs_inode.i_size);
                ceph_encode_timespec(&rec->mtime, &ci->vfs_inode.i_mtime);
                ceph_encode_timespec(&rec->atime, &ci->vfs_inode.i_atime);
@@ -1610,13 +1610,16 @@ int __ceph_mdsc_send_cap(struct ceph_mds_client *mdsc,
        __u64 size, max_size;
        struct timespec mtime, atime;
        int removed_last = 0;
+       int wake = 0;
 
        dout(10, "__send_cap cap %p session %p %d -> %d\n", cap, cap->session,
             cap->issued, cap->issued & wanted);
        cap->issued &= wanted;  /* drop bits we don't want */
        
-       if (revoking && (revoking && used) == 0)
+       if (revoking && (revoking && used) == 0) {
                cap->implemented = cap->issued;
+               wake = 1;  /* for waiters on wanted -> needed transition */
+       }
        
        keep = cap->issued;
        seq = cap->seq;
@@ -1649,6 +1652,8 @@ int __ceph_mdsc_send_cap(struct ceph_mds_client *mdsc,
                     size, max_size, &mtime, &atime, time_warp_seq,
                     session->s_mds);
 
+       if (wake)
+               wake_up(&ci->i_cap_wq);
        if (wanted == 0)
                iput(inode);  /* removed cap */
 

diff --git a/src/kernel/super.h b/src/kernel/super.h
index 698bdccd751b6c524004cdf6d8ba6ed8ed4cb3db..465bfd587077f30cb3f52c7a157d04277174d36e 100644 (file)
--- a/src/kernel/super.h
+++ b/src/kernel/super.h
@@ -337,13 +337,13 @@ static inline struct inode *ceph_find_inode(struct super_block *sb, __u64 ino)
 /*
  * caps helpers
  */
-extern int __ceph_caps_issued(struct ceph_inode_info *ci);
+extern int __ceph_caps_issued(struct ceph_inode_info *ci, int *implemented);
 
 static inline int ceph_caps_issued(struct ceph_inode_info *ci)
 {
        int issued;
        spin_lock(&ci->vfs_inode.i_lock);
-       issued = __ceph_caps_issued(ci);
+       issued = __ceph_caps_issued(ci, 0);
        spin_unlock(&ci->vfs_inode.i_lock);
        return issued;
 }