ceph: add object version support for sync read

Turn the guts of ceph_sync_read into a new helper that takes an inode
and an offset instead of a kiocb struct, and make ceph_sync_read call
the helper as a wrapper.

Make the new helper always return the last object's version.

Signed-off-by: Xiubo Li <xiubli@redhat.com>
Reviewed-by: Jeff Layton <jlayton@kernel.org>
Reviewed-and-tested-by: Luís Henriques <lhenriques@suse.de>
Reviewed-by: Milind Changire <mchangir@redhat.com>
Signed-off-by: Ilya Dryomov <idryomov@gmail.com>

diff --git a/fs/ceph/file.c b/fs/ceph/file.c
index c3ce224..24ee84f 100644 (file)
--- a/fs/ceph/file.c
+++ b/fs/ceph/file.c
@@ -959,22 +959,24 @@ enum {
  * If we get a short result from the OSD, check against i_size; we need to
  * only return a short read to the caller if we hit EOF.
  */
-static ssize_t ceph_sync_read(struct kiocb *iocb, struct iov_iter *to,
-                             int *retry_op)
+ssize_t __ceph_sync_read(struct inode *inode, loff_t *ki_pos,
+                        struct iov_iter *to, int *retry_op,
+                        u64 *last_objver)
 {
-       struct file *file = iocb->ki_filp;
-       struct inode *inode = file_inode(file);
        struct ceph_inode_info *ci = ceph_inode(inode);
        struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
        struct ceph_osd_client *osdc = &fsc->client->osdc;
        ssize_t ret;
-       u64 off = iocb->ki_pos;
+       u64 off = *ki_pos;
        u64 len = iov_iter_count(to);
        u64 i_size = i_size_read(inode);
        bool sparse = ceph_test_mount_opt(fsc, SPARSEREAD);
+       u64 objver = 0;
 
-       dout("sync_read on file %p %llu~%u %s\n", file, off, (unsigned)len,
-            (file->f_flags & O_DIRECT) ? "O_DIRECT" : "");
+       dout("sync_read on inode %p %llx~%llx\n", inode, *ki_pos, len);
+
+       if (ceph_inode_is_shutdown(inode))
+               return -EIO;
 
        if (!len)
                return 0;
@@ -1042,6 +1044,9 @@ static ssize_t ceph_sync_read(struct kiocb *iocb, struct iov_iter *to,
                                         req->r_end_latency,
                                         len, ret);
 
+               if (ret > 0)
+                       objver = req->r_version;
+
                i_size = i_size_read(inode);
                dout("sync_read %llu~%llu got %zd i_size %llu%s\n",
                     off, len, ret, i_size, (more ? " MORE" : ""));
@@ -1092,21 +1097,36 @@ static ssize_t ceph_sync_read(struct kiocb *iocb, struct iov_iter *to,
                        break;
        }
 
-       if (off > iocb->ki_pos) {
+       if (off > *ki_pos) {
                if (off >= i_size) {
                        *retry_op = CHECK_EOF;
-                       ret = i_size - iocb->ki_pos;
-                       iocb->ki_pos = i_size;
+                       ret = i_size - *ki_pos;
+                       *ki_pos = i_size;
                } else {
-                       ret = off - iocb->ki_pos;
-                       iocb->ki_pos = off;
+                       ret = off - *ki_pos;
+                       *ki_pos = off;
                }
        }
 
+       if (last_objver && ret > 0)
+               *last_objver = objver;
+
        dout("sync_read result %zd retry_op %d\n", ret, *retry_op);
        return ret;
 }
 
+static ssize_t ceph_sync_read(struct kiocb *iocb, struct iov_iter *to,
+                             int *retry_op)
+{
+       struct file *file = iocb->ki_filp;
+       struct inode *inode = file_inode(file);
+
+       dout("sync_read on file %p %llx~%zx %s\n", file, iocb->ki_pos,
+            iov_iter_count(to), (file->f_flags & O_DIRECT) ? "O_DIRECT" : "");
+
+       return __ceph_sync_read(inode, &iocb->ki_pos, to, retry_op, NULL);
+}
+
 struct ceph_aio_request {
        struct kiocb *iocb;
        size_t total_len;

diff --git a/fs/ceph/super.h b/fs/ceph/super.h
index 1148385..d60342c 100644 (file)
--- a/fs/ceph/super.h
+++ b/fs/ceph/super.h
@@ -1300,6 +1300,9 @@ extern int ceph_renew_caps(struct inode *inode, int fmode);
 extern int ceph_open(struct inode *inode, struct file *file);
 extern int ceph_atomic_open(struct inode *dir, struct dentry *dentry,
                            struct file *file, unsigned flags, umode_t mode);
+extern ssize_t __ceph_sync_read(struct inode *inode, loff_t *ki_pos,
+                               struct iov_iter *to, int *retry_op,
+                               u64 *last_objver);
 extern int ceph_release(struct inode *inode, struct file *filp);
 extern void ceph_fill_inline_data(struct inode *inode, struct page *locked_page,
                                  char *data, size_t len);