drivers/staging/lustre/lustre/obdclass/cl_object.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.gnu.org/licenses/gpl-2.0.html
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2008, 2010, Oracle and/or its affiliates. All rights reserved.
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2011, 2015, Intel Corporation.
  27  */
  28 /*
  29  * This file is part of Lustre, http://www.lustre.org/
  30  * Lustre is a trademark of Sun Microsystems, Inc.
  31  *
  32  * Client Lustre Object.
  33  *
  34  *   Author: Nikita Danilov <nikita.danilov@sun.com>
  35  *   Author: Jinshan Xiong <jinshan.xiong@intel.com>
  36  */
  37
  38 /*
  39  * Locking.
  40  *
  41  *  i_mutex
  42  *      PG_locked
  43  *        ->coh_attr_guard
  44  *        ->ls_guard
  45  */
  46
  47 #define DEBUG_SUBSYSTEM S_CLASS
  48
  49 #include "../../include/linux/libcfs/libcfs.h"
  50 /* class_put_type() */
  51 #include "../include/obd_class.h"
  52 #include "../include/obd_support.h"
  53 #include "../include/lustre_fid.h"
  54 #include <linux/list.h>
  55 #include "../../include/linux/libcfs/libcfs_hash.h"     /* for cfs_hash stuff */
  56 #include "../include/cl_object.h"
  57 #include "cl_internal.h"
  58
  59 static struct kmem_cache *cl_env_kmem;
  60
  61 /** Lock class of cl_object_header::coh_attr_guard */
  62 static struct lock_class_key cl_attr_guard_class;
  63
  64 extern __u32 lu_context_tags_default;
  65 extern __u32 lu_session_tags_default;
  66 /**
  67  * Initialize cl_object_header.
  68  */
  69 int cl_object_header_init(struct cl_object_header *h)
  70 {
  71         int result;
  72
  73         result = lu_object_header_init(&h->coh_lu);
  74         if (result == 0) {
  75                 spin_lock_init(&h->coh_attr_guard);
  76                 lockdep_set_class(&h->coh_attr_guard, &cl_attr_guard_class);
  77                 h->coh_page_bufsize = 0;
  78         }
  79         return result;
  80 }
  81 EXPORT_SYMBOL(cl_object_header_init);
  82
  83 /**
  84  * Returns a cl_object with a given \a fid.
  85  *
  86  * Returns either cached or newly created object. Additional reference on the
  87  * returned object is acquired.
  88  *
  89  * \see lu_object_find(), cl_page_find(), cl_lock_find()
  90  */
  91 struct cl_object *cl_object_find(const struct lu_env *env,
  92                                  struct cl_device *cd, const struct lu_fid *fid,
  93                                  const struct cl_object_conf *c)
  94 {
  95         might_sleep();
  96         return lu2cl(lu_object_find_slice(env, cl2lu_dev(cd), fid, &c->coc_lu));
  97 }
  98 EXPORT_SYMBOL(cl_object_find);
  99
 100 /**
 101  * Releases a reference on \a o.
 102  *
 103  * When last reference is released object is returned to the cache, unless
 104  * lu_object_header_flags::LU_OBJECT_HEARD_BANSHEE bit is set in its header.
 105  *
 106  * \see cl_page_put(), cl_lock_put().
 107  */
 108 void cl_object_put(const struct lu_env *env, struct cl_object *o)
 109 {
 110         lu_object_put(env, &o->co_lu);
 111 }
 112 EXPORT_SYMBOL(cl_object_put);
 113
 114 /**
 115  * Acquire an additional reference to the object \a o.
 116  *
 117  * This can only be used to acquire _additional_ reference, i.e., caller
 118  * already has to possess at least one reference to \a o before calling this.
 119  *
 120  * \see cl_page_get(), cl_lock_get().
 121  */
 122 void cl_object_get(struct cl_object *o)
 123 {
 124         lu_object_get(&o->co_lu);
 125 }
 126 EXPORT_SYMBOL(cl_object_get);
 127
 128 /**
 129  * Returns the top-object for a given \a o.
 130  *
 131  * \see cl_io_top()
 132  */
 133 struct cl_object *cl_object_top(struct cl_object *o)
 134 {
 135         struct cl_object_header *hdr = cl_object_header(o);
 136         struct cl_object *top;
 137
 138         while (hdr->coh_parent)
 139                 hdr = hdr->coh_parent;
 140
 141         top = lu2cl(lu_object_top(&hdr->coh_lu));
 142         CDEBUG(D_TRACE, "%p -> %p\n", o, top);
 143         return top;
 144 }
 145 EXPORT_SYMBOL(cl_object_top);
 146
 147 /**
 148  * Returns pointer to the lock protecting data-attributes for the given object
 149  * \a o.
 150  *
 151  * Data-attributes are protected by the cl_object_header::coh_attr_guard
 152  * spin-lock in the top-object.
 153  *
 154  * \see cl_attr, cl_object_attr_lock(), cl_object_operations::coo_attr_get().
 155  */
 156 static spinlock_t *cl_object_attr_guard(struct cl_object *o)
 157 {
 158         return &cl_object_header(cl_object_top(o))->coh_attr_guard;
 159 }
 160
 161 /**
 162  * Locks data-attributes.
 163  *
 164  * Prevents data-attributes from changing, until lock is released by
 165  * cl_object_attr_unlock(). This has to be called before calls to
 166  * cl_object_attr_get(), cl_object_attr_set().
 167  */
 168 void cl_object_attr_lock(struct cl_object *o)
 169         __acquires(cl_object_attr_guard(o))
 170 {
 171         spin_lock(cl_object_attr_guard(o));
 172 }
 173 EXPORT_SYMBOL(cl_object_attr_lock);
 174
 175 /**
 176  * Releases data-attributes lock, acquired by cl_object_attr_lock().
 177  */
 178 void cl_object_attr_unlock(struct cl_object *o)
 179         __releases(cl_object_attr_guard(o))
 180 {
 181         spin_unlock(cl_object_attr_guard(o));
 182 }
 183 EXPORT_SYMBOL(cl_object_attr_unlock);
 184
 185 /**
 186  * Returns data-attributes of an object \a obj.
 187  *
 188  * Every layer is asked (by calling cl_object_operations::coo_attr_get())
 189  * top-to-bottom to fill in parts of \a attr that this layer is responsible
 190  * for.
 191  */
 192 int cl_object_attr_get(const struct lu_env *env, struct cl_object *obj,
 193                        struct cl_attr *attr)
 194 {
 195         struct lu_object_header *top;
 196         int result;
 197
 198         assert_spin_locked(cl_object_attr_guard(obj));
 199
 200         top = obj->co_lu.lo_header;
 201         result = 0;
 202         list_for_each_entry(obj, &top->loh_layers, co_lu.lo_linkage) {
 203                 if (obj->co_ops->coo_attr_get) {
 204                         result = obj->co_ops->coo_attr_get(env, obj, attr);
 205                         if (result != 0) {
 206                                 if (result > 0)
 207                                         result = 0;
 208                                 break;
 209                         }
 210                 }
 211         }
 212         return result;
 213 }
 214 EXPORT_SYMBOL(cl_object_attr_get);
 215
 216 /**
 217  * Updates data-attributes of an object \a obj.
 218  *
 219  * Only attributes, mentioned in a validness bit-mask \a v are
 220  * updated. Calls cl_object_operations::coo_attr_set() on every layer, bottom
 221  * to top.
 222  */
 223 int cl_object_attr_set(const struct lu_env *env, struct cl_object *obj,
 224                        const struct cl_attr *attr, unsigned v)
 225 {
 226         struct lu_object_header *top;
 227         int result;
 228
 229         assert_spin_locked(cl_object_attr_guard(obj));
 230
 231         top = obj->co_lu.lo_header;
 232         result = 0;
 233         list_for_each_entry_reverse(obj, &top->loh_layers, co_lu.lo_linkage) {
 234                 if (obj->co_ops->coo_attr_set) {
 235                         result = obj->co_ops->coo_attr_set(env, obj, attr, v);
 236                         if (result != 0) {
 237                                 if (result > 0)
 238                                         result = 0;
 239                                 break;
 240                         }
 241                 }
 242         }
 243         return result;
 244 }
 245 EXPORT_SYMBOL(cl_object_attr_set);
 246
 247 /**
 248  * Notifies layers (bottom-to-top) that glimpse AST was received.
 249  *
 250  * Layers have to fill \a lvb fields with information that will be shipped
 251  * back to glimpse issuer.
 252  *
 253  * \see cl_lock_operations::clo_glimpse()
 254  */
 255 int cl_object_glimpse(const struct lu_env *env, struct cl_object *obj,
 256                       struct ost_lvb *lvb)
 257 {
 258         struct lu_object_header *top;
 259         int result;
 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_glimpse) {
 265                         result = obj->co_ops->coo_glimpse(env, obj, lvb);
 266                         if (result != 0)
 267                                 break;
 268                 }
 269         }
 270         LU_OBJECT_HEADER(D_DLMTRACE, env, lu_object_top(top),
 271                          "size: %llu mtime: %llu atime: %llu ctime: %llu blocks: %llu\n",
 272                          lvb->lvb_size, lvb->lvb_mtime, lvb->lvb_atime,
 273                          lvb->lvb_ctime, lvb->lvb_blocks);
 274         return result;
 275 }
 276 EXPORT_SYMBOL(cl_object_glimpse);
 277
 278 /**
 279  * Updates a configuration of an object \a obj.
 280  */
 281 int cl_conf_set(const struct lu_env *env, struct cl_object *obj,
 282                 const struct cl_object_conf *conf)
 283 {
 284         struct lu_object_header *top;
 285         int result;
 286
 287         top = obj->co_lu.lo_header;
 288         result = 0;
 289         list_for_each_entry(obj, &top->loh_layers, co_lu.lo_linkage) {
 290                 if (obj->co_ops->coo_conf_set) {
 291                         result = obj->co_ops->coo_conf_set(env, obj, conf);
 292                         if (result != 0)
 293                                 break;
 294                 }
 295         }
 296         return result;
 297 }
 298 EXPORT_SYMBOL(cl_conf_set);
 299
 300 /**
 301  * Prunes caches of pages and locks for this object.
 302  */
 303 int cl_object_prune(const struct lu_env *env, struct cl_object *obj)
 304 {
 305         struct lu_object_header *top;
 306         struct cl_object *o;
 307         int result;
 308
 309         top = obj->co_lu.lo_header;
 310         result = 0;
 311         list_for_each_entry(o, &top->loh_layers, co_lu.lo_linkage) {
 312                 if (o->co_ops->coo_prune) {
 313                         result = o->co_ops->coo_prune(env, o);
 314                         if (result != 0)
 315                                 break;
 316                 }
 317         }
 318
 319         return result;
 320 }
 321 EXPORT_SYMBOL(cl_object_prune);
 322
 323 /**
 324  * Helper function removing all object locks, and marking object for
 325  * deletion. All object pages must have been deleted at this point.
 326  *
 327  * This is called by cl_inode_fini() and lov_object_delete() to destroy top-
 328  * and sub- objects respectively.
 329  */
 330 void cl_object_kill(const struct lu_env *env, struct cl_object *obj)
 331 {
 332         struct cl_object_header *hdr = cl_object_header(obj);
 333
 334         set_bit(LU_OBJECT_HEARD_BANSHEE, &hdr->coh_lu.loh_flags);
 335 }
 336 EXPORT_SYMBOL(cl_object_kill);
 337
 338 void cache_stats_init(struct cache_stats *cs, const char *name)
 339 {
 340         int i;
 341
 342         cs->cs_name = name;
 343         for (i = 0; i < CS_NR; i++)
 344                 atomic_set(&cs->cs_stats[i], 0);
 345 }
 346
 347 static int cache_stats_print(const struct cache_stats *cs,
 348                              struct seq_file *m, int h)
 349 {
 350         int i;
 351         /*
 352          *   lookup    hit    total  cached create
 353          * env: ...... ...... ...... ...... ......
 354          */
 355         if (h) {
 356                 const char *names[CS_NR] = CS_NAMES;
 357
 358                 seq_printf(m, "%6s", " ");
 359                 for (i = 0; i < CS_NR; i++)
 360                         seq_printf(m, "%8s", names[i]);
 361                 seq_printf(m, "\n");
 362         }
 363
 364         seq_printf(m, "%5.5s:", cs->cs_name);
 365         for (i = 0; i < CS_NR; i++)
 366                 seq_printf(m, "%8u", atomic_read(&cs->cs_stats[i]));
 367         return 0;
 368 }
 369
 370 static void cl_env_percpu_refill(void);
 371
 372 /**
 373  * Initialize client site.
 374  *
 375  * Perform common initialization (lu_site_init()), and initialize statistical
 376  * counters. Also perform global initializations on the first call.
 377  */
 378 int cl_site_init(struct cl_site *s, struct cl_device *d)
 379 {
 380         int i;
 381         int result;
 382
 383         result = lu_site_init(&s->cs_lu, &d->cd_lu_dev);
 384         if (result == 0) {
 385                 cache_stats_init(&s->cs_pages, "pages");
 386                 for (i = 0; i < ARRAY_SIZE(s->cs_pages_state); ++i)
 387                         atomic_set(&s->cs_pages_state[0], 0);
 388                 cl_env_percpu_refill();
 389         }
 390         return result;
 391 }
 392 EXPORT_SYMBOL(cl_site_init);
 393
 394 /**
 395  * Finalize client site. Dual to cl_site_init().
 396  */
 397 void cl_site_fini(struct cl_site *s)
 398 {
 399         lu_site_fini(&s->cs_lu);
 400 }
 401 EXPORT_SYMBOL(cl_site_fini);
 402
 403 static struct cache_stats cl_env_stats = {
 404         .cs_name    = "envs",
 405         .cs_stats = { ATOMIC_INIT(0), }
 406 };
 407
 408 /**
 409  * Outputs client site statistical counters into a buffer. Suitable for
 410  * ll_rd_*()-style functions.
 411  */
 412 int cl_site_stats_print(const struct cl_site *site, struct seq_file *m)
 413 {
 414         int i;
 415         static const char *pstate[] = {
 416                 [CPS_CACHED]  = "c",
 417                 [CPS_OWNED]   = "o",
 418                 [CPS_PAGEOUT] = "w",
 419                 [CPS_PAGEIN]  = "r",
 420                 [CPS_FREEING] = "f"
 421         };
 422 /*
 423        lookup    hit  total   busy create
 424 pages: ...... ...... ...... ...... ...... [...... ...... ...... ......]
 425 locks: ...... ...... ...... ...... ...... [...... ...... ...... ...... ......]
 426   env: ...... ...... ...... ...... ......
 427  */
 428         lu_site_stats_print(&site->cs_lu, m);
 429         cache_stats_print(&site->cs_pages, m, 1);
 430         seq_printf(m, " [");
 431         for (i = 0; i < ARRAY_SIZE(site->cs_pages_state); ++i)
 432                 seq_printf(m, "%s: %u ", pstate[i],
 433                            atomic_read(&site->cs_pages_state[i]));
 434         seq_printf(m, "]\n");
 435         cache_stats_print(&cl_env_stats, m, 0);
 436         seq_printf(m, "\n");
 437         return 0;
 438 }
 439 EXPORT_SYMBOL(cl_site_stats_print);
 440
 441 /*****************************************************************************
 442  *
 443  * lu_env handling on client.
 444  *
 445  */
 446
 447 /**
 448  * The most efficient way is to store cl_env pointer in task specific
 449  * structures. On Linux, it wont' be easy to use task_struct->journal_info
 450  * because Lustre code may call into other fs which has certain assumptions
 451  * about journal_info. Currently following fields in task_struct are identified
 452  * can be used for this purpose:
 453  *  - tux_info: only on RedHat kernel.
 454  *  - ...
 455  * \note As long as we use task_struct to store cl_env, we assume that once
 456  * called into Lustre, we'll never call into the other part of the kernel
 457  * which will use those fields in task_struct without explicitly exiting
 458  * Lustre.
 459  *
 460  * If there's no space in task_struct is available, hash will be used.
 461  * bz20044, bz22683.
 462  */
 463
 464 static LIST_HEAD(cl_envs);
 465 static unsigned int cl_envs_cached_nr;
 466 static unsigned int cl_envs_cached_max = 128; /* XXX: prototype: arbitrary limit
 467                                                * for now.
 468                                                */
 469 static DEFINE_SPINLOCK(cl_envs_guard);
 470
 471 struct cl_env {
 472         void         *ce_magic;
 473         struct lu_env     ce_lu;
 474         struct lu_context ce_ses;
 475
 476         /**
 477          * This allows cl_env to be entered into cl_env_hash which implements
 478          * the current thread -> client environment lookup.
 479          */
 480         struct hlist_node  ce_node;
 481         /**
 482          * Owner for the current cl_env.
 483          *
 484          * If LL_TASK_CL_ENV is defined, this point to the owning current,
 485          * only for debugging purpose ;
 486          * Otherwise hash is used, and this is the key for cfs_hash.
 487          * Now current thread pid is stored. Note using thread pointer would
 488          * lead to unbalanced hash because of its specific allocation locality
 489          * and could be varied for different platforms and OSes, even different
 490          * OS versions.
 491          */
 492         void         *ce_owner;
 493
 494         /*
 495          * Linkage into global list of all client environments. Used for
 496          * garbage collection.
 497          */
 498         struct list_head        ce_linkage;
 499         /*
 500          *
 501          */
 502         int            ce_ref;
 503         /*
 504          * Debugging field: address of the caller who made original
 505          * allocation.
 506          */
 507         void         *ce_debug;
 508 };
 509
 510 #define CL_ENV_INC(counter)
 511 #define CL_ENV_DEC(counter)
 512
 513 static void cl_env_init0(struct cl_env *cle, void *debug)
 514 {
 515         LASSERT(cle->ce_ref == 0);
 516         LASSERT(cle->ce_magic == &cl_env_init0);
 517         LASSERT(!cle->ce_debug && !cle->ce_owner);
 518
 519         cle->ce_ref = 1;
 520         cle->ce_debug = debug;
 521         CL_ENV_INC(busy);
 522 }
 523
 524 /*
 525  * The implementation of using hash table to connect cl_env and thread
 526  */
 527
 528 static struct cfs_hash *cl_env_hash;
 529
 530 static unsigned cl_env_hops_hash(struct cfs_hash *lh,
 531                                  const void *key, unsigned mask)
 532 {
 533 #if BITS_PER_LONG == 64
 534         return cfs_hash_u64_hash((__u64)key, mask);
 535 #else
 536         return cfs_hash_u32_hash((__u32)key, mask);
 537 #endif
 538 }
 539
 540 static void *cl_env_hops_obj(struct hlist_node *hn)
 541 {
 542         struct cl_env *cle = hlist_entry(hn, struct cl_env, ce_node);
 543
 544         LASSERT(cle->ce_magic == &cl_env_init0);
 545         return (void *)cle;
 546 }
 547
 548 static int cl_env_hops_keycmp(const void *key, struct hlist_node *hn)
 549 {
 550         struct cl_env *cle = cl_env_hops_obj(hn);
 551
 552         LASSERT(cle->ce_owner);
 553         return (key == cle->ce_owner);
 554 }
 555
 556 static void cl_env_hops_noop(struct cfs_hash *hs, struct hlist_node *hn)
 557 {
 558         struct cl_env *cle = hlist_entry(hn, struct cl_env, ce_node);
 559
 560         LASSERT(cle->ce_magic == &cl_env_init0);
 561 }
 562
 563 static struct cfs_hash_ops cl_env_hops = {
 564         .hs_hash        = cl_env_hops_hash,
 565         .hs_key         = cl_env_hops_obj,
 566         .hs_keycmp      = cl_env_hops_keycmp,
 567         .hs_object      = cl_env_hops_obj,
 568         .hs_get         = cl_env_hops_noop,
 569         .hs_put_locked  = cl_env_hops_noop,
 570 };
 571
 572 static inline struct cl_env *cl_env_fetch(void)
 573 {
 574         struct cl_env *cle;
 575
 576         cle = cfs_hash_lookup(cl_env_hash, (void *)(long)current->pid);
 577         LASSERT(ergo(cle, cle->ce_magic == &cl_env_init0));
 578         return cle;
 579 }
 580
 581 static inline void cl_env_attach(struct cl_env *cle)
 582 {
 583         if (cle) {
 584                 int rc;
 585
 586                 LASSERT(!cle->ce_owner);
 587                 cle->ce_owner = (void *)(long)current->pid;
 588                 rc = cfs_hash_add_unique(cl_env_hash, cle->ce_owner,
 589                                          &cle->ce_node);
 590                 LASSERT(rc == 0);
 591         }
 592 }
 593
 594 static inline void cl_env_do_detach(struct cl_env *cle)
 595 {
 596         void *cookie;
 597
 598         LASSERT(cle->ce_owner == (void *)(long)current->pid);
 599         cookie = cfs_hash_del(cl_env_hash, cle->ce_owner,
 600                               &cle->ce_node);
 601         LASSERT(cookie == cle);
 602         cle->ce_owner = NULL;
 603 }
 604
 605 static int cl_env_store_init(void)
 606 {
 607         cl_env_hash = cfs_hash_create("cl_env",
 608                                       HASH_CL_ENV_BITS, HASH_CL_ENV_BITS,
 609                                       HASH_CL_ENV_BKT_BITS, 0,
 610                                       CFS_HASH_MIN_THETA,
 611                                       CFS_HASH_MAX_THETA,
 612                                       &cl_env_hops,
 613                                       CFS_HASH_RW_BKTLOCK);
 614         return cl_env_hash ? 0 : -ENOMEM;
 615 }
 616
 617 static void cl_env_store_fini(void)
 618 {
 619         cfs_hash_putref(cl_env_hash);
 620 }
 621
 622 static inline struct cl_env *cl_env_detach(struct cl_env *cle)
 623 {
 624         if (!cle)
 625                 cle = cl_env_fetch();
 626
 627         if (cle && cle->ce_owner)
 628                 cl_env_do_detach(cle);
 629
 630         return cle;
 631 }
 632
 633 static struct lu_env *cl_env_new(__u32 ctx_tags, __u32 ses_tags, void *debug)
 634 {
 635         struct lu_env *env;
 636         struct cl_env *cle;
 637
 638         cle = kmem_cache_zalloc(cl_env_kmem, GFP_NOFS);
 639         if (cle) {
 640                 int rc;
 641
 642                 INIT_LIST_HEAD(&cle->ce_linkage);
 643                 cle->ce_magic = &cl_env_init0;
 644                 env = &cle->ce_lu;
 645                 rc = lu_env_init(env, ctx_tags | LCT_CL_THREAD);
 646                 if (rc == 0) {
 647                         rc = lu_context_init(&cle->ce_ses,
 648                                              ses_tags | LCT_SESSION);
 649                         if (rc == 0) {
 650                                 lu_context_enter(&cle->ce_ses);
 651                                 env->le_ses = &cle->ce_ses;
 652                                 cl_env_init0(cle, debug);
 653                         } else {
 654                                 lu_env_fini(env);
 655                         }
 656                 }
 657                 if (rc != 0) {
 658                         kmem_cache_free(cl_env_kmem, cle);
 659                         env = ERR_PTR(rc);
 660                 } else {
 661                         CL_ENV_INC(create);
 662                         CL_ENV_INC(total);
 663                 }
 664         } else {
 665                 env = ERR_PTR(-ENOMEM);
 666         }
 667         return env;
 668 }
 669
 670 static void cl_env_fini(struct cl_env *cle)
 671 {
 672         CL_ENV_DEC(total);
 673         lu_context_fini(&cle->ce_lu.le_ctx);
 674         lu_context_fini(&cle->ce_ses);
 675         kmem_cache_free(cl_env_kmem, cle);
 676 }
 677
 678 static struct lu_env *cl_env_obtain(void *debug)
 679 {
 680         struct cl_env *cle;
 681         struct lu_env *env;
 682
 683         spin_lock(&cl_envs_guard);
 684         LASSERT(equi(cl_envs_cached_nr == 0, list_empty(&cl_envs)));
 685         if (cl_envs_cached_nr > 0) {
 686                 int rc;
 687
 688                 cle = container_of(cl_envs.next, struct cl_env, ce_linkage);
 689                 list_del_init(&cle->ce_linkage);
 690                 cl_envs_cached_nr--;
 691                 spin_unlock(&cl_envs_guard);
 692
 693                 env = &cle->ce_lu;
 694                 rc = lu_env_refill(env);
 695                 if (rc == 0) {
 696                         cl_env_init0(cle, debug);
 697                         lu_context_enter(&env->le_ctx);
 698                         lu_context_enter(&cle->ce_ses);
 699                 } else {
 700                         cl_env_fini(cle);
 701                         env = ERR_PTR(rc);
 702                 }
 703         } else {
 704                 spin_unlock(&cl_envs_guard);
 705                 env = cl_env_new(lu_context_tags_default,
 706                                  lu_session_tags_default, debug);
 707         }
 708         return env;
 709 }
 710
 711 static inline struct cl_env *cl_env_container(struct lu_env *env)
 712 {
 713         return container_of(env, struct cl_env, ce_lu);
 714 }
 715
 716 static struct lu_env *cl_env_peek(int *refcheck)
 717 {
 718         struct lu_env *env;
 719         struct cl_env *cle;
 720
 721         CL_ENV_INC(lookup);
 722
 723         /* check that we don't go far from untrusted pointer */
 724         CLASSERT(offsetof(struct cl_env, ce_magic) == 0);
 725
 726         env = NULL;
 727         cle = cl_env_fetch();
 728         if (cle) {
 729                 CL_ENV_INC(hit);
 730                 env = &cle->ce_lu;
 731                 *refcheck = ++cle->ce_ref;
 732         }
 733         CDEBUG(D_OTHER, "%d@%p\n", cle ? cle->ce_ref : 0, cle);
 734         return env;
 735 }
 736
 737 /**
 738  * Returns lu_env: if there already is an environment associated with the
 739  * current thread, it is returned, otherwise, new environment is allocated.
 740  *
 741  * Allocations are amortized through the global cache of environments.
 742  *
 743  * \param refcheck pointer to a counter used to detect environment leaks. In
 744  * the usual case cl_env_get() and cl_env_put() are called in the same lexical
 745  * scope and pointer to the same integer is passed as \a refcheck. This is
 746  * used to detect missed cl_env_put().
 747  *
 748  * \see cl_env_put()
 749  */
 750 struct lu_env *cl_env_get(int *refcheck)
 751 {
 752         struct lu_env *env;
 753
 754         env = cl_env_peek(refcheck);
 755         if (!env) {
 756                 env = cl_env_obtain(__builtin_return_address(0));
 757                 if (!IS_ERR(env)) {
 758                         struct cl_env *cle;
 759
 760                         cle = cl_env_container(env);
 761                         cl_env_attach(cle);
 762                         *refcheck = cle->ce_ref;
 763                         CDEBUG(D_OTHER, "%d@%p\n", cle->ce_ref, cle);
 764                 }
 765         }
 766         return env;
 767 }
 768 EXPORT_SYMBOL(cl_env_get);
 769
 770 /**
 771  * Forces an allocation of a fresh environment with given tags.
 772  *
 773  * \see cl_env_get()
 774  */
 775 struct lu_env *cl_env_alloc(int *refcheck, __u32 tags)
 776 {
 777         struct lu_env *env;
 778
 779         LASSERT(!cl_env_peek(refcheck));
 780         env = cl_env_new(tags, tags, __builtin_return_address(0));
 781         if (!IS_ERR(env)) {
 782                 struct cl_env *cle;
 783
 784                 cle = cl_env_container(env);
 785                 *refcheck = cle->ce_ref;
 786                 CDEBUG(D_OTHER, "%d@%p\n", cle->ce_ref, cle);
 787         }
 788         return env;
 789 }
 790 EXPORT_SYMBOL(cl_env_alloc);
 791
 792 static void cl_env_exit(struct cl_env *cle)
 793 {
 794         LASSERT(!cle->ce_owner);
 795         lu_context_exit(&cle->ce_lu.le_ctx);
 796         lu_context_exit(&cle->ce_ses);
 797 }
 798
 799 /**
 800  * Finalizes and frees a given number of cached environments. This is done to
 801  * (1) free some memory (not currently hooked into VM), or (2) release
 802  * references to modules.
 803  */
 804 unsigned int cl_env_cache_purge(unsigned int nr)
 805 {
 806         struct cl_env *cle;
 807
 808         spin_lock(&cl_envs_guard);
 809         for (; !list_empty(&cl_envs) && nr > 0; --nr) {
 810                 cle = container_of(cl_envs.next, struct cl_env, ce_linkage);
 811                 list_del_init(&cle->ce_linkage);
 812                 LASSERT(cl_envs_cached_nr > 0);
 813                 cl_envs_cached_nr--;
 814                 spin_unlock(&cl_envs_guard);
 815
 816                 cl_env_fini(cle);
 817                 spin_lock(&cl_envs_guard);
 818         }
 819         LASSERT(equi(cl_envs_cached_nr == 0, list_empty(&cl_envs)));
 820         spin_unlock(&cl_envs_guard);
 821         return nr;
 822 }
 823 EXPORT_SYMBOL(cl_env_cache_purge);
 824
 825 /**
 826  * Release an environment.
 827  *
 828  * Decrement \a env reference counter. When counter drops to 0, nothing in
 829  * this thread is using environment and it is returned to the allocation
 830  * cache, or freed straight away, if cache is large enough.
 831  */
 832 void cl_env_put(struct lu_env *env, int *refcheck)
 833 {
 834         struct cl_env *cle;
 835
 836         cle = cl_env_container(env);
 837
 838         LASSERT(cle->ce_ref > 0);
 839         LASSERT(ergo(refcheck, cle->ce_ref == *refcheck));
 840
 841         CDEBUG(D_OTHER, "%d@%p\n", cle->ce_ref, cle);
 842         if (--cle->ce_ref == 0) {
 843                 CL_ENV_DEC(busy);
 844                 cl_env_detach(cle);
 845                 cle->ce_debug = NULL;
 846                 cl_env_exit(cle);
 847                 /*
 848                  * Don't bother to take a lock here.
 849                  *
 850                  * Return environment to the cache only when it was allocated
 851                  * with the standard tags.
 852                  */
 853                 if (cl_envs_cached_nr < cl_envs_cached_max &&
 854                     (env->le_ctx.lc_tags & ~LCT_HAS_EXIT) == LCT_CL_THREAD &&
 855                     (env->le_ses->lc_tags & ~LCT_HAS_EXIT) == LCT_SESSION) {
 856                         spin_lock(&cl_envs_guard);
 857                         list_add(&cle->ce_linkage, &cl_envs);
 858                         cl_envs_cached_nr++;
 859                         spin_unlock(&cl_envs_guard);
 860                 } else {
 861                         cl_env_fini(cle);
 862                 }
 863         }
 864 }
 865 EXPORT_SYMBOL(cl_env_put);
 866
 867 /**
 868  * Declares a point of re-entrancy.
 869  *
 870  * \see cl_env_reexit()
 871  */
 872 void *cl_env_reenter(void)
 873 {
 874         return cl_env_detach(NULL);
 875 }
 876 EXPORT_SYMBOL(cl_env_reenter);
 877
 878 /**
 879  * Exits re-entrancy.
 880  */
 881 void cl_env_reexit(void *cookie)
 882 {
 883         cl_env_detach(NULL);
 884         cl_env_attach(cookie);
 885 }
 886 EXPORT_SYMBOL(cl_env_reexit);
 887
 888 /**
 889  * Setup user-supplied \a env as a current environment. This is to be used to
 890  * guaranteed that environment exists even when cl_env_get() fails. It is up
 891  * to user to ensure proper concurrency control.
 892  *
 893  * \see cl_env_unplant()
 894  */
 895 void cl_env_implant(struct lu_env *env, int *refcheck)
 896 {
 897         struct cl_env *cle = cl_env_container(env);
 898
 899         LASSERT(cle->ce_ref > 0);
 900
 901         cl_env_attach(cle);
 902         cl_env_get(refcheck);
 903         CDEBUG(D_OTHER, "%d@%p\n", cle->ce_ref, cle);
 904 }
 905 EXPORT_SYMBOL(cl_env_implant);
 906
 907 /**
 908  * Detach environment installed earlier by cl_env_implant().
 909  */
 910 void cl_env_unplant(struct lu_env *env, int *refcheck)
 911 {
 912         struct cl_env *cle = cl_env_container(env);
 913
 914         LASSERT(cle->ce_ref > 1);
 915
 916         CDEBUG(D_OTHER, "%d@%p\n", cle->ce_ref, cle);
 917
 918         cl_env_detach(cle);
 919         cl_env_put(env, refcheck);
 920 }
 921 EXPORT_SYMBOL(cl_env_unplant);
 922
 923 struct lu_env *cl_env_nested_get(struct cl_env_nest *nest)
 924 {
 925         struct lu_env *env;
 926
 927         nest->cen_cookie = NULL;
 928         env = cl_env_peek(&nest->cen_refcheck);
 929         if (env) {
 930                 if (!cl_io_is_going(env))
 931                         return env;
 932                 cl_env_put(env, &nest->cen_refcheck);
 933                 nest->cen_cookie = cl_env_reenter();
 934         }
 935         env = cl_env_get(&nest->cen_refcheck);
 936         if (IS_ERR(env)) {
 937                 cl_env_reexit(nest->cen_cookie);
 938                 return env;
 939         }
 940
 941         LASSERT(!cl_io_is_going(env));
 942         return env;
 943 }
 944 EXPORT_SYMBOL(cl_env_nested_get);
 945
 946 void cl_env_nested_put(struct cl_env_nest *nest, struct lu_env *env)
 947 {
 948         cl_env_put(env, &nest->cen_refcheck);
 949         cl_env_reexit(nest->cen_cookie);
 950 }
 951 EXPORT_SYMBOL(cl_env_nested_put);
 952
 953 /**
 954  * Converts struct ost_lvb to struct cl_attr.
 955  *
 956  * \see cl_attr2lvb
 957  */
 958 void cl_lvb2attr(struct cl_attr *attr, const struct ost_lvb *lvb)
 959 {
 960         attr->cat_size   = lvb->lvb_size;
 961         attr->cat_mtime  = lvb->lvb_mtime;
 962         attr->cat_atime  = lvb->lvb_atime;
 963         attr->cat_ctime  = lvb->lvb_ctime;
 964         attr->cat_blocks = lvb->lvb_blocks;
 965 }
 966 EXPORT_SYMBOL(cl_lvb2attr);
 967
 968 static struct cl_env cl_env_percpu[NR_CPUS];
 969
 970 static int cl_env_percpu_init(void)
 971 {
 972         struct cl_env *cle;
 973         int tags = LCT_REMEMBER | LCT_NOREF;
 974         int i, j;
 975         int rc = 0;
 976
 977         for_each_possible_cpu(i) {
 978                 struct lu_env *env;
 979
 980                 cle = &cl_env_percpu[i];
 981                 env = &cle->ce_lu;
 982
 983                 INIT_LIST_HEAD(&cle->ce_linkage);
 984                 cle->ce_magic = &cl_env_init0;
 985                 rc = lu_env_init(env, LCT_CL_THREAD | tags);
 986                 if (rc == 0) {
 987                         rc = lu_context_init(&cle->ce_ses, LCT_SESSION | tags);
 988                         if (rc == 0) {
 989                                 lu_context_enter(&cle->ce_ses);
 990                                 env->le_ses = &cle->ce_ses;
 991                         } else {
 992                                 lu_env_fini(env);
 993                         }
 994                 }
 995                 if (rc != 0)
 996                         break;
 997         }
 998         if (rc != 0) {
 999                 /* Indices 0 to i (excluding i) were correctly initialized,
1000                  * thus we must uninitialize up to i, the rest are undefined.
1001                  */
1002                 for (j = 0; j < i; j++) {
1003                         cle = &cl_env_percpu[i];
1004                         lu_context_exit(&cle->ce_ses);
1005                         lu_context_fini(&cle->ce_ses);
1006                         lu_env_fini(&cle->ce_lu);
1007                 }
1008         }
1009
1010         return rc;
1011 }
1012
1013 static void cl_env_percpu_fini(void)
1014 {
1015         int i;
1016
1017         for_each_possible_cpu(i) {
1018                 struct cl_env *cle = &cl_env_percpu[i];
1019
1020                 lu_context_exit(&cle->ce_ses);
1021                 lu_context_fini(&cle->ce_ses);
1022                 lu_env_fini(&cle->ce_lu);
1023         }
1024 }
1025
1026 static void cl_env_percpu_refill(void)
1027 {
1028         int i;
1029
1030         for_each_possible_cpu(i)
1031                 lu_env_refill(&cl_env_percpu[i].ce_lu);
1032 }
1033
1034 void cl_env_percpu_put(struct lu_env *env)
1035 {
1036         struct cl_env *cle;
1037         int cpu;
1038
1039         cpu = smp_processor_id();
1040         cle = cl_env_container(env);
1041         LASSERT(cle == &cl_env_percpu[cpu]);
1042
1043         cle->ce_ref--;
1044         LASSERT(cle->ce_ref == 0);
1045
1046         CL_ENV_DEC(busy);
1047         cl_env_detach(cle);
1048         cle->ce_debug = NULL;
1049
1050         put_cpu();
1051 }
1052 EXPORT_SYMBOL(cl_env_percpu_put);
1053
1054 struct lu_env *cl_env_percpu_get(void)
1055 {
1056         struct cl_env *cle;
1057
1058         cle = &cl_env_percpu[get_cpu()];
1059         cl_env_init0(cle, __builtin_return_address(0));
1060
1061         cl_env_attach(cle);
1062         return &cle->ce_lu;
1063 }
1064 EXPORT_SYMBOL(cl_env_percpu_get);
1065
1066 /*****************************************************************************
1067  *
1068  * Temporary prototype thing: mirror obd-devices into cl devices.
1069  *
1070  */
1071
1072 struct cl_device *cl_type_setup(const struct lu_env *env, struct lu_site *site,
1073                                 struct lu_device_type *ldt,
1074                                 struct lu_device *next)
1075 {
1076         const char       *typename;
1077         struct lu_device *d;
1078
1079         typename = ldt->ldt_name;
1080         d = ldt->ldt_ops->ldto_device_alloc(env, ldt, NULL);
1081         if (!IS_ERR(d)) {
1082                 int rc;
1083
1084                 if (site)
1085                         d->ld_site = site;
1086                 rc = ldt->ldt_ops->ldto_device_init(env, d, typename, next);
1087                 if (rc == 0) {
1088                         lu_device_get(d);
1089                         lu_ref_add(&d->ld_reference,
1090                                    "lu-stack", &lu_site_init);
1091                 } else {
1092                         ldt->ldt_ops->ldto_device_free(env, d);
1093                         CERROR("can't init device '%s', %d\n", typename, rc);
1094                         d = ERR_PTR(rc);
1095                 }
1096         } else {
1097                 CERROR("Cannot allocate device: '%s'\n", typename);
1098         }
1099         return lu2cl_dev(d);
1100 }
1101 EXPORT_SYMBOL(cl_type_setup);
1102
1103 /**
1104  * Finalize device stack by calling lu_stack_fini().
1105  */
1106 void cl_stack_fini(const struct lu_env *env, struct cl_device *cl)
1107 {
1108         lu_stack_fini(env, cl2lu_dev(cl));
1109 }
1110 EXPORT_SYMBOL(cl_stack_fini);
1111
1112 static struct lu_context_key cl_key;
1113
1114 struct cl_thread_info *cl_env_info(const struct lu_env *env)
1115 {
1116         return lu_context_key_get(&env->le_ctx, &cl_key);
1117 }
1118
1119 /* defines cl0_key_{init,fini}() */
1120 LU_KEY_INIT_FINI(cl0, struct cl_thread_info);
1121
1122 static void *cl_key_init(const struct lu_context *ctx,
1123                          struct lu_context_key *key)
1124 {
1125         struct cl_thread_info *info;
1126
1127         info = cl0_key_init(ctx, key);
1128         if (!IS_ERR(info)) {
1129                 int i;
1130
1131                 for (i = 0; i < ARRAY_SIZE(info->clt_counters); ++i)
1132                         lu_ref_init(&info->clt_counters[i].ctc_locks_locked);
1133         }
1134         return info;
1135 }
1136
1137 static void cl_key_fini(const struct lu_context *ctx,
1138                         struct lu_context_key *key, void *data)
1139 {
1140         struct cl_thread_info *info;
1141         int i;
1142
1143         info = data;
1144         for (i = 0; i < ARRAY_SIZE(info->clt_counters); ++i)
1145                 lu_ref_fini(&info->clt_counters[i].ctc_locks_locked);
1146         cl0_key_fini(ctx, key, data);
1147 }
1148
1149 static void cl_key_exit(const struct lu_context *ctx,
1150                         struct lu_context_key *key, void *data)
1151 {
1152         struct cl_thread_info *info = data;
1153         int i;
1154
1155         for (i = 0; i < ARRAY_SIZE(info->clt_counters); ++i) {
1156                 LASSERT(info->clt_counters[i].ctc_nr_held == 0);
1157                 LASSERT(info->clt_counters[i].ctc_nr_used == 0);
1158                 LASSERT(info->clt_counters[i].ctc_nr_locks_acquired == 0);
1159                 LASSERT(info->clt_counters[i].ctc_nr_locks_locked == 0);
1160                 lu_ref_fini(&info->clt_counters[i].ctc_locks_locked);
1161                 lu_ref_init(&info->clt_counters[i].ctc_locks_locked);
1162         }
1163 }
1164
1165 static struct lu_context_key cl_key = {
1166         .lct_tags = LCT_CL_THREAD,
1167         .lct_init = cl_key_init,
1168         .lct_fini = cl_key_fini,
1169         .lct_exit = cl_key_exit
1170 };
1171
1172 static struct lu_kmem_descr cl_object_caches[] = {
1173         {
1174                 .ckd_cache = &cl_env_kmem,
1175                 .ckd_name  = "cl_env_kmem",
1176                 .ckd_size  = sizeof(struct cl_env)
1177         },
1178         {
1179                 .ckd_cache = NULL
1180         }
1181 };
1182
1183 /**
1184  * Global initialization of cl-data. Create kmem caches, register
1185  * lu_context_key's, etc.
1186  *
1187  * \see cl_global_fini()
1188  */
1189 int cl_global_init(void)
1190 {
1191         int result;
1192
1193         result = cl_env_store_init();
1194         if (result)
1195                 return result;
1196
1197         result = lu_kmem_init(cl_object_caches);
1198         if (result)
1199                 goto out_store;
1200
1201         LU_CONTEXT_KEY_INIT(&cl_key);
1202         result = lu_context_key_register(&cl_key);
1203         if (result)
1204                 goto out_kmem;
1205
1206         result = cl_env_percpu_init();
1207         if (result)
1208                 /* no cl_env_percpu_fini on error */
1209                 goto out_context;
1210
1211         return 0;
1212
1213 out_context:
1214         lu_context_key_degister(&cl_key);
1215 out_kmem:
1216         lu_kmem_fini(cl_object_caches);
1217 out_store:
1218         cl_env_store_fini();
1219         return result;
1220 }
1221
1222 /**
1223  * Finalization of global cl-data. Dual to cl_global_init().
1224  */
1225 void cl_global_fini(void)
1226 {
1227         cl_env_percpu_fini();
1228         lu_context_key_degister(&cl_key);
1229         lu_kmem_fini(cl_object_caches);
1230         cl_env_store_fini();
1231 }