lustre/obdclass/cl_page.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.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2011, 2013, Intel Corporation.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * Client Lustre Page.
  37  *
  38  *   Author: Nikita Danilov <nikita.danilov@sun.com>
  39  *   Author: Jinshan Xiong <jinshan.xiong@intel.com>
  40  */
  41
  42 #define DEBUG_SUBSYSTEM S_CLASS
  43
  44 #include <libcfs/libcfs.h>
  45 #include <obd_class.h>
  46 #include <obd_support.h>
  47 #include <libcfs/list.h>
  48
  49 #include <cl_object.h>
  50 #include "cl_internal.h"
  51
  52 static void cl_page_delete0(const struct lu_env *env, struct cl_page *pg);
  53
  54 #ifdef LIBCFS_DEBUG
  55 # define PASSERT(env, page, expr)                                       \
  56   do {                                                                    \
  57           if (unlikely(!(expr))) {                                      \
  58                   CL_PAGE_DEBUG(D_ERROR, (env), (page), #expr "\n");    \
  59                   LASSERT(0);                                           \
  60           }                                                             \
  61   } while (0)
  62 #else /* !LIBCFS_DEBUG */
  63 # define PASSERT(env, page, exp) \
  64         ((void)sizeof(env), (void)sizeof(page), (void)sizeof !!(exp))
  65 #endif /* !LIBCFS_DEBUG */
  66
  67 #ifdef CONFIG_LUSTRE_DEBUG_EXPENSIVE_CHECK
  68 # define PINVRNT(env, page, expr)                                       \
  69   do {                                                                    \
  70           if (unlikely(!(expr))) {                                      \
  71                   CL_PAGE_DEBUG(D_ERROR, (env), (page), #expr "\n");    \
  72                   LINVRNT(0);                                           \
  73           }                                                             \
  74   } while (0)
  75 #else /* !CONFIG_LUSTRE_DEBUG_EXPENSIVE_CHECK */
  76 # define PINVRNT(env, page, exp) \
  77          ((void)sizeof(env), (void)sizeof(page), (void)sizeof !!(exp))
  78 #endif /* !CONFIG_LUSTRE_DEBUG_EXPENSIVE_CHECK */
  79
  80 /* Disable page statistic by default due to huge performance penalty. */
  81 #ifdef CONFIG_DEBUG_PAGESTATE_TRACKING
  82 #define CS_PAGE_INC(o, item) \
  83         atomic_inc(&cl_object_site(o)->cs_pages.cs_stats[CS_##item])
  84 #define CS_PAGE_DEC(o, item) \
  85         atomic_dec(&cl_object_site(o)->cs_pages.cs_stats[CS_##item])
  86 #define CS_PAGESTATE_INC(o, state) \
  87         atomic_inc(&cl_object_site(o)->cs_pages_state[state])
  88 #define CS_PAGESTATE_DEC(o, state) \
  89         atomic_dec(&cl_object_site(o)->cs_pages_state[state])
  90 #else
  91 #define CS_PAGE_INC(o, item)
  92 #define CS_PAGE_DEC(o, item)
  93 #define CS_PAGESTATE_INC(o, state)
  94 #define CS_PAGESTATE_DEC(o, state)
  95 #endif
  96
  97 /**
  98  * Internal version of cl_page_get().
  99  *
 100  * This function can be used to obtain initial reference to previously
 101  * unreferenced cached object. It can be called only if concurrent page
 102  * reclamation is somehow prevented, e.g., by keeping a lock on a VM page,
 103  * associated with \a page.
 104  *
 105  * Use with care! Not exported.
 106  */
 107 static void cl_page_get_trust(struct cl_page *page)
 108 {
 109         LASSERT(atomic_read(&page->cp_ref) > 0);
 110         atomic_inc(&page->cp_ref);
 111 }
 112
 113 /**
 114  * Returns a slice within a page, corresponding to the given layer in the
 115  * device stack.
 116  *
 117  * \see cl_lock_at()
 118  */
 119 static const struct cl_page_slice *
 120 cl_page_at_trusted(const struct cl_page *page,
 121                    const struct lu_device_type *dtype)
 122 {
 123         const struct cl_page_slice *slice;
 124         ENTRY;
 125
 126         list_for_each_entry(slice, &page->cp_layers, cpl_linkage) {
 127                 if (slice->cpl_obj->co_lu.lo_dev->ld_type == dtype)
 128                         RETURN(slice);
 129         }
 130         RETURN(NULL);
 131 }
 132
 133 static void cl_page_free(const struct lu_env *env, struct cl_page *page)
 134 {
 135         struct cl_object *obj  = page->cp_obj;
 136         int pagesize = cl_object_header(obj)->coh_page_bufsize;
 137
 138         PASSERT(env, page, list_empty(&page->cp_batch));
 139         PASSERT(env, page, page->cp_owner == NULL);
 140         PASSERT(env, page, page->cp_req == NULL);
 141         PASSERT(env, page, page->cp_state == CPS_FREEING);
 142
 143         ENTRY;
 144         while (!list_empty(&page->cp_layers)) {
 145                 struct cl_page_slice *slice;
 146
 147                 slice = list_entry(page->cp_layers.next,
 148                                    struct cl_page_slice, cpl_linkage);
 149                 list_del_init(page->cp_layers.next);
 150                 if (unlikely(slice->cpl_ops->cpo_fini != NULL))
 151                         slice->cpl_ops->cpo_fini(env, slice);
 152         }
 153         CS_PAGE_DEC(obj, total);
 154         CS_PAGESTATE_DEC(obj, page->cp_state);
 155         lu_object_ref_del_at(&obj->co_lu, &page->cp_obj_ref, "cl_page", page);
 156         cl_object_put(env, obj);
 157         lu_ref_fini(&page->cp_reference);
 158         OBD_FREE(page, pagesize);
 159         EXIT;
 160 }
 161
 162 /**
 163  * Helper function updating page state. This is the only place in the code
 164  * where cl_page::cp_state field is mutated.
 165  */
 166 static inline void cl_page_state_set_trust(struct cl_page *page,
 167                                            enum cl_page_state state)
 168 {
 169         /* bypass const. */
 170         *(enum cl_page_state *)&page->cp_state = state;
 171 }
 172
 173 struct cl_page *cl_page_alloc(const struct lu_env *env,
 174                 struct cl_object *o, pgoff_t ind, struct page *vmpage,
 175                 enum cl_page_type type)
 176 {
 177         struct cl_page          *page;
 178         struct lu_object_header *head;
 179
 180         ENTRY;
 181         OBD_ALLOC_GFP(page, cl_object_header(o)->coh_page_bufsize,
 182                         GFP_NOFS);
 183         if (page != NULL) {
 184                 int result = 0;
 185                 atomic_set(&page->cp_ref, 1);
 186                 page->cp_obj = o;
 187                 cl_object_get(o);
 188                 lu_object_ref_add_at(&o->co_lu, &page->cp_obj_ref, "cl_page",
 189                                      page);
 190                 page->cp_vmpage = vmpage;
 191                 cl_page_state_set_trust(page, CPS_CACHED);
 192                 page->cp_type = type;
 193                 INIT_LIST_HEAD(&page->cp_layers);
 194                 INIT_LIST_HEAD(&page->cp_batch);
 195                 INIT_LIST_HEAD(&page->cp_flight);
 196                 lu_ref_init(&page->cp_reference);
 197                 head = o->co_lu.lo_header;
 198                 list_for_each_entry(o, &head->loh_layers,
 199                                     co_lu.lo_linkage) {
 200                         if (o->co_ops->coo_page_init != NULL) {
 201                                 result = o->co_ops->coo_page_init(env, o, page,
 202                                                                   ind);
 203                                 if (result != 0) {
 204                                         cl_page_delete0(env, page);
 205                                         cl_page_free(env, page);
 206                                         page = ERR_PTR(result);
 207                                         break;
 208                                 }
 209                         }
 210                 }
 211                 if (result == 0) {
 212                         CS_PAGE_INC(o, total);
 213                         CS_PAGE_INC(o, create);
 214                         CS_PAGESTATE_DEC(o, CPS_CACHED);
 215                 }
 216         } else {
 217                 page = ERR_PTR(-ENOMEM);
 218         }
 219         RETURN(page);
 220 }
 221 EXPORT_SYMBOL(cl_page_alloc);
 222
 223 /**
 224  * Returns a cl_page with index \a idx at the object \a o, and associated with
 225  * the VM page \a vmpage.
 226  *
 227  * This is the main entry point into the cl_page caching interface. First, a
 228  * cache (implemented as a per-object radix tree) is consulted. If page is
 229  * found there, it is returned immediately. Otherwise new page is allocated
 230  * and returned. In any case, additional reference to page is acquired.
 231  *
 232  * \see cl_object_find(), cl_lock_find()
 233  */
 234 struct cl_page *cl_page_find(const struct lu_env *env,
 235                              struct cl_object *o,
 236                              pgoff_t idx, struct page *vmpage,
 237                              enum cl_page_type type)
 238 {
 239         struct cl_page          *page = NULL;
 240         struct cl_object_header *hdr;
 241
 242         LASSERT(type == CPT_CACHEABLE || type == CPT_TRANSIENT);
 243         might_sleep();
 244
 245         ENTRY;
 246
 247         hdr = cl_object_header(o);
 248         CS_PAGE_INC(o, lookup);
 249
 250         CDEBUG(D_PAGE, "%lu@"DFID" %p %lx %d\n",
 251                idx, PFID(&hdr->coh_lu.loh_fid), vmpage, vmpage->private, type);
 252         /* fast path. */
 253         if (type == CPT_CACHEABLE) {
 254                 /* vmpage lock is used to protect the child/parent
 255                  * relationship */
 256                 KLASSERT(PageLocked(vmpage));
 257                 /*
 258                  * cl_vmpage_page() can be called here without any locks as
 259                  *
 260                  *     - "vmpage" is locked (which prevents ->private from
 261                  *       concurrent updates), and
 262                  *
 263                  *     - "o" cannot be destroyed while current thread holds a
 264                  *       reference on it.
 265                  */
 266                 page = cl_vmpage_page(vmpage, o);
 267                 if (page != NULL) {
 268                         CS_PAGE_INC(o, hit);
 269                         RETURN(page);
 270                 }
 271         }
 272
 273         /* allocate and initialize cl_page */
 274         page = cl_page_alloc(env, o, idx, vmpage, type);
 275         RETURN(page);
 276 }
 277 EXPORT_SYMBOL(cl_page_find);
 278
 279 static inline int cl_page_invariant(const struct cl_page *pg)
 280 {
 281         return cl_page_in_use_noref(pg);
 282 }
 283
 284 static void cl_page_state_set0(const struct lu_env *env,
 285                                struct cl_page *page, enum cl_page_state state)
 286 {
 287         enum cl_page_state old;
 288
 289         /*
 290          * Matrix of allowed state transitions [old][new], for sanity
 291          * checking.
 292          */
 293         static const int allowed_transitions[CPS_NR][CPS_NR] = {
 294                 [CPS_CACHED] = {
 295                         [CPS_CACHED]  = 0,
 296                         [CPS_OWNED]   = 1, /* io finds existing cached page */
 297                         [CPS_PAGEIN]  = 0,
 298                         [CPS_PAGEOUT] = 1, /* write-out from the cache */
 299                         [CPS_FREEING] = 1, /* eviction on the memory pressure */
 300                 },
 301                 [CPS_OWNED] = {
 302                         [CPS_CACHED]  = 1, /* release to the cache */
 303                         [CPS_OWNED]   = 0,
 304                         [CPS_PAGEIN]  = 1, /* start read immediately */
 305                         [CPS_PAGEOUT] = 1, /* start write immediately */
 306                         [CPS_FREEING] = 1, /* lock invalidation or truncate */
 307                 },
 308                 [CPS_PAGEIN] = {
 309                         [CPS_CACHED]  = 1, /* io completion */
 310                         [CPS_OWNED]   = 0,
 311                         [CPS_PAGEIN]  = 0,
 312                         [CPS_PAGEOUT] = 0,
 313                         [CPS_FREEING] = 0,
 314                 },
 315                 [CPS_PAGEOUT] = {
 316                         [CPS_CACHED]  = 1, /* io completion */
 317                         [CPS_OWNED]   = 0,
 318                         [CPS_PAGEIN]  = 0,
 319                         [CPS_PAGEOUT] = 0,
 320                         [CPS_FREEING] = 0,
 321                 },
 322                 [CPS_FREEING] = {
 323                         [CPS_CACHED]  = 0,
 324                         [CPS_OWNED]   = 0,
 325                         [CPS_PAGEIN]  = 0,
 326                         [CPS_PAGEOUT] = 0,
 327                         [CPS_FREEING] = 0,
 328                 }
 329         };
 330
 331         ENTRY;
 332         old = page->cp_state;
 333         PASSERT(env, page, allowed_transitions[old][state]);
 334         CL_PAGE_HEADER(D_TRACE, env, page, "%d -> %d\n", old, state);
 335         PASSERT(env, page, page->cp_state == old);
 336         PASSERT(env, page, equi(state == CPS_OWNED, page->cp_owner != NULL));
 337
 338         CS_PAGESTATE_DEC(page->cp_obj, page->cp_state);
 339         CS_PAGESTATE_INC(page->cp_obj, state);
 340         cl_page_state_set_trust(page, state);
 341         EXIT;
 342 }
 343
 344 static void cl_page_state_set(const struct lu_env *env,
 345                               struct cl_page *page, enum cl_page_state state)
 346 {
 347         cl_page_state_set0(env, page, state);
 348 }
 349
 350 /**
 351  * Acquires an additional reference to a page.
 352  *
 353  * This can be called only by caller already possessing a reference to \a
 354  * page.
 355  *
 356  * \see cl_object_get(), cl_lock_get().
 357  */
 358 void cl_page_get(struct cl_page *page)
 359 {
 360         ENTRY;
 361         cl_page_get_trust(page);
 362         EXIT;
 363 }
 364 EXPORT_SYMBOL(cl_page_get);
 365
 366 /**
 367  * Releases a reference to a page.
 368  *
 369  * When last reference is released, page is returned to the cache, unless it
 370  * is in cl_page_state::CPS_FREEING state, in which case it is immediately
 371  * destroyed.
 372  *
 373  * \see cl_object_put(), cl_lock_put().
 374  */
 375 void cl_page_put(const struct lu_env *env, struct cl_page *page)
 376 {
 377         ENTRY;
 378         CL_PAGE_HEADER(D_TRACE, env, page, "%d\n",
 379                        atomic_read(&page->cp_ref));
 380
 381         if (atomic_dec_and_test(&page->cp_ref)) {
 382                 LASSERT(page->cp_state == CPS_FREEING);
 383
 384                 LASSERT(atomic_read(&page->cp_ref) == 0);
 385                 PASSERT(env, page, page->cp_owner == NULL);
 386                 PASSERT(env, page, list_empty(&page->cp_batch));
 387                 /*
 388                  * Page is no longer reachable by other threads. Tear
 389                  * it down.
 390                  */
 391                 cl_page_free(env, page);
 392         }
 393
 394         EXIT;
 395 }
 396 EXPORT_SYMBOL(cl_page_put);
 397
 398 /**
 399  * Returns a cl_page associated with a VM page, and given cl_object.
 400  */
 401 struct cl_page *cl_vmpage_page(struct page *vmpage, struct cl_object *obj)
 402 {
 403         struct cl_page *page;
 404
 405         ENTRY;
 406         KLASSERT(PageLocked(vmpage));
 407
 408         /*
 409          * NOTE: absence of races and liveness of data are guaranteed by page
 410          *       lock on a "vmpage". That works because object destruction has
 411          *       bottom-to-top pass.
 412          */
 413
 414         page = (struct cl_page *)vmpage->private;
 415         if (page != NULL) {
 416                 cl_page_get_trust(page);
 417                 LASSERT(page->cp_type == CPT_CACHEABLE);
 418         }
 419         RETURN(page);
 420 }
 421 EXPORT_SYMBOL(cl_vmpage_page);
 422
 423 const struct cl_page_slice *cl_page_at(const struct cl_page *page,
 424                                        const struct lu_device_type *dtype)
 425 {
 426         return cl_page_at_trusted(page, dtype);
 427 }
 428 EXPORT_SYMBOL(cl_page_at);
 429
 430 #define CL_PAGE_OP(opname) offsetof(struct cl_page_operations, opname)
 431
 432 #define CL_PAGE_INVOKE(_env, _page, _op, _proto, ...)                   \
 433 ({                                                                      \
 434         const struct lu_env        *__env  = (_env);                    \
 435         struct cl_page             *__page = (_page);                   \
 436         const struct cl_page_slice *__scan;                             \
 437         int                         __result;                           \
 438         ptrdiff_t                   __op   = (_op);                     \
 439         int                        (*__method)_proto;                   \
 440                                                                         \
 441         __result = 0;                                                   \
 442         list_for_each_entry(__scan, &__page->cp_layers, cpl_linkage) {  \
 443                 __method = *(void **)((char *)__scan->cpl_ops +  __op); \
 444                 if (__method != NULL) {                                 \
 445                         __result = (*__method)(__env, __scan, ## __VA_ARGS__); \
 446                         if (__result != 0)                              \
 447                                 break;                                  \
 448                 }                                                       \
 449         }                                                               \
 450         if (__result > 0)                                               \
 451                 __result = 0;                                           \
 452         __result;                                                       \
 453 })
 454
 455 #define CL_PAGE_INVOKE_REVERSE(_env, _page, _op, _proto, ...)           \
 456 ({                                                                      \
 457         const struct lu_env        *__env  = (_env);                    \
 458         struct cl_page             *__page = (_page);                   \
 459         const struct cl_page_slice *__scan;                             \
 460         int                         __result;                           \
 461         ptrdiff_t                   __op   = (_op);                     \
 462         int                       (*__method)_proto;                    \
 463                                                                         \
 464         __result = 0;                                                   \
 465         list_for_each_entry_reverse(__scan, &__page->cp_layers,         \
 466                                     cpl_linkage) {                      \
 467                 __method = *(void **)((char *)__scan->cpl_ops +  __op); \
 468                 if (__method != NULL) {                                 \
 469                         __result = (*__method)(__env, __scan, ## __VA_ARGS__); \
 470                         if (__result != 0)                              \
 471                                 break;                                  \
 472                 }                                                       \
 473         }                                                               \
 474         if (__result > 0)                                               \
 475                 __result = 0;                                           \
 476         __result;                                                       \
 477 })
 478
 479 #define CL_PAGE_INVOID(_env, _page, _op, _proto, ...)                   \
 480 do {                                                                    \
 481         const struct lu_env        *__env  = (_env);                    \
 482         struct cl_page             *__page = (_page);                   \
 483         const struct cl_page_slice *__scan;                             \
 484         ptrdiff_t                   __op   = (_op);                     \
 485         void                      (*__method)_proto;                    \
 486                                                                         \
 487         list_for_each_entry(__scan, &__page->cp_layers, cpl_linkage) {  \
 488                 __method = *(void **)((char *)__scan->cpl_ops +  __op); \
 489                 if (__method != NULL)                                   \
 490                         (*__method)(__env, __scan, ## __VA_ARGS__);     \
 491         }                                                               \
 492 } while (0)
 493
 494 #define CL_PAGE_INVOID_REVERSE(_env, _page, _op, _proto, ...)           \
 495 do {                                                                    \
 496         const struct lu_env        *__env  = (_env);                    \
 497         struct cl_page             *__page = (_page);                   \
 498         const struct cl_page_slice *__scan;                             \
 499         ptrdiff_t                   __op   = (_op);                     \
 500         void                      (*__method)_proto;                    \
 501                                                                         \
 502         /* get to the bottom page. */                                   \
 503         list_for_each_entry_reverse(__scan, &__page->cp_layers,         \
 504                                     cpl_linkage) {                      \
 505                 __method = *(void **)((char *)__scan->cpl_ops + __op);  \
 506                 if (__method != NULL)                                   \
 507                         (*__method)(__env, __scan, ## __VA_ARGS__);     \
 508         }                                                               \
 509 } while (0)
 510
 511 static int cl_page_invoke(const struct lu_env *env,
 512                           struct cl_io *io, struct cl_page *page, ptrdiff_t op)
 513
 514 {
 515         PINVRNT(env, page, cl_object_same(page->cp_obj, io->ci_obj));
 516         ENTRY;
 517         RETURN(CL_PAGE_INVOKE(env, page, op,
 518                               (const struct lu_env *,
 519                                const struct cl_page_slice *, struct cl_io *),
 520                               io));
 521 }
 522
 523 static void cl_page_invoid(const struct lu_env *env,
 524                            struct cl_io *io, struct cl_page *page, ptrdiff_t op)
 525
 526 {
 527         PINVRNT(env, page, cl_object_same(page->cp_obj, io->ci_obj));
 528         ENTRY;
 529         CL_PAGE_INVOID(env, page, op,
 530                        (const struct lu_env *,
 531                         const struct cl_page_slice *, struct cl_io *), io);
 532         EXIT;
 533 }
 534
 535 static void cl_page_owner_clear(struct cl_page *page)
 536 {
 537         ENTRY;
 538         if (page->cp_owner != NULL) {
 539                 LASSERT(page->cp_owner->ci_owned_nr > 0);
 540                 page->cp_owner->ci_owned_nr--;
 541                 page->cp_owner = NULL;
 542         }
 543         EXIT;
 544 }
 545
 546 static void cl_page_owner_set(struct cl_page *page)
 547 {
 548         ENTRY;
 549         LASSERT(page->cp_owner != NULL);
 550         page->cp_owner->ci_owned_nr++;
 551         EXIT;
 552 }
 553
 554 void cl_page_disown0(const struct lu_env *env,
 555                      struct cl_io *io, struct cl_page *pg)
 556 {
 557         enum cl_page_state state;
 558
 559         ENTRY;
 560         state = pg->cp_state;
 561         PINVRNT(env, pg, state == CPS_OWNED || state == CPS_FREEING);
 562         PINVRNT(env, pg, cl_page_invariant(pg) || state == CPS_FREEING);
 563         cl_page_owner_clear(pg);
 564
 565         if (state == CPS_OWNED)
 566                 cl_page_state_set(env, pg, CPS_CACHED);
 567         /*
 568          * Completion call-backs are executed in the bottom-up order, so that
 569          * uppermost layer (llite), responsible for VFS/VM interaction runs
 570          * last and can release locks safely.
 571          */
 572         CL_PAGE_INVOID_REVERSE(env, pg, CL_PAGE_OP(cpo_disown),
 573                                (const struct lu_env *,
 574                                 const struct cl_page_slice *, struct cl_io *),
 575                                io);
 576         EXIT;
 577 }
 578
 579 /**
 580  * returns true, iff page is owned by the given io.
 581  */
 582 int cl_page_is_owned(const struct cl_page *pg, const struct cl_io *io)
 583 {
 584         struct cl_io *top = cl_io_top((struct cl_io *)io);
 585         LINVRNT(cl_object_same(pg->cp_obj, io->ci_obj));
 586         ENTRY;
 587         RETURN(pg->cp_state == CPS_OWNED && pg->cp_owner == top);
 588 }
 589 EXPORT_SYMBOL(cl_page_is_owned);
 590
 591 /**
 592  * Try to own a page by IO.
 593  *
 594  * Waits until page is in cl_page_state::CPS_CACHED state, and then switch it
 595  * into cl_page_state::CPS_OWNED state.
 596  *
 597  * \pre  !cl_page_is_owned(pg, io)
 598  * \post result == 0 iff cl_page_is_owned(pg, io)
 599  *
 600  * \retval 0   success
 601  *
 602  * \retval -ve failure, e.g., page was destroyed (and landed in
 603  *             cl_page_state::CPS_FREEING instead of cl_page_state::CPS_CACHED).
 604  *             or, page was owned by another thread, or in IO.
 605  *
 606  * \see cl_page_disown()
 607  * \see cl_page_operations::cpo_own()
 608  * \see cl_page_own_try()
 609  * \see cl_page_own
 610  */
 611 static int cl_page_own0(const struct lu_env *env, struct cl_io *io,
 612                         struct cl_page *pg, int nonblock)
 613 {
 614         int result;
 615
 616         PINVRNT(env, pg, !cl_page_is_owned(pg, io));
 617
 618         ENTRY;
 619         io = cl_io_top(io);
 620
 621         if (pg->cp_state == CPS_FREEING) {
 622                 result = -ENOENT;
 623         } else {
 624                 result = CL_PAGE_INVOKE(env, pg, CL_PAGE_OP(cpo_own),
 625                                         (const struct lu_env *,
 626                                          const struct cl_page_slice *,
 627                                          struct cl_io *, int),
 628                                         io, nonblock);
 629                 if (result == 0) {
 630                         PASSERT(env, pg, pg->cp_owner == NULL);
 631                         PASSERT(env, pg, pg->cp_req == NULL);
 632                         pg->cp_owner = cl_io_top(io);;
 633                         cl_page_owner_set(pg);
 634                         if (pg->cp_state != CPS_FREEING) {
 635                                 cl_page_state_set(env, pg, CPS_OWNED);
 636                         } else {
 637                                 cl_page_disown0(env, io, pg);
 638                                 result = -ENOENT;
 639                         }
 640                 }
 641         }
 642         PINVRNT(env, pg, ergo(result == 0, cl_page_invariant(pg)));
 643         RETURN(result);
 644 }
 645
 646 /**
 647  * Own a page, might be blocked.
 648  *
 649  * \see cl_page_own0()
 650  */
 651 int cl_page_own(const struct lu_env *env, struct cl_io *io, struct cl_page *pg)
 652 {
 653         return cl_page_own0(env, io, pg, 0);
 654 }
 655 EXPORT_SYMBOL(cl_page_own);
 656
 657 /**
 658  * Nonblock version of cl_page_own().
 659  *
 660  * \see cl_page_own0()
 661  */
 662 int cl_page_own_try(const struct lu_env *env, struct cl_io *io,
 663                     struct cl_page *pg)
 664 {
 665         return cl_page_own0(env, io, pg, 1);
 666 }
 667 EXPORT_SYMBOL(cl_page_own_try);
 668
 669
 670 /**
 671  * Assume page ownership.
 672  *
 673  * Called when page is already locked by the hosting VM.
 674  *
 675  * \pre !cl_page_is_owned(pg, io)
 676  * \post cl_page_is_owned(pg, io)
 677  *
 678  * \see cl_page_operations::cpo_assume()
 679  */
 680 void cl_page_assume(const struct lu_env *env,
 681                     struct cl_io *io, struct cl_page *pg)
 682 {
 683         PINVRNT(env, pg, cl_object_same(pg->cp_obj, io->ci_obj));
 684
 685         ENTRY;
 686         io = cl_io_top(io);
 687
 688         cl_page_invoid(env, io, pg, CL_PAGE_OP(cpo_assume));
 689         PASSERT(env, pg, pg->cp_owner == NULL);
 690         pg->cp_owner = cl_io_top(io);
 691         cl_page_owner_set(pg);
 692         cl_page_state_set(env, pg, CPS_OWNED);
 693         EXIT;
 694 }
 695 EXPORT_SYMBOL(cl_page_assume);
 696
 697 /**
 698  * Releases page ownership without unlocking the page.
 699  *
 700  * Moves page into cl_page_state::CPS_CACHED without releasing a lock on the
 701  * underlying VM page (as VM is supposed to do this itself).
 702  *
 703  * \pre   cl_page_is_owned(pg, io)
 704  * \post !cl_page_is_owned(pg, io)
 705  *
 706  * \see cl_page_assume()
 707  */
 708 void cl_page_unassume(const struct lu_env *env,
 709                       struct cl_io *io, struct cl_page *pg)
 710 {
 711         PINVRNT(env, pg, cl_page_is_owned(pg, io));
 712         PINVRNT(env, pg, cl_page_invariant(pg));
 713
 714         ENTRY;
 715         io = cl_io_top(io);
 716         cl_page_owner_clear(pg);
 717         cl_page_state_set(env, pg, CPS_CACHED);
 718         CL_PAGE_INVOID_REVERSE(env, pg, CL_PAGE_OP(cpo_unassume),
 719                                (const struct lu_env *,
 720                                 const struct cl_page_slice *, struct cl_io *),
 721                                io);
 722         EXIT;
 723 }
 724 EXPORT_SYMBOL(cl_page_unassume);
 725
 726 /**
 727  * Releases page ownership.
 728  *
 729  * Moves page into cl_page_state::CPS_CACHED.
 730  *
 731  * \pre   cl_page_is_owned(pg, io)
 732  * \post !cl_page_is_owned(pg, io)
 733  *
 734  * \see cl_page_own()
 735  * \see cl_page_operations::cpo_disown()
 736  */
 737 void cl_page_disown(const struct lu_env *env,
 738                     struct cl_io *io, struct cl_page *pg)
 739 {
 740         PINVRNT(env, pg, cl_page_is_owned(pg, io) ||
 741                 pg->cp_state == CPS_FREEING);
 742
 743         ENTRY;
 744         io = cl_io_top(io);
 745         cl_page_disown0(env, io, pg);
 746         EXIT;
 747 }
 748 EXPORT_SYMBOL(cl_page_disown);
 749
 750 /**
 751  * Called when page is to be removed from the object, e.g., as a result of
 752  * truncate.
 753  *
 754  * Calls cl_page_operations::cpo_discard() top-to-bottom.
 755  *
 756  * \pre cl_page_is_owned(pg, io)
 757  *
 758  * \see cl_page_operations::cpo_discard()
 759  */
 760 void cl_page_discard(const struct lu_env *env,
 761                      struct cl_io *io, struct cl_page *pg)
 762 {
 763         PINVRNT(env, pg, cl_page_is_owned(pg, io));
 764         PINVRNT(env, pg, cl_page_invariant(pg));
 765
 766         cl_page_invoid(env, io, pg, CL_PAGE_OP(cpo_discard));
 767 }
 768 EXPORT_SYMBOL(cl_page_discard);
 769
 770 /**
 771  * Version of cl_page_delete() that can be called for not fully constructed
 772  * pages, e.g,. in a error handling cl_page_find()->cl_page_delete0()
 773  * path. Doesn't check page invariant.
 774  */
 775 static void cl_page_delete0(const struct lu_env *env, struct cl_page *pg)
 776 {
 777         ENTRY;
 778
 779         PASSERT(env, pg, pg->cp_state != CPS_FREEING);
 780
 781         /*
 782          * Severe all ways to obtain new pointers to @pg.
 783          */
 784         cl_page_owner_clear(pg);
 785
 786         cl_page_state_set0(env, pg, CPS_FREEING);
 787
 788         CL_PAGE_INVOID_REVERSE(env, pg, CL_PAGE_OP(cpo_delete),
 789                        (const struct lu_env *, const struct cl_page_slice *));
 790
 791         EXIT;
 792 }
 793
 794 /**
 795  * Called when a decision is made to throw page out of memory.
 796  *
 797  * Notifies all layers about page destruction by calling
 798  * cl_page_operations::cpo_delete() method top-to-bottom.
 799  *
 800  * Moves page into cl_page_state::CPS_FREEING state (this is the only place
 801  * where transition to this state happens).
 802  *
 803  * Eliminates all venues through which new references to the page can be
 804  * obtained:
 805  *
 806  *     - removes page from the radix trees,
 807  *
 808  *     - breaks linkage from VM page to cl_page.
 809  *
 810  * Once page reaches cl_page_state::CPS_FREEING, all remaining references will
 811  * drain after some time, at which point page will be recycled.
 812  *
 813  * \pre  VM page is locked
 814  * \post pg->cp_state == CPS_FREEING
 815  *
 816  * \see cl_page_operations::cpo_delete()
 817  */
 818 void cl_page_delete(const struct lu_env *env, struct cl_page *pg)
 819 {
 820         PINVRNT(env, pg, cl_page_invariant(pg));
 821         ENTRY;
 822         cl_page_delete0(env, pg);
 823         EXIT;
 824 }
 825 EXPORT_SYMBOL(cl_page_delete);
 826
 827 /**
 828  * Marks page up-to-date.
 829  *
 830  * Call cl_page_operations::cpo_export() through all layers top-to-bottom. The
 831  * layer responsible for VM interaction has to mark/clear page as up-to-date
 832  * by the \a uptodate argument.
 833  *
 834  * \see cl_page_operations::cpo_export()
 835  */
 836 void cl_page_export(const struct lu_env *env, struct cl_page *pg, int uptodate)
 837 {
 838         PINVRNT(env, pg, cl_page_invariant(pg));
 839         CL_PAGE_INVOID(env, pg, CL_PAGE_OP(cpo_export),
 840                        (const struct lu_env *,
 841                         const struct cl_page_slice *, int), uptodate);
 842 }
 843 EXPORT_SYMBOL(cl_page_export);
 844
 845 /**
 846  * Returns true, iff \a pg is VM locked in a suitable sense by the calling
 847  * thread.
 848  */
 849 int cl_page_is_vmlocked(const struct lu_env *env, const struct cl_page *pg)
 850 {
 851         int result;
 852         const struct cl_page_slice *slice;
 853
 854         ENTRY;
 855         slice = container_of(pg->cp_layers.next,
 856                              const struct cl_page_slice, cpl_linkage);
 857         PASSERT(env, pg, slice->cpl_ops->cpo_is_vmlocked != NULL);
 858         /*
 859          * Call ->cpo_is_vmlocked() directly instead of going through
 860          * CL_PAGE_INVOKE(), because cl_page_is_vmlocked() is used by
 861          * cl_page_invariant().
 862          */
 863         result = slice->cpl_ops->cpo_is_vmlocked(env, slice);
 864         PASSERT(env, pg, result == -EBUSY || result == -ENODATA);
 865         RETURN(result == -EBUSY);
 866 }
 867 EXPORT_SYMBOL(cl_page_is_vmlocked);
 868
 869 static enum cl_page_state cl_req_type_state(enum cl_req_type crt)
 870 {
 871         ENTRY;
 872         RETURN(crt == CRT_WRITE ? CPS_PAGEOUT : CPS_PAGEIN);
 873 }
 874
 875 static void cl_page_io_start(const struct lu_env *env,
 876                              struct cl_page *pg, enum cl_req_type crt)
 877 {
 878         /*
 879          * Page is queued for IO, change its state.
 880          */
 881         ENTRY;
 882         cl_page_owner_clear(pg);
 883         cl_page_state_set(env, pg, cl_req_type_state(crt));
 884         EXIT;
 885 }
 886
 887 /**
 888  * Prepares page for immediate transfer. cl_page_operations::cpo_prep() is
 889  * called top-to-bottom. Every layer either agrees to submit this page (by
 890  * returning 0), or requests to omit this page (by returning -EALREADY). Layer
 891  * handling interactions with the VM also has to inform VM that page is under
 892  * transfer now.
 893  */
 894 int cl_page_prep(const struct lu_env *env, struct cl_io *io,
 895                  struct cl_page *pg, enum cl_req_type crt)
 896 {
 897         int result;
 898
 899         PINVRNT(env, pg, cl_page_is_owned(pg, io));
 900         PINVRNT(env, pg, cl_page_invariant(pg));
 901         PINVRNT(env, pg, crt < CRT_NR);
 902
 903         /*
 904          * XXX this has to be called bottom-to-top, so that llite can set up
 905          * PG_writeback without risking other layers deciding to skip this
 906          * page.
 907          */
 908         if (crt >= CRT_NR)
 909                 return -EINVAL;
 910         result = cl_page_invoke(env, io, pg, CL_PAGE_OP(io[crt].cpo_prep));
 911         if (result == 0)
 912                 cl_page_io_start(env, pg, crt);
 913
 914         KLASSERT(ergo(crt == CRT_WRITE && pg->cp_type == CPT_CACHEABLE,
 915                       equi(result == 0,
 916                            PageWriteback(cl_page_vmpage(pg)))));
 917         CL_PAGE_HEADER(D_TRACE, env, pg, "%d %d\n", crt, result);
 918         return result;
 919 }
 920 EXPORT_SYMBOL(cl_page_prep);
 921
 922 /**
 923  * Notify layers about transfer completion.
 924  *
 925  * Invoked by transfer sub-system (which is a part of osc) to notify layers
 926  * that a transfer, of which this page is a part of has completed.
 927  *
 928  * Completion call-backs are executed in the bottom-up order, so that
 929  * uppermost layer (llite), responsible for the VFS/VM interaction runs last
 930  * and can release locks safely.
 931  *
 932  * \pre  pg->cp_state == CPS_PAGEIN || pg->cp_state == CPS_PAGEOUT
 933  * \post pg->cp_state == CPS_CACHED
 934  *
 935  * \see cl_page_operations::cpo_completion()
 936  */
 937 void cl_page_completion(const struct lu_env *env,
 938                         struct cl_page *pg, enum cl_req_type crt, int ioret)
 939 {
 940         struct cl_sync_io *anchor = pg->cp_sync_io;
 941
 942         PASSERT(env, pg, crt < CRT_NR);
 943         /* cl_page::cp_req already cleared by the caller (osc_completion()) */
 944         PASSERT(env, pg, pg->cp_req == NULL);
 945         PASSERT(env, pg, pg->cp_state == cl_req_type_state(crt));
 946
 947         ENTRY;
 948         CL_PAGE_HEADER(D_TRACE, env, pg, "%d %d\n", crt, ioret);
 949         cl_page_state_set(env, pg, CPS_CACHED);
 950         if (crt >= CRT_NR)
 951                 return;
 952         CL_PAGE_INVOID_REVERSE(env, pg, CL_PAGE_OP(io[crt].cpo_completion),
 953                                (const struct lu_env *,
 954                                 const struct cl_page_slice *, int), ioret);
 955         if (anchor) {
 956                 LASSERT(pg->cp_sync_io == anchor);
 957                 pg->cp_sync_io = NULL;
 958         }
 959         /*
 960          * As page->cp_obj is pinned by a reference from page->cp_req, it is
 961          * safe to call cl_page_put() without risking object destruction in a
 962          * non-blocking context.
 963          */
 964         cl_page_put(env, pg);
 965
 966         if (anchor)
 967                 cl_sync_io_note(anchor, ioret);
 968
 969         EXIT;
 970 }
 971 EXPORT_SYMBOL(cl_page_completion);
 972
 973 /**
 974  * Notify layers that transfer formation engine decided to yank this page from
 975  * the cache and to make it a part of a transfer.
 976  *
 977  * \pre  pg->cp_state == CPS_CACHED
 978  * \post pg->cp_state == CPS_PAGEIN || pg->cp_state == CPS_PAGEOUT
 979  *
 980  * \see cl_page_operations::cpo_make_ready()
 981  */
 982 int cl_page_make_ready(const struct lu_env *env, struct cl_page *pg,
 983                        enum cl_req_type crt)
 984 {
 985         int result;
 986
 987         PINVRNT(env, pg, crt < CRT_NR);
 988
 989         ENTRY;
 990         if (crt >= CRT_NR)
 991                 RETURN(-EINVAL);
 992         result = CL_PAGE_INVOKE(env, pg, CL_PAGE_OP(io[crt].cpo_make_ready),
 993                                 (const struct lu_env *,
 994                                  const struct cl_page_slice *));
 995         if (result == 0) {
 996                 PASSERT(env, pg, pg->cp_state == CPS_CACHED);
 997                 cl_page_io_start(env, pg, crt);
 998         }
 999         CL_PAGE_HEADER(D_TRACE, env, pg, "%d %d\n", crt, result);
1000         RETURN(result);
1001 }
1002 EXPORT_SYMBOL(cl_page_make_ready);
1003
1004 /**
1005  * Called if a pge is being written back by kernel's intention.
1006  *
1007  * \pre  cl_page_is_owned(pg, io)
1008  * \post ergo(result == 0, pg->cp_state == CPS_PAGEOUT)
1009  *
1010  * \see cl_page_operations::cpo_flush()
1011  */
1012 int cl_page_flush(const struct lu_env *env, struct cl_io *io,
1013                   struct cl_page *pg)
1014 {
1015         int result;
1016
1017         PINVRNT(env, pg, cl_page_is_owned(pg, io));
1018         PINVRNT(env, pg, cl_page_invariant(pg));
1019
1020         ENTRY;
1021
1022         result = cl_page_invoke(env, io, pg, CL_PAGE_OP(cpo_flush));
1023
1024         CL_PAGE_HEADER(D_TRACE, env, pg, "%d\n", result);
1025         RETURN(result);
1026 }
1027 EXPORT_SYMBOL(cl_page_flush);
1028
1029 /**
1030  * Checks whether page is protected by any extent lock is at least required
1031  * mode.
1032  *
1033  * \return the same as in cl_page_operations::cpo_is_under_lock() method.
1034  * \see cl_page_operations::cpo_is_under_lock()
1035  */
1036 int cl_page_is_under_lock(const struct lu_env *env, struct cl_io *io,
1037                           struct cl_page *page, pgoff_t *max_index)
1038 {
1039         int rc;
1040
1041         PINVRNT(env, page, cl_page_invariant(page));
1042
1043         ENTRY;
1044         rc = CL_PAGE_INVOKE_REVERSE(env, page, CL_PAGE_OP(cpo_is_under_lock),
1045                                     (const struct lu_env *,
1046                                      const struct cl_page_slice *,
1047                                      struct cl_io *, pgoff_t *),
1048                                     io, max_index);
1049         RETURN(rc);
1050 }
1051 EXPORT_SYMBOL(cl_page_is_under_lock);
1052
1053 /**
1054  * Tells transfer engine that only part of a page is to be transmitted.
1055  *
1056  * \see cl_page_operations::cpo_clip()
1057  */
1058 void cl_page_clip(const struct lu_env *env, struct cl_page *pg,
1059                   int from, int to)
1060 {
1061         PINVRNT(env, pg, cl_page_invariant(pg));
1062
1063         CL_PAGE_HEADER(D_TRACE, env, pg, "%d %d\n", from, to);
1064         CL_PAGE_INVOID(env, pg, CL_PAGE_OP(cpo_clip),
1065                        (const struct lu_env *,
1066                         const struct cl_page_slice *,int, int),
1067                        from, to);
1068 }
1069 EXPORT_SYMBOL(cl_page_clip);
1070
1071 /**
1072  * Prints human readable representation of \a pg to the \a f.
1073  */
1074 void cl_page_header_print(const struct lu_env *env, void *cookie,
1075                           lu_printer_t printer, const struct cl_page *pg)
1076 {
1077         (*printer)(env, cookie,
1078                    "page@%p[%d %p %d %d %d %p %p]\n",
1079                    pg, atomic_read(&pg->cp_ref), pg->cp_obj,
1080                    pg->cp_state, pg->cp_error, pg->cp_type,
1081                    pg->cp_owner, pg->cp_req);
1082 }
1083 EXPORT_SYMBOL(cl_page_header_print);
1084
1085 /**
1086  * Prints human readable representation of \a pg to the \a f.
1087  */
1088 void cl_page_print(const struct lu_env *env, void *cookie,
1089                    lu_printer_t printer, const struct cl_page *pg)
1090 {
1091         cl_page_header_print(env, cookie, printer, pg);
1092         CL_PAGE_INVOKE(env, (struct cl_page *)pg, CL_PAGE_OP(cpo_print),
1093                        (const struct lu_env *env,
1094                         const struct cl_page_slice *slice,
1095                         void *cookie, lu_printer_t p), cookie, printer);
1096         (*printer)(env, cookie, "end page@%p\n", pg);
1097 }
1098 EXPORT_SYMBOL(cl_page_print);
1099
1100 /**
1101  * Cancel a page which is still in a transfer.
1102  */
1103 int cl_page_cancel(const struct lu_env *env, struct cl_page *page)
1104 {
1105         return CL_PAGE_INVOKE(env, page, CL_PAGE_OP(cpo_cancel),
1106                               (const struct lu_env *,
1107                                const struct cl_page_slice *));
1108 }
1109 EXPORT_SYMBOL(cl_page_cancel);
1110
1111 /**
1112  * Converts a byte offset within object \a obj into a page index.
1113  */
1114 loff_t cl_offset(const struct cl_object *obj, pgoff_t idx)
1115 {
1116         return (loff_t)idx << PAGE_CACHE_SHIFT;
1117 }
1118 EXPORT_SYMBOL(cl_offset);
1119
1120 /**
1121  * Converts a page index into a byte offset within object \a obj.
1122  */
1123 pgoff_t cl_index(const struct cl_object *obj, loff_t offset)
1124 {
1125         return offset >> PAGE_CACHE_SHIFT;
1126 }
1127 EXPORT_SYMBOL(cl_index);
1128
1129 int cl_page_size(const struct cl_object *obj)
1130 {
1131         return 1 << PAGE_CACHE_SHIFT;
1132 }
1133 EXPORT_SYMBOL(cl_page_size);
1134
1135 /**
1136  * Adds page slice to the compound page.
1137  *
1138  * This is called by cl_object_operations::coo_page_init() methods to add a
1139  * per-layer state to the page. New state is added at the end of
1140  * cl_page::cp_layers list, that is, it is at the bottom of the stack.
1141  *
1142  * \see cl_lock_slice_add(), cl_req_slice_add(), cl_io_slice_add()
1143  */
1144 void cl_page_slice_add(struct cl_page *page, struct cl_page_slice *slice,
1145                        struct cl_object *obj, pgoff_t index,
1146                        const struct cl_page_operations *ops)
1147 {
1148         ENTRY;
1149         list_add_tail(&slice->cpl_linkage, &page->cp_layers);
1150         slice->cpl_obj  = obj;
1151         slice->cpl_index = index;
1152         slice->cpl_ops  = ops;
1153         slice->cpl_page = page;
1154         EXIT;
1155 }
1156 EXPORT_SYMBOL(cl_page_slice_add);
1157
1158 int  cl_page_init(void)
1159 {
1160         return 0;
1161 }
1162
1163 void cl_page_fini(void)
1164 {
1165 }