lustre/obdclass/cl_page.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved.
  30  * Use is subject to license terms.
  31  *
  32  * Copyright (c) 2011, 2012, Whamcloud, Inc.
  33  */
  34 /*
  35  * This file is part of Lustre, http://www.lustre.org/
  36  * Lustre is a trademark of Sun Microsystems, Inc.
  37  *
  38  * Client Lustre Page.
  39  *
  40  *   Author: Nikita Danilov <nikita.danilov@sun.com>
  41  */
  42
  43 #define DEBUG_SUBSYSTEM S_CLASS
  44 #ifndef EXPORT_SYMTAB
  45 # define EXPORT_SYMTAB
  46 #endif
  47
  48 #include <libcfs/libcfs.h>
  49 #include <obd_class.h>
  50 #include <obd_support.h>
  51 #include <libcfs/list.h>
  52
  53 #include <cl_object.h>
  54 #include "cl_internal.h"
  55
  56 static void cl_page_delete0(const struct lu_env *env, struct cl_page *pg,
  57                             int radix);
  58
  59 static cfs_mem_cache_t      *cl_page_kmem = NULL;
  60
  61 static struct lu_kmem_descr cl_page_caches[] = {
  62         {
  63                 .ckd_cache = &cl_page_kmem,
  64                 .ckd_name  = "cl_page_kmem",
  65                 .ckd_size  = sizeof (struct cl_page)
  66         },
  67         {
  68                 .ckd_cache = NULL
  69         }
  70 };
  71
  72 #ifdef LIBCFS_DEBUG
  73 # define PASSERT(env, page, expr)                                       \
  74   do {                                                                    \
  75           if (unlikely(!(expr))) {                                      \
  76                   CL_PAGE_DEBUG(D_ERROR, (env), (page), #expr "\n");    \
  77                   LASSERT(0);                                           \
  78           }                                                             \
  79   } while (0)
  80 #else /* !LIBCFS_DEBUG */
  81 # define PASSERT(env, page, exp) \
  82         ((void)sizeof(env), (void)sizeof(page), (void)sizeof !!(exp))
  83 #endif /* !LIBCFS_DEBUG */
  84
  85 #ifdef INVARIANT_CHECK
  86 # define PINVRNT(env, page, expr)                                       \
  87   do {                                                                    \
  88           if (unlikely(!(expr))) {                                      \
  89                   CL_PAGE_DEBUG(D_ERROR, (env), (page), #expr "\n");    \
  90                   LINVRNT(0);                                           \
  91           }                                                             \
  92   } while (0)
  93 #else /* !INVARIANT_CHECK */
  94 # define PINVRNT(env, page, exp) \
  95         ((void)sizeof(env), (void)sizeof(page), (void)sizeof !!(exp))
  96 #endif /* !INVARIANT_CHECK */
  97
  98 /**
  99  * Internal version of cl_page_top, it should be called with page referenced,
 100  * or coh_page_guard held.
 101  */
 102 static struct cl_page *cl_page_top_trusted(struct cl_page *page)
 103 {
 104         while (page->cp_parent != NULL)
 105                 page = page->cp_parent;
 106         return page;
 107 }
 108
 109 /**
 110  * Internal version of cl_page_get().
 111  *
 112  * This function can be used to obtain initial reference to previously
 113  * unreferenced cached object. It can be called only if concurrent page
 114  * reclamation is somehow prevented, e.g., by locking page radix-tree
 115  * (cl_object_header::hdr->coh_page_guard), or by keeping a lock on a VM page,
 116  * associated with \a page.
 117  *
 118  * Use with care! Not exported.
 119  */
 120 static void cl_page_get_trust(struct cl_page *page)
 121 {
 122         /*
 123          * Checkless version for trusted users.
 124          */
 125         if (cfs_atomic_inc_return(&page->cp_ref) == 1)
 126                 cfs_atomic_inc(&cl_object_site(page->cp_obj)->cs_pages.cs_busy);
 127 }
 128
 129 /**
 130  * Returns a slice within a page, corresponding to the given layer in the
 131  * device stack.
 132  *
 133  * \see cl_lock_at()
 134  */
 135 static const struct cl_page_slice *
 136 cl_page_at_trusted(const struct cl_page *page,
 137                    const struct lu_device_type *dtype)
 138 {
 139         const struct cl_page_slice *slice;
 140
 141 #ifdef INVARIANT_CHECK
 142         struct cl_object_header *ch = cl_object_header(page->cp_obj);
 143
 144         if (!cfs_atomic_read(&page->cp_ref))
 145                 LASSERT_SPIN_LOCKED(&ch->coh_page_guard);
 146 #endif
 147         ENTRY;
 148
 149         page = cl_page_top_trusted((struct cl_page *)page);
 150         do {
 151                 cfs_list_for_each_entry(slice, &page->cp_layers, cpl_linkage) {
 152                         if (slice->cpl_obj->co_lu.lo_dev->ld_type == dtype)
 153                                 RETURN(slice);
 154                 }
 155                 page = page->cp_child;
 156         } while (page != NULL);
 157         RETURN(NULL);
 158 }
 159
 160 /**
 161  * Returns a page with given index in the given object, or NULL if no page is
 162  * found. Acquires a reference on \a page.
 163  *
 164  * Locking: called under cl_object_header::coh_page_guard spin-lock.
 165  */
 166 struct cl_page *cl_page_lookup(struct cl_object_header *hdr, pgoff_t index)
 167 {
 168         struct cl_page *page;
 169
 170         LASSERT_SPIN_LOCKED(&hdr->coh_page_guard);
 171
 172         page = radix_tree_lookup(&hdr->coh_tree, index);
 173         if (page != NULL) {
 174                 cl_page_get_trust(page);
 175         }
 176         return page;
 177 }
 178 EXPORT_SYMBOL(cl_page_lookup);
 179
 180 /**
 181  * Returns a list of pages by a given [start, end] of \a obj.
 182  *
 183  * \param resched If not NULL, then we give up before hogging CPU for too
 184  * long and set *resched = 1, in that case caller should implement a retry
 185  * logic.
 186  *
 187  * Gang tree lookup (radix_tree_gang_lookup()) optimization is absolutely
 188  * crucial in the face of [offset, EOF] locks.
 189  *
 190  * Return at least one page in @queue unless there is no covered page.
 191  */
 192 int cl_page_gang_lookup(const struct lu_env *env, struct cl_object *obj,
 193                         struct cl_io *io, pgoff_t start, pgoff_t end,
 194                         cl_page_gang_cb_t cb, void *cbdata)
 195 {
 196         struct cl_object_header *hdr;
 197         struct cl_page          *page;
 198         struct cl_page         **pvec;
 199         const struct cl_page_slice  *slice;
 200         const struct lu_device_type *dtype;
 201         pgoff_t                  idx;
 202         unsigned int             nr;
 203         unsigned int             i;
 204         unsigned int             j;
 205         int                      res = CLP_GANG_OKAY;
 206         int                      tree_lock = 1;
 207         ENTRY;
 208
 209         idx = start;
 210         hdr = cl_object_header(obj);
 211         pvec = cl_env_info(env)->clt_pvec;
 212         dtype = cl_object_top(obj)->co_lu.lo_dev->ld_type;
 213         cfs_spin_lock(&hdr->coh_page_guard);
 214         while ((nr = radix_tree_gang_lookup(&hdr->coh_tree, (void **)pvec,
 215                                             idx, CLT_PVEC_SIZE)) > 0) {
 216                 int end_of_region = 0;
 217                 idx = pvec[nr - 1]->cp_index + 1;
 218                 for (i = 0, j = 0; i < nr; ++i) {
 219                         page = pvec[i];
 220                         pvec[i] = NULL;
 221
 222                         LASSERT(page->cp_type == CPT_CACHEABLE);
 223                         if (page->cp_index > end) {
 224                                 end_of_region = 1;
 225                                 break;
 226                         }
 227                         if (page->cp_state == CPS_FREEING)
 228                                 continue;
 229
 230                         slice = cl_page_at_trusted(page, dtype);
 231                         /*
 232                          * Pages for lsm-less file has no underneath sub-page
 233                          * for osc, in case of ...
 234                          */
 235                         PASSERT(env, page, slice != NULL);
 236
 237                         page = slice->cpl_page;
 238                         /*
 239                          * Can safely call cl_page_get_trust() under
 240                          * radix-tree spin-lock.
 241                          *
 242                          * XXX not true, because @page is from object another
 243                          * than @hdr and protected by different tree lock.
 244                          */
 245                         cl_page_get_trust(page);
 246                         lu_ref_add_atomic(&page->cp_reference,
 247                                           "gang_lookup", cfs_current());
 248                         pvec[j++] = page;
 249                 }
 250
 251                 /*
 252                  * Here a delicate locking dance is performed. Current thread
 253                  * holds a reference to a page, but has to own it before it
 254                  * can be placed into queue. Owning implies waiting, so
 255                  * radix-tree lock is to be released. After a wait one has to
 256                  * check that pages weren't truncated (cl_page_own() returns
 257                  * error in the latter case).
 258                  */
 259                 cfs_spin_unlock(&hdr->coh_page_guard);
 260                 tree_lock = 0;
 261
 262                 for (i = 0; i < j; ++i) {
 263                         page = pvec[i];
 264                         if (res == CLP_GANG_OKAY)
 265                                 res = (*cb)(env, io, page, cbdata);
 266                         lu_ref_del(&page->cp_reference,
 267                                    "gang_lookup", cfs_current());
 268                         cl_page_put(env, page);
 269                 }
 270                 if (nr < CLT_PVEC_SIZE || end_of_region)
 271                         break;
 272
 273                 if (res == CLP_GANG_OKAY && cfs_need_resched())
 274                         res = CLP_GANG_RESCHED;
 275                 if (res != CLP_GANG_OKAY)
 276                         break;
 277
 278                 cfs_spin_lock(&hdr->coh_page_guard);
 279                 tree_lock = 1;
 280         }
 281         if (tree_lock)
 282                 cfs_spin_unlock(&hdr->coh_page_guard);
 283         RETURN(res);
 284 }
 285 EXPORT_SYMBOL(cl_page_gang_lookup);
 286
 287 static void cl_page_free(const struct lu_env *env, struct cl_page *page)
 288 {
 289         struct cl_object *obj  = page->cp_obj;
 290         struct cl_site   *site = cl_object_site(obj);
 291
 292         PASSERT(env, page, cfs_list_empty(&page->cp_batch));
 293         PASSERT(env, page, page->cp_owner == NULL);
 294         PASSERT(env, page, page->cp_req == NULL);
 295         PASSERT(env, page, page->cp_parent == NULL);
 296         PASSERT(env, page, page->cp_state == CPS_FREEING);
 297
 298         ENTRY;
 299         cfs_might_sleep();
 300         while (!cfs_list_empty(&page->cp_layers)) {
 301                 struct cl_page_slice *slice;
 302
 303                 slice = cfs_list_entry(page->cp_layers.next,
 304                                        struct cl_page_slice, cpl_linkage);
 305                 cfs_list_del_init(page->cp_layers.next);
 306                 slice->cpl_ops->cpo_fini(env, slice);
 307         }
 308         cfs_atomic_dec(&site->cs_pages.cs_total);
 309
 310 #ifdef LUSTRE_PAGESTATE_TRACKING
 311         cfs_atomic_dec(&site->cs_pages_state[page->cp_state]);
 312 #endif
 313         lu_object_ref_del_at(&obj->co_lu, page->cp_obj_ref, "cl_page", page);
 314         cl_object_put(env, obj);
 315         lu_ref_fini(&page->cp_reference);
 316         OBD_SLAB_FREE_PTR(page, cl_page_kmem);
 317         EXIT;
 318 }
 319
 320 /**
 321  * Helper function updating page state. This is the only place in the code
 322  * where cl_page::cp_state field is mutated.
 323  */
 324 static inline void cl_page_state_set_trust(struct cl_page *page,
 325                                            enum cl_page_state state)
 326 {
 327         /* bypass const. */
 328         *(enum cl_page_state *)&page->cp_state = state;
 329 }
 330
 331 static int cl_page_alloc(const struct lu_env *env, struct cl_object *o,
 332                          pgoff_t ind, struct page *vmpage,
 333                          enum cl_page_type type, struct cl_page **out)
 334 {
 335         struct cl_page          *page;
 336         struct cl_page          *err  = NULL;
 337         struct lu_object_header *head;
 338         struct cl_site          *site = cl_object_site(o);
 339         int                      result;
 340
 341         ENTRY;
 342         result = +1;
 343         OBD_SLAB_ALLOC_PTR_GFP(page, cl_page_kmem, CFS_ALLOC_IO);
 344         if (page != NULL) {
 345                 cfs_atomic_set(&page->cp_ref, 1);
 346                 page->cp_obj = o;
 347                 cl_object_get(o);
 348                 page->cp_obj_ref = lu_object_ref_add(&o->co_lu,
 349                                                      "cl_page", page);
 350                 page->cp_index = ind;
 351                 cl_page_state_set_trust(page, CPS_CACHED);
 352                 page->cp_type = type;
 353                 CFS_INIT_LIST_HEAD(&page->cp_layers);
 354                 CFS_INIT_LIST_HEAD(&page->cp_batch);
 355                 CFS_INIT_LIST_HEAD(&page->cp_flight);
 356                 cfs_mutex_init(&page->cp_mutex);
 357                 lu_ref_init(&page->cp_reference);
 358                 head = o->co_lu.lo_header;
 359                 cfs_list_for_each_entry(o, &head->loh_layers,
 360                                         co_lu.lo_linkage) {
 361                         if (o->co_ops->coo_page_init != NULL) {
 362                                 err = o->co_ops->coo_page_init(env, o,
 363                                                                page, vmpage);
 364                                 if (err != NULL) {
 365                                         cl_page_delete0(env, page, 0);
 366                                         cl_page_free(env, page);
 367                                         page = err;
 368                                         break;
 369                                 }
 370                         }
 371                 }
 372                 if (err == NULL) {
 373                         cfs_atomic_inc(&site->cs_pages.cs_busy);
 374                         cfs_atomic_inc(&site->cs_pages.cs_total);
 375
 376 #ifdef LUSTRE_PAGESTATE_TRACKING
 377                         cfs_atomic_inc(&site->cs_pages_state[CPS_CACHED]);
 378 #endif
 379                         cfs_atomic_inc(&site->cs_pages.cs_created);
 380                         result = 0;
 381                 }
 382         } else
 383                 page = ERR_PTR(-ENOMEM);
 384         *out = page;
 385         RETURN(result);
 386 }
 387
 388 /**
 389  * Returns a cl_page with index \a idx at the object \a o, and associated with
 390  * the VM page \a vmpage.
 391  *
 392  * This is the main entry point into the cl_page caching interface. First, a
 393  * cache (implemented as a per-object radix tree) is consulted. If page is
 394  * found there, it is returned immediately. Otherwise new page is allocated
 395  * and returned. In any case, additional reference to page is acquired.
 396  *
 397  * \see cl_object_find(), cl_lock_find()
 398  */
 399 static struct cl_page *cl_page_find0(const struct lu_env *env,
 400                                      struct cl_object *o,
 401                                      pgoff_t idx, struct page *vmpage,
 402                                      enum cl_page_type type,
 403                                      struct cl_page *parent)
 404 {
 405         struct cl_page          *page = NULL;
 406         struct cl_page          *ghost = NULL;
 407         struct cl_object_header *hdr;
 408         struct cl_site          *site = cl_object_site(o);
 409         int err;
 410
 411         LASSERT(type == CPT_CACHEABLE || type == CPT_TRANSIENT);
 412         cfs_might_sleep();
 413
 414         ENTRY;
 415
 416         hdr = cl_object_header(o);
 417         cfs_atomic_inc(&site->cs_pages.cs_lookup);
 418
 419         CDEBUG(D_PAGE, "%lu@"DFID" %p %lx %d\n",
 420                idx, PFID(&hdr->coh_lu.loh_fid), vmpage, vmpage->private, type);
 421         /* fast path. */
 422         if (type == CPT_CACHEABLE) {
 423                 /*
 424                  * cl_vmpage_page() can be called here without any locks as
 425                  *
 426                  *     - "vmpage" is locked (which prevents ->private from
 427                  *       concurrent updates), and
 428                  *
 429                  *     - "o" cannot be destroyed while current thread holds a
 430                  *       reference on it.
 431                  */
 432                 page = cl_vmpage_page(vmpage, o);
 433                 PINVRNT(env, page,
 434                         ergo(page != NULL,
 435                              cl_page_vmpage(env, page) == vmpage &&
 436                              (void *)radix_tree_lookup(&hdr->coh_tree,
 437                                                        idx) == page));
 438         }
 439
 440         if (page != NULL) {
 441                 cfs_atomic_inc(&site->cs_pages.cs_hit);
 442                 RETURN(page);
 443         }
 444
 445         /* allocate and initialize cl_page */
 446         err = cl_page_alloc(env, o, idx, vmpage, type, &page);
 447         if (err != 0)
 448                 RETURN(page);
 449
 450         if (type == CPT_TRANSIENT) {
 451                 if (parent) {
 452                         LASSERT(page->cp_parent == NULL);
 453                         page->cp_parent = parent;
 454                         parent->cp_child = page;
 455                 }
 456                 RETURN(page);
 457         }
 458
 459         /*
 460          * XXX optimization: use radix_tree_preload() here, and change tree
 461          * gfp mask to GFP_KERNEL in cl_object_header_init().
 462          */
 463         cfs_spin_lock(&hdr->coh_page_guard);
 464         err = radix_tree_insert(&hdr->coh_tree, idx, page);
 465         if (err != 0) {
 466                 ghost = page;
 467                 /*
 468                  * Noted by Jay: a lock on \a vmpage protects cl_page_find()
 469                  * from this race, but
 470                  *
 471                  *     0. it's better to have cl_page interface "locally
 472                  *     consistent" so that its correctness can be reasoned
 473                  *     about without appealing to the (obscure world of) VM
 474                  *     locking.
 475                  *
 476                  *     1. handling this race allows ->coh_tree to remain
 477                  *     consistent even when VM locking is somehow busted,
 478                  *     which is very useful during diagnosing and debugging.
 479                  */
 480                 page = ERR_PTR(err);
 481                 CL_PAGE_DEBUG(D_ERROR, env, ghost,
 482                               "fail to insert into radix tree: %d\n", err);
 483         } else {
 484                 if (parent) {
 485                         LASSERT(page->cp_parent == NULL);
 486                         page->cp_parent = parent;
 487                         parent->cp_child = page;
 488                 }
 489                 hdr->coh_pages++;
 490         }
 491         cfs_spin_unlock(&hdr->coh_page_guard);
 492
 493         if (unlikely(ghost != NULL)) {
 494                 cfs_atomic_dec(&site->cs_pages.cs_busy);
 495                 cl_page_delete0(env, ghost, 0);
 496                 cl_page_free(env, ghost);
 497         }
 498         RETURN(page);
 499 }
 500
 501 struct cl_page *cl_page_find(const struct lu_env *env, struct cl_object *o,
 502                              pgoff_t idx, struct page *vmpage,
 503                              enum cl_page_type type)
 504 {
 505         return cl_page_find0(env, o, idx, vmpage, type, NULL);
 506 }
 507 EXPORT_SYMBOL(cl_page_find);
 508
 509
 510 struct cl_page *cl_page_find_sub(const struct lu_env *env, struct cl_object *o,
 511                                  pgoff_t idx, struct page *vmpage,
 512                                  struct cl_page *parent)
 513 {
 514         return cl_page_find0(env, o, idx, vmpage, parent->cp_type, parent);
 515 }
 516 EXPORT_SYMBOL(cl_page_find_sub);
 517
 518 static inline int cl_page_invariant(const struct cl_page *pg)
 519 {
 520         struct cl_object_header *header;
 521         struct cl_page          *parent;
 522         struct cl_page          *child;
 523         struct cl_io            *owner;
 524
 525         /*
 526          * Page invariant is protected by a VM lock.
 527          */
 528         LINVRNT(cl_page_is_vmlocked(NULL, pg));
 529
 530         header = cl_object_header(pg->cp_obj);
 531         parent = pg->cp_parent;
 532         child  = pg->cp_child;
 533         owner  = pg->cp_owner;
 534
 535         return cfs_atomic_read(&pg->cp_ref) > 0 &&
 536                 ergo(parent != NULL, parent->cp_child == pg) &&
 537                 ergo(child != NULL, child->cp_parent == pg) &&
 538                 ergo(child != NULL, pg->cp_obj != child->cp_obj) &&
 539                 ergo(parent != NULL, pg->cp_obj != parent->cp_obj) &&
 540                 ergo(owner != NULL && parent != NULL,
 541                      parent->cp_owner == pg->cp_owner->ci_parent) &&
 542                 ergo(owner != NULL && child != NULL,
 543                      child->cp_owner->ci_parent == owner) &&
 544                 /*
 545                  * Either page is early in initialization (has neither child
 546                  * nor parent yet), or it is in the object radix tree.
 547                  */
 548                 ergo(pg->cp_state < CPS_FREEING && pg->cp_type == CPT_CACHEABLE,
 549                      (void *)radix_tree_lookup(&header->coh_tree,
 550                                                pg->cp_index) == pg ||
 551                      (child == NULL && parent == NULL));
 552 }
 553
 554 static void cl_page_state_set0(const struct lu_env *env,
 555                                struct cl_page *page, enum cl_page_state state)
 556 {
 557         enum cl_page_state old;
 558 #ifdef LUSTRE_PAGESTATE_TRACKING
 559         struct cl_site *site = cl_object_site(page->cp_obj);
 560 #endif
 561
 562         /*
 563          * Matrix of allowed state transitions [old][new], for sanity
 564          * checking.
 565          */
 566         static const int allowed_transitions[CPS_NR][CPS_NR] = {
 567                 [CPS_CACHED] = {
 568                         [CPS_CACHED]  = 0,
 569                         [CPS_OWNED]   = 1, /* io finds existing cached page */
 570                         [CPS_PAGEIN]  = 0,
 571                         [CPS_PAGEOUT] = 1, /* write-out from the cache */
 572                         [CPS_FREEING] = 1, /* eviction on the memory pressure */
 573                 },
 574                 [CPS_OWNED] = {
 575                         [CPS_CACHED]  = 1, /* release to the cache */
 576                         [CPS_OWNED]   = 0,
 577                         [CPS_PAGEIN]  = 1, /* start read immediately */
 578                         [CPS_PAGEOUT] = 1, /* start write immediately */
 579                         [CPS_FREEING] = 1, /* lock invalidation or truncate */
 580                 },
 581                 [CPS_PAGEIN] = {
 582                         [CPS_CACHED]  = 1, /* io completion */
 583                         [CPS_OWNED]   = 0,
 584                         [CPS_PAGEIN]  = 0,
 585                         [CPS_PAGEOUT] = 0,
 586                         [CPS_FREEING] = 0,
 587                 },
 588                 [CPS_PAGEOUT] = {
 589                         [CPS_CACHED]  = 1, /* io completion */
 590                         [CPS_OWNED]   = 0,
 591                         [CPS_PAGEIN]  = 0,
 592                         [CPS_PAGEOUT] = 0,
 593                         [CPS_FREEING] = 0,
 594                 },
 595                 [CPS_FREEING] = {
 596                         [CPS_CACHED]  = 0,
 597                         [CPS_OWNED]   = 0,
 598                         [CPS_PAGEIN]  = 0,
 599                         [CPS_PAGEOUT] = 0,
 600                         [CPS_FREEING] = 0,
 601                 }
 602         };
 603
 604         ENTRY;
 605         old = page->cp_state;
 606         PASSERT(env, page, allowed_transitions[old][state]);
 607         CL_PAGE_HEADER(D_TRACE, env, page, "%d -> %d\n", old, state);
 608         for (; page != NULL; page = page->cp_child) {
 609                 PASSERT(env, page, page->cp_state == old);
 610                 PASSERT(env, page,
 611                         equi(state == CPS_OWNED, page->cp_owner != NULL));
 612
 613 #ifdef LUSTRE_PAGESTATE_TRACKING
 614                 cfs_atomic_dec(&site->cs_pages_state[page->cp_state]);
 615                 cfs_atomic_inc(&site->cs_pages_state[state]);
 616 #endif
 617                 cl_page_state_set_trust(page, state);
 618         }
 619         EXIT;
 620 }
 621
 622 static void cl_page_state_set(const struct lu_env *env,
 623                               struct cl_page *page, enum cl_page_state state)
 624 {
 625         cl_page_state_set0(env, page, state);
 626 }
 627
 628 /**
 629  * Acquires an additional reference to a page.
 630  *
 631  * This can be called only by caller already possessing a reference to \a
 632  * page.
 633  *
 634  * \see cl_object_get(), cl_lock_get().
 635  */
 636 void cl_page_get(struct cl_page *page)
 637 {
 638         ENTRY;
 639         LASSERT(page->cp_state != CPS_FREEING);
 640         cl_page_get_trust(page);
 641         EXIT;
 642 }
 643 EXPORT_SYMBOL(cl_page_get);
 644
 645 /**
 646  * Releases a reference to a page.
 647  *
 648  * When last reference is released, page is returned to the cache, unless it
 649  * is in cl_page_state::CPS_FREEING state, in which case it is immediately
 650  * destroyed.
 651  *
 652  * \see cl_object_put(), cl_lock_put().
 653  */
 654 void cl_page_put(const struct lu_env *env, struct cl_page *page)
 655 {
 656         struct cl_object_header *hdr;
 657         struct cl_site *site = cl_object_site(page->cp_obj);
 658
 659         PASSERT(env, page, cfs_atomic_read(&page->cp_ref) > !!page->cp_parent);
 660
 661         ENTRY;
 662         CL_PAGE_HEADER(D_TRACE, env, page, "%d\n",
 663                        cfs_atomic_read(&page->cp_ref));
 664
 665         hdr = cl_object_header(cl_object_top(page->cp_obj));
 666         if (cfs_atomic_dec_and_lock(&page->cp_ref, &hdr->coh_page_guard)) {
 667                 cfs_atomic_dec(&site->cs_pages.cs_busy);
 668                 /* We're going to access the page w/o a reference, but it's
 669                  * ok because we have grabbed the lock coh_page_guard, which
 670                  * means nobody is able to free this page behind us.
 671                  */
 672                 if (page->cp_state == CPS_FREEING) {
 673                         /* We drop the page reference and check the page state
 674                          * inside the coh_page_guard. So that if it gets here,
 675                          * it is the REALLY last reference to this page.
 676                          */
 677                         cfs_spin_unlock(&hdr->coh_page_guard);
 678
 679                         LASSERT(cfs_atomic_read(&page->cp_ref) == 0);
 680                         PASSERT(env, page, page->cp_owner == NULL);
 681                         PASSERT(env, page, cfs_list_empty(&page->cp_batch));
 682                         /*
 683                          * Page is no longer reachable by other threads. Tear
 684                          * it down.
 685                          */
 686                         cl_page_free(env, page);
 687
 688                         EXIT;
 689                         return;
 690                 }
 691                 cfs_spin_unlock(&hdr->coh_page_guard);
 692         }
 693
 694         EXIT;
 695 }
 696 EXPORT_SYMBOL(cl_page_put);
 697
 698 /**
 699  * Returns a VM page associated with a given cl_page.
 700  */
 701 cfs_page_t *cl_page_vmpage(const struct lu_env *env, struct cl_page *page)
 702 {
 703         const struct cl_page_slice *slice;
 704
 705         /*
 706          * Find uppermost layer with ->cpo_vmpage() method, and return its
 707          * result.
 708          */
 709         page = cl_page_top(page);
 710         do {
 711                 cfs_list_for_each_entry(slice, &page->cp_layers, cpl_linkage) {
 712                         if (slice->cpl_ops->cpo_vmpage != NULL)
 713                                 RETURN(slice->cpl_ops->cpo_vmpage(env, slice));
 714                 }
 715                 page = page->cp_child;
 716         } while (page != NULL);
 717         LBUG(); /* ->cpo_vmpage() has to be defined somewhere in the stack */
 718 }
 719 EXPORT_SYMBOL(cl_page_vmpage);
 720
 721 /**
 722  * Returns a cl_page associated with a VM page, and given cl_object.
 723  */
 724 struct cl_page *cl_vmpage_page(cfs_page_t *vmpage, struct cl_object *obj)
 725 {
 726         struct cl_page *page;
 727         struct cl_object_header *hdr;
 728
 729         ENTRY;
 730         KLASSERT(PageLocked(vmpage));
 731
 732         /*
 733          * NOTE: absence of races and liveness of data are guaranteed by page
 734          *       lock on a "vmpage". That works because object destruction has
 735          *       bottom-to-top pass.
 736          */
 737
 738         /*
 739          * This loop assumes that ->private points to the top-most page. This
 740          * can be rectified easily.
 741          */
 742         hdr = cl_object_header(cl_object_top(obj));
 743         cfs_spin_lock(&hdr->coh_page_guard);
 744         for (page = (void *)vmpage->private;
 745              page != NULL; page = page->cp_child) {
 746                 if (cl_object_same(page->cp_obj, obj)) {
 747                         cl_page_get_trust(page);
 748                         break;
 749                 }
 750         }
 751         cfs_spin_unlock(&hdr->coh_page_guard);
 752         LASSERT(ergo(page, page->cp_type == CPT_CACHEABLE));
 753         RETURN(page);
 754 }
 755 EXPORT_SYMBOL(cl_vmpage_page);
 756
 757 /**
 758  * Returns the top-page for a given page.
 759  *
 760  * \see cl_object_top(), cl_io_top()
 761  */
 762 struct cl_page *cl_page_top(struct cl_page *page)
 763 {
 764         return cl_page_top_trusted(page);
 765 }
 766 EXPORT_SYMBOL(cl_page_top);
 767
 768 const struct cl_page_slice *cl_page_at(const struct cl_page *page,
 769                                        const struct lu_device_type *dtype)
 770 {
 771         return cl_page_at_trusted(page, dtype);
 772 }
 773 EXPORT_SYMBOL(cl_page_at);
 774
 775 #define CL_PAGE_OP(opname) offsetof(struct cl_page_operations, opname)
 776
 777 #define CL_PAGE_INVOKE(_env, _page, _op, _proto, ...)                   \
 778 ({                                                                      \
 779         const struct lu_env        *__env  = (_env);                    \
 780         struct cl_page             *__page = (_page);                   \
 781         const struct cl_page_slice *__scan;                             \
 782         int                         __result;                           \
 783         ptrdiff_t                   __op   = (_op);                     \
 784         int                       (*__method)_proto;                    \
 785                                                                         \
 786         __result = 0;                                                   \
 787         __page = cl_page_top(__page);                                   \
 788         do {                                                            \
 789                 cfs_list_for_each_entry(__scan, &__page->cp_layers,     \
 790                                         cpl_linkage) {                  \
 791                         __method = *(void **)((char *)__scan->cpl_ops + \
 792                                               __op);                    \
 793                         if (__method != NULL) {                         \
 794                                 __result = (*__method)(__env, __scan,   \
 795                                                        ## __VA_ARGS__); \
 796                                 if (__result != 0)                      \
 797                                         break;                          \
 798                         }                                               \
 799                 }                                                       \
 800                 __page = __page->cp_child;                              \
 801         } while (__page != NULL && __result == 0);                      \
 802         if (__result > 0)                                               \
 803                 __result = 0;                                           \
 804         __result;                                                       \
 805 })
 806
 807 #define CL_PAGE_INVOID(_env, _page, _op, _proto, ...)                   \
 808 do {                                                                    \
 809         const struct lu_env        *__env  = (_env);                    \
 810         struct cl_page             *__page = (_page);                   \
 811         const struct cl_page_slice *__scan;                             \
 812         ptrdiff_t                   __op   = (_op);                     \
 813         void                      (*__method)_proto;                    \
 814                                                                         \
 815         __page = cl_page_top(__page);                                   \
 816         do {                                                            \
 817                 cfs_list_for_each_entry(__scan, &__page->cp_layers,     \
 818                                         cpl_linkage) {                  \
 819                         __method = *(void **)((char *)__scan->cpl_ops + \
 820                                               __op);                    \
 821                         if (__method != NULL)                           \
 822                                 (*__method)(__env, __scan,              \
 823                                             ## __VA_ARGS__);            \
 824                 }                                                       \
 825                 __page = __page->cp_child;                              \
 826         } while (__page != NULL);                                       \
 827 } while (0)
 828
 829 #define CL_PAGE_INVOID_REVERSE(_env, _page, _op, _proto, ...)               \
 830 do {                                                                        \
 831         const struct lu_env        *__env  = (_env);                        \
 832         struct cl_page             *__page = (_page);                       \
 833         const struct cl_page_slice *__scan;                                 \
 834         ptrdiff_t                   __op   = (_op);                         \
 835         void                      (*__method)_proto;                        \
 836                                                                             \
 837         /* get to the bottom page. */                                       \
 838         while (__page->cp_child != NULL)                                    \
 839                 __page = __page->cp_child;                                  \
 840         do {                                                                \
 841                 cfs_list_for_each_entry_reverse(__scan, &__page->cp_layers, \
 842                                                 cpl_linkage) {              \
 843                         __method = *(void **)((char *)__scan->cpl_ops +     \
 844                                               __op);                        \
 845                         if (__method != NULL)                               \
 846                                 (*__method)(__env, __scan,                  \
 847                                             ## __VA_ARGS__);                \
 848                 }                                                           \
 849                 __page = __page->cp_parent;                                 \
 850         } while (__page != NULL);                                           \
 851 } while (0)
 852
 853 static int cl_page_invoke(const struct lu_env *env,
 854                           struct cl_io *io, struct cl_page *page, ptrdiff_t op)
 855
 856 {
 857         PINVRNT(env, page, cl_object_same(page->cp_obj, io->ci_obj));
 858         ENTRY;
 859         RETURN(CL_PAGE_INVOKE(env, page, op,
 860                               (const struct lu_env *,
 861                                const struct cl_page_slice *, struct cl_io *),
 862                               io));
 863 }
 864
 865 static void cl_page_invoid(const struct lu_env *env,
 866                            struct cl_io *io, struct cl_page *page, ptrdiff_t op)
 867
 868 {
 869         PINVRNT(env, page, cl_object_same(page->cp_obj, io->ci_obj));
 870         ENTRY;
 871         CL_PAGE_INVOID(env, page, op,
 872                        (const struct lu_env *,
 873                         const struct cl_page_slice *, struct cl_io *), io);
 874         EXIT;
 875 }
 876
 877 static void cl_page_owner_clear(struct cl_page *page)
 878 {
 879         ENTRY;
 880         for (page = cl_page_top(page); page != NULL; page = page->cp_child) {
 881                 if (page->cp_owner != NULL) {
 882                         LASSERT(page->cp_owner->ci_owned_nr > 0);
 883                         page->cp_owner->ci_owned_nr--;
 884                         page->cp_owner = NULL;
 885                         page->cp_task = NULL;
 886                 }
 887         }
 888         EXIT;
 889 }
 890
 891 static void cl_page_owner_set(struct cl_page *page)
 892 {
 893         ENTRY;
 894         for (page = cl_page_top(page); page != NULL; page = page->cp_child) {
 895                 LASSERT(page->cp_owner != NULL);
 896                 page->cp_owner->ci_owned_nr++;
 897         }
 898         EXIT;
 899 }
 900
 901 void cl_page_disown0(const struct lu_env *env,
 902                      struct cl_io *io, struct cl_page *pg)
 903 {
 904         enum cl_page_state state;
 905
 906         ENTRY;
 907         state = pg->cp_state;
 908         PINVRNT(env, pg, state == CPS_OWNED || state == CPS_FREEING);
 909         PINVRNT(env, pg, cl_page_invariant(pg));
 910         cl_page_owner_clear(pg);
 911
 912         if (state == CPS_OWNED)
 913                 cl_page_state_set(env, pg, CPS_CACHED);
 914         /*
 915          * Completion call-backs are executed in the bottom-up order, so that
 916          * uppermost layer (llite), responsible for VFS/VM interaction runs
 917          * last and can release locks safely.
 918          */
 919         CL_PAGE_INVOID_REVERSE(env, pg, CL_PAGE_OP(cpo_disown),
 920                                (const struct lu_env *,
 921                                 const struct cl_page_slice *, struct cl_io *),
 922                                io);
 923         EXIT;
 924 }
 925
 926 /**
 927  * returns true, iff page is owned by the given io.
 928  */
 929 int cl_page_is_owned(const struct cl_page *pg, const struct cl_io *io)
 930 {
 931         LINVRNT(cl_object_same(pg->cp_obj, io->ci_obj));
 932         ENTRY;
 933         RETURN(pg->cp_state == CPS_OWNED && pg->cp_owner == io);
 934 }
 935 EXPORT_SYMBOL(cl_page_is_owned);
 936
 937 /**
 938  * Try to own a page by IO.
 939  *
 940  * Waits until page is in cl_page_state::CPS_CACHED state, and then switch it
 941  * into cl_page_state::CPS_OWNED state.
 942  *
 943  * \pre  !cl_page_is_owned(pg, io)
 944  * \post result == 0 iff cl_page_is_owned(pg, io)
 945  *
 946  * \retval 0   success
 947  *
 948  * \retval -ve failure, e.g., page was destroyed (and landed in
 949  *             cl_page_state::CPS_FREEING instead of cl_page_state::CPS_CACHED).
 950  *             or, page was owned by another thread, or in IO.
 951  *
 952  * \see cl_page_disown()
 953  * \see cl_page_operations::cpo_own()
 954  * \see cl_page_own_try()
 955  * \see cl_page_own
 956  */
 957 static int cl_page_own0(const struct lu_env *env, struct cl_io *io,
 958                         struct cl_page *pg, int nonblock)
 959 {
 960         int result;
 961
 962         PINVRNT(env, pg, !cl_page_is_owned(pg, io));
 963
 964         ENTRY;
 965         pg = cl_page_top(pg);
 966         io = cl_io_top(io);
 967
 968         if (pg->cp_state == CPS_FREEING) {
 969                 result = -ENOENT;
 970         } else {
 971                 result = CL_PAGE_INVOKE(env, pg, CL_PAGE_OP(cpo_own),
 972                                         (const struct lu_env *,
 973                                          const struct cl_page_slice *,
 974                                          struct cl_io *, int),
 975                                         io, nonblock);
 976                 if (result == 0) {
 977                         PASSERT(env, pg, pg->cp_owner == NULL);
 978                         PASSERT(env, pg, pg->cp_req == NULL);
 979                         pg->cp_owner = io;
 980                         pg->cp_task  = current;
 981                         cl_page_owner_set(pg);
 982                         if (pg->cp_state != CPS_FREEING) {
 983                                 cl_page_state_set(env, pg, CPS_OWNED);
 984                         } else {
 985                                 cl_page_disown0(env, io, pg);
 986                                 result = -ENOENT;
 987                         }
 988                 }
 989         }
 990         PINVRNT(env, pg, ergo(result == 0, cl_page_invariant(pg)));
 991         RETURN(result);
 992 }
 993
 994 /**
 995  * Own a page, might be blocked.
 996  *
 997  * \see cl_page_own0()
 998  */
 999 int cl_page_own(const struct lu_env *env, struct cl_io *io, struct cl_page *pg)
1000 {
1001         return cl_page_own0(env, io, pg, 0);
1002 }
1003 EXPORT_SYMBOL(cl_page_own);
1004
1005 /**
1006  * Nonblock version of cl_page_own().
1007  *
1008  * \see cl_page_own0()
1009  */
1010 int cl_page_own_try(const struct lu_env *env, struct cl_io *io,
1011                     struct cl_page *pg)
1012 {
1013         return cl_page_own0(env, io, pg, 1);
1014 }
1015 EXPORT_SYMBOL(cl_page_own_try);
1016
1017
1018 /**
1019  * Assume page ownership.
1020  *
1021  * Called when page is already locked by the hosting VM.
1022  *
1023  * \pre !cl_page_is_owned(pg, io)
1024  * \post cl_page_is_owned(pg, io)
1025  *
1026  * \see cl_page_operations::cpo_assume()
1027  */
1028 void cl_page_assume(const struct lu_env *env,
1029                     struct cl_io *io, struct cl_page *pg)
1030 {
1031         PASSERT(env, pg, pg->cp_owner == NULL);
1032         PINVRNT(env, pg, cl_object_same(pg->cp_obj, io->ci_obj));
1033         PINVRNT(env, pg, cl_page_invariant(pg));
1034
1035         ENTRY;
1036         pg = cl_page_top(pg);
1037         io = cl_io_top(io);
1038
1039         cl_page_invoid(env, io, pg, CL_PAGE_OP(cpo_assume));
1040         pg->cp_owner = io;
1041         pg->cp_task = current;
1042         cl_page_owner_set(pg);
1043         cl_page_state_set(env, pg, CPS_OWNED);
1044         EXIT;
1045 }
1046 EXPORT_SYMBOL(cl_page_assume);
1047
1048 /**
1049  * Releases page ownership without unlocking the page.
1050  *
1051  * Moves page into cl_page_state::CPS_CACHED without releasing a lock on the
1052  * underlying VM page (as VM is supposed to do this itself).
1053  *
1054  * \pre   cl_page_is_owned(pg, io)
1055  * \post !cl_page_is_owned(pg, io)
1056  *
1057  * \see cl_page_assume()
1058  */
1059 void cl_page_unassume(const struct lu_env *env,
1060                       struct cl_io *io, struct cl_page *pg)
1061 {
1062         PINVRNT(env, pg, cl_page_is_owned(pg, io));
1063         PINVRNT(env, pg, cl_page_invariant(pg));
1064
1065         ENTRY;
1066         pg = cl_page_top(pg);
1067         io = cl_io_top(io);
1068         cl_page_owner_clear(pg);
1069         cl_page_state_set(env, pg, CPS_CACHED);
1070         CL_PAGE_INVOID_REVERSE(env, pg, CL_PAGE_OP(cpo_unassume),
1071                                (const struct lu_env *,
1072                                 const struct cl_page_slice *, struct cl_io *),
1073                                io);
1074         EXIT;
1075 }
1076 EXPORT_SYMBOL(cl_page_unassume);
1077
1078 /**
1079  * Releases page ownership.
1080  *
1081  * Moves page into cl_page_state::CPS_CACHED.
1082  *
1083  * \pre   cl_page_is_owned(pg, io)
1084  * \post !cl_page_is_owned(pg, io)
1085  *
1086  * \see cl_page_own()
1087  * \see cl_page_operations::cpo_disown()
1088  */
1089 void cl_page_disown(const struct lu_env *env,
1090                     struct cl_io *io, struct cl_page *pg)
1091 {
1092         PINVRNT(env, pg, cl_page_is_owned(pg, io));
1093
1094         ENTRY;
1095         pg = cl_page_top(pg);
1096         io = cl_io_top(io);
1097         cl_page_disown0(env, io, pg);
1098         EXIT;
1099 }
1100 EXPORT_SYMBOL(cl_page_disown);
1101
1102 /**
1103  * Called when page is to be removed from the object, e.g., as a result of
1104  * truncate.
1105  *
1106  * Calls cl_page_operations::cpo_discard() top-to-bottom.
1107  *
1108  * \pre cl_page_is_owned(pg, io)
1109  *
1110  * \see cl_page_operations::cpo_discard()
1111  */
1112 void cl_page_discard(const struct lu_env *env,
1113                      struct cl_io *io, struct cl_page *pg)
1114 {
1115         PINVRNT(env, pg, cl_page_is_owned(pg, io));
1116         PINVRNT(env, pg, cl_page_invariant(pg));
1117
1118         cl_page_invoid(env, io, pg, CL_PAGE_OP(cpo_discard));
1119 }
1120 EXPORT_SYMBOL(cl_page_discard);
1121
1122 /**
1123  * Version of cl_page_delete() that can be called for not fully constructed
1124  * pages, e.g,. in a error handling cl_page_find()->cl_page_delete0()
1125  * path. Doesn't check page invariant.
1126  */
1127 static void cl_page_delete0(const struct lu_env *env, struct cl_page *pg,
1128                             int radix)
1129 {
1130         struct cl_page *tmp = pg;
1131         ENTRY;
1132
1133         PASSERT(env, pg, pg == cl_page_top(pg));
1134         PASSERT(env, pg, pg->cp_state != CPS_FREEING);
1135
1136         /*
1137          * Severe all ways to obtain new pointers to @pg.
1138          */
1139         cl_page_owner_clear(pg);
1140
1141         /*
1142          * unexport the page firstly before freeing it so that
1143          * the page content is considered to be invalid.
1144          * We have to do this because a CPS_FREEING cl_page may
1145          * be NOT under the protection of a cl_lock.
1146          * Afterwards, if this page is found by other threads, then this
1147          * page will be forced to reread.
1148          */
1149         cl_page_export(env, pg, 0);
1150         cl_page_state_set0(env, pg, CPS_FREEING);
1151
1152         if (tmp->cp_type == CPT_CACHEABLE) {
1153                 if (!radix)
1154                         /* !radix means that @pg is not yet in the radix tree,
1155                          * skip removing it.
1156                          */
1157                         tmp = pg->cp_child;
1158                 for (; tmp != NULL; tmp = tmp->cp_child) {
1159                         void                    *value;
1160                         struct cl_object_header *hdr;
1161
1162                         hdr = cl_object_header(tmp->cp_obj);
1163                         cfs_spin_lock(&hdr->coh_page_guard);
1164                         value = radix_tree_delete(&hdr->coh_tree,
1165                                                   tmp->cp_index);
1166                         PASSERT(env, tmp, value == tmp);
1167                         PASSERT(env, tmp, hdr->coh_pages > 0);
1168                         hdr->coh_pages--;
1169                         cfs_spin_unlock(&hdr->coh_page_guard);
1170                 }
1171         }
1172
1173         CL_PAGE_INVOID(env, pg, CL_PAGE_OP(cpo_delete),
1174                        (const struct lu_env *, const struct cl_page_slice *));
1175         EXIT;
1176 }
1177
1178 /**
1179  * Called when a decision is made to throw page out of memory.
1180  *
1181  * Notifies all layers about page destruction by calling
1182  * cl_page_operations::cpo_delete() method top-to-bottom.
1183  *
1184  * Moves page into cl_page_state::CPS_FREEING state (this is the only place
1185  * where transition to this state happens).
1186  *
1187  * Eliminates all venues through which new references to the page can be
1188  * obtained:
1189  *
1190  *     - removes page from the radix trees,
1191  *
1192  *     - breaks linkage from VM page to cl_page.
1193  *
1194  * Once page reaches cl_page_state::CPS_FREEING, all remaining references will
1195  * drain after some time, at which point page will be recycled.
1196  *
1197  * \pre  pg == cl_page_top(pg)
1198  * \pre  VM page is locked
1199  * \post pg->cp_state == CPS_FREEING
1200  *
1201  * \see cl_page_operations::cpo_delete()
1202  */
1203 void cl_page_delete(const struct lu_env *env, struct cl_page *pg)
1204 {
1205         PINVRNT(env, pg, cl_page_invariant(pg));
1206         ENTRY;
1207         cl_page_delete0(env, pg, 1);
1208         EXIT;
1209 }
1210 EXPORT_SYMBOL(cl_page_delete);
1211
1212 /**
1213  * Unmaps page from user virtual memory.
1214  *
1215  * Calls cl_page_operations::cpo_unmap() through all layers top-to-bottom. The
1216  * layer responsible for VM interaction has to unmap page from user space
1217  * virtual memory.
1218  *
1219  * \see cl_page_operations::cpo_unmap()
1220  */
1221 int cl_page_unmap(const struct lu_env *env,
1222                   struct cl_io *io, struct cl_page *pg)
1223 {
1224         PINVRNT(env, pg, cl_page_is_owned(pg, io));
1225         PINVRNT(env, pg, cl_page_invariant(pg));
1226
1227         return cl_page_invoke(env, io, pg, CL_PAGE_OP(cpo_unmap));
1228 }
1229 EXPORT_SYMBOL(cl_page_unmap);
1230
1231 /**
1232  * Marks page up-to-date.
1233  *
1234  * Call cl_page_operations::cpo_export() through all layers top-to-bottom. The
1235  * layer responsible for VM interaction has to mark/clear page as up-to-date
1236  * by the \a uptodate argument.
1237  *
1238  * \see cl_page_operations::cpo_export()
1239  */
1240 void cl_page_export(const struct lu_env *env, struct cl_page *pg, int uptodate)
1241 {
1242         PINVRNT(env, pg, cl_page_invariant(pg));
1243         CL_PAGE_INVOID(env, pg, CL_PAGE_OP(cpo_export),
1244                        (const struct lu_env *,
1245                         const struct cl_page_slice *, int), uptodate);
1246 }
1247 EXPORT_SYMBOL(cl_page_export);
1248
1249 /**
1250  * Returns true, iff \a pg is VM locked in a suitable sense by the calling
1251  * thread.
1252  */
1253 int cl_page_is_vmlocked(const struct lu_env *env, const struct cl_page *pg)
1254 {
1255         int result;
1256         const struct cl_page_slice *slice;
1257
1258         ENTRY;
1259         pg = cl_page_top_trusted((struct cl_page *)pg);
1260         slice = container_of(pg->cp_layers.next,
1261                              const struct cl_page_slice, cpl_linkage);
1262         PASSERT(env, pg, slice->cpl_ops->cpo_is_vmlocked != NULL);
1263         /*
1264          * Call ->cpo_is_vmlocked() directly instead of going through
1265          * CL_PAGE_INVOKE(), because cl_page_is_vmlocked() is used by
1266          * cl_page_invariant().
1267          */
1268         result = slice->cpl_ops->cpo_is_vmlocked(env, slice);
1269         PASSERT(env, pg, result == -EBUSY || result == -ENODATA);
1270         RETURN(result == -EBUSY);
1271 }
1272 EXPORT_SYMBOL(cl_page_is_vmlocked);
1273
1274 static enum cl_page_state cl_req_type_state(enum cl_req_type crt)
1275 {
1276         ENTRY;
1277         RETURN(crt == CRT_WRITE ? CPS_PAGEOUT : CPS_PAGEIN);
1278 }
1279
1280 static void cl_page_io_start(const struct lu_env *env,
1281                              struct cl_page *pg, enum cl_req_type crt)
1282 {
1283         /*
1284          * Page is queued for IO, change its state.
1285          */
1286         ENTRY;
1287         cl_page_owner_clear(pg);
1288         cl_page_state_set(env, pg, cl_req_type_state(crt));
1289         EXIT;
1290 }
1291
1292 /**
1293  * Prepares page for immediate transfer. cl_page_operations::cpo_prep() is
1294  * called top-to-bottom. Every layer either agrees to submit this page (by
1295  * returning 0), or requests to omit this page (by returning -EALREADY). Layer
1296  * handling interactions with the VM also has to inform VM that page is under
1297  * transfer now.
1298  */
1299 int cl_page_prep(const struct lu_env *env, struct cl_io *io,
1300                  struct cl_page *pg, enum cl_req_type crt)
1301 {
1302         int result;
1303
1304         PINVRNT(env, pg, cl_page_is_owned(pg, io));
1305         PINVRNT(env, pg, cl_page_invariant(pg));
1306         PINVRNT(env, pg, crt < CRT_NR);
1307
1308         /*
1309          * XXX this has to be called bottom-to-top, so that llite can set up
1310          * PG_writeback without risking other layers deciding to skip this
1311          * page.
1312          */
1313         result = cl_page_invoke(env, io, pg, CL_PAGE_OP(io[crt].cpo_prep));
1314         if (result == 0)
1315                 cl_page_io_start(env, pg, crt);
1316
1317         KLASSERT(ergo(crt == CRT_WRITE && pg->cp_type == CPT_CACHEABLE,
1318                       equi(result == 0,
1319                            PageWriteback(cl_page_vmpage(env, pg)))));
1320         CL_PAGE_HEADER(D_TRACE, env, pg, "%d %d\n", crt, result);
1321         return result;
1322 }
1323 EXPORT_SYMBOL(cl_page_prep);
1324
1325 /**
1326  * Notify layers about transfer completion.
1327  *
1328  * Invoked by transfer sub-system (which is a part of osc) to notify layers
1329  * that a transfer, of which this page is a part of has completed.
1330  *
1331  * Completion call-backs are executed in the bottom-up order, so that
1332  * uppermost layer (llite), responsible for the VFS/VM interaction runs last
1333  * and can release locks safely.
1334  *
1335  * \pre  pg->cp_state == CPS_PAGEIN || pg->cp_state == CPS_PAGEOUT
1336  * \post pg->cp_state == CPS_CACHED
1337  *
1338  * \see cl_page_operations::cpo_completion()
1339  */
1340 void cl_page_completion(const struct lu_env *env,
1341                         struct cl_page *pg, enum cl_req_type crt, int ioret)
1342 {
1343         struct cl_sync_io *anchor = pg->cp_sync_io;
1344
1345         PASSERT(env, pg, crt < CRT_NR);
1346         /* cl_page::cp_req already cleared by the caller (osc_completion()) */
1347         PASSERT(env, pg, pg->cp_req == NULL);
1348         PASSERT(env, pg, pg->cp_state == cl_req_type_state(crt));
1349
1350         ENTRY;
1351         CL_PAGE_HEADER(D_TRACE, env, pg, "%d %d\n", crt, ioret);
1352         if (crt == CRT_READ && ioret == 0) {
1353                 PASSERT(env, pg, !(pg->cp_flags & CPF_READ_COMPLETED));
1354                 pg->cp_flags |= CPF_READ_COMPLETED;
1355         }
1356
1357         cl_page_state_set(env, pg, CPS_CACHED);
1358         CL_PAGE_INVOID_REVERSE(env, pg, CL_PAGE_OP(io[crt].cpo_completion),
1359                                (const struct lu_env *,
1360                                 const struct cl_page_slice *, int), ioret);
1361         if (anchor) {
1362                 LASSERT(cl_page_is_vmlocked(env, pg));
1363                 LASSERT(pg->cp_sync_io == anchor);
1364                 pg->cp_sync_io = NULL;
1365                 cl_sync_io_note(anchor, ioret);
1366         }
1367
1368         /* Don't assert the page writeback bit here because the lustre file
1369          * may be as a backend of swap space. in this case, the page writeback
1370          * is set by VM, and obvious we shouldn't clear it at all. Fortunately
1371          * this type of pages are all TRANSIENT pages. */
1372         KLASSERT(ergo(pg->cp_type == CPT_CACHEABLE,
1373                       !PageWriteback(cl_page_vmpage(env, pg))));
1374         EXIT;
1375 }
1376 EXPORT_SYMBOL(cl_page_completion);
1377
1378 /**
1379  * Notify layers that transfer formation engine decided to yank this page from
1380  * the cache and to make it a part of a transfer.
1381  *
1382  * \pre  pg->cp_state == CPS_CACHED
1383  * \post pg->cp_state == CPS_PAGEIN || pg->cp_state == CPS_PAGEOUT
1384  *
1385  * \see cl_page_operations::cpo_make_ready()
1386  */
1387 int cl_page_make_ready(const struct lu_env *env, struct cl_page *pg,
1388                        enum cl_req_type crt)
1389 {
1390         int result;
1391
1392         PINVRNT(env, pg, crt < CRT_NR);
1393
1394         ENTRY;
1395         result = CL_PAGE_INVOKE(env, pg, CL_PAGE_OP(io[crt].cpo_make_ready),
1396                                 (const struct lu_env *,
1397                                  const struct cl_page_slice *));
1398         if (result == 0) {
1399                 PASSERT(env, pg, pg->cp_state == CPS_CACHED);
1400                 cl_page_io_start(env, pg, crt);
1401         }
1402         CL_PAGE_HEADER(D_TRACE, env, pg, "%d %d\n", crt, result);
1403         RETURN(result);
1404 }
1405 EXPORT_SYMBOL(cl_page_make_ready);
1406
1407 /**
1408  * Notify layers that high level io decided to place this page into a cache
1409  * for future transfer.
1410  *
1411  * The layer implementing transfer engine (osc) has to register this page in
1412  * its queues.
1413  *
1414  * \pre  cl_page_is_owned(pg, io)
1415  * \post ergo(result == 0,
1416  *            pg->cp_state == CPS_PAGEIN || pg->cp_state == CPS_PAGEOUT)
1417  *
1418  * \see cl_page_operations::cpo_cache_add()
1419  */
1420 int cl_page_cache_add(const struct lu_env *env, struct cl_io *io,
1421                       struct cl_page *pg, enum cl_req_type crt)
1422 {
1423         int result;
1424
1425         PINVRNT(env, pg, crt < CRT_NR);
1426         PINVRNT(env, pg, cl_page_is_owned(pg, io));
1427         PINVRNT(env, pg, cl_page_invariant(pg));
1428
1429         ENTRY;
1430         result = cl_page_invoke(env, io, pg, CL_PAGE_OP(io[crt].cpo_cache_add));
1431         if (result == 0) {
1432                 cl_page_owner_clear(pg);
1433                 cl_page_state_set(env, pg, CPS_CACHED);
1434         }
1435         CL_PAGE_HEADER(D_TRACE, env, pg, "%d %d\n", crt, result);
1436         RETURN(result);
1437 }
1438 EXPORT_SYMBOL(cl_page_cache_add);
1439
1440 /**
1441  * Checks whether page is protected by any extent lock is at least required
1442  * mode.
1443  *
1444  * \return the same as in cl_page_operations::cpo_is_under_lock() method.
1445  * \see cl_page_operations::cpo_is_under_lock()
1446  */
1447 int cl_page_is_under_lock(const struct lu_env *env, struct cl_io *io,
1448                           struct cl_page *page)
1449 {
1450         int rc;
1451
1452         PINVRNT(env, page, cl_page_invariant(page));
1453
1454         ENTRY;
1455         rc = CL_PAGE_INVOKE(env, page, CL_PAGE_OP(cpo_is_under_lock),
1456                             (const struct lu_env *,
1457                              const struct cl_page_slice *, struct cl_io *),
1458                             io);
1459         PASSERT(env, page, rc != 0);
1460         RETURN(rc);
1461 }
1462 EXPORT_SYMBOL(cl_page_is_under_lock);
1463
1464 static int page_prune_cb(const struct lu_env *env, struct cl_io *io,
1465                          struct cl_page *page, void *cbdata)
1466 {
1467         cl_page_own(env, io, page);
1468         cl_page_unmap(env, io, page);
1469         cl_page_discard(env, io, page);
1470         cl_page_disown(env, io, page);
1471         return CLP_GANG_OKAY;
1472 }
1473
1474 /**
1475  * Purges all cached pages belonging to the object \a obj.
1476  */
1477 int cl_pages_prune(const struct lu_env *env, struct cl_object *clobj)
1478 {
1479         struct cl_thread_info   *info;
1480         struct cl_object        *obj = cl_object_top(clobj);
1481         struct cl_io            *io;
1482         int                      result;
1483
1484         ENTRY;
1485         info  = cl_env_info(env);
1486         io    = &info->clt_io;
1487
1488         /*
1489          * initialize the io. This is ugly since we never do IO in this
1490          * function, we just make cl_page_list functions happy. -jay
1491          */
1492         io->ci_obj = obj;
1493         result = cl_io_init(env, io, CIT_MISC, obj);
1494         if (result != 0) {
1495                 cl_io_fini(env, io);
1496                 RETURN(io->ci_result);
1497         }
1498
1499         do {
1500                 result = cl_page_gang_lookup(env, obj, io, 0, CL_PAGE_EOF,
1501                                              page_prune_cb, NULL);
1502                 if (result == CLP_GANG_RESCHED)
1503                         cfs_cond_resched();
1504         } while (result != CLP_GANG_OKAY);
1505
1506         cl_io_fini(env, io);
1507         RETURN(result);
1508 }
1509 EXPORT_SYMBOL(cl_pages_prune);
1510
1511 /**
1512  * Tells transfer engine that only part of a page is to be transmitted.
1513  *
1514  * \see cl_page_operations::cpo_clip()
1515  */
1516 void cl_page_clip(const struct lu_env *env, struct cl_page *pg,
1517                   int from, int to)
1518 {
1519         PINVRNT(env, pg, cl_page_invariant(pg));
1520
1521         CL_PAGE_HEADER(D_TRACE, env, pg, "%d %d\n", from, to);
1522         CL_PAGE_INVOID(env, pg, CL_PAGE_OP(cpo_clip),
1523                        (const struct lu_env *,
1524                         const struct cl_page_slice *,int, int),
1525                        from, to);
1526 }
1527 EXPORT_SYMBOL(cl_page_clip);
1528
1529 /**
1530  * Prints human readable representation of \a pg to the \a f.
1531  */
1532 void cl_page_header_print(const struct lu_env *env, void *cookie,
1533                           lu_printer_t printer, const struct cl_page *pg)
1534 {
1535         (*printer)(env, cookie,
1536                    "page@%p[%d %p:%lu ^%p_%p %d %d %d %p %p %#x]\n",
1537                    pg, cfs_atomic_read(&pg->cp_ref), pg->cp_obj,
1538                    pg->cp_index, pg->cp_parent, pg->cp_child,
1539                    pg->cp_state, pg->cp_error, pg->cp_type,
1540                    pg->cp_owner, pg->cp_req, pg->cp_flags);
1541 }
1542 EXPORT_SYMBOL(cl_page_header_print);
1543
1544 /**
1545  * Prints human readable representation of \a pg to the \a f.
1546  */
1547 void cl_page_print(const struct lu_env *env, void *cookie,
1548                    lu_printer_t printer, const struct cl_page *pg)
1549 {
1550         struct cl_page *scan;
1551
1552         for (scan = cl_page_top((struct cl_page *)pg);
1553              scan != NULL; scan = scan->cp_child)
1554                 cl_page_header_print(env, cookie, printer, scan);
1555         CL_PAGE_INVOKE(env, (struct cl_page *)pg, CL_PAGE_OP(cpo_print),
1556                        (const struct lu_env *env,
1557                         const struct cl_page_slice *slice,
1558                         void *cookie, lu_printer_t p), cookie, printer);
1559         (*printer)(env, cookie, "end page@%p\n", pg);
1560 }
1561 EXPORT_SYMBOL(cl_page_print);
1562
1563 /**
1564  * Cancel a page which is still in a transfer.
1565  */
1566 int cl_page_cancel(const struct lu_env *env, struct cl_page *page)
1567 {
1568         return CL_PAGE_INVOKE(env, page, CL_PAGE_OP(cpo_cancel),
1569                               (const struct lu_env *,
1570                                const struct cl_page_slice *));
1571 }
1572 EXPORT_SYMBOL(cl_page_cancel);
1573
1574 /**
1575  * Converts a byte offset within object \a obj into a page index.
1576  */
1577 loff_t cl_offset(const struct cl_object *obj, pgoff_t idx)
1578 {
1579         /*
1580          * XXX for now.
1581          */
1582         return (loff_t)idx << CFS_PAGE_SHIFT;
1583 }
1584 EXPORT_SYMBOL(cl_offset);
1585
1586 /**
1587  * Converts a page index into a byte offset within object \a obj.
1588  */
1589 pgoff_t cl_index(const struct cl_object *obj, loff_t offset)
1590 {
1591         /*
1592          * XXX for now.
1593          */
1594         return offset >> CFS_PAGE_SHIFT;
1595 }
1596 EXPORT_SYMBOL(cl_index);
1597
1598 int cl_page_size(const struct cl_object *obj)
1599 {
1600         return 1 << CFS_PAGE_SHIFT;
1601 }
1602 EXPORT_SYMBOL(cl_page_size);
1603
1604 /**
1605  * Adds page slice to the compound page.
1606  *
1607  * This is called by cl_object_operations::coo_page_init() methods to add a
1608  * per-layer state to the page. New state is added at the end of
1609  * cl_page::cp_layers list, that is, it is at the bottom of the stack.
1610  *
1611  * \see cl_lock_slice_add(), cl_req_slice_add(), cl_io_slice_add()
1612  */
1613 void cl_page_slice_add(struct cl_page *page, struct cl_page_slice *slice,
1614                        struct cl_object *obj,
1615                        const struct cl_page_operations *ops)
1616 {
1617         ENTRY;
1618         cfs_list_add_tail(&slice->cpl_linkage, &page->cp_layers);
1619         slice->cpl_obj  = obj;
1620         slice->cpl_ops  = ops;
1621         slice->cpl_page = page;
1622         EXIT;
1623 }
1624 EXPORT_SYMBOL(cl_page_slice_add);
1625
1626 int  cl_page_init(void)
1627 {
1628         return lu_kmem_init(cl_page_caches);
1629 }
1630
1631 void cl_page_fini(void)
1632 {
1633         lu_kmem_fini(cl_page_caches);
1634 }