ceph: add new mount option to enable sparse reads

Add a new mount option that has the client issue sparse reads instead of
normal ones. The callers now preallocate an sparse extent buffer that
the libceph receive code can populate and hand back after the operation
completes.

After a successful sparse read, we can't use the req->r_result value to
determine the amount of data "read", so instead we set the received
length to be from the end of the last extent in the buffer. Any
interstitial holes will have been filled by the receive code.

[ xiubli: fix a double free on req reported by Ilya ]

Signed-off-by: Jeff Layton <jlayton@kernel.org>
Reviewed-by: Xiubo Li <xiubli@redhat.com>
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/addr.c b/fs/ceph/addr.c
index 59cbfb80edbda9071e2960cb02deb8b6fbf2dae3..f45ebe957bbebcc4d0f2741d2598d90440598fe7 100644 (file)
--- a/fs/ceph/addr.c
+++ b/fs/ceph/addr.c
@@ -245,8 +245,10 @@ static void finish_netfs_read(struct ceph_osd_request *req)
        struct ceph_fs_client *fsc = ceph_inode_to_client(req->r_inode);
        struct ceph_osd_data *osd_data = osd_req_op_extent_osd_data(req, 0);
        struct netfs_io_subrequest *subreq = req->r_priv;
+       struct ceph_osd_req_op *op = &req->r_ops[0];
        int num_pages;
        int err = req->r_result;
+       bool sparse = (op->op == CEPH_OSD_OP_SPARSE_READ);
 
        ceph_update_read_metrics(&fsc->mdsc->metric, req->r_start_latency,
                                 req->r_end_latency, osd_data->length, err);
@@ -255,7 +257,9 @@ static void finish_netfs_read(struct ceph_osd_request *req)
             subreq->len, i_size_read(req->r_inode));
 
        /* no object means success but no data */
-       if (err == -ENOENT)
+       if (sparse && err >= 0)
+               err = ceph_sparse_ext_map_end(op);
+       else if (err == -ENOENT)
                err = 0;
        else if (err == -EBLOCKLISTED)
                fsc->blocklisted = true;
@@ -338,6 +342,7 @@ static void ceph_netfs_issue_read(struct netfs_io_subrequest *subreq)
        size_t page_off;
        int err = 0;
        u64 len = subreq->len;
+       bool sparse = ceph_test_mount_opt(fsc, SPARSEREAD);
 
        if (ceph_inode_is_shutdown(inode)) {
                err = -EIO;
@@ -348,7 +353,7 @@ static void ceph_netfs_issue_read(struct netfs_io_subrequest *subreq)
                return;
 
        req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout, vino, subreq->start, &len,
-                       0, 1, CEPH_OSD_OP_READ,
+                       0, 1, sparse ? CEPH_OSD_OP_SPARSE_READ : CEPH_OSD_OP_READ,
                        CEPH_OSD_FLAG_READ | fsc->client->osdc.client->options->read_from_replica,
                        NULL, ci->i_truncate_seq, ci->i_truncate_size, false);
        if (IS_ERR(req)) {
@@ -357,6 +362,12 @@ static void ceph_netfs_issue_read(struct netfs_io_subrequest *subreq)
                goto out;
        }
 
+       if (sparse) {
+               err = ceph_alloc_sparse_ext_map(&req->r_ops[0]);
+               if (err)
+                       goto out;
+       }
+
        dout("%s: pos=%llu orig_len=%zu len=%llu\n", __func__, subreq->start, subreq->len, len);
        iov_iter_xarray(&iter, ITER_DEST, &rreq->mapping->i_pages, subreq->start, len);
        err = iov_iter_get_pages_alloc2(&iter, &pages, len, &page_off);

diff --git a/fs/ceph/file.c b/fs/ceph/file.c
index 63efe5389783ce7aa3d61959188e00fb47e328dc..e3671678b2339d0bf5c7992c3f0a30ffc1a822f5 100644 (file)
--- a/fs/ceph/file.c
+++ b/fs/ceph/file.c
@@ -936,6 +936,7 @@ static ssize_t ceph_sync_read(struct kiocb *iocb, struct iov_iter *to,
        u64 off = iocb->ki_pos;
        u64 len = iov_iter_count(to);
        u64 i_size = i_size_read(inode);
+       bool sparse = ceph_test_mount_opt(fsc, SPARSEREAD);
 
        dout("sync_read on file %p %llu~%u %s\n", file, off, (unsigned)len,
             (file->f_flags & O_DIRECT) ? "O_DIRECT" : "");
@@ -962,10 +963,12 @@ static ssize_t ceph_sync_read(struct kiocb *iocb, struct iov_iter *to,
                bool more;
                int idx;
                size_t left;
+               struct ceph_osd_req_op *op;
 
                req = ceph_osdc_new_request(osdc, &ci->i_layout,
                                        ci->i_vino, off, &len, 0, 1,
-                                       CEPH_OSD_OP_READ, CEPH_OSD_FLAG_READ,
+                                       sparse ? CEPH_OSD_OP_SPARSE_READ : CEPH_OSD_OP_READ,
+                                       CEPH_OSD_FLAG_READ,
                                        NULL, ci->i_truncate_seq,
                                        ci->i_truncate_size, false);
                if (IS_ERR(req)) {
@@ -986,6 +989,16 @@ static ssize_t ceph_sync_read(struct kiocb *iocb, struct iov_iter *to,
 
                osd_req_op_extent_osd_data_pages(req, 0, pages, len, page_off,
                                                 false, false);
+
+               op = &req->r_ops[0];
+               if (sparse) {
+                       ret = ceph_alloc_sparse_ext_map(op);
+                       if (ret) {
+                               ceph_osdc_put_request(req);
+                               break;
+                       }
+               }
+
                ceph_osdc_start_request(osdc, req);
                ret = ceph_osdc_wait_request(osdc, req);
 
@@ -994,19 +1007,24 @@ static ssize_t ceph_sync_read(struct kiocb *iocb, struct iov_iter *to,
                                         req->r_end_latency,
                                         len, ret);
 
-               ceph_osdc_put_request(req);
-
                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" : ""));
 
-               if (ret == -ENOENT)
+               /* Fix it to go to end of extent map */
+               if (sparse && ret >= 0)
+                       ret = ceph_sparse_ext_map_end(op);
+               else if (ret == -ENOENT)
                        ret = 0;
+
+               ceph_osdc_put_request(req);
+
                if (ret >= 0 && ret < len && (off + ret < i_size)) {
                        int zlen = min(len - ret, i_size - off - ret);
                        int zoff = page_off + ret;
+
                        dout("sync_read zero gap %llu~%llu\n",
-                             off + ret, off + ret + zlen);
+                               off + ret, off + ret + zlen);
                        ceph_zero_page_vector_range(zoff, zlen, pages);
                        ret += zlen;
                }
@@ -1125,8 +1143,10 @@ static void ceph_aio_complete_req(struct ceph_osd_request *req)
        struct inode *inode = req->r_inode;
        struct ceph_aio_request *aio_req = req->r_priv;
        struct ceph_osd_data *osd_data = osd_req_op_extent_osd_data(req, 0);
+       struct ceph_osd_req_op *op = &req->r_ops[0];
        struct ceph_client_metric *metric = &ceph_sb_to_mdsc(inode->i_sb)->metric;
        unsigned int len = osd_data->bvec_pos.iter.bi_size;
+       bool sparse = (op->op == CEPH_OSD_OP_SPARSE_READ);
 
        BUG_ON(osd_data->type != CEPH_OSD_DATA_TYPE_BVECS);
        BUG_ON(!osd_data->num_bvecs);
@@ -1147,6 +1167,8 @@ static void ceph_aio_complete_req(struct ceph_osd_request *req)
                }
                rc = -ENOMEM;
        } else if (!aio_req->write) {
+               if (sparse && rc >= 0)
+                       rc = ceph_sparse_ext_map_end(op);
                if (rc == -ENOENT)
                        rc = 0;
                if (rc >= 0 && len > rc) {
@@ -1283,6 +1305,7 @@ ceph_direct_read_write(struct kiocb *iocb, struct iov_iter *iter,
        loff_t pos = iocb->ki_pos;
        bool write = iov_iter_rw(iter) == WRITE;
        bool should_dirty = !write && user_backed_iter(iter);
+       bool sparse = ceph_test_mount_opt(fsc, SPARSEREAD);
 
        if (write && ceph_snap(file_inode(file)) != CEPH_NOSNAP)
                return -EROFS;
@@ -1310,6 +1333,8 @@ ceph_direct_read_write(struct kiocb *iocb, struct iov_iter *iter,
        while (iov_iter_count(iter) > 0) {
                u64 size = iov_iter_count(iter);
                ssize_t len;
+               struct ceph_osd_req_op *op;
+               int readop = sparse ? CEPH_OSD_OP_SPARSE_READ : CEPH_OSD_OP_READ;
 
                if (write)
                        size = min_t(u64, size, fsc->mount_options->wsize);
@@ -1320,8 +1345,7 @@ ceph_direct_read_write(struct kiocb *iocb, struct iov_iter *iter,
                req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
                                            vino, pos, &size, 0,
                                            1,
-                                           write ? CEPH_OSD_OP_WRITE :
-                                                   CEPH_OSD_OP_READ,
+                                           write ? CEPH_OSD_OP_WRITE : readop,
                                            flags, snapc,
                                            ci->i_truncate_seq,
                                            ci->i_truncate_size,
@@ -1372,6 +1396,14 @@ ceph_direct_read_write(struct kiocb *iocb, struct iov_iter *iter,
                }
 
                osd_req_op_extent_osd_data_bvecs(req, 0, bvecs, num_pages, len);
+               op = &req->r_ops[0];
+               if (sparse) {
+                       ret = ceph_alloc_sparse_ext_map(op);
+                       if (ret) {
+                               ceph_osdc_put_request(req);
+                               break;
+                       }
+               }
 
                if (aio_req) {
                        aio_req->total_len += len;
@@ -1399,8 +1431,11 @@ ceph_direct_read_write(struct kiocb *iocb, struct iov_iter *iter,
 
                size = i_size_read(inode);
                if (!write) {
-                       if (ret == -ENOENT)
+                       if (sparse && ret >= 0)
+                               ret = ceph_sparse_ext_map_end(op);
+                       else if (ret == -ENOENT)
                                ret = 0;
+
                        if (ret >= 0 && ret < len && pos + ret < size) {
                                struct iov_iter i;
                                int zlen = min_t(size_t, len - ret,

diff --git a/fs/ceph/super.c b/fs/ceph/super.c
index a5f52013314d65d2c4ff576a01c08deffba872ae..d6a1790f69233cca2d96547cf0fee5c5803b5713 100644 (file)
--- a/fs/ceph/super.c
+++ b/fs/ceph/super.c
@@ -165,6 +165,7 @@ enum {
        Opt_copyfrom,
        Opt_wsync,
        Opt_pagecache,
+       Opt_sparseread,
 };
 
 enum ceph_recover_session_mode {
@@ -207,6 +208,7 @@ static const struct fs_parameter_spec ceph_mount_parameters[] = {
        fsparam_u32     ("wsize",                       Opt_wsize),
        fsparam_flag_no ("wsync",                       Opt_wsync),
        fsparam_flag_no ("pagecache",                   Opt_pagecache),
+       fsparam_flag_no ("sparseread",                  Opt_sparseread),
        {}
 };
 
@@ -576,6 +578,12 @@ static int ceph_parse_mount_param(struct fs_context *fc,
                else
                        fsopt->flags &= ~CEPH_MOUNT_OPT_NOPAGECACHE;
                break;
+       case Opt_sparseread:
+               if (result.negated)
+                       fsopt->flags &= ~CEPH_MOUNT_OPT_SPARSEREAD;
+               else
+                       fsopt->flags |= CEPH_MOUNT_OPT_SPARSEREAD;
+               break;
        default:
                BUG();
        }
@@ -710,9 +718,10 @@ static int ceph_show_options(struct seq_file *m, struct dentry *root)
 
        if (!(fsopt->flags & CEPH_MOUNT_OPT_ASYNC_DIROPS))
                seq_puts(m, ",wsync");
-
        if (fsopt->flags & CEPH_MOUNT_OPT_NOPAGECACHE)
                seq_puts(m, ",nopagecache");
+       if (fsopt->flags & CEPH_MOUNT_OPT_SPARSEREAD)
+               seq_puts(m, ",sparseread");
 
        if (fsopt->wsize != CEPH_MAX_WRITE_SIZE)
                seq_printf(m, ",wsize=%u", fsopt->wsize);
@@ -1296,6 +1305,11 @@ static int ceph_reconfigure_fc(struct fs_context *fc)
        else
                ceph_clear_mount_opt(fsc, ASYNC_DIROPS);
 
+       if (fsopt->flags & CEPH_MOUNT_OPT_SPARSEREAD)
+               ceph_set_mount_opt(fsc, SPARSEREAD);
+       else
+               ceph_clear_mount_opt(fsc, SPARSEREAD);
+
        if (strcmp_null(fsc->mount_options->mon_addr, fsopt->mon_addr)) {
                kfree(fsc->mount_options->mon_addr);
                fsc->mount_options->mon_addr = fsopt->mon_addr;

diff --git a/fs/ceph/super.h b/fs/ceph/super.h
index 3bfddf34d488b3216bcbea56a881a20f8311df19..b9cad57ce6063c86d10e438e1777ee0aa21ec30f 100644 (file)
--- a/fs/ceph/super.h
+++ b/fs/ceph/super.h
@@ -42,6 +42,7 @@
 #define CEPH_MOUNT_OPT_NOCOPYFROM      (1<<14) /* don't use RADOS 'copy-from' op */
 #define CEPH_MOUNT_OPT_ASYNC_DIROPS    (1<<15) /* allow async directory ops */
 #define CEPH_MOUNT_OPT_NOPAGECACHE     (1<<16) /* bypass pagecache altogether */
+#define CEPH_MOUNT_OPT_SPARSEREAD      (1<<17) /* always do sparse reads */
 
 #define CEPH_MOUNT_OPT_DEFAULT                 \
        (CEPH_MOUNT_OPT_DCACHE |                \