kclient: add ceph_aio_write (instead of generic) to take cap refs

diff --git a/src/TODO b/src/TODO
index 080432d3ebb2806d040416a836d8afc4f2b4b324..f56542fdbe24327c92d10df80c71b8ee8c127448 100644 (file)
--- a/src/TODO
+++ b/src/TODO
@@ -27,6 +27,8 @@ userspace client
   - also needs cope with mds failures
 
 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 d932ff8f6620007d4a061b74e49971cc76de42f2..bfa1ba32dfcfd881b047c1a0e94ac589a0b81dc2 100644 (file)
--- a/src/kernel/file.c
+++ b/src/kernel/file.c
@@ -286,18 +286,11 @@ out:
 /*
  * ditto
  */
-ssize_t ceph_write(struct file *filp, const char __user *buf,
-                  size_t len, loff_t *ppos)
+static void check_max_size(struct inode *inode, loff_t endoff)
 {
-       struct inode *inode = filp->f_dentry->d_inode;
        struct ceph_inode_info *ci = ceph_inode(inode);
-       ssize_t ret;
-       int got = 0;
        int check = 0;
-       loff_t endoff = *ppos + len;
-
-       __ceph_do_pending_vmtruncate(inode);
-
+       
        /* do we need to explicitly request a larger max_size? */
        spin_lock(&inode->i_lock);
        if ((endoff >= ci->i_max_size ||
@@ -311,13 +304,26 @@ ssize_t ceph_write(struct file *filp, const char __user *buf,
        spin_unlock(&inode->i_lock);
        if (check)
                ceph_check_caps(ci, 0);
+}
+
+ssize_t ceph_write(struct file *filp, const char __user *buf,
+                  size_t len, loff_t *ppos)
+{
+       struct inode *inode = filp->f_dentry->d_inode;
+       struct ceph_inode_info *ci = ceph_inode(inode);
+       ssize_t ret;
+       int got = 0;
+       loff_t endoff = *ppos + len;
+
+       __ceph_do_pending_vmtruncate(inode);
+       check_max_size(inode, endoff);
 
        dout(10, "write %p %llu~%u getting caps. i_size %llu\n",
             inode, *ppos, (unsigned)len, inode->i_size);
        ret = wait_event_interruptible(ci->i_cap_wq,
                                       ceph_get_cap_refs(ci, CEPH_CAP_WR,
                                                         CEPH_CAP_WRBUFFER,
-                                                        &got, *ppos+len));
+                                                        &got, endoff));
        if (ret < 0)
                goto out;
        dout(10, "write %p %llu~%u  got cap refs on %d\n",
@@ -336,6 +342,46 @@ out:
        return ret;
 }
 
+/*
+ * verify caps in aio_write as well!
+ */
+ssize_t ceph_aio_write(struct kiocb *iocb, const struct iovec *iov,
+                      unsigned long nr_segs, loff_t pos)
+{
+       struct file *file = iocb->ki_filp;
+       struct inode *inode = file->f_dentry->d_inode;
+       struct ceph_inode_info *ci = ceph_inode(inode);
+       loff_t endoff = pos + iov->iov_len;
+       int got = 0;
+       int ret;
+
+       check_max_size(inode, endoff);
+       dout(10, "aio_write %p %llu~%u getting caps. i_size %llu\n",
+            inode, pos, (unsigned)iov->iov_len, inode->i_size);
+       ret = wait_event_interruptible(ci->i_cap_wq,
+                                      ceph_get_cap_refs(ci, CEPH_CAP_WR,
+                                                        CEPH_CAP_WRBUFFER,
+                                                        &got, endoff));
+       if (ret < 0)
+               goto out;
+
+       dout(10, "aio_write %p %llu~%u  got cap refs on %d\n",
+            inode, pos, (unsigned)iov->iov_len, got);
+
+       if ((got & CEPH_CAP_WRBUFFER) == 0 ||
+           (inode->i_sb->s_flags & MS_SYNCHRONOUS))
+               /* fixme, this isn't async */
+               ret = ceph_sync_write(file, iov->iov_base, iov->iov_len, &pos);
+       else
+               ret = generic_file_aio_write(iocb, iov, nr_segs, pos);
+       
+out:
+       dout(10, "aio_write %p %llu~%u  dropping cap refs on %d\n",
+            inode, pos, (unsigned)iov->iov_len, got);
+       ceph_put_cap_refs(ci, got);
+       return ret;
+}
+
 static int ceph_fsync(struct file *file, struct dentry *dentry, int datasync)
 {
        struct inode *inode = dentry->d_inode;
@@ -347,13 +393,15 @@ static int ceph_fsync(struct file *file, struct dentry *dentry, int datasync)
                return ret;
 
        /*
-        * fixme: also ensure that caps are flushed to mds
+        * fixme: also ensure that caps are flushed to mds? 
+        * well, not strictly necessary, since with the data on the osds
+        * the mds can always reconstruct the file size.
         */
 
        return 0;
 }
 
-static int ceph_ioctl(struct file *file, unsigned int cmd, unsigned long arg)
+static long ceph_ioctl(struct file *file, unsigned int cmd, unsigned long arg)
 {
        return -EINVAL;
 }
@@ -365,7 +413,7 @@ const struct file_operations ceph_file_fops = {
        .read = ceph_read,
        .write = ceph_write,
        .aio_read = generic_file_aio_read,
-       .aio_write = generic_file_aio_write,
+       .aio_write = ceph_aio_write,
        .mmap = generic_file_mmap,
        .fsync = ceph_fsync,
        .splice_read = generic_file_splice_read,