LU-12043 llite,readahead: don't always use max RPC size

[fs/lustre-release.git] / lustre / llite / rw.c

diff --git a/lustre/llite/rw.c b/lustre/llite/rw.c
index 0b0b0c2..a5f3f9c 100644 (file)
--- a/lustre/llite/rw.c
+++ b/lustre/llite/rw.c
@@ -23,7 +23,7 @@
  * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
  * Use is subject to license terms.
  *
- * Copyright (c) 2011, 2016, Intel Corporation.
+ * Copyright (c) 2011, 2017, Intel Corporation.
  */
 /*
  * This file is part of Lustre, http://www.lustre.org/
@@ -50,6 +50,7 @@
 #include <linux/pagemap.h>
 /* current_is_kswapd() */
 #include <linux/swap.h>
+#include <linux/task_io_accounting_ops.h>
 
 #define DEBUG_SUBSYSTEM S_LLITE
 
@@ -368,7 +369,7 @@ ll_read_ahead_pages(const struct lu_env *env, struct cl_io *io,
                                         io->ci_obj, ra.cra_end, page_idx);
                                /* update read ahead RPC size.
                                 * NB: it's racy but doesn't matter */
-                               if (ras->ras_rpc_size > ra.cra_rpc_size &&
+                               if (ras->ras_rpc_size != ra.cra_rpc_size &&
                                    ra.cra_rpc_size > 0)
                                        ras->ras_rpc_size = ra.cra_rpc_size;
                                /* trim it to align with optimal RPC size */
@@ -1076,7 +1077,7 @@ void ll_cl_remove(struct file *file, const struct lu_env *env)
        write_unlock(&fd->fd_lock);
 }
 
-static int ll_io_read_page(const struct lu_env *env, struct cl_io *io,
+int ll_io_read_page(const struct lu_env *env, struct cl_io *io,
                           struct cl_page *page, struct file *file)
 {
        struct inode              *inode  = vvp_object_inode(page->cp_obj);
@@ -1084,6 +1085,7 @@ static int ll_io_read_page(const struct lu_env *env, struct cl_io *io,
        struct ll_file_data       *fd     = LUSTRE_FPRIVATE(file);
        struct ll_readahead_state *ras    = &fd->fd_ras;
        struct cl_2queue          *queue  = &io->ci_queue;
+       struct cl_sync_io         *anchor = NULL;
        struct vvp_page           *vpg;
        int                        rc = 0;
        bool                       uptodate;
@@ -1111,6 +1113,10 @@ static int ll_io_read_page(const struct lu_env *env, struct cl_io *io,
                cl_page_export(env, page, 1);
                cl_page_disown(env, io, page);
        } else {
+               anchor = &vvp_env_info(env)->vti_anchor;
+               cl_sync_io_init(anchor, 1, &cl_sync_io_end);
+               page->cp_sync_io = anchor;
+
                cl_2queue_add(queue, page);
        }
 
@@ -1124,13 +1130,37 @@ static int ll_io_read_page(const struct lu_env *env, struct cl_io *io,
                       PFID(ll_inode2fid(inode)), rc2, vvp_index(vpg));
        }
 
-       if (queue->c2_qin.pl_nr > 0)
+       if (queue->c2_qin.pl_nr > 0) {
+               int count = queue->c2_qin.pl_nr;
                rc = cl_io_submit_rw(env, io, CRT_READ, queue);
+               if (rc == 0)
+                       task_io_account_read(PAGE_SIZE * count);
+       }
 
-       /*
-        * Unlock unsent pages in case of error.
-        */
+
+       if (anchor != NULL && !cl_page_is_owned(page, io)) { /* have sent */
+               rc = cl_sync_io_wait(env, anchor, 0);
+
+               cl_page_assume(env, io, page);
+               cl_page_list_del(env, &queue->c2_qout, page);
+
+               if (!PageUptodate(cl_page_vmpage(page))) {
+                       /* Failed to read a mirror, discard this page so that
+                        * new page can be created with new mirror.
+                        *
+                        * TODO: this is not needed after page reinit
+                        * route is implemented */
+                       cl_page_discard(env, io, page);
+               }
+               cl_page_disown(env, io, page);
+       }
+
+       /* TODO: discard all pages until page reinit route is implemented */
+       cl_page_list_discard(env, io, &queue->c2_qin);
+
+       /* Unlock unsent read pages in case of error. */
        cl_page_list_disown(env, io, &queue->c2_qin);
+
        cl_2queue_fini(env, queue);
 
        RETURN(rc);
@@ -1158,6 +1188,8 @@ int ll_readpage(struct file *file, struct page *vmpage)
                struct ll_file_data *fd = LUSTRE_FPRIVATE(file);
                struct ll_readahead_state *ras = &fd->fd_ras;
                struct lu_env  *local_env = NULL;
+               unsigned long fast_read_pages =
+                       max(RA_REMAIN_WINDOW_MIN, ras->ras_rpc_size);
                struct vvp_page *vpg;
 
                result = -ENODATA;
@@ -1194,7 +1226,7 @@ int ll_readpage(struct file *file, struct page *vmpage)
                         * the case, we can't do fast IO because we will need
                         * a cl_io to issue the RPC. */
                        if (ras->ras_window_start + ras->ras_window_len <
-                           ras->ras_next_readahead + PTLRPC_MAX_BRW_PAGES) {
+                           ras->ras_next_readahead + fast_read_pages) {
                                /* export the page and skip io stack */
                                vpg->vpg_ra_used = 1;
                                cl_page_export(env, page, 1);
@@ -1219,6 +1251,7 @@ int ll_readpage(struct file *file, struct page *vmpage)
                LASSERT(page->cp_type == CPT_CACHEABLE);
                if (likely(!PageUptodate(vmpage))) {
                        cl_page_assume(env, io, page);
+
                        result = ll_io_read_page(env, io, page, file);
                } else {
                        /* Page from a non-object file. */
@@ -1232,28 +1265,3 @@ int ll_readpage(struct file *file, struct page *vmpage)
         }
        RETURN(result);
 }
-
-int ll_page_sync_io(const struct lu_env *env, struct cl_io *io,
-                   struct cl_page *page, enum cl_req_type crt)
-{
-       struct cl_2queue  *queue;
-       int result;
-
-       LASSERT(io->ci_type == CIT_READ || io->ci_type == CIT_WRITE);
-
-       queue = &io->ci_queue;
-       cl_2queue_init_page(queue, page);
-
-       result = cl_io_submit_sync(env, io, crt, queue, 0);
-       LASSERT(cl_page_is_owned(page, io));
-
-       if (crt == CRT_READ)
-               /*
-                * in CRT_WRITE case page is left locked even in case of
-                * error.
-                */
-               cl_page_list_disown(env, io, &queue->c2_qin);
-       cl_2queue_fini(env, queue);
-
-       return result;
-}