lustre/obdclass/cl_object.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 2008 Sun Microsystems, Inc.  All rights reserved.
  30  * Use is subject to license terms.
  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 Object.
  37  *
  38  *   Author: Nikita Danilov <nikita.danilov@sun.com>
  39  */
  40
  41 /*
  42  * Locking.
  43  *
  44  *  i_mutex
  45  *      PG_locked
  46  *          ->coh_page_guard
  47  *          ->coh_lock_guard
  48  *          ->coh_attr_guard
  49  *          ->ls_guard
  50  */
  51
  52 #define DEBUG_SUBSYSTEM S_CLASS
  53 #ifndef EXPORT_SYMTAB
  54 # define EXPORT_SYMTAB
  55 #endif
  56
  57 #include <libcfs/libcfs.h>
  58 /* class_put_type() */
  59 #include <obd_class.h>
  60 #include <obd_support.h>
  61 #include <lustre_fid.h>
  62 #include <libcfs/list.h>
  63 /* lu_time_global_{init,fini}() */
  64 #include <lu_time.h>
  65
  66 #include <cl_object.h>
  67 #include "cl_internal.h"
  68
  69 static cfs_mem_cache_t *cl_env_kmem;
  70
  71 /** Lock class of cl_object_header::coh_page_guard */
  72 static struct lock_class_key cl_page_guard_class;
  73 /** Lock class of cl_object_header::coh_lock_guard */
  74 static struct lock_class_key cl_lock_guard_class;
  75 /** Lock class of cl_object_header::coh_attr_guard */
  76 static struct lock_class_key cl_attr_guard_class;
  77
  78 /**
  79  * Initialize cl_object_header.
  80  */
  81 int cl_object_header_init(struct cl_object_header *h)
  82 {
  83         int result;
  84
  85         ENTRY;
  86         result = lu_object_header_init(&h->coh_lu);
  87         if (result == 0) {
  88                 spin_lock_init(&h->coh_page_guard);
  89                 spin_lock_init(&h->coh_lock_guard);
  90                 spin_lock_init(&h->coh_attr_guard);
  91                 lockdep_set_class(&h->coh_attr_guard, &cl_page_guard_class);
  92                 lockdep_set_class(&h->coh_attr_guard, &cl_lock_guard_class);
  93                 lockdep_set_class(&h->coh_attr_guard, &cl_attr_guard_class);
  94                 h->coh_pages = 0;
  95                 /* XXX hard coded GFP_* mask. */
  96                 INIT_RADIX_TREE(&h->coh_tree, GFP_ATOMIC);
  97                 CFS_INIT_LIST_HEAD(&h->coh_locks);
  98         }
  99         RETURN(result);
 100 }
 101 EXPORT_SYMBOL(cl_object_header_init);
 102
 103 /**
 104  * Finalize cl_object_header.
 105  */
 106 void cl_object_header_fini(struct cl_object_header *h)
 107 {
 108         LASSERT(list_empty(&h->coh_locks));
 109         lu_object_header_fini(&h->coh_lu);
 110 }
 111 EXPORT_SYMBOL(cl_object_header_fini);
 112
 113 /**
 114  * Returns a cl_object with a given \a fid.
 115  *
 116  * Returns either cached or newly created object. Additional reference on the
 117  * returned object is acquired.
 118  *
 119  * \see lu_object_find(), cl_page_find(), cl_lock_find()
 120  */
 121 struct cl_object *cl_object_find(const struct lu_env *env,
 122                                  struct cl_device *cd, const struct lu_fid *fid,
 123                                  const struct cl_object_conf *c)
 124 {
 125         might_sleep();
 126         return lu2cl(lu_object_find_slice(env, cl2lu_dev(cd), fid, &c->coc_lu));
 127 }
 128 EXPORT_SYMBOL(cl_object_find);
 129
 130 /**
 131  * Releases a reference on \a o.
 132  *
 133  * When last reference is released object is returned to the cache, unless
 134  * lu_object_header_flags::LU_OBJECT_HEARD_BANSHEE bit is set in its header.
 135  *
 136  * \see cl_page_put(), cl_lock_put().
 137  */
 138 void cl_object_put(const struct lu_env *env, struct cl_object *o)
 139 {
 140         lu_object_put(env, &o->co_lu);
 141 }
 142 EXPORT_SYMBOL(cl_object_put);
 143
 144 /**
 145  * Acquire an additional reference to the object \a o.
 146  *
 147  * This can only be used to acquire _additional_ reference, i.e., caller
 148  * already has to possess at least one reference to \a o before calling this.
 149  *
 150  * \see cl_page_get(), cl_lock_get().
 151  */
 152 void cl_object_get(struct cl_object *o)
 153 {
 154         lu_object_get(&o->co_lu);
 155 }
 156 EXPORT_SYMBOL(cl_object_get);
 157
 158 /**
 159  * Returns the top-object for a given \a o.
 160  *
 161  * \see cl_page_top(), cl_io_top()
 162  */
 163 struct cl_object *cl_object_top(struct cl_object *o)
 164 {
 165         struct cl_object_header *hdr = cl_object_header(o);
 166         struct cl_object *top;
 167
 168         while (hdr->coh_parent != NULL)
 169                 hdr = hdr->coh_parent;
 170
 171         top = lu2cl(lu_object_top(&hdr->coh_lu));
 172         CDEBUG(D_TRACE, "%p -> %p\n", o, top);
 173         return top;
 174 }
 175 EXPORT_SYMBOL(cl_object_top);
 176
 177 /**
 178  * Returns pointer to the lock protecting data-attributes for the given object
 179  * \a o.
 180  *
 181  * Data-attributes are protected by the cl_object_header::coh_attr_guard
 182  * spin-lock in the top-object.
 183  *
 184  * \see cl_attr, cl_object_attr_lock(), cl_object_operations::coo_attr_get().
 185  */
 186 static spinlock_t *cl_object_attr_guard(struct cl_object *o)
 187 {
 188         return &cl_object_header(cl_object_top(o))->coh_attr_guard;
 189 }
 190
 191 /**
 192  * Locks data-attributes.
 193  *
 194  * Prevents data-attributes from changing, until lock is released by
 195  * cl_object_attr_unlock(). This has to be called before calls to
 196  * cl_object_attr_get(), cl_object_attr_set().
 197  */
 198 void cl_object_attr_lock(struct cl_object *o)
 199 {
 200         spin_lock(cl_object_attr_guard(o));
 201 }
 202 EXPORT_SYMBOL(cl_object_attr_lock);
 203
 204 /**
 205  * Releases data-attributes lock, acquired by cl_object_attr_lock().
 206  */
 207 void cl_object_attr_unlock(struct cl_object *o)
 208 {
 209         spin_unlock(cl_object_attr_guard(o));
 210 }
 211 EXPORT_SYMBOL(cl_object_attr_unlock);
 212
 213 /**
 214  * Returns data-attributes of an object \a obj.
 215  *
 216  * Every layer is asked (by calling cl_object_operations::coo_attr_get())
 217  * top-to-bottom to fill in parts of \a attr that this layer is responsible
 218  * for.
 219  */
 220 int cl_object_attr_get(const struct lu_env *env, struct cl_object *obj,
 221                        struct cl_attr *attr)
 222 {
 223         struct lu_object_header *top;
 224         int result;
 225
 226         LASSERT_SPIN_LOCKED(cl_object_attr_guard(obj));
 227         ENTRY;
 228
 229         top = obj->co_lu.lo_header;
 230         result = 0;
 231         list_for_each_entry(obj, &top->loh_layers, co_lu.lo_linkage) {
 232                 if (obj->co_ops->coo_attr_get != NULL) {
 233                         result = obj->co_ops->coo_attr_get(env, obj, attr);
 234                         if (result != 0) {
 235                                 if (result > 0)
 236                                         result = 0;
 237                                 break;
 238                         }
 239                 }
 240         }
 241         RETURN(result);
 242 }
 243 EXPORT_SYMBOL(cl_object_attr_get);
 244
 245 /**
 246  * Updates data-attributes of an object \a obj.
 247  *
 248  * Only attributes, mentioned in a validness bit-mask \a v are
 249  * updated. Calls cl_object_operations::coo_attr_set() on every layer, bottom
 250  * to top.
 251  */
 252 int cl_object_attr_set(const struct lu_env *env, struct cl_object *obj,
 253                        const struct cl_attr *attr, unsigned v)
 254 {
 255         struct lu_object_header *top;
 256         int result;
 257
 258         LASSERT_SPIN_LOCKED(cl_object_attr_guard(obj));
 259         ENTRY;
 260
 261         top = obj->co_lu.lo_header;
 262         result = 0;
 263         list_for_each_entry_reverse(obj, &top->loh_layers, co_lu.lo_linkage) {
 264                 if (obj->co_ops->coo_attr_set != NULL) {
 265                         result = obj->co_ops->coo_attr_set(env, obj, attr, v);
 266                         if (result != 0) {
 267                                 if (result > 0)
 268                                         result = 0;
 269                                 break;
 270                         }
 271                 }
 272         }
 273         RETURN(result);
 274 }
 275 EXPORT_SYMBOL(cl_object_attr_set);
 276
 277 /**
 278  * Notifies layers (bottom-to-top) that glimpse AST was received.
 279  *
 280  * Layers have to fill \a lvb fields with information that will be shipped
 281  * back to glimpse issuer.
 282  *
 283  * \see cl_lock_operations::clo_glimpse()
 284  */
 285 int cl_object_glimpse(const struct lu_env *env, struct cl_object *obj,
 286                       struct ost_lvb *lvb)
 287 {
 288         struct lu_object_header *top;
 289         int result;
 290
 291         ENTRY;
 292         top = obj->co_lu.lo_header;
 293         result = 0;
 294         list_for_each_entry_reverse(obj, &top->loh_layers, co_lu.lo_linkage) {
 295                 if (obj->co_ops->coo_glimpse != NULL) {
 296                         result = obj->co_ops->coo_glimpse(env, obj, lvb);
 297                         if (result != 0)
 298                                 break;
 299                 }
 300         }
 301         LU_OBJECT_HEADER(D_DLMTRACE, env, lu_object_top(top),
 302                          "size: "LPU64" mtime: "LPU64" atime: "LPU64" "
 303                          "ctime: "LPU64" blocks: "LPU64"\n",
 304                          lvb->lvb_size, lvb->lvb_mtime, lvb->lvb_atime,
 305                          lvb->lvb_ctime, lvb->lvb_blocks);
 306         RETURN(result);
 307 }
 308 EXPORT_SYMBOL(cl_object_glimpse);
 309
 310 /**
 311  * Updates a configuration of an object \a obj.
 312  */
 313 int cl_conf_set(const struct lu_env *env, struct cl_object *obj,
 314                 const struct cl_object_conf *conf)
 315 {
 316         struct lu_object_header *top;
 317         int result;
 318
 319         ENTRY;
 320         top = obj->co_lu.lo_header;
 321         result = 0;
 322         list_for_each_entry(obj, &top->loh_layers, co_lu.lo_linkage) {
 323                 if (obj->co_ops->coo_conf_set != NULL) {
 324                         result = obj->co_ops->coo_conf_set(env, obj, conf);
 325                         if (result != 0)
 326                                 break;
 327                 }
 328         }
 329         RETURN(result);
 330 }
 331 EXPORT_SYMBOL(cl_conf_set);
 332
 333 /**
 334  * Helper function removing all object locks, and marking object for
 335  * deletion. All object pages must have been deleted at this point.
 336  *
 337  * This is called by cl_inode_fini() and lov_object_delete() to destroy top-
 338  * and sub- objects respectively.
 339  */
 340 void cl_object_kill(const struct lu_env *env, struct cl_object *obj)
 341 {
 342         struct cl_object_header *hdr;
 343
 344         hdr = cl_object_header(obj);
 345         LASSERT(hdr->coh_tree.rnode == NULL);
 346         LASSERT(hdr->coh_pages == 0);
 347
 348         set_bit(LU_OBJECT_HEARD_BANSHEE, &hdr->coh_lu.loh_flags);
 349         /*
 350          * Destroy all locks. Object destruction (including cl_inode_fini())
 351          * cannot cancel the locks, because in the case of a local client,
 352          * where client and server share the same thread running
 353          * prune_icache(), this can dead-lock with ldlm_cancel_handler()
 354          * waiting on __wait_on_freeing_inode().
 355          */
 356         cl_locks_prune(env, obj, 0);
 357 }
 358 EXPORT_SYMBOL(cl_object_kill);
 359
 360 /**
 361  * Prunes caches of pages and locks for this object.
 362  */
 363 void cl_object_prune(const struct lu_env *env, struct cl_object *obj)
 364 {
 365         ENTRY;
 366         cl_pages_prune(env, obj);
 367         cl_locks_prune(env, obj, 1);
 368         EXIT;
 369 }
 370 EXPORT_SYMBOL(cl_object_prune);
 371
 372 void cache_stats_init(struct cache_stats *cs, const char *name)
 373 {
 374         cs->cs_name = name;
 375         atomic_set(&cs->cs_lookup, 0);
 376         atomic_set(&cs->cs_hit,    0);
 377         atomic_set(&cs->cs_total,  0);
 378         atomic_set(&cs->cs_busy,   0);
 379 }
 380
 381 int cache_stats_print(const struct cache_stats *cs,
 382                       char *page, int count, int h)
 383 {
 384         int nob = 0;
 385 /*
 386        lookup    hit  total cached create
 387   env: ...... ...... ...... ...... ......
 388 */
 389         if (h)
 390                 nob += snprintf(page, count,
 391                                 "       lookup    hit  total   busy create\n");
 392
 393         nob += snprintf(page + nob, count - nob,
 394                         "%5.5s: %6u %6u %6u %6u %6u",
 395                         cs->cs_name,
 396                         atomic_read(&cs->cs_lookup),
 397                         atomic_read(&cs->cs_hit),
 398                         atomic_read(&cs->cs_total),
 399                         atomic_read(&cs->cs_busy),
 400                         atomic_read(&cs->cs_created));
 401         return nob;
 402 }
 403
 404 /**
 405  * Initialize client site.
 406  *
 407  * Perform common initialization (lu_site_init()), and initialize statistical
 408  * counters. Also perform global initializations on the first call.
 409  */
 410 int cl_site_init(struct cl_site *s, struct cl_device *d)
 411 {
 412         int i;
 413         int result;
 414
 415         result = lu_site_init(&s->cs_lu, &d->cd_lu_dev);
 416         if (result == 0) {
 417                 cache_stats_init(&s->cs_pages, "pages");
 418                 cache_stats_init(&s->cs_locks, "locks");
 419                 for (i = 0; i < ARRAY_SIZE(s->cs_pages_state); ++i)
 420                         atomic_set(&s->cs_pages_state[0], 0);
 421                 for (i = 0; i < ARRAY_SIZE(s->cs_locks_state); ++i)
 422                         atomic_set(&s->cs_locks_state[i], 0);
 423         }
 424         return result;
 425 }
 426 EXPORT_SYMBOL(cl_site_init);
 427
 428 /**
 429  * Finalize client site. Dual to cl_site_init().
 430  */
 431 void cl_site_fini(struct cl_site *s)
 432 {
 433         lu_site_fini(&s->cs_lu);
 434 }
 435 EXPORT_SYMBOL(cl_site_fini);
 436
 437 static struct cache_stats cl_env_stats = {
 438         .cs_name    = "envs",
 439         .cs_created = ATOMIC_INIT(0),
 440         .cs_lookup  = ATOMIC_INIT(0),
 441         .cs_hit     = ATOMIC_INIT(0),
 442         .cs_total   = ATOMIC_INIT(0),
 443         .cs_busy    = ATOMIC_INIT(0)
 444 };
 445
 446 /**
 447  * Outputs client site statistical counters into a buffer. Suitable for
 448  * ll_rd_*()-style functions.
 449  */
 450 int cl_site_stats_print(const struct cl_site *site, char *page, int count)
 451 {
 452         int nob;
 453         int i;
 454         static const char *pstate[] = {
 455                 [CPS_CACHED]  = "c",
 456                 [CPS_OWNED]   = "o",
 457                 [CPS_PAGEOUT] = "w",
 458                 [CPS_PAGEIN]  = "r",
 459                 [CPS_FREEING] = "f"
 460         };
 461         static const char *lstate[] = {
 462                 [CLS_NEW]       = "n",
 463                 [CLS_QUEUING]   = "q",
 464                 [CLS_ENQUEUED]  = "e",
 465                 [CLS_HELD]      = "h",
 466                 [CLS_UNLOCKING] = "u",
 467                 [CLS_CACHED]    = "c",
 468                 [CLS_FREEING]   = "f"
 469         };
 470 /*
 471        lookup    hit  total   busy create
 472 pages: ...... ...... ...... ...... ...... [...... ...... ...... ......]
 473 locks: ...... ...... ...... ...... ...... [...... ...... ...... ...... ......]
 474   env: ...... ...... ...... ...... ......
 475  */
 476         nob = lu_site_stats_print(&site->cs_lu, page, count);
 477         nob += cache_stats_print(&site->cs_pages, page + nob, count - nob, 1);
 478         nob += snprintf(page + nob, count - nob, " [");
 479         for (i = 0; i < ARRAY_SIZE(site->cs_pages_state); ++i)
 480                 nob += snprintf(page + nob, count - nob, "%s: %u ",
 481                                 pstate[i],
 482                                 atomic_read(&site->cs_pages_state[i]));
 483         nob += snprintf(page + nob, count - nob, "]\n");
 484         nob += cache_stats_print(&site->cs_locks, page + nob, count - nob, 0);
 485         nob += snprintf(page + nob, count - nob, " [");
 486         for (i = 0; i < ARRAY_SIZE(site->cs_locks_state); ++i)
 487                 nob += snprintf(page + nob, count - nob, "%s: %u ",
 488                                 lstate[i],
 489                                 atomic_read(&site->cs_locks_state[i]));
 490         nob += snprintf(page + nob, count - nob, "]\n");
 491         nob += cache_stats_print(&cl_env_stats, page + nob, count - nob, 0);
 492         nob += snprintf(page + nob, count - nob, "\n");
 493         return nob;
 494 }
 495 EXPORT_SYMBOL(cl_site_stats_print);
 496
 497 /*****************************************************************************
 498  *
 499  * lu_env handling on client.
 500  *
 501  */
 502
 503 /*
 504  * TBD: Description.
 505  *
 506  * XXX: this assumes that re-entrant file system calls (e.g., ->writepage())
 507  * do not modify already existing current->journal_info.
 508  */
 509
 510 static CFS_LIST_HEAD(cl_envs);
 511 static unsigned cl_envs_cached_nr  = 0;
 512 static unsigned cl_envs_cached_max = 128; /* XXX: prototype: arbitrary limit
 513                                            * for now. */
 514 static spinlock_t cl_envs_guard = SPIN_LOCK_UNLOCKED;
 515
 516 struct cl_env {
 517         void             *ce_magic;
 518         struct lu_env     ce_lu;
 519         struct lu_context ce_ses;
 520         /*
 521          * Linkage into global list of all client environments. Used for
 522          * garbage collection.
 523          */
 524         struct list_head  ce_linkage;
 525         /*
 526          *
 527          */
 528         int               ce_ref;
 529         void             *ce_prev;
 530         /*
 531          * Debugging field: address of the caller who made original
 532          * allocation.
 533          */
 534         void             *ce_debug;
 535         void             *ce_owner;
 536 };
 537
 538 #define CL_ENV_INC(counter) atomic_inc(&cl_env_stats.counter)
 539
 540 #define CL_ENV_DEC(counter)                                             \
 541         do {                                                            \
 542                 LASSERT(atomic_read(&cl_env_stats.counter) > 0);        \
 543                 atomic_dec(&cl_env_stats.counter);                      \
 544         } while (0)
 545
 546 static void cl_env_init0(struct cl_env *cle, void *debug)
 547 {
 548         LASSERT(cle->ce_ref == 0);
 549         LASSERT(cle->ce_magic == &cl_env_init0);
 550         LASSERT(cle->ce_debug == NULL && cle->ce_owner == NULL);
 551
 552         cle->ce_ref = 1;
 553         cle->ce_prev = current->journal_info;
 554         cle->ce_debug = debug;
 555         cle->ce_owner = current;
 556         current->journal_info = cle;
 557         CL_ENV_INC(cs_busy);
 558 }
 559
 560 static struct lu_env *cl_env_new(__u32 tags, void *debug)
 561 {
 562         struct lu_env *env;
 563         struct cl_env *cle;
 564
 565         OBD_SLAB_ALLOC_PTR_GFP(cle, cl_env_kmem, CFS_ALLOC_IO);
 566         if (cle != NULL) {
 567                 int rc;
 568
 569                 CFS_INIT_LIST_HEAD(&cle->ce_linkage);
 570                 cle->ce_magic = &cl_env_init0;
 571                 env = &cle->ce_lu;
 572                 rc = lu_env_init(env, LCT_CL_THREAD|tags);
 573                 if (rc == 0) {
 574                         rc = lu_context_init(&cle->ce_ses, LCT_SESSION|tags);
 575                         if (rc == 0) {
 576                                 lu_context_enter(&cle->ce_ses);
 577                                 env->le_ses = &cle->ce_ses;
 578                                 cl_env_init0(cle, debug);
 579                         } else
 580                                 lu_env_fini(env);
 581                 }
 582                 if (rc != 0) {
 583                         OBD_SLAB_FREE_PTR(cle, cl_env_kmem);
 584                         env = ERR_PTR(rc);
 585                 } else {
 586                         CL_ENV_INC(cs_created);
 587                         CL_ENV_INC(cs_total);
 588                 }
 589         } else
 590                 env = ERR_PTR(-ENOMEM);
 591         return env;
 592 }
 593
 594 static void cl_env_fini(struct cl_env *cle)
 595 {
 596         CL_ENV_DEC(cs_total);
 597         lu_context_fini(&cle->ce_lu.le_ctx);
 598         lu_context_fini(&cle->ce_ses);
 599         OBD_SLAB_FREE_PTR(cle, cl_env_kmem);
 600 }
 601
 602 static struct lu_env *cl_env_obtain(void *debug)
 603 {
 604         struct cl_env *cle;
 605         struct lu_env *env;
 606
 607         ENTRY;
 608         spin_lock(&cl_envs_guard);
 609         LASSERT(equi(cl_envs_cached_nr == 0, list_empty(&cl_envs)));
 610         if (cl_envs_cached_nr > 0) {
 611                 int rc;
 612
 613                 cle = container_of(cl_envs.next, struct cl_env, ce_linkage);
 614                 list_del_init(&cle->ce_linkage);
 615                 cl_envs_cached_nr--;
 616                 spin_unlock(&cl_envs_guard);
 617
 618                 env = &cle->ce_lu;
 619                 rc = lu_env_refill(env);
 620                 if (rc == 0) {
 621                         cl_env_init0(cle, debug);
 622                         lu_context_enter(&env->le_ctx);
 623                         lu_context_enter(&cle->ce_ses);
 624                 } else {
 625                         cl_env_fini(cle);
 626                         env = ERR_PTR(rc);
 627                 }
 628         } else {
 629                 spin_unlock(&cl_envs_guard);
 630                 env = cl_env_new(0, debug);
 631         }
 632         RETURN(env);
 633 }
 634
 635 static inline struct cl_env *cl_env_container(struct lu_env *env)
 636 {
 637         return container_of(env, struct cl_env, ce_lu);
 638 }
 639
 640 struct lu_env *cl_env_peek(int *refcheck)
 641 {
 642         struct lu_env *env;
 643         struct cl_env *cle;
 644
 645         CL_ENV_INC(cs_lookup);
 646
 647         /* check that we don't go far from untrusted pointer */
 648         CLASSERT(offsetof(struct cl_env, ce_magic) == 0);
 649
 650         env = NULL;
 651         cle = current->journal_info;
 652         if (cle != NULL && cle->ce_magic == &cl_env_init0) {
 653                 CL_ENV_INC(cs_hit);
 654                 env = &cle->ce_lu;
 655                 *refcheck = ++cle->ce_ref;
 656         }
 657         CDEBUG(D_OTHER, "%i@%p\n", cle ? cle->ce_ref : 0, cle);
 658         return env;
 659 }
 660 EXPORT_SYMBOL(cl_env_peek);
 661
 662 /**
 663  * Returns lu_env: if there already is an environment associated with the
 664  * current thread, it is returned, otherwise, new environment is allocated.
 665  *
 666  * Allocations are amortized through the global cache of environments.
 667  *
 668  * \param refcheck pointer to a counter used to detect environment leaks. In
 669  * the usual case cl_env_get() and cl_env_put() are called in the same lexical
 670  * scope and pointer to the same integer is passed as \a refcheck. This is
 671  * used to detect missed cl_env_put().
 672  *
 673  * \see cl_env_put()
 674  */
 675 struct lu_env *cl_env_get(int *refcheck)
 676 {
 677         struct lu_env *env;
 678
 679         env = cl_env_peek(refcheck);
 680         if (env == NULL) {
 681                 env = cl_env_obtain(__builtin_return_address(0));
 682                 if (!IS_ERR(env)) {
 683                         struct cl_env *cle;
 684
 685                         cle = cl_env_container(env);
 686                         *refcheck = cle->ce_ref;
 687                         CDEBUG(D_OTHER, "%i@%p\n", cle->ce_ref, cle);
 688                 }
 689         }
 690         return env;
 691 }
 692 EXPORT_SYMBOL(cl_env_get);
 693
 694 /**
 695  * Forces an allocation of a fresh environment with given tags.
 696  *
 697  * \see cl_env_get()
 698  */
 699 struct lu_env *cl_env_alloc(int *refcheck, __u32 tags)
 700 {
 701         struct lu_env *env;
 702
 703         LASSERT(cl_env_peek(refcheck) == NULL);
 704         env = cl_env_new(tags, __builtin_return_address(0));
 705         if (!IS_ERR(env)) {
 706                 struct cl_env *cle;
 707
 708                 cle = cl_env_container(env);
 709                 *refcheck = cle->ce_ref;
 710                 CDEBUG(D_OTHER, "%i@%p\n", cle->ce_ref, cle);
 711         }
 712         return env;
 713 }
 714 EXPORT_SYMBOL(cl_env_alloc);
 715
 716 static void cl_env_exit(struct cl_env *cle)
 717 {
 718         lu_context_exit(&cle->ce_lu.le_ctx);
 719         lu_context_exit(&cle->ce_ses);
 720 }
 721
 722 /**
 723  * Finalizes and frees a given number of cached environments. This is done to
 724  * (1) free some memory (not currently hooked into VM), or (2) release
 725  * references to modules.
 726  */
 727 unsigned cl_env_cache_purge(unsigned nr)
 728 {
 729         struct cl_env *cle;
 730
 731         ENTRY;
 732         spin_lock(&cl_envs_guard);
 733         for (; !list_empty(&cl_envs) && nr > 0; --nr) {
 734                 cle = container_of(cl_envs.next, struct cl_env, ce_linkage);
 735                 list_del_init(&cle->ce_linkage);
 736                 LASSERT(cl_envs_cached_nr > 0);
 737                 cl_envs_cached_nr--;
 738                 spin_unlock(&cl_envs_guard);
 739
 740                 cl_env_fini(cle);
 741                 spin_lock(&cl_envs_guard);
 742         }
 743         LASSERT(equi(cl_envs_cached_nr == 0, list_empty(&cl_envs)));
 744         spin_unlock(&cl_envs_guard);
 745         RETURN(nr);
 746 }
 747 EXPORT_SYMBOL(cl_env_cache_purge);
 748
 749 /**
 750  * Release an environment.
 751  *
 752  * Decrement \a env reference counter. When counter drops to 0, nothing in
 753  * this thread is using environment and it is returned to the allocation
 754  * cache, or freed straight away, if cache is large enough.
 755  */
 756 void cl_env_put(struct lu_env *env, int *refcheck)
 757 {
 758         struct cl_env *cle;
 759
 760         cle = cl_env_container(env);
 761
 762         LASSERT(cle->ce_ref > 0);
 763         LASSERT(ergo(refcheck != NULL, cle->ce_ref == *refcheck));
 764
 765         CDEBUG(D_OTHER, "%i@%p\n", cle->ce_ref, cle);
 766         if (--cle->ce_ref == 0) {
 767                 CL_ENV_DEC(cs_busy);
 768                 current->journal_info = cle->ce_prev;
 769                 LASSERT(cle->ce_prev == NULL ||
 770                         cl_env_container(cle->ce_prev)->ce_magic !=
 771                         &cl_env_init0);
 772                 cle->ce_debug = NULL;
 773                 cle->ce_owner = NULL;
 774                 cl_env_exit(cle);
 775                 /*
 776                  * Don't bother to take a lock here.
 777                  *
 778                  * Return environment to the cache only when it was allocated
 779                  * with the standard tags.
 780                  */
 781                 if (cl_envs_cached_nr < cl_envs_cached_max &&
 782                     (env->le_ctx.lc_tags & ~LCT_HAS_EXIT) == LCT_CL_THREAD &&
 783                     (env->le_ses->lc_tags & ~LCT_HAS_EXIT) == LCT_SESSION) {
 784                         spin_lock(&cl_envs_guard);
 785                         list_add(&cle->ce_linkage, &cl_envs);
 786                         cl_envs_cached_nr++;
 787                         spin_unlock(&cl_envs_guard);
 788                 } else
 789                         cl_env_fini(cle);
 790         }
 791 }
 792 EXPORT_SYMBOL(cl_env_put);
 793
 794 /**
 795  * Declares a point of re-entrancy.
 796  *
 797  * In Linux kernel environments are attached to the thread through
 798  * current->journal_info pointer that is used by other sub-systems also. When
 799  * lustre code is invoked in the situation where current->journal_info is
 800  * potentially already set, cl_env_reenter() is called to save
 801  * current->journal_info value, so that current->journal_info field can be
 802  * used to store pointer to the environment.
 803  *
 804  * \see cl_env_reexit()
 805  */
 806 void *cl_env_reenter(void)
 807 {
 808         void *cookie;
 809
 810         cookie = current->journal_info;
 811         current->journal_info = NULL;
 812         CDEBUG(D_OTHER, "cookie: %p\n", cookie);
 813         return cookie;
 814 }
 815 EXPORT_SYMBOL(cl_env_reenter);
 816
 817 /**
 818  * Exits re-entrancy.
 819  *
 820  * This restores old value of current->journal_info that was saved by
 821  * cl_env_reenter().
 822  */
 823 void cl_env_reexit(void *cookie)
 824 {
 825         current->journal_info = cookie;
 826         CDEBUG(D_OTHER, "cookie: %p\n", cookie);
 827 }
 828 EXPORT_SYMBOL(cl_env_reexit);
 829
 830 /**
 831  * Setup user-supplied \a env as a current environment. This is to be used to
 832  * guaranteed that environment exists even when cl_env_get() fails. It is up
 833  * to user to ensure proper concurrency control.
 834  *
 835  * \see cl_env_unplant()
 836  */
 837 void cl_env_implant(struct lu_env *env, int *refcheck)
 838 {
 839         struct cl_env *cle = cl_env_container(env);
 840
 841         LASSERT(current->journal_info == NULL);
 842         LASSERT(cle->ce_ref > 0);
 843
 844         current->journal_info = cle;
 845         cl_env_get(refcheck);
 846         CDEBUG(D_OTHER, "%i@%p\n", cle->ce_ref, cle);
 847 }
 848 EXPORT_SYMBOL(cl_env_implant);
 849
 850 /**
 851  * Detach environment installed earlier by cl_env_implant().
 852  */
 853 void cl_env_unplant(struct lu_env *env, int *refcheck)
 854 {
 855         struct cl_env *cle = cl_env_container(env);
 856
 857         LASSERT(cle == current->journal_info);
 858         LASSERT(cle->ce_ref > 1);
 859
 860         CDEBUG(D_OTHER, "%i@%p\n", cle->ce_ref, cle);
 861
 862         cl_env_put(env, refcheck);
 863         current->journal_info = NULL;
 864 }
 865 EXPORT_SYMBOL(cl_env_unplant);
 866
 867 struct lu_env *cl_env_nested_get(struct cl_env_nest *nest)
 868 {
 869         struct lu_env *env;
 870
 871         nest->cen_cookie = NULL;
 872         env = cl_env_peek(&nest->cen_refcheck);
 873         if (env != NULL) {
 874                 if (!cl_io_is_going(env))
 875                         return env;
 876                 else {
 877                         cl_env_put(env, &nest->cen_refcheck);
 878                         nest->cen_cookie = cl_env_reenter();
 879                 }
 880         }
 881         env = cl_env_get(&nest->cen_refcheck);
 882         if (IS_ERR(env)) {
 883                 cl_env_reexit(nest->cen_cookie);
 884                 return env;
 885         }
 886
 887         LASSERT(!cl_io_is_going(env));
 888         return env;
 889 }
 890 EXPORT_SYMBOL(cl_env_nested_get);
 891
 892 void cl_env_nested_put(struct cl_env_nest *nest, struct lu_env *env)
 893 {
 894         cl_env_put(env, &nest->cen_refcheck);
 895         cl_env_reexit(nest->cen_cookie);
 896 }
 897 EXPORT_SYMBOL(cl_env_nested_put);
 898
 899 /**
 900  * Converts struct cl_attr to struct ost_lvb.
 901  *
 902  * \see cl_lvb2attr
 903  */
 904 void cl_attr2lvb(struct ost_lvb *lvb, const struct cl_attr *attr)
 905 {
 906         ENTRY;
 907         lvb->lvb_size   = attr->cat_size;
 908         lvb->lvb_mtime  = attr->cat_mtime;
 909         lvb->lvb_atime  = attr->cat_atime;
 910         lvb->lvb_ctime  = attr->cat_ctime;
 911         lvb->lvb_blocks = attr->cat_blocks;
 912         EXIT;
 913 }
 914 EXPORT_SYMBOL(cl_attr2lvb);
 915
 916 /**
 917  * Converts struct ost_lvb to struct cl_attr.
 918  *
 919  * \see cl_attr2lvb
 920  */
 921 void cl_lvb2attr(struct cl_attr *attr, const struct ost_lvb *lvb)
 922 {
 923         ENTRY;
 924         attr->cat_size   = lvb->lvb_size;
 925         attr->cat_mtime  = lvb->lvb_mtime;
 926         attr->cat_atime  = lvb->lvb_atime;
 927         attr->cat_ctime  = lvb->lvb_ctime;
 928         attr->cat_blocks = lvb->lvb_blocks;
 929         EXIT;
 930 }
 931 EXPORT_SYMBOL(cl_lvb2attr);
 932
 933
 934 /*****************************************************************************
 935  *
 936  * Temporary prototype thing: mirror obd-devices into cl devices.
 937  *
 938  */
 939
 940 struct cl_device *cl_type_setup(const struct lu_env *env, struct lu_site *site,
 941                                 struct lu_device_type *ldt,
 942                                 struct lu_device *next)
 943 {
 944         const char       *typename;
 945         struct lu_device *d;
 946
 947         LASSERT(ldt != NULL);
 948
 949         typename = ldt->ldt_name;
 950         d = ldt->ldt_ops->ldto_device_alloc(env, ldt, NULL);
 951         if (!IS_ERR(d)) {
 952                 int rc;
 953
 954                 if (site != NULL)
 955                         d->ld_site = site;
 956                 rc = ldt->ldt_ops->ldto_device_init(env, d, typename, next);
 957                 if (rc == 0) {
 958                         lu_device_get(d);
 959                         lu_ref_add(&d->ld_reference,
 960                                    "lu-stack", &lu_site_init);
 961                 } else {
 962                         ldt->ldt_ops->ldto_device_free(env, d);
 963                         CERROR("can't init device '%s', %d\n", typename, rc);
 964                         d = ERR_PTR(rc);
 965                 }
 966         } else
 967                 CERROR("Cannot allocate device: '%s'\n", typename);
 968         return lu2cl_dev(d);
 969 }
 970 EXPORT_SYMBOL(cl_type_setup);
 971
 972 /**
 973  * Finalize device stack by calling lu_stack_fini().
 974  */
 975 void cl_stack_fini(const struct lu_env *env, struct cl_device *cl)
 976 {
 977         lu_stack_fini(env, cl2lu_dev(cl));
 978 }
 979 EXPORT_SYMBOL(cl_stack_fini);
 980
 981 int  cl_lock_init(void);
 982 void cl_lock_fini(void);
 983
 984 int  cl_page_init(void);
 985 void cl_page_fini(void);
 986
 987 static struct lu_context_key cl_key;
 988
 989 struct cl_thread_info *cl_env_info(const struct lu_env *env)
 990 {
 991         return lu_context_key_get(&env->le_ctx, &cl_key);
 992 }
 993
 994 /* defines cl0_key_{init,fini}() */
 995 LU_KEY_INIT_FINI(cl0, struct cl_thread_info);
 996
 997 static void *cl_key_init(const struct lu_context *ctx,
 998                          struct lu_context_key *key)
 999 {
1000         struct cl_thread_info *info;
1001
1002         info = cl0_key_init(ctx, key);
1003         if (!IS_ERR(info)) {
1004                 int i;
1005
1006                 for (i = 0; i < ARRAY_SIZE(info->clt_counters); ++i)
1007                         lu_ref_init(&info->clt_counters[i].ctc_locks_locked);
1008         }
1009         return info;
1010 }
1011
1012 static void cl_key_fini(const struct lu_context *ctx,
1013                         struct lu_context_key *key, void *data)
1014 {
1015         struct cl_thread_info *info;
1016         int i;
1017
1018         info = data;
1019         for (i = 0; i < ARRAY_SIZE(info->clt_counters); ++i)
1020                 lu_ref_fini(&info->clt_counters[i].ctc_locks_locked);
1021         cl0_key_fini(ctx, key, data);
1022 }
1023
1024 static void cl_key_exit(const struct lu_context *ctx,
1025                         struct lu_context_key *key, void *data)
1026 {
1027         struct cl_thread_info *info = data;
1028         int i;
1029
1030         for (i = 0; i < ARRAY_SIZE(info->clt_counters); ++i) {
1031                 LASSERT(info->clt_counters[i].ctc_nr_held == 0);
1032                 LASSERT(info->clt_counters[i].ctc_nr_used == 0);
1033                 LASSERT(info->clt_counters[i].ctc_nr_locks_acquired == 0);
1034                 LASSERT(info->clt_counters[i].ctc_nr_locks_locked == 0);
1035                 lu_ref_fini(&info->clt_counters[i].ctc_locks_locked);
1036                 lu_ref_init(&info->clt_counters[i].ctc_locks_locked);
1037         }
1038 }
1039
1040 static struct lu_context_key cl_key = {
1041         .lct_tags = LCT_CL_THREAD,
1042         .lct_init = cl_key_init,
1043         .lct_fini = cl_key_fini,
1044         .lct_exit = cl_key_exit
1045 };
1046
1047 static struct lu_kmem_descr cl_object_caches[] = {
1048         {
1049                 .ckd_cache = &cl_env_kmem,
1050                 .ckd_name  = "cl_env_kmem",
1051                 .ckd_size  = sizeof (struct cl_env)
1052         },
1053         {
1054                 .ckd_cache = NULL
1055         }
1056 };
1057
1058 /**
1059  * Global initialization of cl-data. Create kmem caches, register
1060  * lu_context_key's, etc.
1061  *
1062  * \see cl_global_fini()
1063  */
1064 int cl_global_init(void)
1065 {
1066         int result;
1067
1068         result = lu_kmem_init(cl_object_caches);
1069         if (result == 0) {
1070                 LU_CONTEXT_KEY_INIT(&cl_key);
1071                 result = lu_context_key_register(&cl_key);
1072                 if (result == 0) {
1073                         result = cl_lock_init();
1074                         if (result == 0)
1075                                 result = cl_page_init();
1076                 }
1077         }
1078         return result;
1079 }
1080
1081 /**
1082  * Finalization of global cl-data. Dual to cl_global_init().
1083  */
1084 void cl_global_fini(void)
1085 {
1086         cl_lock_fini();
1087         cl_page_fini();
1088         lu_context_key_degister(&cl_key);
1089         lu_kmem_fini(cl_object_caches);
1090 }