kclient: obey max_size, and return current size/mtime/atime when acking with re-used...

diff --git a/src/TODO b/src/TODO
index f5c02d08e2a27ca3c763de6d61fcf23687748f3d..5e3a6842746e2b951c7ab65b29e930564aab92ae 100644 (file)
--- a/src/TODO
+++ b/src/TODO
@@ -15,12 +15,14 @@ userspace client
   - also needs cope with mds failures
 - reference count lease validations on path lookup?
 
+yehuda:
+- apply_truncate needs to truncate page cache, or something?
+
 kernel client
 - why is mds_client delayed work not working?
 - file_data_version stuff!
 - deal with CAP_RDCACHE properly: invalidate cache pages?
 - trim expired leases so we don't indefinitely hold dcache/icache refs...
-- apply_truncate needs to truncate page cache, or something?
 - obey file_max
 - send mds a file_cap message when we approach the current file_max
 - procfs/debugfs
@@ -32,10 +34,9 @@ kernel client
 - vfs
  - can we use dentry_path(), if it gets merged into mainline?
 - io / osd client
+  - fix writeback_now on cap revocation
   - kick requests when new map arrives
   - osd ack vs commit handling.  hmm!
-- mon client
-  - work out message resend logic..?
 - unmount
 /  - close open file handles, mds sessions
     - flush data before unmount.

diff --git a/src/kernel/file.c b/src/kernel/file.c
index 48e30465b2ea869e8f05adc558aa3354daa35304..57b161c83fceaf8ae28ff4c493fe4fbc4c79214d 100644 (file)
--- a/src/kernel/file.c
+++ b/src/kernel/file.c
@@ -243,7 +243,7 @@ ssize_t ceph_read(struct file *filp, char __user *buf, size_t len, loff_t *ppos)
        ret = wait_event_interruptible(ci->i_cap_wq,
                                       ceph_get_cap_refs(ci, CEPH_CAP_RD, 
                                                         CEPH_CAP_RDCACHE, 
-                                                        &got));
+                                                        &got, -1));
        if (ret < 0) 
                goto out;
        dout(10, "read %llx %llu~%u got cap refs %d\n",
@@ -272,11 +272,11 @@ ssize_t ceph_write(struct file *filp, const char __user *buf,
        ssize_t ret;
        int got = 0;
 
-       dout(10, "write trying to get caps\n");
+       dout(10, "write trying to get caps. i_size %llu\n", inode->i_size);
        ret = wait_event_interruptible(ci->i_cap_wq,
                                       ceph_get_cap_refs(ci, CEPH_CAP_WR, 
                                                         CEPH_CAP_WRBUFFER,
-                                                        &got));
+                                                        &got, *ppos));
        if (ret < 0) 
                goto out;
        dout(10, "write got cap refs on %d\n", got);

diff --git a/src/kernel/inode.c b/src/kernel/inode.c
index ba8af2e9ac9c1f767695eaa4579cc343cfe41985..c30ff5f81eda3f086138299084f625237dc4c498 100644 (file)
--- a/src/kernel/inode.c
+++ b/src/kernel/inode.c
@@ -880,7 +880,7 @@ int ceph_handle_cap_grant(struct inode *inode, struct ceph_mds_file_caps *grant,
        int used;
        int issued; /* to me, before */
        int wanted;
-       int ret = 0;
+       int reply = 0;
        u64 size = le64_to_cpu(grant->size);
        u64 max_size = le64_to_cpu(grant->max_size);
        struct timespec mtime, atime, ctime;
@@ -890,10 +890,25 @@ int ceph_handle_cap_grant(struct inode *inode, struct ceph_mds_file_caps *grant,
        
        dout(10, "handle_cap_grant inode %p ci %p mds%d seq %d\n", 
             inode, ci, mds, seq);
-       dout(10, " size %llu max_size %llu\n", size, max_size);
+       dout(10, " size %llu max_size %llu, i_size %llu\n", size, max_size,
+               inode->i_size);
 
-       /* size change? */
        spin_lock(&inode->i_lock);
+
+       /* do we have this cap? */
+       cap = __get_cap_for_mds(inode, mds);
+       if (!cap) {
+               /*
+                * then ignore.  never reply to cap messages out of turn,
+                * or we'll be mixing up different instances of caps on the
+                * same inode, and confuse the mds.
+                */
+               dout(10, "no cap on ino %llx from mds%d, ignoring\n",
+                    ci->i_ceph_ino, mds);
+               goto out;
+       } 
+
+       /* size change? */
        if (size > inode->i_size) {
                dout(10, "size %lld -> %llu\n", inode->i_size, size);
                inode->i_size = size;
@@ -903,6 +918,7 @@ int ceph_handle_cap_grant(struct inode *inode, struct ceph_mds_file_caps *grant,
        if (max_size != ci->i_max_size) {
                dout(10, "max_size %lld -> %llu\n", ci->i_max_size, max_size);
                ci->i_max_size = max_size;
+               wake = 1;
        }
 
        /* mtime/atime? */
@@ -931,19 +947,7 @@ int ceph_handle_cap_grant(struct inode *inode, struct ceph_mds_file_caps *grant,
                }
        }
 
-       /* do we have this cap? */
-       cap = __get_cap_for_mds(inode, mds);
-       if (!cap) {
-               /*
-                * then ignore.  never reply to cap messages out of turn,
-                * or we'll be mixing up different instances of caps on the
-                * same inode, and confuse the mds.
-                */
-               dout(10, "no cap on ino %llx from mds%d, ignoring\n",
-                    ci->i_ceph_ino, mds);
-               goto out;
-       } 
-
+       /* check cap bits */
        wanted = __ceph_caps_wanted(ci);
        used = __ceph_caps_used(ci);
        dout(10, " my wanted = %d, used = %d\n", wanted, used);
@@ -965,7 +969,11 @@ int ceph_handle_cap_grant(struct inode *inode, struct ceph_mds_file_caps *grant,
                        writeback_now = 1; /* will delay ack */
                else {
                        cap->implemented = newcaps;
-                       ret = 1; /* ack now */
+                       /* ack now.  re-use incoming message. */
+                       grant->size = le64_to_cpu(inode->i_size);
+                       ceph_encode_timespec(&grant->mtime, &inode->i_mtime);
+                       ceph_encode_timespec(&grant->atime, &inode->i_atime);
+                       reply = 1; 
                }
                cap->issued = newcaps;
                goto out;
@@ -988,7 +996,7 @@ out:
                write_inode_now(inode, 0);
        if (invalidate)
                invalidate_mapping_pages(&inode->i_data, 0, -1);
-       return ret;     
+       return reply;
 }
 
 void apply_truncate(struct inode *inode, loff_t size)
@@ -1038,13 +1046,19 @@ void ceph_take_cap_refs(struct ceph_inode_info *ci, int got)
        spin_unlock(&ci->vfs_inode.i_lock);
 }      
 
-int ceph_get_cap_refs(struct ceph_inode_info *ci, int need, int want, int *got)
+int ceph_get_cap_refs(struct ceph_inode_info *ci, int need, int want, int *got,
+                     loff_t offset)
 {
        int ret = 0;
        int have;
        dout(10, "get_cap_refs on %p need %d want %d\n", &ci->vfs_inode,
             need, want);
        spin_lock(&ci->vfs_inode.i_lock);
+       if (offset >= 0 && offset >= (loff_t)ci->i_max_size) {
+               dout(20, "get_cap_refs offset %llu >= max_size %llu\n",
+                    offset, ci->i_max_size);
+               goto sorry;
+       }
        have = __ceph_caps_issued(ci);
        dout(20, "get_cap_refs have %d\n", have);
        if ((have & need) == need) {
@@ -1052,6 +1066,7 @@ int ceph_get_cap_refs(struct ceph_inode_info *ci, int need, int want, int *got)
                __take_cap_refs(ci, *got);
                ret = 1;
        }
+sorry:
        spin_unlock(&ci->vfs_inode.i_lock);
        dout(10, "get_cap_refs on %p ret %d got %d\n", &ci->vfs_inode,
             ret, *got);

diff --git a/src/kernel/super.h b/src/kernel/super.h
index 4fdb4e3531f9e02a6d39522a069cefadd46078d6..407dc43fd57dddbe8b23c8fbbc8a3a9dbc799f34 100644 (file)
--- a/src/kernel/super.h
+++ b/src/kernel/super.h
@@ -372,7 +372,7 @@ extern int ceph_handle_cap_grant(struct inode *inode,
 extern int ceph_handle_cap_trunc(struct inode *inode,
                                 struct ceph_mds_file_caps *grant,
                                 struct ceph_mds_session *session);
-extern int ceph_get_cap_refs(struct ceph_inode_info *ci, int need, int want, int *got);
+extern int ceph_get_cap_refs(struct ceph_inode_info *ci, int need, int want, int *got, loff_t offset);
 extern void ceph_take_cap_refs(struct ceph_inode_info *ci, int got);
 extern void ceph_put_cap_refs(struct ceph_inode_info *ci, int had);
 extern void ceph_put_wrbuffer_cap_refs(struct ceph_inode_info *ci, int nr);