1 files changed, 310 insertions, 127 deletions

diff --git a/fs/ceph/file.c b/fs/ceph/file.c
index 3de89829e2a1..dfd2ce3419f8 100644
--- a/fs/ceph/file.c
+++ b/fs/ceph/file.c
@@ -408,51 +408,92 @@ more:
  *
  * If the read spans object boundary, just do multiple reads.
  */
-static ssize_t ceph_sync_read(struct file *file, char __user *data,
-                              unsigned len, loff_t *poff, int *checkeof)
+static ssize_t ceph_sync_read(struct kiocb *iocb, struct iov_iter *i,
+                                int *checkeof)
 {
+        struct file *file = iocb->ki_filp;
         struct inode *inode = file_inode(file);
         struct page **pages;
-        u64 off = *poff;
+        u64 off = iocb->ki_pos;
         int num_pages, ret;
+        size_t len = i->count;
 
-        dout("sync_read on file %p %llu~%u %s\n", file, off, len,
+        dout("sync_read on file %p %llu~%u %s\n", file, off,
+             (unsigned)len,
              (file->f_flags & O_DIRECT) ? "O_DIRECT" : "");
-
-        if (file->f_flags & O_DIRECT) {
-                num_pages = calc_pages_for((unsigned long)data, len);
-                pages = ceph_get_direct_page_vector(data, num_pages, true);
-        } else {
-                num_pages = calc_pages_for(off, len);
-                pages = ceph_alloc_page_vector(num_pages, GFP_NOFS);
-        }
-        if (IS_ERR(pages))
-                return PTR_ERR(pages);
-
         /*
          * flush any page cache pages in this range.  this
          * will make concurrent normal and sync io slow,
          * but it will at least behave sensibly when they are
          * in sequence.
          */
-        ret = filemap_write_and_wait(inode->i_mapping);
+        ret = filemap_write_and_wait_range(inode->i_mapping, off,
+                                                off + len);
         if (ret < 0)
-                goto done;
+                return ret;
 
-        ret = striped_read(inode, off, len, pages, num_pages, checkeof,
-                           file->f_flags & O_DIRECT,
-                           (unsigned long)data & ~PAGE_MASK);
+        if (file->f_flags & O_DIRECT) {
+                while (iov_iter_count(i)) {
+                        void __user *data = i->iov[0].iov_base + i->iov_offset;
+                        size_t len = i->iov[0].iov_len - i->iov_offset;
+
+                        num_pages = calc_pages_for((unsigned long)data, len);
+                        pages = ceph_get_direct_page_vector(data,
+                                                            num_pages, true);
+                        if (IS_ERR(pages))
+                                return PTR_ERR(pages);
+
+                        ret = striped_read(inode, off, len,
+                                           pages, num_pages, checkeof,
+                                           1, (unsigned long)data & ~PAGE_MASK);
+                        ceph_put_page_vector(pages, num_pages, true);
+
+                        if (ret <= 0)
+                                break;
+                        off += ret;
+                        iov_iter_advance(i, ret);
+                        if (ret < len)
+                                break;
+                }
+        } else {
+                num_pages = calc_pages_for(off, len);
+                pages = ceph_alloc_page_vector(num_pages, GFP_NOFS);
+                if (IS_ERR(pages))
+                        return PTR_ERR(pages);
+                ret = striped_read(inode, off, len, pages,
+                                        num_pages, checkeof, 0, 0);
+                if (ret > 0) {
+                        int l, k = 0;
+                        size_t left = len = ret;
+
+                        while (left) {
+                                void __user *data = i->iov[0].iov_base
+                                                        + i->iov_offset;
+                                l = min(i->iov[0].iov_len - i->iov_offset,
+                                        left);
+
+                                ret = ceph_copy_page_vector_to_user(&pages[k],
+                                                                    data, off,
+                                                                    l);
+                                if (ret > 0) {
+                                        iov_iter_advance(i, ret);
+                                        left -= ret;
+                                        off += ret;
+                                        k = calc_pages_for(iocb->ki_pos,
+                                                           len - left + 1) - 1;
+                                        BUG_ON(k >= num_pages && left);
+                                } else
+                                        break;
+                        }
+                }
+                ceph_release_page_vector(pages, num_pages);
+        }
 
-        if (ret >= 0 && (file->f_flags & O_DIRECT) == 0)
-                ret = ceph_copy_page_vector_to_user(pages, data, off, ret);
-        if (ret >= 0)
-                *poff = off + ret;
+        if (off > iocb->ki_pos) {
+                ret = off - iocb->ki_pos;
+                iocb->ki_pos = off;
+        }
 
-done:
-        if (file->f_flags & O_DIRECT)
-                ceph_put_page_vector(pages, num_pages, true);
-        else
-                ceph_release_page_vector(pages, num_pages);
         dout("sync_read result %d\n", ret);
         return ret;
 }
@@ -489,83 +530,79 @@ static void ceph_sync_write_unsafe(struct ceph_osd_request *req, bool unsafe)
         }
 }
 
+
 /*
- * Synchronous write, straight from __user pointer or user pages (if
- * O_DIRECT).
+ * Synchronous write, straight from __user pointer or user pages.
  *
  * If write spans object boundary, just do multiple writes.  (For a
  * correct atomic write, we should e.g. take write locks on all
  * objects, rollback on failure, etc.)
  */
-static ssize_t ceph_sync_write(struct file *file, const char __user *data,
-                               size_t left, loff_t pos, loff_t *ppos)
+static ssize_t
+ceph_sync_direct_write(struct kiocb *iocb, const struct iovec *iov,
+                       unsigned long nr_segs, size_t count)
 {
+        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_snap_context *snapc;
         struct ceph_vino vino;
         struct ceph_osd_request *req;
-        int num_ops = 1;
         struct page **pages;
         int num_pages;
-        u64 len;
         int written = 0;
         int flags;
         int check_caps = 0;
-        int page_align, io_align;
-        unsigned long buf_align;
+        int page_align;
         int ret;
         struct timespec mtime = CURRENT_TIME;
-        bool own_pages = false;
+        loff_t pos = iocb->ki_pos;
+        struct iov_iter i;
 
         if (ceph_snap(file_inode(file)) != CEPH_NOSNAP)
                 return -EROFS;
 
-        dout("sync_write on file %p %lld~%u %s\n", file, pos,
-             (unsigned)left, (file->f_flags & O_DIRECT) ? "O_DIRECT" : "");
+        dout("sync_direct_write on file %p %lld~%u\n", file, pos,
+             (unsigned)count);
 
-        ret = filemap_write_and_wait_range(inode->i_mapping, pos, pos + left);
+        ret = filemap_write_and_wait_range(inode->i_mapping, pos, pos + count);
         if (ret < 0)
                 return ret;
 
         ret = invalidate_inode_pages2_range(inode->i_mapping,
                                             pos >> PAGE_CACHE_SHIFT,
-                                            (pos + left) >> PAGE_CACHE_SHIFT);
+                                            (pos + count) >> PAGE_CACHE_SHIFT);
         if (ret < 0)
                 dout("invalidate_inode_pages2_range returned %d\n", ret);
 
         flags = CEPH_OSD_FLAG_ORDERSNAP |
                 CEPH_OSD_FLAG_ONDISK |
                 CEPH_OSD_FLAG_WRITE;
-        if ((file->f_flags & (O_SYNC|O_DIRECT)) == 0)
-                flags |= CEPH_OSD_FLAG_ACK;
-        else
-                num_ops++;      /* Also include a 'startsync' command. */
 
-        /*
-         * we may need to do multiple writes here if we span an object
-         * boundary.  this isn't atomic, unfortunately.  :(
-         */
-more:
-        io_align = pos & ~PAGE_MASK;
-        buf_align = (unsigned long)data & ~PAGE_MASK;
-        len = left;
-
-        snapc = ci->i_snap_realm->cached_context;
-        vino = ceph_vino(inode);
-        req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
-                                    vino, pos, &len, num_ops,
-                                    CEPH_OSD_OP_WRITE, flags, snapc,
-                                    ci->i_truncate_seq, ci->i_truncate_size,
-                                    false);
-        if (IS_ERR(req))
-                return PTR_ERR(req);
+        iov_iter_init(&i, iov, nr_segs, count, 0);
+
+        while (iov_iter_count(&i) > 0) {
+                void __user *data = i.iov->iov_base + i.iov_offset;
+                u64 len = i.iov->iov_len - i.iov_offset;
+
+                page_align = (unsigned long)data & ~PAGE_MASK;
+
+                snapc = ci->i_snap_realm->cached_context;
+                vino = ceph_vino(inode);
+                req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
+                                            vino, pos, &len,
+                                            2,/*include a 'startsync' command*/
+                                            CEPH_OSD_OP_WRITE, flags, snapc,
+                                            ci->i_truncate_seq,
+                                            ci->i_truncate_size,
+                                            false);
+                if (IS_ERR(req)) {
+                        ret = PTR_ERR(req);
+                        goto out;
+                }
 
-        /* write from beginning of first page, regardless of io alignment */
-        page_align = file->f_flags & O_DIRECT ? buf_align : io_align;
-        num_pages = calc_pages_for(page_align, len);
-        if (file->f_flags & O_DIRECT) {
+                num_pages = calc_pages_for(page_align, len);
                 pages = ceph_get_direct_page_vector(data, num_pages, false);
                 if (IS_ERR(pages)) {
                         ret = PTR_ERR(pages);
@@ -577,60 +614,175 @@ more:
                  * may block.
                  */
                 truncate_inode_pages_range(inode->i_mapping, pos,
-                                           (pos+len) | (PAGE_CACHE_SIZE-1));
-        } else {
+                                   (pos+len) | (PAGE_CACHE_SIZE-1));
+                osd_req_op_extent_osd_data_pages(req, 0, pages, len, page_align,
+                                                false, false);
+
+                /* BUG_ON(vino.snap != CEPH_NOSNAP); */
+                ceph_osdc_build_request(req, pos, snapc, vino.snap, &mtime);
+
+                ret = ceph_osdc_start_request(&fsc->client->osdc, req, false);
+                if (!ret)
+                        ret = ceph_osdc_wait_request(&fsc->client->osdc, req);
+
+                ceph_put_page_vector(pages, num_pages, false);
+
+out:
+                ceph_osdc_put_request(req);
+                if (ret == 0) {
+                        pos += len;
+                        written += len;
+                        iov_iter_advance(&i, (size_t)len);
+
+                        if (pos > i_size_read(inode)) {
+                                check_caps = ceph_inode_set_size(inode, pos);
+                                if (check_caps)
+                                        ceph_check_caps(ceph_inode(inode),
+                                                        CHECK_CAPS_AUTHONLY,
+                                                        NULL);
+                        }
+                } else
+                        break;
+        }
+
+        if (ret != -EOLDSNAPC && written > 0) {
+                iocb->ki_pos = pos;
+                ret = written;
+        }
+        return ret;
+}
+
+
+/*
+ * Synchronous write, straight from __user pointer or user pages.
+ *
+ * If write spans object boundary, just do multiple writes.  (For a
+ * correct atomic write, we should e.g. take write locks on all
+ * objects, rollback on failure, etc.)
+ */
+static ssize_t ceph_sync_write(struct kiocb *iocb, const struct iovec *iov,
+                               unsigned long nr_segs, size_t count)
+{
+        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_snap_context *snapc;
+        struct ceph_vino vino;
+        struct ceph_osd_request *req;
+        struct page **pages;
+        u64 len;
+        int num_pages;
+        int written = 0;
+        int flags;
+        int check_caps = 0;
+        int ret;
+        struct timespec mtime = CURRENT_TIME;
+        loff_t pos = iocb->ki_pos;
+        struct iov_iter i;
+
+        if (ceph_snap(file_inode(file)) != CEPH_NOSNAP)
+                return -EROFS;
+
+        dout("sync_write on file %p %lld~%u\n", file, pos, (unsigned)count);
+
+        ret = filemap_write_and_wait_range(inode->i_mapping, pos, pos + count);
+        if (ret < 0)
+                return ret;
+
+        ret = invalidate_inode_pages2_range(inode->i_mapping,
+                                            pos >> PAGE_CACHE_SHIFT,
+                                            (pos + count) >> PAGE_CACHE_SHIFT);
+        if (ret < 0)
+                dout("invalidate_inode_pages2_range returned %d\n", ret);
+
+        flags = CEPH_OSD_FLAG_ORDERSNAP |
+                CEPH_OSD_FLAG_ONDISK |
+                CEPH_OSD_FLAG_WRITE |
+                CEPH_OSD_FLAG_ACK;
+
+        iov_iter_init(&i, iov, nr_segs, count, 0);
+
+        while ((len = iov_iter_count(&i)) > 0) {
+                size_t left;
+                int n;
+
+                snapc = ci->i_snap_realm->cached_context;
+                vino = ceph_vino(inode);
+                req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
+                                            vino, pos, &len, 1,
+                                            CEPH_OSD_OP_WRITE, flags, snapc,
+                                            ci->i_truncate_seq,
+                                            ci->i_truncate_size,
+                                            false);
+                if (IS_ERR(req)) {
+                        ret = PTR_ERR(req);
+                        goto out;
+                }
+
+                /*
+                 * write from beginning of first page,
+                 * regardless of io alignment
+                 */
+                num_pages = (len + PAGE_CACHE_SIZE - 1) >> PAGE_CACHE_SHIFT;
+
                 pages = ceph_alloc_page_vector(num_pages, GFP_NOFS);
                 if (IS_ERR(pages)) {
                         ret = PTR_ERR(pages);
                         goto out;
                 }
-                ret = ceph_copy_user_to_page_vector(pages, data, pos, len);
+
+                left = len;
+                for (n = 0; n < num_pages; n++) {
+                        size_t plen = min_t(size_t, left, PAGE_SIZE);
+                        ret = iov_iter_copy_from_user(pages[n], &i, 0, plen);
+                        if (ret != plen) {
+                                ret = -EFAULT;
+                                break;
+                        }
+                        left -= ret;
+                        iov_iter_advance(&i, ret);
+                }
+
                 if (ret < 0) {
                         ceph_release_page_vector(pages, num_pages);
                         goto out;
                 }
 
-                if ((file->f_flags & O_SYNC) == 0) {
-                        /* get a second commit callback */
-                        req->r_unsafe_callback = ceph_sync_write_unsafe;
-                        req->r_inode = inode;
-                        own_pages = true;
-                }
-        }
-        osd_req_op_extent_osd_data_pages(req, 0, pages, len, page_align,
-                                        false, own_pages);
+                /* get a second commit callback */
+                req->r_unsafe_callback = ceph_sync_write_unsafe;
+                req->r_inode = inode;
 
-        /* BUG_ON(vino.snap != CEPH_NOSNAP); */
-        ceph_osdc_build_request(req, pos, snapc, vino.snap, &mtime);
+                osd_req_op_extent_osd_data_pages(req, 0, pages, len, 0,
+                                                false, true);
 
-        ret = ceph_osdc_start_request(&fsc->client->osdc, req, false);
-        if (!ret)
-                ret = ceph_osdc_wait_request(&fsc->client->osdc, req);
+                /* BUG_ON(vino.snap != CEPH_NOSNAP); */
+                ceph_osdc_build_request(req, pos, snapc, vino.snap, &mtime);
 
-        if (file->f_flags & O_DIRECT)
-                ceph_put_page_vector(pages, num_pages, false);
-        else if (file->f_flags & O_SYNC)
-                ceph_release_page_vector(pages, num_pages);
+                ret = ceph_osdc_start_request(&fsc->client->osdc, req, false);
+                if (!ret)
+                        ret = ceph_osdc_wait_request(&fsc->client->osdc, req);
 
 out:
-        ceph_osdc_put_request(req);
-        if (ret == 0) {
-                pos += len;
-                written += len;
-                left -= len;
-                data += len;
-                if (left)
-                        goto more;
+                ceph_osdc_put_request(req);
+                if (ret == 0) {
+                        pos += len;
+                        written += len;
+
+                        if (pos > i_size_read(inode)) {
+                                check_caps = ceph_inode_set_size(inode, pos);
+                                if (check_caps)
+                                        ceph_check_caps(ceph_inode(inode),
+                                                        CHECK_CAPS_AUTHONLY,
+                                                        NULL);
+                        }
+                } else
+                        break;
+        }
 
+        if (ret != -EOLDSNAPC && written > 0) {
                 ret = written;
-                *ppos = pos;
-                if (pos > i_size_read(inode))
-                        check_caps = ceph_inode_set_size(inode, pos);
-                if (check_caps)
-                        ceph_check_caps(ceph_inode(inode), CHECK_CAPS_AUTHONLY,
-                                        NULL);
-        } else if (ret != -EOLDSNAPC && written > 0) {
-                ret = written;
+                iocb->ki_pos = pos;
         }
         return ret;
 }
@@ -647,55 +799,84 @@ static ssize_t ceph_aio_read(struct kiocb *iocb, const struct iovec *iov,
 {
         struct file *filp = iocb->ki_filp;
         struct ceph_file_info *fi = filp->private_data;
-        loff_t *ppos = &iocb->ki_pos;
-        size_t len = iov->iov_len;
+        size_t len = iocb->ki_nbytes;
         struct inode *inode = file_inode(filp);
         struct ceph_inode_info *ci = ceph_inode(inode);
-        void __user *base = iov->iov_base;
         ssize_t ret;
         int want, got = 0;
         int checkeof = 0, read = 0;
 
-        dout("aio_read %p %llx.%llx %llu~%u trying to get caps on %p\n",
-             inode, ceph_vinop(inode), pos, (unsigned)len, inode);
 again:
+        dout("aio_read %p %llx.%llx %llu~%u trying to get caps on %p\n",
+             inode, ceph_vinop(inode), iocb->ki_pos, (unsigned)len, inode);
+
         if (fi->fmode & CEPH_FILE_MODE_LAZY)
                 want = CEPH_CAP_FILE_CACHE | CEPH_CAP_FILE_LAZYIO;
         else
                 want = CEPH_CAP_FILE_CACHE;
         ret = ceph_get_caps(ci, CEPH_CAP_FILE_RD, want, &got, -1);
         if (ret < 0)
-                goto out;
-        dout("aio_read %p %llx.%llx %llu~%u got cap refs on %s\n",
-             inode, ceph_vinop(inode), pos, (unsigned)len,
-             ceph_cap_string(got));
+                return ret;
 
         if ((got & (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)) == 0 ||
             (iocb->ki_filp->f_flags & O_DIRECT) ||
-            (fi->flags & CEPH_F_SYNC))
+            (fi->flags & CEPH_F_SYNC)) {
+                struct iov_iter i;
+
+                dout("aio_sync_read %p %llx.%llx %llu~%u got cap refs on %s\n",
+                     inode, ceph_vinop(inode), iocb->ki_pos, (unsigned)len,
+                     ceph_cap_string(got));
+
+                if (!read) {
+                        ret = generic_segment_checks(iov, &nr_segs,
+                                                        &len, VERIFY_WRITE);
+                        if (ret)
+                                goto out;
+                }
+
+                iov_iter_init(&i, iov, nr_segs, len, read);
+
                 /* hmm, this isn't really async... */
-                ret = ceph_sync_read(filp, base, len, ppos, &checkeof);
-        else
-                ret = generic_file_aio_read(iocb, iov, nr_segs, pos);
+                ret = ceph_sync_read(iocb, &i, &checkeof);
+        } else {
+                /*
+                 * We can't modify the content of iov,
+                 * so we only read from beginning.
+                 */
+                if (read) {
+                        iocb->ki_pos = pos;
+                        len = iocb->ki_nbytes;
+                        read = 0;
+                }
+                dout("aio_read %p %llx.%llx %llu~%u got cap refs on %s\n",
+                     inode, ceph_vinop(inode), pos, (unsigned)len,
+                     ceph_cap_string(got));
 
+                ret = generic_file_aio_read(iocb, iov, nr_segs, pos);
+        }
 out:
         dout("aio_read %p %llx.%llx dropping cap refs on %s = %d\n",
              inode, ceph_vinop(inode), ceph_cap_string(got), (int)ret);
         ceph_put_cap_refs(ci, got);
 
         if (checkeof && ret >= 0) {
-                int statret = ceph_do_getattr(inode, CEPH_STAT_CAP_SIZE);
+                int statret = ceph_do_getattr(inode,
+                                              CEPH_STAT_CAP_SIZE);
 
                 /* hit EOF or hole? */
-                if (statret == 0 && *ppos < inode->i_size) {
-                        dout("aio_read sync_read hit hole, ppos %lld < size %lld, reading more\n", *ppos, inode->i_size);
+                if (statret == 0 && iocb->ki_pos < inode->i_size &&
+                        ret < len) {
+                        dout("sync_read hit hole, ppos %lld < size %lld"
+                             ", reading more\n", iocb->ki_pos,
+                             inode->i_size);
+
                         read += ret;
-                        base += ret;
                         len -= ret;
                         checkeof = 0;
                         goto again;
                 }
         }
+
         if (ret >= 0)
                 ret += read;
 
@@ -772,11 +953,13 @@ retry_snap:
              inode, ceph_vinop(inode), pos, count, ceph_cap_string(got));
 
         if ((got & (CEPH_CAP_FILE_BUFFER|CEPH_CAP_FILE_LAZYIO)) == 0 ||
-            (iocb->ki_filp->f_flags & O_DIRECT) ||
-            (fi->flags & CEPH_F_SYNC)) {
+            (file->f_flags & O_DIRECT) || (fi->flags & CEPH_F_SYNC)) {
                 mutex_unlock(&inode->i_mutex);
-                written = ceph_sync_write(file, iov->iov_base, count,
-                                          pos, &iocb->ki_pos);
+                if (file->f_flags & O_DIRECT)
+                        written = ceph_sync_direct_write(iocb, iov,
+                                                         nr_segs, count);
+                else
+                        written = ceph_sync_write(iocb, iov, nr_segs, count);
                 if (written == -EOLDSNAPC) {
                         dout("aio_write %p %llx.%llx %llu~%u"
                                 "got EOLDSNAPC, retrying\n",
@@ -1018,7 +1201,7 @@ static long ceph_fallocate(struct file *file, int mode,
                                 loff_t offset, loff_t length)
 {
         struct ceph_file_info *fi = file->private_data;
-        struct inode *inode = file->f_dentry->d_inode;
+        struct inode *inode = file_inode(file);
         struct ceph_inode_info *ci = ceph_inode(inode);
         struct ceph_osd_client *osdc =
                 &ceph_inode_to_client(inode)->client->osdc;