lustre/llite/rw.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2002, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2011, 2017, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  *
  31  * lustre/llite/rw.c
  32  *
  33  * Lustre Lite I/O page cache routines shared by different kernel revs
  34  */
  35
  36 #include <linux/kernel.h>
  37 #include <linux/mm.h>
  38 #include <linux/string.h>
  39 #include <linux/stat.h>
  40 #include <linux/errno.h>
  41 #include <linux/unistd.h>
  42 #include <linux/writeback.h>
  43 #include <asm/uaccess.h>
  44
  45 #include <linux/fs.h>
  46 #include <linux/file.h>
  47 #include <linux/stat.h>
  48 #include <asm/uaccess.h>
  49 #include <linux/mm.h>
  50 #include <linux/pagemap.h>
  51 /* current_is_kswapd() */
  52 #include <linux/swap.h>
  53 #include <linux/task_io_accounting_ops.h>
  54
  55 #define DEBUG_SUBSYSTEM S_LLITE
  56
  57 #include <obd_cksum.h>
  58 #include "llite_internal.h"
  59 #include <lustre_compat.h>
  60
  61 static void ll_ra_stats_inc_sbi(struct ll_sb_info *sbi, enum ra_stat which);
  62
  63 /**
  64  * Get readahead pages from the filesystem readahead pool of the client for a
  65  * thread.
  66  *
  67  * /param sbi superblock for filesystem readahead state ll_ra_info
  68  * /param ria per-thread readahead state
  69  * /param pages number of pages requested for readahead for the thread.
  70  *
  71  * WARNING: This algorithm is used to reduce contention on sbi->ll_lock.
  72  * It should work well if the ra_max_pages is much greater than the single
  73  * file's read-ahead window, and not too many threads contending for
  74  * these readahead pages.
  75  *
  76  * TODO: There may be a 'global sync problem' if many threads are trying
  77  * to get an ra budget that is larger than the remaining readahead pages
  78  * and reach here at exactly the same time. They will compute /a ret to
  79  * consume the remaining pages, but will fail at atomic_add_return() and
  80  * get a zero ra window, although there is still ra space remaining. - Jay */
  81
  82 static unsigned long ll_ra_count_get(struct ll_sb_info *sbi,
  83                                      struct ra_io_arg *ria,
  84                                      unsigned long pages,
  85                                      unsigned long pages_min)
  86 {
  87         struct ll_ra_info *ra = &sbi->ll_ra_info;
  88         long ret;
  89
  90         ENTRY;
  91
  92         WARN_ON_ONCE(pages_min > pages);
  93         /**
  94          * Don't try readahead aggresively if we are limited
  95          * LRU pages, otherwise, it could cause deadlock.
  96          */
  97         pages = min(sbi->ll_cache->ccc_lru_max >> 2, pages);
  98         /**
  99          * if this happen, we reserve more pages than needed,
 100          * this will make us leak @ra_cur_pages, because
 101          * ll_ra_count_put() acutally freed @pages.
 102          */
 103         if (unlikely(pages_min > pages))
 104                 pages_min = pages;
 105
 106         /*
 107          * If read-ahead pages left are less than 1M, do not do read-ahead,
 108          * otherwise it will form small read RPC(< 1M), which hurt server
 109          * performance a lot.
 110          */
 111         ret = min(ra->ra_max_pages - atomic_read(&ra->ra_cur_pages),
 112                   pages);
 113         if (ret < 0 || ret < min_t(long, PTLRPC_MAX_BRW_PAGES, pages))
 114                 GOTO(out, ret = 0);
 115
 116         if (atomic_add_return(ret, &ra->ra_cur_pages) > ra->ra_max_pages) {
 117                 atomic_sub(ret, &ra->ra_cur_pages);
 118                 ret = 0;
 119         }
 120
 121 out:
 122         if (ret < pages_min) {
 123                 /* override ra limit for maximum performance */
 124                 atomic_add(pages_min - ret, &ra->ra_cur_pages);
 125                 ret = pages_min;
 126         }
 127         RETURN(ret);
 128 }
 129
 130 void ll_ra_count_put(struct ll_sb_info *sbi, unsigned long pages)
 131 {
 132         struct ll_ra_info *ra = &sbi->ll_ra_info;
 133         atomic_sub(pages, &ra->ra_cur_pages);
 134 }
 135
 136 static void ll_ra_stats_inc_sbi(struct ll_sb_info *sbi, enum ra_stat which)
 137 {
 138         LASSERTF(which < _NR_RA_STAT, "which: %u\n", which);
 139         lprocfs_counter_incr(sbi->ll_ra_stats, which);
 140 }
 141
 142 static inline bool ll_readahead_enabled(struct ll_sb_info *sbi)
 143 {
 144         return sbi->ll_ra_info.ra_max_pages_per_file > 0 &&
 145                 sbi->ll_ra_info.ra_max_pages > 0;
 146 }
 147
 148 void ll_ra_stats_inc(struct inode *inode, enum ra_stat which)
 149 {
 150         struct ll_sb_info *sbi = ll_i2sbi(inode);
 151
 152         ll_ra_stats_inc_sbi(sbi, which);
 153 }
 154
 155 static void ll_ra_stats_add(struct inode *inode, enum ra_stat which, long count)
 156 {
 157         struct ll_sb_info *sbi = ll_i2sbi(inode);
 158
 159         LASSERTF(which < _NR_RA_STAT, "which: %u\n", which);
 160         lprocfs_counter_add(sbi->ll_ra_stats, which, count);
 161 }
 162
 163 #define RAS_CDEBUG(ras) \
 164         CDEBUG(D_READA,                                                      \
 165                "lre %llu cr %lu cb %llu wsi %lu wp %lu nra %lu rpc %lu "     \
 166                "r %lu csr %lu so %llu sb %llu sl %llu lr %lu\n",             \
 167                ras->ras_last_read_end_bytes, ras->ras_consecutive_requests,  \
 168                ras->ras_consecutive_bytes, ras->ras_window_start_idx,        \
 169                ras->ras_window_pages, ras->ras_next_readahead_idx,           \
 170                ras->ras_rpc_pages, ras->ras_requests,                        \
 171                ras->ras_consecutive_stride_requests, ras->ras_stride_offset, \
 172                ras->ras_stride_bytes, ras->ras_stride_length,                \
 173                ras->ras_async_last_readpage_idx)
 174
 175 static bool pos_in_window(loff_t pos, loff_t point,
 176                           unsigned long before, unsigned long after)
 177 {
 178         loff_t start = point - before;
 179         loff_t end = point + after;
 180
 181         if (start > point)
 182                 start = 0;
 183         if (end < point)
 184                 end = ~0;
 185
 186         return start <= pos && pos <= end;
 187 }
 188
 189 enum ll_ra_page_hint {
 190         MAYNEED = 0, /* this page possibly accessed soon */
 191         WILLNEED /* this page is gurateed to be needed */
 192 };
 193
 194 /**
 195  * Initiates read-ahead of a page with given index.
 196  *
 197  * \retval +ve: page was already uptodate so it will be skipped
 198  *              from being added;
 199  * \retval -ve: page wasn't added to \a queue for error;
 200  * \retval   0: page was added into \a queue for read ahead.
 201  */
 202 static int ll_read_ahead_page(const struct lu_env *env, struct cl_io *io,
 203                               struct cl_page_list *queue, pgoff_t index,
 204                               enum ll_ra_page_hint hint)
 205 {
 206         struct cl_object *clob  = io->ci_obj;
 207         struct inode     *inode = vvp_object_inode(clob);
 208         struct page      *vmpage = NULL;
 209         struct cl_page   *cp;
 210         enum ra_stat      which = _NR_RA_STAT; /* keep gcc happy */
 211         int               rc    = 0;
 212         const char       *msg   = NULL;
 213
 214         ENTRY;
 215
 216         switch (hint) {
 217         case MAYNEED:
 218                 /*
 219                  * We need __GFP_NORETRY here for read-ahead page, otherwise
 220                  * the process will fail with OOM killed due to memcg limit.
 221                  * See @readahead_gfp_mask for an example.
 222                  */
 223                 vmpage = pagecache_get_page(inode->i_mapping, index,
 224                                             FGP_LOCK | FGP_CREAT |
 225                                             FGP_NOFS | FGP_NOWAIT,
 226                                             mapping_gfp_mask(inode->i_mapping) |
 227                                             __GFP_NORETRY | __GFP_NOWARN);
 228                 if (vmpage == NULL) {
 229                         which = RA_STAT_FAILED_GRAB_PAGE;
 230                         msg   = "g_c_p_n failed";
 231                         GOTO(out, rc = -EBUSY);
 232                 }
 233                 break;
 234         case WILLNEED:
 235                 vmpage = find_or_create_page(inode->i_mapping, index,
 236                                              GFP_NOFS);
 237                 if (vmpage == NULL)
 238                         GOTO(out, rc = -ENOMEM);
 239                 break;
 240         default:
 241                 /* should not come here */
 242                 GOTO(out, rc = -EINVAL);
 243         }
 244
 245         /* Check if vmpage was truncated or reclaimed */
 246         if (vmpage->mapping != inode->i_mapping) {
 247                 which = RA_STAT_WRONG_GRAB_PAGE;
 248                 msg   = "g_c_p_n returned invalid page";
 249                 GOTO(out, rc = -EBUSY);
 250         }
 251
 252         cp = cl_page_find(env, clob, vmpage->index, vmpage, CPT_CACHEABLE);
 253         if (IS_ERR(cp)) {
 254                 which = RA_STAT_FAILED_GRAB_PAGE;
 255                 msg   = "cl_page_find failed";
 256                 GOTO(out, rc = PTR_ERR(cp));
 257         }
 258
 259         lu_ref_add(&cp->cp_reference, "ra", current);
 260         cl_page_assume(env, io, cp);
 261
 262         if (!cp->cp_defer_uptodate && !PageUptodate(vmpage)) {
 263                 if (hint == MAYNEED) {
 264                         cp->cp_defer_uptodate = 1;
 265                         cp->cp_ra_used = 0;
 266                 }
 267
 268                 cl_page_list_add(queue, cp, true);
 269         } else {
 270                 /* skip completed pages */
 271                 cl_page_unassume(env, io, cp);
 272                 /* This page is already uptodate, returning a positive number
 273                  * to tell the callers about this */
 274                 rc = 1;
 275         }
 276
 277         lu_ref_del(&cp->cp_reference, "ra", current);
 278         cl_page_put(env, cp);
 279
 280 out:
 281         if (vmpage != NULL) {
 282                 if (rc != 0)
 283                         unlock_page(vmpage);
 284                 put_page(vmpage);
 285         }
 286         if (msg != NULL && hint == MAYNEED) {
 287                 ll_ra_stats_inc(inode, which);
 288                 CDEBUG(D_READA, "%s\n", msg);
 289
 290         }
 291
 292         RETURN(rc);
 293 }
 294
 295 #define RIA_DEBUG(ria)                                                  \
 296         CDEBUG(D_READA, "rs %lu re %lu ro %llu rl %llu rb %llu\n",      \
 297                ria->ria_start_idx, ria->ria_end_idx, ria->ria_stoff,    \
 298                ria->ria_length, ria->ria_bytes)
 299
 300 static inline int stride_io_mode(struct ll_readahead_state *ras)
 301 {
 302         return ras->ras_consecutive_stride_requests > 1;
 303 }
 304
 305 /* The function calculates how many bytes will be read in
 306  * [off, off + length], in such stride IO area,
 307  * stride_offset = st_off, stride_lengh = st_len,
 308  * stride_bytes = st_bytes
 309  *
 310  *   |------------------|*****|------------------|*****|------------|*****|....
 311  * st_off
 312  *   |--- st_bytes     ---|
 313  *   |-----     st_len   -----|
 314  *
 315  *              How many bytes it should read in such pattern
 316  *              |-------------------------------------------------------------|
 317  *              off
 318  *              |<------                  length                      ------->|
 319  *
 320  *          =   |<----->|  +  |-------------------------------------| +   |---|
 321  *             start_left                 st_bytes * i                 end_left
 322  */
 323 static loff_t stride_byte_count(loff_t st_off, loff_t st_len, loff_t st_bytes,
 324                                 loff_t off, loff_t length)
 325 {
 326         u64 start = off > st_off ? off - st_off : 0;
 327         u64 end = off + length > st_off ? off + length - st_off : 0;
 328         u64 start_left;
 329         u64 end_left;
 330         u64 bytes_count;
 331
 332         if (st_len == 0 || length == 0 || end == 0)
 333                 return length;
 334
 335         start = div64_u64_rem(start, st_len, &start_left);
 336         if (start_left < st_bytes)
 337                 start_left = st_bytes - start_left;
 338         else
 339                 start_left = 0;
 340
 341         end = div64_u64_rem(end, st_len, &end_left);
 342         if (end_left > st_bytes)
 343                 end_left = st_bytes;
 344
 345         CDEBUG(D_READA, "start %llu, end %llu start_left %llu end_left %llu\n",
 346                start, end, start_left, end_left);
 347
 348         if (start == end)
 349                 bytes_count = end_left - (st_bytes - start_left);
 350         else
 351                 bytes_count = start_left +
 352                         st_bytes * (end - start - 1) + end_left;
 353
 354         CDEBUG(D_READA,
 355                "st_off %llu, st_len %llu st_bytes %llu off %llu length %llu bytescount %llu\n",
 356                st_off, st_len, st_bytes, off, length, bytes_count);
 357
 358         return bytes_count;
 359 }
 360
 361 static unsigned long ria_page_count(struct ra_io_arg *ria)
 362 {
 363         loff_t length_bytes = ria->ria_end_idx >= ria->ria_start_idx ?
 364                 (loff_t)(ria->ria_end_idx -
 365                          ria->ria_start_idx + 1) << PAGE_SHIFT : 0;
 366         loff_t bytes_count;
 367
 368         if (ria->ria_length > ria->ria_bytes && ria->ria_bytes &&
 369             (ria->ria_length & ~PAGE_MASK || ria->ria_bytes & ~PAGE_MASK ||
 370              ria->ria_stoff & ~PAGE_MASK)) {
 371                 /* Over-estimate un-aligned page stride read */
 372                 unsigned long pg_count = ((ria->ria_bytes +
 373                                            PAGE_SIZE - 1) >> PAGE_SHIFT) + 1;
 374                 pg_count *= length_bytes / ria->ria_length + 1;
 375
 376                 return pg_count;
 377         }
 378         bytes_count = stride_byte_count(ria->ria_stoff, ria->ria_length,
 379                                         ria->ria_bytes,
 380                                         (loff_t)ria->ria_start_idx<<PAGE_SHIFT,
 381                                         length_bytes);
 382         return (bytes_count + PAGE_SIZE - 1) >> PAGE_SHIFT;
 383 }
 384
 385 static pgoff_t ras_align(struct ll_readahead_state *ras, pgoff_t index)
 386 {
 387         unsigned opt_size = min(ras->ras_window_pages, ras->ras_rpc_pages);
 388
 389         if (opt_size == 0)
 390                 opt_size = 1;
 391         return index - (index % opt_size);
 392 }
 393
 394 /* Check whether the index is in the defined ra-window */
 395 static bool ras_inside_ra_window(pgoff_t idx, struct ra_io_arg *ria)
 396 {
 397         loff_t pos = (loff_t)idx << PAGE_SHIFT;
 398
 399         /* If ria_length == ria_bytes, it means non-stride I/O mode,
 400          * idx should always inside read-ahead window in this case
 401          * For stride I/O mode, just check whether the idx is inside
 402          * the ria_bytes.
 403          */
 404         if (ria->ria_length == 0 || ria->ria_length == ria->ria_bytes)
 405                 return true;
 406
 407         if (pos >= ria->ria_stoff) {
 408                 u64 offset;
 409
 410                 div64_u64_rem(pos - ria->ria_stoff, ria->ria_length, &offset);
 411
 412                 if (offset < ria->ria_bytes ||
 413                     (ria->ria_length - offset) < PAGE_SIZE)
 414                         return true;
 415         } else if (pos + PAGE_SIZE > ria->ria_stoff) {
 416                 return true;
 417         }
 418
 419         return false;
 420 }
 421
 422 static unsigned long
 423 ll_read_ahead_pages(const struct lu_env *env, struct cl_io *io,
 424                     struct cl_page_list *queue, struct ll_readahead_state *ras,
 425                     struct ra_io_arg *ria, pgoff_t *ra_end, pgoff_t skip_index)
 426 {
 427         struct cl_read_ahead ra = { 0 };
 428         /* busy page count is per stride */
 429         int rc = 0, count = 0, busy_page_count = 0;
 430         pgoff_t page_idx;
 431
 432         LASSERT(ria != NULL);
 433         RIA_DEBUG(ria);
 434
 435         for (page_idx = ria->ria_start_idx;
 436              page_idx <= ria->ria_end_idx && ria->ria_reserved > 0;
 437              page_idx++) {
 438                 if (skip_index && page_idx == skip_index)
 439                         continue;
 440                 if (ras_inside_ra_window(page_idx, ria)) {
 441                         if (ra.cra_end_idx == 0 || ra.cra_end_idx < page_idx) {
 442                                 pgoff_t end_idx;
 443
 444                                 /*
 445                                  * Do not shrink ria_end_idx at any case until
 446                                  * the minimum end of current read is covered.
 447                                  *
 448                                  * Do not extend read lock accross stripe if
 449                                  * lock contention detected.
 450                                  */
 451                                 if (ra.cra_contention &&
 452                                     page_idx > ria->ria_end_idx_min) {
 453                                         ria->ria_end_idx = *ra_end;
 454                                         break;
 455                                 }
 456
 457                                 cl_read_ahead_release(env, &ra);
 458
 459                                 rc = cl_io_read_ahead(env, io, page_idx, &ra);
 460                                 if (rc < 0)
 461                                         break;
 462
 463                                  /*
 464                                   * Only shrink ria_end_idx if the matched
 465                                   * LDLM lock doesn't cover more.
 466                                   */
 467                                 if (page_idx > ra.cra_end_idx) {
 468                                         ria->ria_end_idx = ra.cra_end_idx;
 469                                         break;
 470                                 }
 471
 472                                 CDEBUG(D_READA, "idx: %lu, ra: %lu, rpc: %lu\n",
 473                                        page_idx, ra.cra_end_idx,
 474                                        ra.cra_rpc_pages);
 475                                 LASSERTF(ra.cra_end_idx >= page_idx,
 476                                          "object: %px, indcies %lu / %lu\n",
 477                                          io->ci_obj, ra.cra_end_idx, page_idx);
 478                                 /* update read ahead RPC size.
 479                                  * NB: it's racy but doesn't matter */
 480                                 if (ras->ras_rpc_pages != ra.cra_rpc_pages &&
 481                                     ra.cra_rpc_pages > 0)
 482                                         ras->ras_rpc_pages = ra.cra_rpc_pages;
 483                                 if (!skip_index) {
 484                                         /* trim it to align with optimal RPC size */
 485                                         end_idx = ras_align(ras, ria->ria_end_idx + 1);
 486                                         if (end_idx > 0 && !ria->ria_eof)
 487                                                 ria->ria_end_idx = end_idx - 1;
 488                                 }
 489                                 if (ria->ria_end_idx < ria->ria_end_idx_min)
 490                                         ria->ria_end_idx = ria->ria_end_idx_min;
 491                         }
 492                         if (page_idx > ria->ria_end_idx)
 493                                 break;
 494
 495                         /* If the page is inside the read-ahead window */
 496                         rc = ll_read_ahead_page(env, io, queue, page_idx,
 497                                                 MAYNEED);
 498                         if (rc < 0 && rc != -EBUSY)
 499                                 break;
 500                         if (rc == -EBUSY) {
 501                                 busy_page_count++;
 502                                 CDEBUG(D_READA,
 503                                        "skip busy page: %lu\n", page_idx);
 504                                 /* For page unaligned readahead the first
 505                                  * last pages of each region can be read by
 506                                  * another reader on the same node, and so
 507                                  * may be busy. So only stop for > 2 busy
 508                                  * pages. */
 509                                 if (busy_page_count > 2)
 510                                         break;
 511                         }
 512
 513                         *ra_end = page_idx;
 514                         /* Only subtract from reserve & count the page if we
 515                          * really did readahead on that page. */
 516                         if (rc == 0) {
 517                                 ria->ria_reserved--;
 518                                 count++;
 519                         }
 520                 } else if (stride_io_mode(ras)) {
 521                         /* If it is not in the read-ahead window, and it is
 522                          * read-ahead mode, then check whether it should skip
 523                          * the stride gap.
 524                          */
 525                         loff_t pos = (loff_t)page_idx << PAGE_SHIFT;
 526                         u64 offset;
 527
 528                         div64_u64_rem(pos - ria->ria_stoff, ria->ria_length,
 529                                       &offset);
 530                         if (offset >= ria->ria_bytes) {
 531                                 pos += (ria->ria_length - offset);
 532                                 if ((pos >> PAGE_SHIFT) >= page_idx + 1)
 533                                         page_idx = (pos >> PAGE_SHIFT) - 1;
 534                                 busy_page_count = 0;
 535                                 CDEBUG(D_READA,
 536                                        "Stride: jump %llu pages to %lu\n",
 537                                        ria->ria_length - offset, page_idx);
 538                                 continue;
 539                         }
 540                 }
 541         }
 542
 543         cl_read_ahead_release(env, &ra);
 544
 545         if (count)
 546                 ll_ra_stats_add(vvp_object_inode(io->ci_obj),
 547                                 RA_STAT_READAHEAD_PAGES, count);
 548
 549         return count;
 550 }
 551
 552 static void ll_readahead_work_free(struct ll_readahead_work *work)
 553 {
 554         fput(work->lrw_file);
 555         OBD_FREE_PTR(work);
 556 }
 557
 558 static void ll_readahead_handle_work(struct work_struct *wq);
 559 static void ll_readahead_work_add(struct inode *inode,
 560                                   struct ll_readahead_work *work)
 561 {
 562         INIT_WORK(&work->lrw_readahead_work, ll_readahead_handle_work);
 563         queue_work(ll_i2sbi(inode)->ll_ra_info.ll_readahead_wq,
 564                    &work->lrw_readahead_work);
 565 }
 566
 567 static int ll_readahead_file_kms(const struct lu_env *env,
 568                                 struct cl_io *io, __u64 *kms)
 569 {
 570         struct cl_object *clob;
 571         struct inode *inode;
 572         struct cl_attr *attr = vvp_env_thread_attr(env);
 573         int ret;
 574
 575         clob = io->ci_obj;
 576         inode = vvp_object_inode(clob);
 577
 578         cl_object_attr_lock(clob);
 579         ret = cl_object_attr_get(env, clob, attr);
 580         cl_object_attr_unlock(clob);
 581
 582         if (ret != 0)
 583                 RETURN(ret);
 584
 585         *kms = attr->cat_kms;
 586         return 0;
 587 }
 588
 589 static void ll_readahead_handle_work(struct work_struct *wq)
 590 {
 591         struct ll_readahead_work *work;
 592         struct lu_env *env;
 593         __u16 refcheck;
 594         struct ra_io_arg *ria;
 595         struct inode *inode;
 596         struct ll_file_data *fd;
 597         struct ll_readahead_state *ras;
 598         struct cl_io *io;
 599         struct cl_2queue *queue;
 600         pgoff_t ra_end_idx = 0;
 601         unsigned long pages, pages_min = 0;
 602         struct file *file;
 603         __u64 kms;
 604         int rc;
 605         pgoff_t eof_index;
 606         struct ll_sb_info *sbi;
 607
 608         work = container_of(wq, struct ll_readahead_work,
 609                             lrw_readahead_work);
 610         fd = work->lrw_file->private_data;
 611         ras = &fd->fd_ras;
 612         file = work->lrw_file;
 613         inode = file_inode(file);
 614         sbi = ll_i2sbi(inode);
 615
 616         CDEBUG(D_READA|D_IOTRACE,
 617                "%s:"DFID": async ra from %lu to %lu triggered by user pid %d\n",
 618                file_dentry(file)->d_name.name, PFID(ll_inode2fid(inode)),
 619                work->lrw_start_idx, work->lrw_end_idx, work->lrw_user_pid);
 620
 621         env = cl_env_alloc(&refcheck, LCT_NOREF);
 622         if (IS_ERR(env))
 623                 GOTO(out_free_work, rc = PTR_ERR(env));
 624
 625         io = vvp_env_thread_io(env);
 626         ll_io_init(io, file, CIT_READ, NULL);
 627
 628         rc = ll_readahead_file_kms(env, io, &kms);
 629         if (rc != 0)
 630                 GOTO(out_put_env, rc);
 631
 632         if (kms == 0) {
 633                 ll_ra_stats_inc(inode, RA_STAT_ZERO_LEN);
 634                 GOTO(out_put_env, rc = 0);
 635         }
 636
 637         ria = &ll_env_info(env)->lti_ria;
 638         memset(ria, 0, sizeof(*ria));
 639
 640         ria->ria_start_idx = work->lrw_start_idx;
 641         /* Truncate RA window to end of file */
 642         eof_index = (pgoff_t)(kms - 1) >> PAGE_SHIFT;
 643         if (eof_index <= work->lrw_end_idx) {
 644                 work->lrw_end_idx = eof_index;
 645                 ria->ria_eof = true;
 646         }
 647         if (work->lrw_end_idx <= work->lrw_start_idx)
 648                 GOTO(out_put_env, rc = 0);
 649
 650         ria->ria_end_idx = work->lrw_end_idx;
 651         pages = ria->ria_end_idx - ria->ria_start_idx + 1;
 652         ria->ria_reserved = ll_ra_count_get(sbi, ria,
 653                                             ria_page_count(ria), pages_min);
 654
 655         CDEBUG(D_READA,
 656                "async reserved pages: %lu/%lu/%lu, ra_cur %d, ra_max %lu\n",
 657                ria->ria_reserved, pages, pages_min,
 658                atomic_read(&ll_i2sbi(inode)->ll_ra_info.ra_cur_pages),
 659                ll_i2sbi(inode)->ll_ra_info.ra_max_pages);
 660
 661         if (ria->ria_reserved < pages) {
 662                 ll_ra_stats_inc(inode, RA_STAT_MAX_IN_FLIGHT);
 663                 if (PAGES_TO_MiB(ria->ria_reserved) < 1) {
 664                         ll_ra_count_put(ll_i2sbi(inode), ria->ria_reserved);
 665                         GOTO(out_put_env, rc = 0);
 666                 }
 667         }
 668
 669         rc = cl_io_rw_init(env, io, CIT_READ, ria->ria_start_idx, pages);
 670         if (rc)
 671                 GOTO(out_put_env, rc);
 672
 673         /* overwrite jobid inited in vvp_io_init() */
 674         if (strncmp(ll_i2info(inode)->lli_jobid, work->lrw_jobid,
 675                     sizeof(work->lrw_jobid)))
 676                 memcpy(ll_i2info(inode)->lli_jobid, work->lrw_jobid,
 677                        sizeof(work->lrw_jobid));
 678
 679         vvp_env_io(env)->vui_fd = fd;
 680         io->ci_state = CIS_LOCKED;
 681         io->ci_async_readahead = true;
 682         rc = cl_io_start(env, io);
 683         if (rc)
 684                 GOTO(out_io_fini, rc);
 685
 686         queue = &io->ci_queue;
 687         cl_2queue_init(queue);
 688
 689         rc = ll_read_ahead_pages(env, io, &queue->c2_qin, ras, ria,
 690                                  &ra_end_idx, 0);
 691         if (ria->ria_reserved != 0)
 692                 ll_ra_count_put(ll_i2sbi(inode), ria->ria_reserved);
 693         if (queue->c2_qin.pl_nr > 0) {
 694                 int count = queue->c2_qin.pl_nr;
 695
 696                 rc = cl_io_submit_rw(env, io, CRT_READ, queue);
 697                 if (rc == 0)
 698                         task_io_account_read(PAGE_SIZE * count);
 699         }
 700         if (ria->ria_end_idx == ra_end_idx && ra_end_idx == (kms >> PAGE_SHIFT))
 701                 ll_ra_stats_inc(inode, RA_STAT_EOF);
 702
 703         if (ra_end_idx != ria->ria_end_idx)
 704                 ll_ra_stats_inc(inode, RA_STAT_FAILED_REACH_END);
 705
 706         /* TODO: discard all pages until page reinit route is implemented */
 707         cl_page_list_discard(env, io, &queue->c2_qin);
 708
 709         /* Unlock unsent read pages in case of error. */
 710         cl_page_list_disown(env, &queue->c2_qin);
 711
 712         cl_2queue_fini(env, queue);
 713 out_io_fini:
 714         cl_io_end(env, io);
 715         cl_io_fini(env, io);
 716 out_put_env:
 717         cl_env_put(env, &refcheck);
 718 out_free_work:
 719         if (ra_end_idx > 0)
 720                 ll_ra_stats_inc_sbi(ll_i2sbi(inode), RA_STAT_ASYNC);
 721         atomic_dec(&sbi->ll_ra_info.ra_async_inflight);
 722         ll_readahead_work_free(work);
 723 }
 724
 725 static int ll_readahead(const struct lu_env *env, struct cl_io *io,
 726                         struct cl_page_list *queue,
 727                         struct ll_readahead_state *ras, bool hit,
 728                         struct file *file, pgoff_t skip_index,
 729                         pgoff_t *start_idx)
 730 {
 731         struct vvp_io *vio = vvp_env_io(env);
 732         struct ll_thread_info *lti = ll_env_info(env);
 733         unsigned long pages, pages_min = 0;
 734         pgoff_t ra_end_idx = 0, end_idx = 0;
 735         struct inode *inode;
 736         struct ra_io_arg *ria = &lti->lti_ria;
 737         struct cl_object *clob;
 738         int ret = 0;
 739         __u64 kms;
 740         struct ll_sb_info *sbi;
 741         struct ll_ra_info *ra;
 742
 743         ENTRY;
 744
 745         clob = io->ci_obj;
 746         inode = vvp_object_inode(clob);
 747         sbi = ll_i2sbi(inode);
 748         ra = &sbi->ll_ra_info;
 749
 750         /**
 751          * In case we have a limited max_cached_mb, readahead
 752          * should be stopped if it have run out of all LRU slots.
 753          */
 754         if (atomic_read(&ra->ra_cur_pages) >= sbi->ll_cache->ccc_lru_max) {
 755                 ll_ra_stats_inc(inode, RA_STAT_MAX_IN_FLIGHT);
 756                 RETURN(0);
 757         }
 758
 759         memset(ria, 0, sizeof(*ria));
 760         ret = ll_readahead_file_kms(env, io, &kms);
 761         if (ret != 0)
 762                 RETURN(ret);
 763
 764         if (kms == 0) {
 765                 ll_ra_stats_inc(inode, RA_STAT_ZERO_LEN);
 766                 RETURN(0);
 767         }
 768
 769         spin_lock(&ras->ras_lock);
 770
 771         /**
 772          * Note: other thread might rollback the ras_next_readahead_idx,
 773          * if it can not get the full size of prepared pages, see the
 774          * end of this function. For stride read ahead, it needs to
 775          * make sure the offset is no less than ras_stride_offset,
 776          * so that stride read ahead can work correctly.
 777          */
 778         if (stride_io_mode(ras))
 779                 *start_idx = max_t(pgoff_t, ras->ras_next_readahead_idx,
 780                                   ras->ras_stride_offset >> PAGE_SHIFT);
 781         else
 782                 *start_idx = ras->ras_next_readahead_idx;
 783
 784         if (ras->ras_window_pages > 0)
 785                 end_idx = ras->ras_window_start_idx + ras->ras_window_pages - 1;
 786
 787         if (skip_index)
 788                 end_idx = *start_idx + ras->ras_window_pages - 1;
 789
 790         /* Enlarge the RA window to encompass the full read */
 791         if (vio->vui_ra_valid &&
 792             end_idx < vio->vui_ra_start_idx + vio->vui_ra_pages - 1)
 793                 end_idx = vio->vui_ra_start_idx + vio->vui_ra_pages - 1;
 794
 795         if (end_idx != 0) {
 796                 pgoff_t eof_index;
 797
 798                 /* Truncate RA window to end of file */
 799                 eof_index = (pgoff_t)((kms - 1) >> PAGE_SHIFT);
 800                 if (eof_index <= end_idx) {
 801                         end_idx = eof_index;
 802                         ria->ria_eof = true;
 803                 }
 804         }
 805         ria->ria_start_idx = *start_idx;
 806         ria->ria_end_idx = end_idx;
 807         /* If stride I/O mode is detected, get stride window*/
 808         if (stride_io_mode(ras)) {
 809                 ria->ria_stoff = ras->ras_stride_offset;
 810                 ria->ria_length = ras->ras_stride_length;
 811                 ria->ria_bytes = ras->ras_stride_bytes;
 812         }
 813         spin_unlock(&ras->ras_lock);
 814
 815         pages = ria_page_count(ria);
 816
 817         RAS_CDEBUG(ras);
 818         CDEBUG(D_READA,
 819                DFID": ria: %lu/%lu, bead: %lu/%lu, pages %lu, hit: %d\n",
 820                PFID(lu_object_fid(&clob->co_lu)),
 821                ria->ria_start_idx, ria->ria_end_idx,
 822                vio->vui_ra_valid ? vio->vui_ra_start_idx : 0,
 823                vio->vui_ra_valid ? vio->vui_ra_pages : 0,
 824                pages, hit);
 825
 826         if (end_idx == 0) {
 827                 ll_ra_stats_inc(inode, RA_STAT_ZERO_WINDOW);
 828                 RETURN(0);
 829         }
 830         if (pages == 0) {
 831                 ll_ra_stats_inc(inode, RA_STAT_ZERO_WINDOW);
 832                 RETURN(0);
 833         }
 834
 835         /* at least to extend the readahead window to cover current read */
 836         if (!hit && vio->vui_ra_valid &&
 837             vio->vui_ra_start_idx + vio->vui_ra_pages > ria->ria_start_idx) {
 838                 ria->ria_end_idx_min =
 839                         vio->vui_ra_start_idx + vio->vui_ra_pages - 1;
 840                 pages_min = vio->vui_ra_start_idx + vio->vui_ra_pages -
 841                                 ria->ria_start_idx;
 842                  /**
 843                   * For performance reason, exceeding @ra_max_pages
 844                   * are allowed, but this should be limited with RPC
 845                   * size in case a large block size read issued. Trim
 846                   * to RPC boundary.
 847                   */
 848                 pages_min = min(pages_min, ras->ras_rpc_pages -
 849                                 (ria->ria_start_idx % ras->ras_rpc_pages));
 850         }
 851
 852         /* don't over reserved for mmap range read */
 853         if (skip_index)
 854                 pages_min = 0;
 855         if (pages_min > pages)
 856                 pages = pages_min;
 857         ria->ria_reserved = ll_ra_count_get(ll_i2sbi(inode), ria, pages,
 858                                             pages_min);
 859         if (ria->ria_reserved < pages)
 860                 ll_ra_stats_inc(inode, RA_STAT_MAX_IN_FLIGHT);
 861
 862         CDEBUG(D_READA, "reserved pages: %lu/%lu/%lu, ra_cur %d, ra_max %lu\n",
 863                ria->ria_reserved, pages, pages_min,
 864                atomic_read(&ll_i2sbi(inode)->ll_ra_info.ra_cur_pages),
 865                ll_i2sbi(inode)->ll_ra_info.ra_max_pages);
 866
 867         ret = ll_read_ahead_pages(env, io, queue, ras, ria, &ra_end_idx,
 868                                   skip_index);
 869         if (ria->ria_reserved != 0)
 870                 ll_ra_count_put(ll_i2sbi(inode), ria->ria_reserved);
 871
 872         if (ra_end_idx == end_idx && ra_end_idx == (kms >> PAGE_SHIFT))
 873                 ll_ra_stats_inc(inode, RA_STAT_EOF);
 874
 875         CDEBUG(D_READA,
 876                "ra_end_idx = %lu end_idx = %lu stride end = %lu pages = %d\n",
 877                ra_end_idx, end_idx, ria->ria_end_idx, ret);
 878
 879         if (ra_end_idx != end_idx)
 880                 ll_ra_stats_inc(inode, RA_STAT_FAILED_REACH_END);
 881         if (ra_end_idx > 0) {
 882                 /* update the ras so that the next read-ahead tries from
 883                  * where we left off. */
 884                 spin_lock(&ras->ras_lock);
 885                 ras->ras_next_readahead_idx = ra_end_idx + 1;
 886                 spin_unlock(&ras->ras_lock);
 887                 RAS_CDEBUG(ras);
 888         }
 889
 890         RETURN(ret);
 891 }
 892
 893 static int ll_readpages(const struct lu_env *env, struct cl_io *io,
 894                         struct cl_page_list *queue,
 895                         pgoff_t start, pgoff_t end)
 896 {
 897         int ret = 0;
 898         __u64 kms;
 899         pgoff_t page_idx;
 900         int count = 0;
 901
 902         ENTRY;
 903
 904         ret = ll_readahead_file_kms(env, io, &kms);
 905         if (ret != 0)
 906                 RETURN(ret);
 907
 908         if (kms == 0)
 909                 RETURN(0);
 910
 911         if (end != 0) {
 912                 unsigned long end_index;
 913
 914                 end_index = (unsigned long)((kms - 1) >> PAGE_SHIFT);
 915                 if (end_index <= end)
 916                         end = end_index;
 917         }
 918
 919         for (page_idx = start; page_idx <= end; page_idx++) {
 920                 ret= ll_read_ahead_page(env, io, queue, page_idx,
 921                                         WILLNEED);
 922                 if (ret < 0)
 923                         break;
 924                 else if (ret == 0) /* ret 1 is already uptodate */
 925                         count++;
 926         }
 927
 928         RETURN(count > 0 ? count : ret);
 929 }
 930
 931 /* called with the ras_lock held or from places where it doesn't matter */
 932 static void ras_reset(struct ll_readahead_state *ras, pgoff_t index)
 933 {
 934         ras->ras_consecutive_requests = 0;
 935         ras->ras_consecutive_bytes = 0;
 936         ras->ras_window_pages = 0;
 937         ras->ras_window_start_idx = ras_align(ras, index);
 938         ras->ras_next_readahead_idx = max(ras->ras_window_start_idx, index + 1);
 939
 940         RAS_CDEBUG(ras);
 941 }
 942
 943 /* called with the ras_lock held or from places where it doesn't matter */
 944 static void ras_stride_reset(struct ll_readahead_state *ras)
 945 {
 946         ras->ras_consecutive_stride_requests = 0;
 947         ras->ras_stride_length = 0;
 948         ras->ras_stride_bytes = 0;
 949         RAS_CDEBUG(ras);
 950 }
 951
 952 void ll_readahead_init(struct inode *inode, struct ll_readahead_state *ras)
 953 {
 954         spin_lock_init(&ras->ras_lock);
 955         ras->ras_rpc_pages = PTLRPC_MAX_BRW_PAGES;
 956         ras_reset(ras, 0);
 957         ras->ras_last_read_end_bytes = 0;
 958         ras->ras_requests = 0;
 959         ras->ras_range_min_start_idx = 0;
 960         ras->ras_range_max_end_idx = 0;
 961         ras->ras_range_requests = 0;
 962         ras->ras_last_range_pages = 0;
 963 }
 964
 965 /*
 966  * Check whether the read request is in the stride window.
 967  * If it is in the stride window, return true, otherwise return false.
 968  */
 969 static bool read_in_stride_window(struct ll_readahead_state *ras,
 970                                   loff_t pos, loff_t bytes)
 971 {
 972         loff_t stride_gap;
 973
 974         if (ras->ras_stride_length == 0 || ras->ras_stride_bytes == 0 ||
 975             ras->ras_stride_bytes == ras->ras_stride_length)
 976                 return false;
 977
 978         stride_gap = pos - ras->ras_last_read_end_bytes - 1;
 979
 980         /* If it is contiguous read */
 981         if (stride_gap == 0)
 982                 return ras->ras_consecutive_bytes + bytes <=
 983                         ras->ras_stride_bytes;
 984
 985         /* Otherwise check the stride by itself */
 986         return (ras->ras_stride_length - ras->ras_stride_bytes) == stride_gap &&
 987                 ras->ras_consecutive_bytes == ras->ras_stride_bytes &&
 988                 bytes <= ras->ras_stride_bytes;
 989 }
 990
 991 static void ras_init_stride_detector(struct ll_readahead_state *ras,
 992                                      loff_t pos, loff_t bytes)
 993 {
 994         loff_t stride_gap = pos - ras->ras_last_read_end_bytes - 1;
 995
 996         LASSERT(ras->ras_consecutive_stride_requests == 0);
 997
 998         if (pos <= ras->ras_last_read_end_bytes) {
 999                 /* Reset stride window for forward read */
1000                 ras_stride_reset(ras);
1001                 return;
1002         }
1003
1004         ras->ras_stride_bytes = ras->ras_consecutive_bytes;
1005         ras->ras_stride_length = stride_gap + ras->ras_consecutive_bytes;
1006         ras->ras_consecutive_stride_requests++;
1007         ras->ras_stride_offset = pos;
1008
1009         RAS_CDEBUG(ras);
1010 }
1011
1012 static unsigned long
1013 stride_page_count(struct ll_readahead_state *ras, loff_t len)
1014 {
1015         loff_t bytes_count =
1016                 stride_byte_count(ras->ras_stride_offset,
1017                                   ras->ras_stride_length, ras->ras_stride_bytes,
1018                                   ras->ras_window_start_idx << PAGE_SHIFT, len);
1019
1020         return (bytes_count + PAGE_SIZE - 1) >> PAGE_SHIFT;
1021 }
1022
1023 /* Stride Read-ahead window will be increased inc_len according to
1024  * stride I/O pattern */
1025 static void ras_stride_increase_window(struct ll_readahead_state *ras,
1026                                        struct ll_ra_info *ra, loff_t inc_bytes)
1027 {
1028         loff_t window_bytes, stride_bytes;
1029         u64 left_bytes;
1030         u64 step;
1031         loff_t end;
1032
1033         /* temporarily store in page units to reduce LASSERT() cost below */
1034         end = ras->ras_window_start_idx + ras->ras_window_pages;
1035
1036         LASSERT(ras->ras_stride_length > 0);
1037         LASSERTF(end >= (ras->ras_stride_offset >> PAGE_SHIFT),
1038                  "window_start_idx %lu, window_pages %lu stride_offset %llu\n",
1039                  ras->ras_window_start_idx, ras->ras_window_pages,
1040                  ras->ras_stride_offset);
1041
1042         end <<= PAGE_SHIFT;
1043         if (end <= ras->ras_stride_offset)
1044                 stride_bytes = 0;
1045         else
1046                 stride_bytes = end - ras->ras_stride_offset;
1047
1048         div64_u64_rem(stride_bytes, ras->ras_stride_length, &left_bytes);
1049         window_bytes = (ras->ras_window_pages << PAGE_SHIFT);
1050         if (left_bytes < ras->ras_stride_bytes) {
1051                 if (ras->ras_stride_bytes - left_bytes >= inc_bytes) {
1052                         window_bytes += inc_bytes;
1053                         goto out;
1054                 } else {
1055                         window_bytes += (ras->ras_stride_bytes - left_bytes);
1056                         inc_bytes -= (ras->ras_stride_bytes - left_bytes);
1057                 }
1058         } else {
1059                 window_bytes += (ras->ras_stride_length - left_bytes);
1060         }
1061
1062         LASSERT(ras->ras_stride_bytes != 0);
1063
1064         step = div64_u64_rem(inc_bytes, ras->ras_stride_bytes, &left_bytes);
1065
1066         window_bytes += step * ras->ras_stride_length + left_bytes;
1067         LASSERT(window_bytes > 0);
1068
1069 out:
1070         if (stride_page_count(ras, window_bytes) <=
1071             ra->ra_max_pages_per_file || ras->ras_window_pages == 0)
1072                 ras->ras_window_pages = (window_bytes >> PAGE_SHIFT);
1073
1074         LASSERT(ras->ras_window_pages > 0);
1075
1076         RAS_CDEBUG(ras);
1077 }
1078
1079 static void ras_increase_window(struct inode *inode,
1080                                 struct ll_readahead_state *ras,
1081                                 struct ll_ra_info *ra)
1082 {
1083         /* The stretch of ra-window should be aligned with max rpc_size
1084          * but current clio architecture does not support retrieve such
1085          * information from lower layer. FIXME later
1086          */
1087         if (stride_io_mode(ras)) {
1088                 ras_stride_increase_window(ras, ra,
1089                                       (loff_t)ras->ras_rpc_pages << PAGE_SHIFT);
1090         } else {
1091                 pgoff_t window_pages;
1092
1093                 window_pages = min(ras->ras_window_pages + ras->ras_rpc_pages,
1094                                    ra->ra_max_pages_per_file);
1095                 if (window_pages < ras->ras_rpc_pages)
1096                         ras->ras_window_pages = window_pages;
1097                 else
1098                         ras->ras_window_pages = ras_align(ras, window_pages);
1099         }
1100 }
1101
1102 /**
1103  * Seek within 8 pages are considered as sequential read for now.
1104  */
1105 static inline bool is_loose_seq_read(struct ll_readahead_state *ras, loff_t pos)
1106 {
1107         return pos_in_window(pos, ras->ras_last_read_end_bytes,
1108                              8UL << PAGE_SHIFT, 8UL << PAGE_SHIFT);
1109 }
1110
1111 static inline bool is_loose_mmap_read(struct ll_sb_info *sbi,
1112                                       struct ll_readahead_state *ras,
1113                                       unsigned long pos)
1114 {
1115         unsigned long range_pages = sbi->ll_ra_info.ra_range_pages;
1116
1117         return pos_in_window(pos, ras->ras_last_read_end_bytes,
1118                              range_pages << PAGE_SHIFT,
1119                              range_pages << PAGE_SHIFT);
1120 }
1121
1122 /**
1123  * We have observed slow mmap read performances for some
1124  * applications. The problem is if access pattern is neither
1125  * sequential nor stride, but could be still adjacent in a
1126  * small range and then seek a random position.
1127  *
1128  * So the pattern could be something like this:
1129  *
1130  * [1M data] [hole] [0.5M data] [hole] [0.7M data] [1M data]
1131  *
1132  *
1133  * Every time an application reads mmap data, it may not only
1134  * read a single 4KB page, but aslo a cluster of nearby pages in
1135  * a range(e.g. 1MB) of the first page after a cache miss.
1136  *
1137  * The readahead engine is modified to track the range size of
1138  * a cluster of mmap reads, so that after a seek and/or cache miss,
1139  * the range size is used to efficiently prefetch multiple pages
1140  * in a single RPC rather than many small RPCs.
1141  */
1142 static void ras_detect_cluster_range(struct ll_readahead_state *ras,
1143                                      struct ll_sb_info *sbi,
1144                                      unsigned long pos, unsigned long count)
1145 {
1146         pgoff_t last_pages, pages;
1147         pgoff_t end_idx = (pos + count - 1) >> PAGE_SHIFT;
1148
1149         last_pages = ras->ras_range_max_end_idx -
1150                         ras->ras_range_min_start_idx + 1;
1151         /* First time come here */
1152         if (!ras->ras_range_max_end_idx)
1153                 goto out;
1154
1155         /* Random or Stride read */
1156         if (!is_loose_mmap_read(sbi, ras, pos))
1157                 goto out;
1158
1159         ras->ras_range_requests++;
1160         if (ras->ras_range_max_end_idx < end_idx)
1161                 ras->ras_range_max_end_idx = end_idx;
1162
1163         if (ras->ras_range_min_start_idx > (pos >> PAGE_SHIFT))
1164                 ras->ras_range_min_start_idx = pos >> PAGE_SHIFT;
1165
1166         /* Out of range, consider it as random or stride */
1167         pages = ras->ras_range_max_end_idx -
1168                         ras->ras_range_min_start_idx + 1;
1169         if (pages <= sbi->ll_ra_info.ra_range_pages)
1170                 return;
1171 out:
1172         ras->ras_last_range_pages = last_pages;
1173         ras->ras_range_requests = 0;
1174         ras->ras_range_min_start_idx = pos >> PAGE_SHIFT;
1175         ras->ras_range_max_end_idx = end_idx;
1176 }
1177
1178 static void ras_detect_read_pattern(struct ll_readahead_state *ras,
1179                                     struct ll_sb_info *sbi,
1180                                     loff_t pos, size_t bytes, bool mmap)
1181 {
1182         bool stride_detect = false;
1183         pgoff_t index = pos >> PAGE_SHIFT;
1184
1185         RAS_CDEBUG(ras);
1186         /*
1187          * Reset the read-ahead window in two cases. First when the app seeks
1188          * or reads to some other part of the file. Secondly if we get a
1189          * read-ahead miss that we think we've previously issued. This can
1190          * be a symptom of there being so many read-ahead pages that the VM
1191          * is reclaiming it before we get to it.
1192          */
1193         if (!is_loose_seq_read(ras, pos)) {
1194                 /* Check whether it is in stride I/O mode */
1195                 if (!read_in_stride_window(ras, pos, bytes)) {
1196                         if (ras->ras_consecutive_stride_requests == 0)
1197                                 ras_init_stride_detector(ras, pos, bytes);
1198                         else
1199                                 ras_stride_reset(ras);
1200                         ras->ras_consecutive_bytes = 0;
1201                         ras_reset(ras, index);
1202                 } else {
1203                         ras->ras_consecutive_bytes = 0;
1204                         ras->ras_consecutive_requests = 0;
1205                         if (++ras->ras_consecutive_stride_requests > 1)
1206                                 stride_detect = true;
1207                         RAS_CDEBUG(ras);
1208                 }
1209                 ll_ra_stats_inc_sbi(sbi, RA_STAT_DISTANT_READPAGE);
1210         } else if (stride_io_mode(ras)) {
1211                 /*
1212                  * If this is contiguous read but in stride I/O mode
1213                  * currently, check whether stride step still is valid,
1214                  * if invalid, it will reset the stride ra window to
1215                  * be zero.
1216                  */
1217                 if (!read_in_stride_window(ras, pos, bytes)) {
1218                         ras_stride_reset(ras);
1219                         ras->ras_window_pages = 0;
1220                         ras->ras_next_readahead_idx = index;
1221                 }
1222         }
1223
1224         ras->ras_consecutive_bytes += bytes;
1225         if (mmap) {
1226                 pgoff_t idx = ras->ras_consecutive_bytes >> PAGE_SHIFT;
1227                 unsigned long ra_range_pages =
1228                                 max_t(unsigned long, RA_MIN_MMAP_RANGE_PAGES,
1229                                       sbi->ll_ra_info.ra_range_pages);
1230
1231                 if ((idx >= ra_range_pages &&
1232                      idx % ra_range_pages == 0) || stride_detect)
1233                         ras->ras_need_increase_window = true;
1234         } else if ((ras->ras_consecutive_requests > 1 || stride_detect)) {
1235                 ras->ras_need_increase_window = true;
1236         }
1237
1238         ras->ras_last_read_end_bytes = pos + bytes - 1;
1239         RAS_CDEBUG(ras);
1240 }
1241
1242 void ll_ras_enter(struct file *f, loff_t pos, size_t bytes)
1243 {
1244         struct ll_file_data *fd = f->private_data;
1245         struct ll_readahead_state *ras = &fd->fd_ras;
1246         struct inode *inode = file_inode(f);
1247         struct ll_sb_info *sbi = ll_i2sbi(inode);
1248
1249         spin_lock(&ras->ras_lock);
1250         ras->ras_requests++;
1251         ras->ras_consecutive_requests++;
1252         ras->ras_need_increase_window = false;
1253         ras->ras_whole_file_read = false;
1254         /*
1255          * On the second access to a file smaller than the tunable
1256          * ra_max_read_ahead_whole_pages trigger RA on all pages in the
1257          * file up to ra_max_pages_per_file.  This is simply a best effort
1258          * and only occurs once per open file. Normal RA behavior is reverted
1259          * to for subsequent IO.
1260          */
1261         if (ras->ras_requests >= 2) {
1262                 __u64 kms_pages;
1263                 struct ll_ra_info *ra = &sbi->ll_ra_info;
1264
1265                 kms_pages = (i_size_read(inode) + PAGE_SIZE - 1) >>
1266                             PAGE_SHIFT;
1267
1268                 CDEBUG(D_READA, "kmsp %llu mwp %lu mp %lu\n", kms_pages,
1269                        ra->ra_max_read_ahead_whole_pages,
1270                        ra->ra_max_pages_per_file);
1271
1272                 if (kms_pages &&
1273                     kms_pages <= ra->ra_max_read_ahead_whole_pages) {
1274                         ras->ras_whole_file_read = true;
1275                         ras->ras_window_start_idx = 0;
1276                         ras->ras_next_readahead_idx = 0;
1277                         ras->ras_window_pages = min(ra->ra_max_pages_per_file,
1278                                             ra->ra_max_read_ahead_whole_pages);
1279                         GOTO(out_unlock, 0);
1280                 }
1281         }
1282         ras_detect_read_pattern(ras, sbi, pos, bytes, false);
1283 out_unlock:
1284         spin_unlock(&ras->ras_lock);
1285 }
1286
1287 static bool index_in_stride_window(struct ll_readahead_state *ras,
1288                                    pgoff_t index)
1289 {
1290         loff_t pos = (loff_t)index << PAGE_SHIFT;
1291
1292         if (ras->ras_stride_length == 0 || ras->ras_stride_bytes == 0 ||
1293             ras->ras_stride_bytes == ras->ras_stride_length)
1294                 return false;
1295
1296         if (pos >= ras->ras_stride_offset) {
1297                 u64 offset;
1298
1299                 div64_u64_rem(pos - ras->ras_stride_offset,
1300                               ras->ras_stride_length, &offset);
1301                 if (offset < ras->ras_stride_bytes ||
1302                     ras->ras_stride_length - offset < PAGE_SIZE)
1303                         return true;
1304         } else if (ras->ras_stride_offset - pos < PAGE_SIZE) {
1305                 return true;
1306         }
1307
1308         return false;
1309 }
1310
1311 /*
1312  * ll_ras_enter() is used to detect read pattern according to pos and count.
1313  *
1314  * ras_update() is used to detect cache miss and
1315  * reset window or increase window accordingly
1316  */
1317 static void ras_update(struct ll_sb_info *sbi, struct inode *inode,
1318                        struct ll_readahead_state *ras, pgoff_t index,
1319                        enum ras_update_flags flags, struct cl_io *io)
1320 {
1321         struct ll_ra_info *ra = &sbi->ll_ra_info;
1322         bool hit = flags & LL_RAS_HIT;
1323
1324         ENTRY;
1325         spin_lock(&ras->ras_lock);
1326
1327         RAS_CDEBUG(ras);
1328
1329         if (!hit)
1330                 CDEBUG(D_READA|D_IOTRACE, DFID " pages at %lu miss.\n",
1331                        PFID(ll_inode2fid(inode)), index);
1332         ll_ra_stats_inc_sbi(sbi, hit ? RA_STAT_HIT : RA_STAT_MISS);
1333
1334         /*
1335          * The readahead window has been expanded to cover whole
1336          * file size, we don't care whether ra miss happen or not.
1337          * Because we will read whole file to page cache even if
1338          * some pages missed.
1339          */
1340         if (ras->ras_whole_file_read)
1341                 GOTO(out_unlock, 0);
1342
1343         if (io && io->ci_rand_read)
1344                 GOTO(out_unlock, 0);
1345
1346         if (io && io->ci_seq_read) {
1347                 if (!hit) {
1348                         /* to avoid many small read RPC here */
1349                         ras->ras_window_pages = sbi->ll_ra_info.ra_range_pages;
1350                         ll_ra_stats_inc_sbi(sbi, RA_STAT_MMAP_RANGE_READ);
1351                 }
1352                 goto skip_miss_checking;
1353         }
1354
1355         if (flags & LL_RAS_MMAP) {
1356                 unsigned long ra_pages;
1357
1358                 ras_detect_cluster_range(ras, sbi, index << PAGE_SHIFT,
1359                                          PAGE_SIZE);
1360                 ras_detect_read_pattern(ras, sbi, (loff_t)index << PAGE_SHIFT,
1361                                         PAGE_SIZE, true);
1362
1363                 /* we did not detect anything but we could prefetch */
1364                 if (!ras->ras_need_increase_window &&
1365                     ras->ras_window_pages <= sbi->ll_ra_info.ra_range_pages &&
1366                     ras->ras_range_requests >= 2) {
1367                         if (!hit) {
1368                                 ra_pages = max_t(unsigned long,
1369                                         RA_MIN_MMAP_RANGE_PAGES,
1370                                         ras->ras_last_range_pages);
1371                                 if (index < ra_pages / 2)
1372                                         index = 0;
1373                                 else
1374                                         index -= ra_pages / 2;
1375                                 ras->ras_window_pages = ra_pages;
1376                                 ll_ra_stats_inc_sbi(sbi,
1377                                         RA_STAT_MMAP_RANGE_READ);
1378                         } else {
1379                                 ras->ras_window_pages = 0;
1380                         }
1381                         goto skip_miss_checking;
1382                 }
1383         }
1384
1385         if (!hit && ras->ras_window_pages &&
1386             index < ras->ras_next_readahead_idx &&
1387             pos_in_window(index, ras->ras_window_start_idx, 0,
1388                           ras->ras_window_pages)) {
1389                 ll_ra_stats_inc_sbi(sbi, RA_STAT_MISS_IN_WINDOW);
1390                 ras->ras_need_increase_window = false;
1391
1392                 if (index_in_stride_window(ras, index) &&
1393                     stride_io_mode(ras)) {
1394                         /*
1395                          * if (index != ras->ras_last_readpage + 1)
1396                          *      ras->ras_consecutive_pages = 0;
1397                          */
1398                         ras_reset(ras, index);
1399
1400                         /*
1401                          * If stride-RA hit cache miss, the stride
1402                          * detector will not be reset to avoid the
1403                          * overhead of redetecting read-ahead mode,
1404                          * but on the condition that the stride window
1405                          * is still intersect with normal sequential
1406                          * read-ahead window.
1407                          */
1408                         if (ras->ras_window_start_idx < ras->ras_stride_offset)
1409                                 ras_stride_reset(ras);
1410                         RAS_CDEBUG(ras);
1411                 } else {
1412                         /*
1413                          * Reset both stride window and normal RA
1414                          * window.
1415                          */
1416                         ras_reset(ras, index);
1417                         /* ras->ras_consecutive_pages++; */
1418                         ras->ras_consecutive_bytes = 0;
1419                         ras_stride_reset(ras);
1420                         GOTO(out_unlock, 0);
1421                 }
1422         }
1423
1424 skip_miss_checking:
1425         ras->ras_window_start_idx = ras_align(ras, index);
1426
1427         if (stride_io_mode(ras)) {
1428                 /* Since stride readahead is sentivite to the offset
1429                  * of read-ahead, so we use original offset here,
1430                  * instead of ras_window_start_idx, which is RPC aligned.
1431                  */
1432                 ras->ras_next_readahead_idx = max(index + 1,
1433                                                   ras->ras_next_readahead_idx);
1434                 ras->ras_window_start_idx =
1435                                 max_t(pgoff_t, ras->ras_window_start_idx,
1436                                       ras->ras_stride_offset >> PAGE_SHIFT);
1437         } else {
1438                 if (ras->ras_next_readahead_idx < ras->ras_window_start_idx)
1439                         ras->ras_next_readahead_idx = ras->ras_window_start_idx;
1440                 if (!hit)
1441                         ras->ras_next_readahead_idx = index + 1;
1442         }
1443
1444         if (ras->ras_need_increase_window) {
1445                 ras_increase_window(inode, ras, ra);
1446                 ras->ras_need_increase_window = false;
1447         }
1448
1449         EXIT;
1450 out_unlock:
1451         RAS_CDEBUG(ras);
1452         spin_unlock(&ras->ras_lock);
1453 }
1454
1455 int ll_writepage(struct page *vmpage, struct writeback_control *wbc)
1456 {
1457         struct inode           *inode = vmpage->mapping->host;
1458         struct ll_inode_info   *lli   = ll_i2info(inode);
1459         struct lu_env          *env;
1460         struct cl_io           *io;
1461         struct cl_page         *page;
1462         struct cl_object       *clob;
1463         bool redirtied = false;
1464         bool unlocked = false;
1465         int result;
1466         __u16 refcheck;
1467         ENTRY;
1468
1469         LASSERT(PageLocked(vmpage));
1470         LASSERT(!PageWriteback(vmpage));
1471
1472         LASSERT(ll_i2dtexp(inode) != NULL);
1473
1474         env = cl_env_get(&refcheck);
1475         if (IS_ERR(env))
1476                 GOTO(out, result = PTR_ERR(env));
1477
1478         clob  = ll_i2info(inode)->lli_clob;
1479         LASSERT(clob != NULL);
1480
1481         io = vvp_env_thread_io(env);
1482         io->ci_obj = clob;
1483         io->ci_ignore_layout = 1;
1484         result = cl_io_init(env, io, CIT_MISC, clob);
1485         if (result == 0) {
1486                 page = cl_page_find(env, clob, vmpage->index,
1487                                     vmpage, CPT_CACHEABLE);
1488                 if (!IS_ERR(page)) {
1489                         lu_ref_add(&page->cp_reference, "writepage",
1490                                    current);
1491                         cl_page_assume(env, io, page);
1492                         result = cl_page_flush(env, io, page);
1493                         if (result != 0) {
1494                                 /*
1495                                  * Re-dirty page on error so it retries write,
1496                                  * but not in case when IO has actually
1497                                  * occurred and completed with an error.
1498                                  */
1499                                 if (!PageError(vmpage)) {
1500                                         redirty_page_for_writepage(wbc, vmpage);
1501                                         result = 0;
1502                                         redirtied = true;
1503                                 }
1504                         }
1505                         cl_page_disown(env, io, page);
1506                         unlocked = true;
1507                         lu_ref_del(&page->cp_reference,
1508                                    "writepage", current);
1509                         cl_page_put(env, page);
1510                 } else {
1511                         result = PTR_ERR(page);
1512                 }
1513         }
1514         cl_io_fini(env, io);
1515
1516         if (redirtied && wbc->sync_mode == WB_SYNC_ALL) {
1517                 loff_t offset = vmpage->index << PAGE_SHIFT;
1518
1519                 /* Flush page failed because the extent is being written out.
1520                  * Wait for the write of extent to be finished to avoid
1521                  * breaking kernel which assumes ->writepage should mark
1522                  * PageWriteback or clean the page. */
1523                 result = cl_sync_file_range(inode, offset,
1524                                             offset + PAGE_SIZE - 1,
1525                                             CL_FSYNC_LOCAL, 1);
1526                 if (result > 0) {
1527                         /* actually we may have written more than one page.
1528                          * decreasing this page because the caller will count
1529                          * it. */
1530                         wbc->nr_to_write -= result - 1;
1531                         result = 0;
1532                 }
1533         }
1534
1535         cl_env_put(env, &refcheck);
1536         GOTO(out, result);
1537
1538 out:
1539         if (result < 0) {
1540                 if (!lli->lli_async_rc)
1541                         lli->lli_async_rc = result;
1542                 SetPageError(vmpage);
1543                 if (!unlocked)
1544                         unlock_page(vmpage);
1545         }
1546         return result;
1547 }
1548
1549 int ll_writepages(struct address_space *mapping, struct writeback_control *wbc)
1550 {
1551         struct inode *inode = mapping->host;
1552         loff_t start;
1553         loff_t end;
1554         enum cl_fsync_mode mode;
1555         int range_whole = 0;
1556         int result;
1557
1558         ENTRY;
1559
1560         if (wbc->range_cyclic) {
1561                 start = (loff_t)mapping->writeback_index << PAGE_SHIFT;
1562                 end = OBD_OBJECT_EOF;
1563         } else {
1564                 start = wbc->range_start;
1565                 end = wbc->range_end;
1566                 if (end == LLONG_MAX) {
1567                         end = OBD_OBJECT_EOF;
1568                         range_whole = start == 0;
1569                 }
1570         }
1571
1572         mode = CL_FSYNC_NONE;
1573         if (wbc->sync_mode == WB_SYNC_ALL)
1574                 mode = CL_FSYNC_LOCAL;
1575
1576         if (wbc->sync_mode == WB_SYNC_NONE) {
1577 #ifdef SB_I_CGROUPWB
1578                 struct bdi_writeback *wb;
1579
1580                 /*
1581                  * As it may break full stripe writes on the inode,
1582                  * disable periodic kupdate writeback (@wbc->for_kupdate)?
1583                  */
1584
1585                 /*
1586                  * The system is under memory pressure and it is now reclaiming
1587                  * cache pages.
1588                  */
1589                 wb = inode_to_wb(inode);
1590                 if (wbc->for_background ||
1591                     (wb->start_all_reason == WB_REASON_VMSCAN &&
1592                      test_bit(WB_start_all, &wb->state)))
1593                         mode = CL_FSYNC_RECLAIM;
1594 #else
1595                 /*
1596                  * We have no idea about writeback reason for memory reclaim
1597                  * WB_REASON_TRY_TO_FREE_PAGES in the old kernel such as rhel7
1598                  * (WB_REASON_VMSCAN in the newer kernel) ...
1599                  * Here set mode with CL_FSYNC_RECLAIM forcely on the old
1600                  * kernel.
1601                  */
1602                 if (!wbc->for_kupdate)
1603                         mode = CL_FSYNC_RECLAIM;
1604 #endif
1605         }
1606
1607         if (ll_i2info(inode)->lli_clob == NULL || (inode->i_state & I_FREEING))
1608                 RETURN(0);
1609
1610         /* for directio, it would call writepages() to evict cached pages
1611          * inside the IO context of write, which will cause deadlock at
1612          * layout_conf since it waits for active IOs to complete. */
1613         result = cl_sync_file_range(inode, start, end, mode, 1);
1614         if (result > 0) {
1615                 wbc->nr_to_write -= result;
1616                 result = 0;
1617          }
1618
1619         if (wbc->range_cyclic || (range_whole && wbc->nr_to_write > 0)) {
1620                 if (end == OBD_OBJECT_EOF)
1621                         mapping->writeback_index = 0;
1622                 else
1623                         mapping->writeback_index = (end >> PAGE_SHIFT) + 1;
1624         }
1625         RETURN(result);
1626 }
1627
1628 struct ll_cl_context *ll_cl_find(struct inode *inode)
1629 {
1630         struct ll_inode_info *lli = ll_i2info(inode);
1631         struct ll_cl_context *lcc;
1632         struct ll_cl_context *found = NULL;
1633
1634         read_lock(&lli->lli_lock);
1635         list_for_each_entry(lcc, &lli->lli_lccs, lcc_list) {
1636                 if (lcc->lcc_cookie == current) {
1637                         found = lcc;
1638                         break;
1639                 }
1640         }
1641         read_unlock(&lli->lli_lock);
1642
1643         return found;
1644 }
1645
1646 void ll_cl_add(struct inode *inode, const struct lu_env *env, struct cl_io *io,
1647                enum lcc_type type)
1648 {
1649         struct ll_inode_info *lli = ll_i2info(inode);
1650         struct ll_cl_context *lcc = &ll_env_info(env)->lti_io_ctx;
1651
1652         memset(lcc, 0, sizeof(*lcc));
1653         INIT_LIST_HEAD(&lcc->lcc_list);
1654         lcc->lcc_cookie = current;
1655         lcc->lcc_env = env;
1656         lcc->lcc_io = io;
1657         lcc->lcc_type = type;
1658
1659         write_lock(&lli->lli_lock);
1660         list_add(&lcc->lcc_list, &lli->lli_lccs);
1661         write_unlock(&lli->lli_lock);
1662 }
1663
1664 void ll_cl_remove(struct inode *inode, const struct lu_env *env)
1665 {
1666         struct ll_inode_info *lli = ll_i2info(inode);
1667         struct ll_cl_context *lcc = &ll_env_info(env)->lti_io_ctx;
1668
1669         write_lock(&lli->lli_lock);
1670         list_del_init(&lcc->lcc_list);
1671         write_unlock(&lli->lli_lock);
1672 }
1673
1674 int ll_io_read_page(const struct lu_env *env, struct cl_io *io,
1675                            struct cl_page *page, struct file *file)
1676 {
1677         struct inode              *inode  = vvp_object_inode(page->cp_obj);
1678         struct ll_sb_info         *sbi    = ll_i2sbi(inode);
1679         struct ll_file_data       *fd     = NULL;
1680         struct ll_readahead_state *ras    = NULL;
1681         struct cl_2queue          *queue  = &io->ci_queue;
1682         struct cl_sync_io         *anchor = NULL;
1683         int                        rc = 0, rc2 = 0;
1684         bool                       uptodate;
1685         struct vvp_io *vio = vvp_env_io(env);
1686         bool mmap = !vio->vui_ra_valid;
1687         pgoff_t ra_start_index = 0;
1688         pgoff_t io_start_index;
1689         pgoff_t io_end_index;
1690         bool unlockpage = true;
1691         ENTRY;
1692
1693         if (file) {
1694                 fd = file->private_data;
1695                 ras = &fd->fd_ras;
1696         }
1697
1698         /* PagePrivate2 is set in ll_io_zero_page() to tell us the vmpage
1699          * must not be unlocked after processing.
1700          */
1701         if (page->cp_vmpage && PagePrivate2(page->cp_vmpage))
1702                 unlockpage = false;
1703
1704         uptodate = page->cp_defer_uptodate;
1705
1706         if (ll_readahead_enabled(sbi) && !page->cp_ra_updated && ras) {
1707                 enum ras_update_flags flags = 0;
1708
1709                 if (uptodate)
1710                         flags |= LL_RAS_HIT;
1711                 if (mmap)
1712                         flags |= LL_RAS_MMAP;
1713                 ras_update(sbi, inode, ras, cl_page_index(page), flags, io);
1714         }
1715
1716         cl_2queue_init(queue);
1717         if (uptodate) {
1718                 page->cp_ra_used = 1;
1719                 SetPageUptodate(page->cp_vmpage);
1720                 cl_page_disown(env, io, page);
1721         } else {
1722                 anchor = &vvp_env_info(env)->vti_anchor;
1723                 cl_sync_io_init(anchor, 1);
1724                 page->cp_sync_io = anchor;
1725
1726                 cl_page_list_add(&queue->c2_qin, page, true);
1727         }
1728
1729         /* mmap does not set the ci_rw fields */
1730         if (!mmap) {
1731                 io_start_index = io->u.ci_rw.crw_pos >> PAGE_SHIFT;
1732                 io_end_index = (io->u.ci_rw.crw_pos +
1733                                 io->u.ci_rw.crw_bytes - 1) >> PAGE_SHIFT;
1734         } else {
1735                 io_start_index = cl_page_index(page);
1736                 io_end_index = cl_page_index(page);
1737         }
1738
1739         if (ll_readahead_enabled(sbi) && ras && !io->ci_rand_read) {
1740                 pgoff_t skip_index = 0;
1741
1742                 if (ras->ras_next_readahead_idx < cl_page_index(page))
1743                         skip_index = cl_page_index(page);
1744                 rc2 = ll_readahead(env, io, &queue->c2_qin, ras,
1745                                    uptodate, file, skip_index,
1746                                    &ra_start_index);
1747                 /* to keep iotrace clean, we only print here if we actually
1748                  * read pages
1749                  */
1750                 CDEBUG(D_READA | (rc2 ? D_IOTRACE : 0),
1751                        DFID " %d pages read ahead at %lu, triggered by user read at %lu, stride offset %lld, stride length %lld, stride bytes %lld\n",
1752                        PFID(ll_inode2fid(inode)), rc2, ra_start_index,
1753                        cl_page_index(page), ras->ras_stride_offset,
1754                        ras->ras_stride_length, ras->ras_stride_bytes);
1755
1756         } else if (cl_page_index(page) == io_start_index &&
1757                    io_end_index - io_start_index > 0) {
1758                 rc2 = ll_readpages(env, io, &queue->c2_qin, io_start_index + 1,
1759                                    io_end_index);
1760                 CDEBUG(D_READA, DFID " %d pages read at %lu\n",
1761                        PFID(ll_inode2fid(inode)), rc2, cl_page_index(page));
1762         }
1763
1764         if (queue->c2_qin.pl_nr > 0) {
1765                 int count = queue->c2_qin.pl_nr;
1766                 rc = cl_io_submit_rw(env, io, CRT_READ, queue);
1767                 if (rc == 0)
1768                         task_io_account_read(PAGE_SIZE * count);
1769         }
1770
1771
1772         if (anchor != NULL && !cl_page_is_owned(page, io)) { /* have sent */
1773                 rc = cl_sync_io_wait(env, anchor, 0);
1774
1775                 cl_page_assume(env, io, page);
1776                 cl_page_list_del(env, &queue->c2_qout, page, true);
1777
1778                 if (!PageUptodate(cl_page_vmpage(page))) {
1779                         /* Failed to read a mirror, discard this page so that
1780                          * new page can be created with new mirror.
1781                          *
1782                          * TODO: this is not needed after page reinit
1783                          * route is implemented */
1784                         cl_page_discard(env, io, page);
1785                 }
1786                 if (unlockpage)
1787                         cl_page_disown(env, io, page);
1788         }
1789
1790         /* TODO: discard all pages until page reinit route is implemented */
1791         cl_page_list_discard(env, io, &queue->c2_qin);
1792
1793         /* Unlock unsent read pages in case of error. */
1794         cl_page_list_disown(env, &queue->c2_qin);
1795
1796         cl_2queue_fini(env, queue);
1797
1798         RETURN(rc);
1799 }
1800
1801 /*
1802  * Possible return value:
1803  * 0 no async readahead triggered and fast read could not be used.
1804  * 1 no async readahead, but fast read could be used.
1805  * 2 async readahead triggered and fast read could be used too.
1806  * < 0 on error.
1807  */
1808 static int kickoff_async_readahead(struct file *file, unsigned long pages)
1809 {
1810         struct ll_readahead_work *lrw;
1811         struct inode *inode = file_inode(file);
1812         struct ll_sb_info *sbi = ll_i2sbi(inode);
1813         struct ll_file_data *fd = file->private_data;
1814         struct ll_readahead_state *ras = &fd->fd_ras;
1815         struct ll_ra_info *ra = &sbi->ll_ra_info;
1816         unsigned long throttle;
1817         pgoff_t start_idx = ras_align(ras, ras->ras_next_readahead_idx);
1818         pgoff_t end_idx = start_idx + pages - 1;
1819
1820         /**
1821          * In case we have a limited max_cached_mb, readahead
1822          * should be stopped if it have run out of all LRU slots.
1823          */
1824         if (atomic_read(&ra->ra_cur_pages) >= sbi->ll_cache->ccc_lru_max) {
1825                 ll_ra_stats_inc(inode, RA_STAT_MAX_IN_FLIGHT);
1826                 return 0;
1827         }
1828
1829         throttle = min(ra->ra_async_pages_per_file_threshold,
1830                        ra->ra_max_pages_per_file);
1831         /*
1832          * If this is strided i/o or the window is smaller than the
1833          * throttle limit, we do not do async readahead. Otherwise,
1834          * we do async readahead, allowing the user thread to do fast i/o.
1835          */
1836         if (stride_io_mode(ras) || !throttle ||
1837             ras->ras_window_pages < throttle ||
1838             atomic_read(&ra->ra_async_inflight) > ra->ra_async_max_active)
1839                 return 0;
1840
1841         if ((atomic_read(&ra->ra_cur_pages) + pages) > ra->ra_max_pages)
1842                 return 0;
1843
1844         if (ras->ras_async_last_readpage_idx == start_idx)
1845                 return 1;
1846
1847         /* ll_readahead_work_free() free it */
1848         OBD_ALLOC_PTR(lrw);
1849         if (lrw) {
1850                 atomic_inc(&sbi->ll_ra_info.ra_async_inflight);
1851                 lrw->lrw_file = get_file(file);
1852                 lrw->lrw_start_idx = start_idx;
1853                 lrw->lrw_end_idx = end_idx;
1854                 lrw->lrw_user_pid = current->pid;
1855                 spin_lock(&ras->ras_lock);
1856                 ras->ras_next_readahead_idx = end_idx + 1;
1857                 ras->ras_async_last_readpage_idx = start_idx;
1858                 spin_unlock(&ras->ras_lock);
1859                 memcpy(lrw->lrw_jobid, ll_i2info(inode)->lli_jobid,
1860                        sizeof(lrw->lrw_jobid));
1861                 ll_readahead_work_add(inode, lrw);
1862         } else {
1863                 return -ENOMEM;
1864         }
1865
1866         return 2;
1867 }
1868
1869 /*
1870  * Check if we can issue a readahead RPC, if that is
1871  * the case, we can't do fast IO because we will need
1872  * a cl_io to issue the RPC.
1873  */
1874 static bool ll_use_fast_io(struct file *file,
1875                            struct ll_readahead_state *ras, pgoff_t index)
1876 {
1877         unsigned long fast_read_pages =
1878                 max(RA_REMAIN_WINDOW_MIN, ras->ras_rpc_pages);
1879         loff_t skip_pages;
1880         loff_t stride_bytes = ras->ras_stride_bytes;
1881
1882         RAS_CDEBUG(ras);
1883
1884         if (stride_io_mode(ras) && stride_bytes) {
1885                 skip_pages = (ras->ras_stride_length +
1886                         ras->ras_stride_bytes - 1) / stride_bytes;
1887                 skip_pages *= fast_read_pages;
1888         } else {
1889                 skip_pages = fast_read_pages;
1890         }
1891
1892         RAS_CDEBUG(ras);
1893
1894         if (ras->ras_whole_file_read ||
1895             ras->ras_window_start_idx + ras->ras_window_pages <
1896             ras->ras_next_readahead_idx + skip_pages ||
1897             kickoff_async_readahead(file, fast_read_pages) > 0) {
1898                 return true;
1899         }
1900
1901         return false;
1902 }
1903
1904 int ll_readpage(struct file *file, struct page *vmpage)
1905 {
1906         struct inode *inode = file_inode(file);
1907         struct cl_object *clob = ll_i2info(inode)->lli_clob;
1908         struct ll_sb_info *sbi = ll_i2sbi(inode);
1909         const struct lu_env *env = NULL;
1910         struct cl_read_ahead ra = { 0 };
1911         struct ll_cl_context *lcc;
1912         struct cl_io *io = NULL;
1913         struct cl_page *page;
1914         struct vvp_io *vio;
1915         int result;
1916         int flags;
1917
1918         ENTRY;
1919
1920         if (CFS_FAIL_PRECHECK(OBD_FAIL_LLITE_READPAGE_PAUSE)) {
1921                 unlock_page(vmpage);
1922                 CFS_FAIL_TIMEOUT(OBD_FAIL_LLITE_READPAGE_PAUSE, cfs_fail_val);
1923                 lock_page(vmpage);
1924         }
1925
1926         /*
1927          * The @vmpage got truncated.
1928          * This is a kernel bug introduced since kernel 5.12:
1929          * comment: cbd59c48ae2bcadc4a7599c29cf32fd3f9b78251
1930          * ("mm/filemap: use head pages in generic_file_buffered_read")
1931          *
1932          * The page end offset calculation in filemap_get_read_batch() was off
1933          * by one.  When a read is submitted with end offset 1048575, then it
1934          * calculates the end page for read of 256 where it should be 255. This
1935          * results in the readpage() for the page with index 256 is over stripe
1936          * boundary and may not covered by a DLM extent lock.
1937          *
1938          * This happens in a corner race case: filemap_get_read_batch() adds
1939          * the page with index 256 for read which is not in the current read
1940          * I/O context, and this page is being invalidated and will be removed
1941          * from page cache due to the lock protected it being revoken. This
1942          * results in this page in the read path not covered by any DLM lock.
1943          *
1944          * The solution is simple. Check whether the page was truncated in
1945          * ->readpage(). If so, just return AOP_TRUNCATED_PAGE to the upper
1946          * caller. Then the kernel will retry to batch pages, and it will not
1947          * add the truncated page into batches as it was removed from page
1948          * cache of the file.
1949          */
1950         if (vmpage->mapping != inode->i_mapping) {
1951                 unlock_page(vmpage);
1952                 RETURN(AOP_TRUNCATED_PAGE);
1953         }
1954
1955         lcc = ll_cl_find(inode);
1956         if (lcc != NULL) {
1957                 env = lcc->lcc_env;
1958                 io  = lcc->lcc_io;
1959         }
1960
1961         if (io == NULL) { /* fast read */
1962                 struct inode *inode = file_inode(file);
1963                 struct ll_file_data *fd = file->private_data;
1964                 struct ll_readahead_state *ras = &fd->fd_ras;
1965                 struct lu_env  *local_env = NULL;
1966
1967                 CDEBUG(D_VFSTRACE, "fast read pgno: %ld\n", vmpage->index);
1968
1969                 result = -ENODATA;
1970
1971                 /* TODO: need to verify the layout version to make sure
1972                  * the page is not invalid due to layout change. */
1973                 page = cl_vmpage_page(vmpage, clob);
1974                 if (page == NULL) {
1975                         unlock_page(vmpage);
1976                         CDEBUG(D_READA, "fast read: failed to find page %ld\n",
1977                                 vmpage->index);
1978                         ll_ra_stats_inc_sbi(sbi, RA_STAT_FAILED_FAST_READ);
1979                         RETURN(result);
1980                 }
1981
1982                 if (page->cp_defer_uptodate) {
1983                         enum ras_update_flags flags = LL_RAS_HIT;
1984
1985                         if (lcc && lcc->lcc_type == LCC_MMAP)
1986                                 flags |= LL_RAS_MMAP;
1987
1988                         /* For fast read, it updates read ahead state only
1989                          * if the page is hit in cache because non cache page
1990                          * case will be handled by slow read later. */
1991                         ras_update(sbi, inode, ras, cl_page_index(page), flags, io);
1992                         /* avoid duplicate ras_update() call */
1993                         page->cp_ra_updated = 1;
1994
1995                         if (ll_use_fast_io(file, ras, cl_page_index(page)))
1996                                 result = 0;
1997                 }
1998
1999                 if (!env) {
2000                         local_env = cl_env_percpu_get();
2001                         env = local_env;
2002                 }
2003
2004                 /* export the page and skip io stack */
2005                 if (result == 0) {
2006                         page->cp_ra_used = 1;
2007                         SetPageUptodate(vmpage);
2008                 } else {
2009                         ll_ra_stats_inc_sbi(sbi, RA_STAT_FAILED_FAST_READ);
2010                 }
2011
2012                 /* release page refcount before unlocking the page to ensure
2013                  * the object won't be destroyed in the calling path of
2014                  * cl_page_put(). Please see comment in ll_releasepage(). */
2015                 cl_page_put(env, page);
2016                 unlock_page(vmpage);
2017                 if (local_env)
2018                         cl_env_percpu_put(local_env);
2019
2020                 RETURN(result);
2021         }
2022
2023         if (lcc && lcc->lcc_type != LCC_MMAP) {
2024                 /*
2025                  * This handles a kernel bug introduced in kernel 5.12:
2026                  * comment: cbd59c48ae2bcadc4a7599c29cf32fd3f9b78251
2027                  * ("mm/filemap: use head pages in generic_file_buffered_read")
2028                  *
2029                  * See above in this function for a full description of the
2030                  * bug.  Briefly, the kernel will try to read 1 more page than
2031                  * was actually requested *if that page is already in cache*.
2032                  *
2033                  * Because this page is beyond the boundary of the requested
2034                  * read, Lustre does not lock it as part of the read.  This
2035                  * means we must check if there is a valid dlmlock on this
2036                  * this page and reference it before we attempt to read in the
2037                  * page.  If there is not a valid dlmlock, then we are racing
2038                  * with dlmlock cancellation and the page is being removed
2039                  * from the cache.
2040                  *
2041                  * That means we should return AOP_TRUNCATED_PAGE, which will
2042                  * cause the kernel to retry the read, which should allow the
2043                  * page to be removed from cache as the lock is cancelled.
2044                  *
2045                  * This should never occur except in kernels with the bug
2046                  * mentioned above.
2047                  */
2048                 if (vmpage->index >= lcc->lcc_end_index) {
2049                         CDEBUG(D_VFSTRACE,
2050                                "pgno:%ld, beyond read end_index:%ld\n",
2051                                vmpage->index, lcc->lcc_end_index);
2052
2053                         result = cl_io_read_ahead(env, io, vmpage->index, &ra);
2054                         if (result < 0 || vmpage->index > ra.cra_end_idx) {
2055                                 cl_read_ahead_release(env, &ra);
2056                                 unlock_page(vmpage);
2057                                 RETURN(AOP_TRUNCATED_PAGE);
2058                         }
2059                 }
2060         }
2061
2062         vio = vvp_env_io(env);
2063         /**
2064          * Direct read can fall back to buffered read, but DIO is done
2065          * with lockless i/o, and buffered requires LDLM locking, so in
2066          * this case we must restart without lockless.
2067          */
2068         flags = iocb_ki_flags_get(file, vio->vui_iocb);
2069         if (iocb_ki_flags_check(flags, DIRECT) &&
2070             lcc && lcc->lcc_type == LCC_RW &&
2071             !io->ci_dio_lock) {
2072                 unlock_page(vmpage);
2073                 io->ci_dio_lock = 1;
2074                 io->ci_need_restart = 1;
2075                 GOTO(out, result = -ENOLCK);
2076         }
2077
2078         LASSERT(io->ci_state == CIS_IO_GOING);
2079         page = cl_page_find(env, clob, vmpage->index, vmpage, CPT_CACHEABLE);
2080         if (!IS_ERR(page)) {
2081                 LASSERT(page->cp_type == CPT_CACHEABLE);
2082                 if (likely(!PageUptodate(vmpage))) {
2083                         cl_page_assume(env, io, page);
2084
2085                         result = ll_io_read_page(env, io, page, file);
2086                 } else {
2087                         /* Page from a non-object file. */
2088                         unlock_page(vmpage);
2089                         result = 0;
2090                 }
2091                 cl_page_put(env, page);
2092         } else {
2093                 unlock_page(vmpage);
2094                 result = PTR_ERR(page);
2095         }
2096
2097 out:
2098         if (ra.cra_release != NULL)
2099                 cl_read_ahead_release(env, &ra);
2100
2101         /* this delay gives time for the actual read of the page to finish and
2102          * unlock the page in vvp_page_completion_read before we return to our
2103          * caller and the caller tries to use the page, allowing us to test
2104          * races with the page being unlocked after readpage() but before it's
2105          * used by the caller
2106          */
2107         CFS_FAIL_TIMEOUT(OBD_FAIL_LLITE_READPAGE_PAUSE2, cfs_fail_val);
2108
2109         RETURN(result);
2110 }
2111
2112 #ifdef HAVE_AOPS_READ_FOLIO
2113 int ll_read_folio(struct file *file, struct folio *folio)
2114 {
2115         return ll_readpage(file, folio_page(folio, 0));
2116 }
2117 #endif