lustre/obdclass/lu_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.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) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2011, 2012, 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  * lustre/obdclass/lu_object.c
  37  *
  38  * Lustre Object.
  39  * These are the only exported functions, they provide some generic
  40  * infrastructure for managing object devices
  41  *
  42  *   Author: Nikita Danilov <nikita.danilov@sun.com>
  43  */
  44
  45 #define DEBUG_SUBSYSTEM S_CLASS
  46
  47 #include <libcfs/libcfs.h>
  48
  49 #ifdef __KERNEL__
  50 # include <linux/module.h>
  51 #endif
  52
  53 /* hash_long() */
  54 #include <libcfs/libcfs_hash.h>
  55 #include <obd_class.h>
  56 #include <obd_support.h>
  57 #include <lustre_disk.h>
  58 #include <lustre_fid.h>
  59 #include <lu_object.h>
  60 #include <libcfs/list.h>
  61
  62 static void lu_object_free(const struct lu_env *env, struct lu_object *o);
  63
  64 /**
  65  * Decrease reference counter on object. If last reference is freed, return
  66  * object to the cache, unless lu_object_is_dying(o) holds. In the latter
  67  * case, free object immediately.
  68  */
  69 void lu_object_put(const struct lu_env *env, struct lu_object *o)
  70 {
  71         struct lu_site_bkt_data *bkt;
  72         struct lu_object_header *top;
  73         struct lu_site          *site;
  74         struct lu_object        *orig;
  75         cfs_hash_bd_t            bd;
  76         const struct lu_fid     *fid;
  77
  78         top  = o->lo_header;
  79         site = o->lo_dev->ld_site;
  80         orig = o;
  81
  82         /*
  83          * till we have full fids-on-OST implemented anonymous objects
  84          * are possible in OSP. such an object isn't listed in the site
  85          * so we should not remove it from the site.
  86          */
  87         fid = lu_object_fid(o);
  88         if (fid_is_zero(fid)) {
  89                 LASSERT(top->loh_hash.next == NULL
  90                         && top->loh_hash.pprev == NULL);
  91                 LASSERT(cfs_list_empty(&top->loh_lru));
  92                 if (!cfs_atomic_dec_and_test(&top->loh_ref))
  93                         return;
  94                 cfs_list_for_each_entry_reverse(o, &top->loh_layers, lo_linkage) {
  95                         if (o->lo_ops->loo_object_release != NULL)
  96                                 o->lo_ops->loo_object_release(env, o);
  97                 }
  98                 lu_object_free(env, orig);
  99                 return;
 100         }
 101
 102         cfs_hash_bd_get(site->ls_obj_hash, &top->loh_fid, &bd);
 103         bkt = cfs_hash_bd_extra_get(site->ls_obj_hash, &bd);
 104
 105         if (!cfs_hash_bd_dec_and_lock(site->ls_obj_hash, &bd, &top->loh_ref)) {
 106                 if (lu_object_is_dying(top)) {
 107
 108                         /*
 109                          * somebody may be waiting for this, currently only
 110                          * used for cl_object, see cl_object_put_last().
 111                          */
 112                         cfs_waitq_broadcast(&bkt->lsb_marche_funebre);
 113                 }
 114                 return;
 115         }
 116
 117         LASSERT(bkt->lsb_busy > 0);
 118         bkt->lsb_busy--;
 119         /*
 120          * When last reference is released, iterate over object
 121          * layers, and notify them that object is no longer busy.
 122          */
 123         cfs_list_for_each_entry_reverse(o, &top->loh_layers, lo_linkage) {
 124                 if (o->lo_ops->loo_object_release != NULL)
 125                         o->lo_ops->loo_object_release(env, o);
 126         }
 127
 128         if (!lu_object_is_dying(top)) {
 129                 LASSERT(cfs_list_empty(&top->loh_lru));
 130                 cfs_list_add_tail(&top->loh_lru, &bkt->lsb_lru);
 131                 cfs_hash_bd_unlock(site->ls_obj_hash, &bd, 1);
 132                 return;
 133         }
 134
 135         /*
 136          * If object is dying (will not be cached), removed it
 137          * from hash table and LRU.
 138          *
 139          * This is done with hash table and LRU lists locked. As the only
 140          * way to acquire first reference to previously unreferenced
 141          * object is through hash-table lookup (lu_object_find()),
 142          * or LRU scanning (lu_site_purge()), that are done under hash-table
 143          * and LRU lock, no race with concurrent object lookup is possible
 144          * and we can safely destroy object below.
 145          */
 146         cfs_hash_bd_del_locked(site->ls_obj_hash, &bd, &top->loh_hash);
 147         cfs_hash_bd_unlock(site->ls_obj_hash, &bd, 1);
 148         /*
 149          * Object was already removed from hash and lru above, can
 150          * kill it.
 151          */
 152         lu_object_free(env, orig);
 153 }
 154 EXPORT_SYMBOL(lu_object_put);
 155
 156 /**
 157  * Put object and don't keep in cache. This is temporary solution for
 158  * multi-site objects when its layering is not constant.
 159  */
 160 void lu_object_put_nocache(const struct lu_env *env, struct lu_object *o)
 161 {
 162         set_bit(LU_OBJECT_HEARD_BANSHEE,
 163                     &o->lo_header->loh_flags);
 164         return lu_object_put(env, o);
 165 }
 166 EXPORT_SYMBOL(lu_object_put_nocache);
 167
 168 /**
 169  * Allocate new object.
 170  *
 171  * This follows object creation protocol, described in the comment within
 172  * struct lu_device_operations definition.
 173  */
 174 static struct lu_object *lu_object_alloc(const struct lu_env *env,
 175                                          struct lu_device *dev,
 176                                          const struct lu_fid *f,
 177                                          const struct lu_object_conf *conf)
 178 {
 179         struct lu_object *scan;
 180         struct lu_object *top;
 181         cfs_list_t *layers;
 182         int clean;
 183         int result;
 184         ENTRY;
 185
 186         /*
 187          * Create top-level object slice. This will also create
 188          * lu_object_header.
 189          */
 190         top = dev->ld_ops->ldo_object_alloc(env, NULL, dev);
 191         if (top == NULL)
 192                 RETURN(ERR_PTR(-ENOMEM));
 193         if (IS_ERR(top))
 194                 RETURN(top);
 195         /*
 196          * This is the only place where object fid is assigned. It's constant
 197          * after this point.
 198          */
 199         top->lo_header->loh_fid = *f;
 200         layers = &top->lo_header->loh_layers;
 201         do {
 202                 /*
 203                  * Call ->loo_object_init() repeatedly, until no more new
 204                  * object slices are created.
 205                  */
 206                 clean = 1;
 207                 cfs_list_for_each_entry(scan, layers, lo_linkage) {
 208                         if (scan->lo_flags & LU_OBJECT_ALLOCATED)
 209                                 continue;
 210                         clean = 0;
 211                         scan->lo_header = top->lo_header;
 212                         result = scan->lo_ops->loo_object_init(env, scan, conf);
 213                         if (result != 0) {
 214                                 lu_object_free(env, top);
 215                                 RETURN(ERR_PTR(result));
 216                         }
 217                         scan->lo_flags |= LU_OBJECT_ALLOCATED;
 218                 }
 219         } while (!clean);
 220
 221         cfs_list_for_each_entry_reverse(scan, layers, lo_linkage) {
 222                 if (scan->lo_ops->loo_object_start != NULL) {
 223                         result = scan->lo_ops->loo_object_start(env, scan);
 224                         if (result != 0) {
 225                                 lu_object_free(env, top);
 226                                 RETURN(ERR_PTR(result));
 227                         }
 228                 }
 229         }
 230
 231         lprocfs_counter_incr(dev->ld_site->ls_stats, LU_SS_CREATED);
 232         RETURN(top);
 233 }
 234
 235 /**
 236  * Free an object.
 237  */
 238 static void lu_object_free(const struct lu_env *env, struct lu_object *o)
 239 {
 240         struct lu_site_bkt_data *bkt;
 241         struct lu_site          *site;
 242         struct lu_object        *scan;
 243         cfs_list_t              *layers;
 244         cfs_list_t               splice;
 245
 246         site   = o->lo_dev->ld_site;
 247         layers = &o->lo_header->loh_layers;
 248         bkt    = lu_site_bkt_from_fid(site, &o->lo_header->loh_fid);
 249         /*
 250          * First call ->loo_object_delete() method to release all resources.
 251          */
 252         cfs_list_for_each_entry_reverse(scan, layers, lo_linkage) {
 253                 if (scan->lo_ops->loo_object_delete != NULL)
 254                         scan->lo_ops->loo_object_delete(env, scan);
 255         }
 256
 257         /*
 258          * Then, splice object layers into stand-alone list, and call
 259          * ->loo_object_free() on all layers to free memory. Splice is
 260          * necessary, because lu_object_header is freed together with the
 261          * top-level slice.
 262          */
 263         CFS_INIT_LIST_HEAD(&splice);
 264         cfs_list_splice_init(layers, &splice);
 265         while (!cfs_list_empty(&splice)) {
 266                 /*
 267                  * Free layers in bottom-to-top order, so that object header
 268                  * lives as long as possible and ->loo_object_free() methods
 269                  * can look at its contents.
 270                  */
 271                 o = container_of0(splice.prev, struct lu_object, lo_linkage);
 272                 cfs_list_del_init(&o->lo_linkage);
 273                 LASSERT(o->lo_ops->loo_object_free != NULL);
 274                 o->lo_ops->loo_object_free(env, o);
 275         }
 276
 277         if (cfs_waitq_active(&bkt->lsb_marche_funebre))
 278                 cfs_waitq_broadcast(&bkt->lsb_marche_funebre);
 279 }
 280
 281 /**
 282  * Free \a nr objects from the cold end of the site LRU list.
 283  */
 284 int lu_site_purge(const struct lu_env *env, struct lu_site *s, int nr)
 285 {
 286         struct lu_object_header *h;
 287         struct lu_object_header *temp;
 288         struct lu_site_bkt_data *bkt;
 289         cfs_hash_bd_t            bd;
 290         cfs_hash_bd_t            bd2;
 291         cfs_list_t               dispose;
 292         int                      did_sth;
 293         int                      start;
 294         int                      count;
 295         int                      bnr;
 296         int                      i;
 297
 298         CFS_INIT_LIST_HEAD(&dispose);
 299         /*
 300          * Under LRU list lock, scan LRU list and move unreferenced objects to
 301          * the dispose list, removing them from LRU and hash table.
 302          */
 303         start = s->ls_purge_start;
 304         bnr = (nr == ~0) ? -1 : nr / CFS_HASH_NBKT(s->ls_obj_hash) + 1;
 305  again:
 306         did_sth = 0;
 307         cfs_hash_for_each_bucket(s->ls_obj_hash, &bd, i) {
 308                 if (i < start)
 309                         continue;
 310                 count = bnr;
 311                 cfs_hash_bd_lock(s->ls_obj_hash, &bd, 1);
 312                 bkt = cfs_hash_bd_extra_get(s->ls_obj_hash, &bd);
 313
 314                 cfs_list_for_each_entry_safe(h, temp, &bkt->lsb_lru, loh_lru) {
 315                         LASSERT(cfs_atomic_read(&h->loh_ref) == 0);
 316
 317                         cfs_hash_bd_get(s->ls_obj_hash, &h->loh_fid, &bd2);
 318                         LASSERT(bd.bd_bucket == bd2.bd_bucket);
 319
 320                         cfs_hash_bd_del_locked(s->ls_obj_hash,
 321                                                &bd2, &h->loh_hash);
 322                         cfs_list_move(&h->loh_lru, &dispose);
 323                         if (did_sth == 0)
 324                                 did_sth = 1;
 325
 326                         if (nr != ~0 && --nr == 0)
 327                                 break;
 328
 329                         if (count > 0 && --count == 0)
 330                                 break;
 331
 332                 }
 333                 cfs_hash_bd_unlock(s->ls_obj_hash, &bd, 1);
 334                 cfs_cond_resched();
 335                 /*
 336                  * Free everything on the dispose list. This is safe against
 337                  * races due to the reasons described in lu_object_put().
 338                  */
 339                 while (!cfs_list_empty(&dispose)) {
 340                         h = container_of0(dispose.next,
 341                                           struct lu_object_header, loh_lru);
 342                         cfs_list_del_init(&h->loh_lru);
 343                         lu_object_free(env, lu_object_top(h));
 344                         lprocfs_counter_incr(s->ls_stats, LU_SS_LRU_PURGED);
 345                 }
 346
 347                 if (nr == 0)
 348                         break;
 349         }
 350
 351         if (nr != 0 && did_sth && start != 0) {
 352                 start = 0; /* restart from the first bucket */
 353                 goto again;
 354         }
 355         /* race on s->ls_purge_start, but nobody cares */
 356         s->ls_purge_start = i % CFS_HASH_NBKT(s->ls_obj_hash);
 357
 358         return nr;
 359 }
 360 EXPORT_SYMBOL(lu_site_purge);
 361
 362 /*
 363  * Object printing.
 364  *
 365  * Code below has to jump through certain loops to output object description
 366  * into libcfs_debug_msg-based log. The problem is that lu_object_print()
 367  * composes object description from strings that are parts of _lines_ of
 368  * output (i.e., strings that are not terminated by newline). This doesn't fit
 369  * very well into libcfs_debug_msg() interface that assumes that each message
 370  * supplied to it is a self-contained output line.
 371  *
 372  * To work around this, strings are collected in a temporary buffer
 373  * (implemented as a value of lu_cdebug_key key), until terminating newline
 374  * character is detected.
 375  *
 376  */
 377
 378 enum {
 379         /**
 380          * Maximal line size.
 381          *
 382          * XXX overflow is not handled correctly.
 383          */
 384         LU_CDEBUG_LINE = 512
 385 };
 386
 387 struct lu_cdebug_data {
 388         /**
 389          * Temporary buffer.
 390          */
 391         char lck_area[LU_CDEBUG_LINE];
 392 };
 393
 394 /* context key constructor/destructor: lu_global_key_init, lu_global_key_fini */
 395 LU_KEY_INIT_FINI(lu_global, struct lu_cdebug_data);
 396
 397 /**
 398  * Key, holding temporary buffer. This key is registered very early by
 399  * lu_global_init().
 400  */
 401 struct lu_context_key lu_global_key = {
 402         .lct_tags = LCT_MD_THREAD | LCT_DT_THREAD |
 403                     LCT_MG_THREAD | LCT_CL_THREAD,
 404         .lct_init = lu_global_key_init,
 405         .lct_fini = lu_global_key_fini
 406 };
 407
 408 /**
 409  * Printer function emitting messages through libcfs_debug_msg().
 410  */
 411 int lu_cdebug_printer(const struct lu_env *env,
 412                       void *cookie, const char *format, ...)
 413 {
 414         struct libcfs_debug_msg_data *msgdata = cookie;
 415         struct lu_cdebug_data        *key;
 416         int used;
 417         int complete;
 418         va_list args;
 419
 420         va_start(args, format);
 421
 422         key = lu_context_key_get(&env->le_ctx, &lu_global_key);
 423         LASSERT(key != NULL);
 424
 425         used = strlen(key->lck_area);
 426         complete = format[strlen(format) - 1] == '\n';
 427         /*
 428          * Append new chunk to the buffer.
 429          */
 430         vsnprintf(key->lck_area + used,
 431                   ARRAY_SIZE(key->lck_area) - used, format, args);
 432         if (complete) {
 433                 if (cfs_cdebug_show(msgdata->msg_mask, msgdata->msg_subsys))
 434                         libcfs_debug_msg(msgdata, "%s", key->lck_area);
 435                 key->lck_area[0] = 0;
 436         }
 437         va_end(args);
 438         return 0;
 439 }
 440 EXPORT_SYMBOL(lu_cdebug_printer);
 441
 442 /**
 443  * Print object header.
 444  */
 445 void lu_object_header_print(const struct lu_env *env, void *cookie,
 446                             lu_printer_t printer,
 447                             const struct lu_object_header *hdr)
 448 {
 449         (*printer)(env, cookie, "header@%p[%#lx, %d, "DFID"%s%s%s]",
 450                    hdr, hdr->loh_flags, cfs_atomic_read(&hdr->loh_ref),
 451                    PFID(&hdr->loh_fid),
 452                    cfs_hlist_unhashed(&hdr->loh_hash) ? "" : " hash",
 453                    cfs_list_empty((cfs_list_t *)&hdr->loh_lru) ? \
 454                    "" : " lru",
 455                    hdr->loh_attr & LOHA_EXISTS ? " exist":"");
 456 }
 457 EXPORT_SYMBOL(lu_object_header_print);
 458
 459 /**
 460  * Print human readable representation of the \a o to the \a printer.
 461  */
 462 void lu_object_print(const struct lu_env *env, void *cookie,
 463                      lu_printer_t printer, const struct lu_object *o)
 464 {
 465         static const char ruler[] = "........................................";
 466         struct lu_object_header *top;
 467         int depth;
 468
 469         top = o->lo_header;
 470         lu_object_header_print(env, cookie, printer, top);
 471         (*printer)(env, cookie, "{ \n");
 472         cfs_list_for_each_entry(o, &top->loh_layers, lo_linkage) {
 473                 depth = o->lo_depth + 4;
 474
 475                 /*
 476                  * print `.' \a depth times followed by type name and address
 477                  */
 478                 (*printer)(env, cookie, "%*.*s%s@%p", depth, depth, ruler,
 479                            o->lo_dev->ld_type->ldt_name, o);
 480                 if (o->lo_ops->loo_object_print != NULL)
 481                         o->lo_ops->loo_object_print(env, cookie, printer, o);
 482                 (*printer)(env, cookie, "\n");
 483         }
 484         (*printer)(env, cookie, "} header@%p\n", top);
 485 }
 486 EXPORT_SYMBOL(lu_object_print);
 487
 488 /**
 489  * Check object consistency.
 490  */
 491 int lu_object_invariant(const struct lu_object *o)
 492 {
 493         struct lu_object_header *top;
 494
 495         top = o->lo_header;
 496         cfs_list_for_each_entry(o, &top->loh_layers, lo_linkage) {
 497                 if (o->lo_ops->loo_object_invariant != NULL &&
 498                     !o->lo_ops->loo_object_invariant(o))
 499                         return 0;
 500         }
 501         return 1;
 502 }
 503 EXPORT_SYMBOL(lu_object_invariant);
 504
 505 static struct lu_object *htable_lookup(struct lu_site *s,
 506                                        cfs_hash_bd_t *bd,
 507                                        const struct lu_fid *f,
 508                                        cfs_waitlink_t *waiter,
 509                                        __u64 *version)
 510 {
 511         struct lu_site_bkt_data *bkt;
 512         struct lu_object_header *h;
 513         cfs_hlist_node_t        *hnode;
 514         __u64  ver = cfs_hash_bd_version_get(bd);
 515
 516         if (*version == ver)
 517                 return NULL;
 518
 519         *version = ver;
 520         bkt = cfs_hash_bd_extra_get(s->ls_obj_hash, bd);
 521         /* cfs_hash_bd_peek_locked is a somehow "internal" function
 522          * of cfs_hash, it doesn't add refcount on object. */
 523         hnode = cfs_hash_bd_peek_locked(s->ls_obj_hash, bd, (void *)f);
 524         if (hnode == NULL) {
 525                 lprocfs_counter_incr(s->ls_stats, LU_SS_CACHE_MISS);
 526                 return NULL;
 527         }
 528
 529         h = container_of0(hnode, struct lu_object_header, loh_hash);
 530         if (likely(!lu_object_is_dying(h))) {
 531                 cfs_hash_get(s->ls_obj_hash, hnode);
 532                 lprocfs_counter_incr(s->ls_stats, LU_SS_CACHE_HIT);
 533                 cfs_list_del_init(&h->loh_lru);
 534                 return lu_object_top(h);
 535         }
 536
 537         /*
 538          * Lookup found an object being destroyed this object cannot be
 539          * returned (to assure that references to dying objects are eventually
 540          * drained), and moreover, lookup has to wait until object is freed.
 541          */
 542
 543         cfs_waitlink_init(waiter);
 544         cfs_waitq_add(&bkt->lsb_marche_funebre, waiter);
 545         cfs_set_current_state(CFS_TASK_UNINT);
 546         lprocfs_counter_incr(s->ls_stats, LU_SS_CACHE_DEATH_RACE);
 547         return ERR_PTR(-EAGAIN);
 548 }
 549
 550 /**
 551  * Search cache for an object with the fid \a f. If such object is found,
 552  * return it. Otherwise, create new object, insert it into cache and return
 553  * it. In any case, additional reference is acquired on the returned object.
 554  */
 555 struct lu_object *lu_object_find(const struct lu_env *env,
 556                                  struct lu_device *dev, const struct lu_fid *f,
 557                                  const struct lu_object_conf *conf)
 558 {
 559         return lu_object_find_at(env, dev->ld_site->ls_top_dev, f, conf);
 560 }
 561 EXPORT_SYMBOL(lu_object_find);
 562
 563 static struct lu_object *lu_object_new(const struct lu_env *env,
 564                                        struct lu_device *dev,
 565                                        const struct lu_fid *f,
 566                                        const struct lu_object_conf *conf)
 567 {
 568         struct lu_object        *o;
 569         cfs_hash_t              *hs;
 570         cfs_hash_bd_t            bd;
 571         struct lu_site_bkt_data *bkt;
 572
 573         o = lu_object_alloc(env, dev, f, conf);
 574         if (unlikely(IS_ERR(o)))
 575                 return o;
 576
 577         hs = dev->ld_site->ls_obj_hash;
 578         cfs_hash_bd_get_and_lock(hs, (void *)f, &bd, 1);
 579         bkt = cfs_hash_bd_extra_get(hs, &bd);
 580         cfs_hash_bd_add_locked(hs, &bd, &o->lo_header->loh_hash);
 581         bkt->lsb_busy++;
 582         cfs_hash_bd_unlock(hs, &bd, 1);
 583         return o;
 584 }
 585
 586 /**
 587  * Core logic of lu_object_find*() functions.
 588  */
 589 static struct lu_object *lu_object_find_try(const struct lu_env *env,
 590                                             struct lu_device *dev,
 591                                             const struct lu_fid *f,
 592                                             const struct lu_object_conf *conf,
 593                                             cfs_waitlink_t *waiter)
 594 {
 595         struct lu_object      *o;
 596         struct lu_object      *shadow;
 597         struct lu_site        *s;
 598         cfs_hash_t            *hs;
 599         cfs_hash_bd_t          bd;
 600         __u64                  version = 0;
 601
 602         /*
 603          * This uses standard index maintenance protocol:
 604          *
 605          *     - search index under lock, and return object if found;
 606          *     - otherwise, unlock index, allocate new object;
 607          *     - lock index and search again;
 608          *     - if nothing is found (usual case), insert newly created
 609          *       object into index;
 610          *     - otherwise (race: other thread inserted object), free
 611          *       object just allocated.
 612          *     - unlock index;
 613          *     - return object.
 614          *
 615          * For "LOC_F_NEW" case, we are sure the object is new established.
 616          * It is unnecessary to perform lookup-alloc-lookup-insert, instead,
 617          * just alloc and insert directly.
 618          *
 619          * If dying object is found during index search, add @waiter to the
 620          * site wait-queue and return ERR_PTR(-EAGAIN).
 621          */
 622         if (conf != NULL && conf->loc_flags & LOC_F_NEW)
 623                 return lu_object_new(env, dev, f, conf);
 624
 625         s  = dev->ld_site;
 626         hs = s->ls_obj_hash;
 627         cfs_hash_bd_get_and_lock(hs, (void *)f, &bd, 1);
 628         o = htable_lookup(s, &bd, f, waiter, &version);
 629         cfs_hash_bd_unlock(hs, &bd, 1);
 630         if (o != NULL)
 631                 return o;
 632
 633         /*
 634          * Allocate new object. This may result in rather complicated
 635          * operations, including fld queries, inode loading, etc.
 636          */
 637         o = lu_object_alloc(env, dev, f, conf);
 638         if (unlikely(IS_ERR(o)))
 639                 return o;
 640
 641         LASSERT(lu_fid_eq(lu_object_fid(o), f));
 642
 643         cfs_hash_bd_lock(hs, &bd, 1);
 644
 645         shadow = htable_lookup(s, &bd, f, waiter, &version);
 646         if (likely(shadow == NULL)) {
 647                 struct lu_site_bkt_data *bkt;
 648
 649                 bkt = cfs_hash_bd_extra_get(hs, &bd);
 650                 cfs_hash_bd_add_locked(hs, &bd, &o->lo_header->loh_hash);
 651                 bkt->lsb_busy++;
 652                 cfs_hash_bd_unlock(hs, &bd, 1);
 653                 return o;
 654         }
 655
 656         lprocfs_counter_incr(s->ls_stats, LU_SS_CACHE_RACE);
 657         cfs_hash_bd_unlock(hs, &bd, 1);
 658         lu_object_free(env, o);
 659         return shadow;
 660 }
 661
 662 /**
 663  * Much like lu_object_find(), but top level device of object is specifically
 664  * \a dev rather than top level device of the site. This interface allows
 665  * objects of different "stacking" to be created within the same site.
 666  */
 667 struct lu_object *lu_object_find_at(const struct lu_env *env,
 668                                     struct lu_device *dev,
 669                                     const struct lu_fid *f,
 670                                     const struct lu_object_conf *conf)
 671 {
 672         struct lu_site_bkt_data *bkt;
 673         struct lu_object        *obj;
 674         cfs_waitlink_t           wait;
 675
 676         while (1) {
 677                 obj = lu_object_find_try(env, dev, f, conf, &wait);
 678                 if (obj != ERR_PTR(-EAGAIN))
 679                         return obj;
 680                 /*
 681                  * lu_object_find_try() already added waiter into the
 682                  * wait queue.
 683                  */
 684                 cfs_waitq_wait(&wait, CFS_TASK_UNINT);
 685                 bkt = lu_site_bkt_from_fid(dev->ld_site, (void *)f);
 686                 cfs_waitq_del(&bkt->lsb_marche_funebre, &wait);
 687         }
 688 }
 689 EXPORT_SYMBOL(lu_object_find_at);
 690
 691 /**
 692  * Find object with given fid, and return its slice belonging to given device.
 693  */
 694 struct lu_object *lu_object_find_slice(const struct lu_env *env,
 695                                        struct lu_device *dev,
 696                                        const struct lu_fid *f,
 697                                        const struct lu_object_conf *conf)
 698 {
 699         struct lu_object *top;
 700         struct lu_object *obj;
 701
 702         top = lu_object_find(env, dev, f, conf);
 703         if (!IS_ERR(top)) {
 704                 obj = lu_object_locate(top->lo_header, dev->ld_type);
 705                 if (obj == NULL)
 706                         lu_object_put(env, top);
 707         } else
 708                 obj = top;
 709         return obj;
 710 }
 711 EXPORT_SYMBOL(lu_object_find_slice);
 712
 713 /**
 714  * Global list of all device types.
 715  */
 716 static CFS_LIST_HEAD(lu_device_types);
 717
 718 int lu_device_type_init(struct lu_device_type *ldt)
 719 {
 720         int result = 0;
 721
 722         CFS_INIT_LIST_HEAD(&ldt->ldt_linkage);
 723         if (ldt->ldt_ops->ldto_init)
 724                 result = ldt->ldt_ops->ldto_init(ldt);
 725         if (result == 0)
 726                 cfs_list_add(&ldt->ldt_linkage, &lu_device_types);
 727         return result;
 728 }
 729 EXPORT_SYMBOL(lu_device_type_init);
 730
 731 void lu_device_type_fini(struct lu_device_type *ldt)
 732 {
 733         cfs_list_del_init(&ldt->ldt_linkage);
 734         if (ldt->ldt_ops->ldto_fini)
 735                 ldt->ldt_ops->ldto_fini(ldt);
 736 }
 737 EXPORT_SYMBOL(lu_device_type_fini);
 738
 739 void lu_types_stop(void)
 740 {
 741         struct lu_device_type *ldt;
 742
 743         cfs_list_for_each_entry(ldt, &lu_device_types, ldt_linkage) {
 744                 if (ldt->ldt_device_nr == 0 && ldt->ldt_ops->ldto_stop)
 745                         ldt->ldt_ops->ldto_stop(ldt);
 746         }
 747 }
 748 EXPORT_SYMBOL(lu_types_stop);
 749
 750 /**
 751  * Global list of all sites on this node
 752  */
 753 static CFS_LIST_HEAD(lu_sites);
 754 static DEFINE_MUTEX(lu_sites_guard);
 755
 756 /**
 757  * Global environment used by site shrinker.
 758  */
 759 static struct lu_env lu_shrink_env;
 760
 761 struct lu_site_print_arg {
 762         struct lu_env   *lsp_env;
 763         void            *lsp_cookie;
 764         lu_printer_t     lsp_printer;
 765 };
 766
 767 static int
 768 lu_site_obj_print(cfs_hash_t *hs, cfs_hash_bd_t *bd,
 769                   cfs_hlist_node_t *hnode, void *data)
 770 {
 771         struct lu_site_print_arg *arg = (struct lu_site_print_arg *)data;
 772         struct lu_object_header  *h;
 773
 774         h = cfs_hlist_entry(hnode, struct lu_object_header, loh_hash);
 775         if (!cfs_list_empty(&h->loh_layers)) {
 776                 const struct lu_object *o;
 777
 778                 o = lu_object_top(h);
 779                 lu_object_print(arg->lsp_env, arg->lsp_cookie,
 780                                 arg->lsp_printer, o);
 781         } else {
 782                 lu_object_header_print(arg->lsp_env, arg->lsp_cookie,
 783                                        arg->lsp_printer, h);
 784         }
 785         return 0;
 786 }
 787
 788 /**
 789  * Print all objects in \a s.
 790  */
 791 void lu_site_print(const struct lu_env *env, struct lu_site *s, void *cookie,
 792                    lu_printer_t printer)
 793 {
 794         struct lu_site_print_arg arg = {
 795                 .lsp_env     = (struct lu_env *)env,
 796                 .lsp_cookie  = cookie,
 797                 .lsp_printer = printer,
 798         };
 799
 800         cfs_hash_for_each(s->ls_obj_hash, lu_site_obj_print, &arg);
 801 }
 802 EXPORT_SYMBOL(lu_site_print);
 803
 804 enum {
 805         LU_CACHE_PERCENT_MAX     = 50,
 806         LU_CACHE_PERCENT_DEFAULT = 20
 807 };
 808
 809 static unsigned int lu_cache_percent = LU_CACHE_PERCENT_DEFAULT;
 810 CFS_MODULE_PARM(lu_cache_percent, "i", int, 0644,
 811                 "Percentage of memory to be used as lu_object cache");
 812
 813 /**
 814  * Return desired hash table order.
 815  */
 816 static int lu_htable_order(void)
 817 {
 818         unsigned long cache_size;
 819         int bits;
 820
 821         /*
 822          * Calculate hash table size, assuming that we want reasonable
 823          * performance when 20% of total memory is occupied by cache of
 824          * lu_objects.
 825          *
 826          * Size of lu_object is (arbitrary) taken as 1K (together with inode).
 827          */
 828         cache_size = cfs_num_physpages;
 829
 830 #if BITS_PER_LONG == 32
 831         /* limit hashtable size for lowmem systems to low RAM */
 832         if (cache_size > 1 << (30 - CFS_PAGE_SHIFT))
 833                 cache_size = 1 << (30 - CFS_PAGE_SHIFT) * 3 / 4;
 834 #endif
 835
 836         /* clear off unreasonable cache setting. */
 837         if (lu_cache_percent == 0 || lu_cache_percent > LU_CACHE_PERCENT_MAX) {
 838                 CWARN("obdclass: invalid lu_cache_percent: %u, it must be in"
 839                       " the range of (0, %u]. Will use default value: %u.\n",
 840                       lu_cache_percent, LU_CACHE_PERCENT_MAX,
 841                       LU_CACHE_PERCENT_DEFAULT);
 842
 843                 lu_cache_percent = LU_CACHE_PERCENT_DEFAULT;
 844         }
 845         cache_size = cache_size / 100 * lu_cache_percent *
 846                 (CFS_PAGE_SIZE / 1024);
 847
 848         for (bits = 1; (1 << bits) < cache_size; ++bits) {
 849                 ;
 850         }
 851         return bits;
 852 }
 853
 854 static unsigned lu_obj_hop_hash(cfs_hash_t *hs,
 855                                 const void *key, unsigned mask)
 856 {
 857         struct lu_fid  *fid = (struct lu_fid *)key;
 858         __u32           hash;
 859
 860         hash = fid_flatten32(fid);
 861         hash += (hash >> 4) + (hash << 12); /* mixing oid and seq */
 862         hash = cfs_hash_long(hash, hs->hs_bkt_bits);
 863
 864         /* give me another random factor */
 865         hash -= cfs_hash_long((unsigned long)hs, fid_oid(fid) % 11 + 3);
 866
 867         hash <<= hs->hs_cur_bits - hs->hs_bkt_bits;
 868         hash |= (fid_seq(fid) + fid_oid(fid)) & (CFS_HASH_NBKT(hs) - 1);
 869
 870         return hash & mask;
 871 }
 872
 873 static void *lu_obj_hop_object(cfs_hlist_node_t *hnode)
 874 {
 875         return cfs_hlist_entry(hnode, struct lu_object_header, loh_hash);
 876 }
 877
 878 static void *lu_obj_hop_key(cfs_hlist_node_t *hnode)
 879 {
 880         struct lu_object_header *h;
 881
 882         h = cfs_hlist_entry(hnode, struct lu_object_header, loh_hash);
 883         return &h->loh_fid;
 884 }
 885
 886 static int lu_obj_hop_keycmp(const void *key, cfs_hlist_node_t *hnode)
 887 {
 888         struct lu_object_header *h;
 889
 890         h = cfs_hlist_entry(hnode, struct lu_object_header, loh_hash);
 891         return lu_fid_eq(&h->loh_fid, (struct lu_fid *)key);
 892 }
 893
 894 static void lu_obj_hop_get(cfs_hash_t *hs, cfs_hlist_node_t *hnode)
 895 {
 896         struct lu_object_header *h;
 897
 898         h = cfs_hlist_entry(hnode, struct lu_object_header, loh_hash);
 899         if (cfs_atomic_add_return(1, &h->loh_ref) == 1) {
 900                 struct lu_site_bkt_data *bkt;
 901                 cfs_hash_bd_t            bd;
 902
 903                 cfs_hash_bd_get(hs, &h->loh_fid, &bd);
 904                 bkt = cfs_hash_bd_extra_get(hs, &bd);
 905                 bkt->lsb_busy++;
 906         }
 907 }
 908
 909 static void lu_obj_hop_put_locked(cfs_hash_t *hs, cfs_hlist_node_t *hnode)
 910 {
 911         LBUG(); /* we should never called it */
 912 }
 913
 914 cfs_hash_ops_t lu_site_hash_ops = {
 915         .hs_hash        = lu_obj_hop_hash,
 916         .hs_key         = lu_obj_hop_key,
 917         .hs_keycmp      = lu_obj_hop_keycmp,
 918         .hs_object      = lu_obj_hop_object,
 919         .hs_get         = lu_obj_hop_get,
 920         .hs_put_locked  = lu_obj_hop_put_locked,
 921 };
 922
 923 void lu_dev_add_linkage(struct lu_site *s, struct lu_device *d)
 924 {
 925         spin_lock(&s->ls_ld_lock);
 926         if (cfs_list_empty(&d->ld_linkage))
 927                 cfs_list_add(&d->ld_linkage, &s->ls_ld_linkage);
 928         spin_unlock(&s->ls_ld_lock);
 929 }
 930 EXPORT_SYMBOL(lu_dev_add_linkage);
 931
 932 void lu_dev_del_linkage(struct lu_site *s, struct lu_device *d)
 933 {
 934         spin_lock(&s->ls_ld_lock);
 935         cfs_list_del_init(&d->ld_linkage);
 936         spin_unlock(&s->ls_ld_lock);
 937 }
 938 EXPORT_SYMBOL(lu_dev_del_linkage);
 939
 940 /**
 941  * Initialize site \a s, with \a d as the top level device.
 942  */
 943 #define LU_SITE_BITS_MIN    12
 944 #define LU_SITE_BITS_MAX    24
 945 /**
 946  * total 256 buckets, we don't want too many buckets because:
 947  * - consume too much memory
 948  * - avoid unbalanced LRU list
 949  */
 950 #define LU_SITE_BKT_BITS    8
 951
 952 int lu_site_init(struct lu_site *s, struct lu_device *top)
 953 {
 954         struct lu_site_bkt_data *bkt;
 955         cfs_hash_bd_t bd;
 956         char name[16];
 957         int bits;
 958         int i;
 959         ENTRY;
 960
 961         memset(s, 0, sizeof *s);
 962         bits = lu_htable_order();
 963         snprintf(name, 16, "lu_site_%s", top->ld_type->ldt_name);
 964         for (bits = min(max(LU_SITE_BITS_MIN, bits), LU_SITE_BITS_MAX);
 965              bits >= LU_SITE_BITS_MIN; bits--) {
 966                 s->ls_obj_hash = cfs_hash_create(name, bits, bits,
 967                                                  bits - LU_SITE_BKT_BITS,
 968                                                  sizeof(*bkt), 0, 0,
 969                                                  &lu_site_hash_ops,
 970                                                  CFS_HASH_SPIN_BKTLOCK |
 971                                                  CFS_HASH_NO_ITEMREF |
 972                                                  CFS_HASH_DEPTH |
 973                                                  CFS_HASH_ASSERT_EMPTY);
 974                 if (s->ls_obj_hash != NULL)
 975                         break;
 976         }
 977
 978         if (s->ls_obj_hash == NULL) {
 979                 CERROR("failed to create lu_site hash with bits: %d\n", bits);
 980                 return -ENOMEM;
 981         }
 982
 983         cfs_hash_for_each_bucket(s->ls_obj_hash, &bd, i) {
 984                 bkt = cfs_hash_bd_extra_get(s->ls_obj_hash, &bd);
 985                 CFS_INIT_LIST_HEAD(&bkt->lsb_lru);
 986                 cfs_waitq_init(&bkt->lsb_marche_funebre);
 987         }
 988
 989         s->ls_stats = lprocfs_alloc_stats(LU_SS_LAST_STAT, 0);
 990         if (s->ls_stats == NULL) {
 991                 cfs_hash_putref(s->ls_obj_hash);
 992                 s->ls_obj_hash = NULL;
 993                 return -ENOMEM;
 994         }
 995
 996         lprocfs_counter_init(s->ls_stats, LU_SS_CREATED,
 997                              0, "created", "created");
 998         lprocfs_counter_init(s->ls_stats, LU_SS_CACHE_HIT,
 999                              0, "cache_hit", "cache_hit");
1000         lprocfs_counter_init(s->ls_stats, LU_SS_CACHE_MISS,
1001                              0, "cache_miss", "cache_miss");
1002         lprocfs_counter_init(s->ls_stats, LU_SS_CACHE_RACE,
1003                              0, "cache_race", "cache_race");
1004         lprocfs_counter_init(s->ls_stats, LU_SS_CACHE_DEATH_RACE,
1005                              0, "cache_death_race", "cache_death_race");
1006         lprocfs_counter_init(s->ls_stats, LU_SS_LRU_PURGED,
1007                              0, "lru_purged", "lru_purged");
1008
1009         CFS_INIT_LIST_HEAD(&s->ls_linkage);
1010         s->ls_top_dev = top;
1011         top->ld_site = s;
1012         lu_device_get(top);
1013         lu_ref_add(&top->ld_reference, "site-top", s);
1014
1015         CFS_INIT_LIST_HEAD(&s->ls_ld_linkage);
1016         spin_lock_init(&s->ls_ld_lock);
1017
1018         lu_dev_add_linkage(s, top);
1019
1020         RETURN(0);
1021 }
1022 EXPORT_SYMBOL(lu_site_init);
1023
1024 /**
1025  * Finalize \a s and release its resources.
1026  */
1027 void lu_site_fini(struct lu_site *s)
1028 {
1029         mutex_lock(&lu_sites_guard);
1030         cfs_list_del_init(&s->ls_linkage);
1031         mutex_unlock(&lu_sites_guard);
1032
1033         if (s->ls_obj_hash != NULL) {
1034                 cfs_hash_putref(s->ls_obj_hash);
1035                 s->ls_obj_hash = NULL;
1036         }
1037
1038         if (s->ls_top_dev != NULL) {
1039                 s->ls_top_dev->ld_site = NULL;
1040                 lu_ref_del(&s->ls_top_dev->ld_reference, "site-top", s);
1041                 lu_device_put(s->ls_top_dev);
1042                 s->ls_top_dev = NULL;
1043         }
1044
1045         if (s->ls_stats != NULL)
1046                 lprocfs_free_stats(&s->ls_stats);
1047 }
1048 EXPORT_SYMBOL(lu_site_fini);
1049
1050 /**
1051  * Called when initialization of stack for this site is completed.
1052  */
1053 int lu_site_init_finish(struct lu_site *s)
1054 {
1055         int result;
1056         mutex_lock(&lu_sites_guard);
1057         result = lu_context_refill(&lu_shrink_env.le_ctx);
1058         if (result == 0)
1059                 cfs_list_add(&s->ls_linkage, &lu_sites);
1060         mutex_unlock(&lu_sites_guard);
1061         return result;
1062 }
1063 EXPORT_SYMBOL(lu_site_init_finish);
1064
1065 /**
1066  * Acquire additional reference on device \a d
1067  */
1068 void lu_device_get(struct lu_device *d)
1069 {
1070         cfs_atomic_inc(&d->ld_ref);
1071 }
1072 EXPORT_SYMBOL(lu_device_get);
1073
1074 /**
1075  * Release reference on device \a d.
1076  */
1077 void lu_device_put(struct lu_device *d)
1078 {
1079         LASSERT(cfs_atomic_read(&d->ld_ref) > 0);
1080         cfs_atomic_dec(&d->ld_ref);
1081 }
1082 EXPORT_SYMBOL(lu_device_put);
1083
1084 /**
1085  * Initialize device \a d of type \a t.
1086  */
1087 int lu_device_init(struct lu_device *d, struct lu_device_type *t)
1088 {
1089         if (t->ldt_device_nr++ == 0 && t->ldt_ops->ldto_start != NULL)
1090                 t->ldt_ops->ldto_start(t);
1091         memset(d, 0, sizeof *d);
1092         cfs_atomic_set(&d->ld_ref, 0);
1093         d->ld_type = t;
1094         lu_ref_init(&d->ld_reference);
1095         CFS_INIT_LIST_HEAD(&d->ld_linkage);
1096         return 0;
1097 }
1098 EXPORT_SYMBOL(lu_device_init);
1099
1100 /**
1101  * Finalize device \a d.
1102  */
1103 void lu_device_fini(struct lu_device *d)
1104 {
1105         struct lu_device_type *t;
1106
1107         t = d->ld_type;
1108         if (d->ld_obd != NULL) {
1109                 d->ld_obd->obd_lu_dev = NULL;
1110                 d->ld_obd = NULL;
1111         }
1112
1113         lu_ref_fini(&d->ld_reference);
1114         LASSERTF(cfs_atomic_read(&d->ld_ref) == 0,
1115                  "Refcount is %u\n", cfs_atomic_read(&d->ld_ref));
1116         LASSERT(t->ldt_device_nr > 0);
1117         if (--t->ldt_device_nr == 0 && t->ldt_ops->ldto_stop != NULL)
1118                 t->ldt_ops->ldto_stop(t);
1119 }
1120 EXPORT_SYMBOL(lu_device_fini);
1121
1122 /**
1123  * Initialize object \a o that is part of compound object \a h and was created
1124  * by device \a d.
1125  */
1126 int lu_object_init(struct lu_object *o,
1127                    struct lu_object_header *h, struct lu_device *d)
1128 {
1129         memset(o, 0, sizeof *o);
1130         o->lo_header = h;
1131         o->lo_dev    = d;
1132         lu_device_get(d);
1133         o->lo_dev_ref = lu_ref_add(&d->ld_reference, "lu_object", o);
1134         CFS_INIT_LIST_HEAD(&o->lo_linkage);
1135         return 0;
1136 }
1137 EXPORT_SYMBOL(lu_object_init);
1138
1139 /**
1140  * Finalize object and release its resources.
1141  */
1142 void lu_object_fini(struct lu_object *o)
1143 {
1144         struct lu_device *dev = o->lo_dev;
1145
1146         LASSERT(cfs_list_empty(&o->lo_linkage));
1147
1148         if (dev != NULL) {
1149                 lu_ref_del_at(&dev->ld_reference,
1150                               o->lo_dev_ref , "lu_object", o);
1151                 lu_device_put(dev);
1152                 o->lo_dev = NULL;
1153         }
1154 }
1155 EXPORT_SYMBOL(lu_object_fini);
1156
1157 /**
1158  * Add object \a o as first layer of compound object \a h
1159  *
1160  * This is typically called by the ->ldo_object_alloc() method of top-level
1161  * device.
1162  */
1163 void lu_object_add_top(struct lu_object_header *h, struct lu_object *o)
1164 {
1165         cfs_list_move(&o->lo_linkage, &h->loh_layers);
1166 }
1167 EXPORT_SYMBOL(lu_object_add_top);
1168
1169 /**
1170  * Add object \a o as a layer of compound object, going after \a before.
1171  *
1172  * This is typically called by the ->ldo_object_alloc() method of \a
1173  * before->lo_dev.
1174  */
1175 void lu_object_add(struct lu_object *before, struct lu_object *o)
1176 {
1177         cfs_list_move(&o->lo_linkage, &before->lo_linkage);
1178 }
1179 EXPORT_SYMBOL(lu_object_add);
1180
1181 /**
1182  * Initialize compound object.
1183  */
1184 int lu_object_header_init(struct lu_object_header *h)
1185 {
1186         memset(h, 0, sizeof *h);
1187         cfs_atomic_set(&h->loh_ref, 1);
1188         CFS_INIT_HLIST_NODE(&h->loh_hash);
1189         CFS_INIT_LIST_HEAD(&h->loh_lru);
1190         CFS_INIT_LIST_HEAD(&h->loh_layers);
1191         lu_ref_init(&h->loh_reference);
1192         return 0;
1193 }
1194 EXPORT_SYMBOL(lu_object_header_init);
1195
1196 /**
1197  * Finalize compound object.
1198  */
1199 void lu_object_header_fini(struct lu_object_header *h)
1200 {
1201         LASSERT(cfs_list_empty(&h->loh_layers));
1202         LASSERT(cfs_list_empty(&h->loh_lru));
1203         LASSERT(cfs_hlist_unhashed(&h->loh_hash));
1204         lu_ref_fini(&h->loh_reference);
1205 }
1206 EXPORT_SYMBOL(lu_object_header_fini);
1207
1208 /**
1209  * Given a compound object, find its slice, corresponding to the device type
1210  * \a dtype.
1211  */
1212 struct lu_object *lu_object_locate(struct lu_object_header *h,
1213                                    const struct lu_device_type *dtype)
1214 {
1215         struct lu_object *o;
1216
1217         cfs_list_for_each_entry(o, &h->loh_layers, lo_linkage) {
1218                 if (o->lo_dev->ld_type == dtype)
1219                         return o;
1220         }
1221         return NULL;
1222 }
1223 EXPORT_SYMBOL(lu_object_locate);
1224
1225
1226
1227 /**
1228  * Finalize and free devices in the device stack.
1229  *
1230  * Finalize device stack by purging object cache, and calling
1231  * lu_device_type_operations::ldto_device_fini() and
1232  * lu_device_type_operations::ldto_device_free() on all devices in the stack.
1233  */
1234 void lu_stack_fini(const struct lu_env *env, struct lu_device *top)
1235 {
1236         struct lu_site   *site = top->ld_site;
1237         struct lu_device *scan;
1238         struct lu_device *next;
1239
1240         lu_site_purge(env, site, ~0);
1241         for (scan = top; scan != NULL; scan = next) {
1242                 next = scan->ld_type->ldt_ops->ldto_device_fini(env, scan);
1243                 lu_ref_del(&scan->ld_reference, "lu-stack", &lu_site_init);
1244                 lu_device_put(scan);
1245         }
1246
1247         /* purge again. */
1248         lu_site_purge(env, site, ~0);
1249
1250         for (scan = top; scan != NULL; scan = next) {
1251                 const struct lu_device_type *ldt = scan->ld_type;
1252                 struct obd_type             *type;
1253
1254                 next = ldt->ldt_ops->ldto_device_free(env, scan);
1255                 type = ldt->ldt_obd_type;
1256                 if (type != NULL) {
1257                         type->typ_refcnt--;
1258                         class_put_type(type);
1259                 }
1260         }
1261 }
1262 EXPORT_SYMBOL(lu_stack_fini);
1263
1264 enum {
1265         /**
1266          * Maximal number of tld slots.
1267          */
1268         LU_CONTEXT_KEY_NR = 40
1269 };
1270
1271 static struct lu_context_key *lu_keys[LU_CONTEXT_KEY_NR] = { NULL, };
1272
1273 static DEFINE_SPINLOCK(lu_keys_guard);
1274
1275 /**
1276  * Global counter incremented whenever key is registered, unregistered,
1277  * revived or quiesced. This is used to void unnecessary calls to
1278  * lu_context_refill(). No locking is provided, as initialization and shutdown
1279  * are supposed to be externally serialized.
1280  */
1281 static unsigned key_set_version = 0;
1282
1283 /**
1284  * Register new key.
1285  */
1286 int lu_context_key_register(struct lu_context_key *key)
1287 {
1288         int result;
1289         int i;
1290
1291         LASSERT(key->lct_init != NULL);
1292         LASSERT(key->lct_fini != NULL);
1293         LASSERT(key->lct_tags != 0);
1294         LASSERT(key->lct_owner != NULL);
1295
1296         result = -ENFILE;
1297         spin_lock(&lu_keys_guard);
1298         for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
1299                 if (lu_keys[i] == NULL) {
1300                         key->lct_index = i;
1301                         cfs_atomic_set(&key->lct_used, 1);
1302                         lu_keys[i] = key;
1303                         lu_ref_init(&key->lct_reference);
1304                         result = 0;
1305                         ++key_set_version;
1306                         break;
1307                 }
1308         }
1309         spin_unlock(&lu_keys_guard);
1310         return result;
1311 }
1312 EXPORT_SYMBOL(lu_context_key_register);
1313
1314 static void key_fini(struct lu_context *ctx, int index)
1315 {
1316         if (ctx->lc_value != NULL && ctx->lc_value[index] != NULL) {
1317                 struct lu_context_key *key;
1318
1319                 key = lu_keys[index];
1320                 LASSERT(key != NULL);
1321                 LASSERT(key->lct_fini != NULL);
1322                 LASSERT(cfs_atomic_read(&key->lct_used) > 1);
1323
1324                 key->lct_fini(ctx, key, ctx->lc_value[index]);
1325                 lu_ref_del(&key->lct_reference, "ctx", ctx);
1326                 cfs_atomic_dec(&key->lct_used);
1327
1328                 LASSERT(key->lct_owner != NULL);
1329                 if ((ctx->lc_tags & LCT_NOREF) == 0) {
1330                         LINVRNT(cfs_module_refcount(key->lct_owner) > 0);
1331                         cfs_module_put(key->lct_owner);
1332                 }
1333                 ctx->lc_value[index] = NULL;
1334         }
1335 }
1336
1337 /**
1338  * Deregister key.
1339  */
1340 void lu_context_key_degister(struct lu_context_key *key)
1341 {
1342         LASSERT(cfs_atomic_read(&key->lct_used) >= 1);
1343         LINVRNT(0 <= key->lct_index && key->lct_index < ARRAY_SIZE(lu_keys));
1344
1345         lu_context_key_quiesce(key);
1346
1347         ++key_set_version;
1348         spin_lock(&lu_keys_guard);
1349         key_fini(&lu_shrink_env.le_ctx, key->lct_index);
1350         if (lu_keys[key->lct_index]) {
1351                 lu_keys[key->lct_index] = NULL;
1352                 lu_ref_fini(&key->lct_reference);
1353         }
1354         spin_unlock(&lu_keys_guard);
1355
1356         LASSERTF(cfs_atomic_read(&key->lct_used) == 1,
1357                  "key has instances: %d\n",
1358                  cfs_atomic_read(&key->lct_used));
1359 }
1360 EXPORT_SYMBOL(lu_context_key_degister);
1361
1362 /**
1363  * Register a number of keys. This has to be called after all keys have been
1364  * initialized by a call to LU_CONTEXT_KEY_INIT().
1365  */
1366 int lu_context_key_register_many(struct lu_context_key *k, ...)
1367 {
1368         struct lu_context_key *key = k;
1369         va_list args;
1370         int result;
1371
1372         va_start(args, k);
1373         do {
1374                 result = lu_context_key_register(key);
1375                 if (result)
1376                         break;
1377                 key = va_arg(args, struct lu_context_key *);
1378         } while (key != NULL);
1379         va_end(args);
1380
1381         if (result != 0) {
1382                 va_start(args, k);
1383                 while (k != key) {
1384                         lu_context_key_degister(k);
1385                         k = va_arg(args, struct lu_context_key *);
1386                 }
1387                 va_end(args);
1388         }
1389
1390         return result;
1391 }
1392 EXPORT_SYMBOL(lu_context_key_register_many);
1393
1394 /**
1395  * De-register a number of keys. This is a dual to
1396  * lu_context_key_register_many().
1397  */
1398 void lu_context_key_degister_many(struct lu_context_key *k, ...)
1399 {
1400         va_list args;
1401
1402         va_start(args, k);
1403         do {
1404                 lu_context_key_degister(k);
1405                 k = va_arg(args, struct lu_context_key*);
1406         } while (k != NULL);
1407         va_end(args);
1408 }
1409 EXPORT_SYMBOL(lu_context_key_degister_many);
1410
1411 /**
1412  * Revive a number of keys.
1413  */
1414 void lu_context_key_revive_many(struct lu_context_key *k, ...)
1415 {
1416         va_list args;
1417
1418         va_start(args, k);
1419         do {
1420                 lu_context_key_revive(k);
1421                 k = va_arg(args, struct lu_context_key*);
1422         } while (k != NULL);
1423         va_end(args);
1424 }
1425 EXPORT_SYMBOL(lu_context_key_revive_many);
1426
1427 /**
1428  * Quiescent a number of keys.
1429  */
1430 void lu_context_key_quiesce_many(struct lu_context_key *k, ...)
1431 {
1432         va_list args;
1433
1434         va_start(args, k);
1435         do {
1436                 lu_context_key_quiesce(k);
1437                 k = va_arg(args, struct lu_context_key*);
1438         } while (k != NULL);
1439         va_end(args);
1440 }
1441 EXPORT_SYMBOL(lu_context_key_quiesce_many);
1442
1443 /**
1444  * Return value associated with key \a key in context \a ctx.
1445  */
1446 void *lu_context_key_get(const struct lu_context *ctx,
1447                          const struct lu_context_key *key)
1448 {
1449         LINVRNT(ctx->lc_state == LCS_ENTERED);
1450         LINVRNT(0 <= key->lct_index && key->lct_index < ARRAY_SIZE(lu_keys));
1451         LASSERT(lu_keys[key->lct_index] == key);
1452         return ctx->lc_value[key->lct_index];
1453 }
1454 EXPORT_SYMBOL(lu_context_key_get);
1455
1456 /**
1457  * List of remembered contexts. XXX document me.
1458  */
1459 static CFS_LIST_HEAD(lu_context_remembered);
1460
1461 /**
1462  * Destroy \a key in all remembered contexts. This is used to destroy key
1463  * values in "shared" contexts (like service threads), when a module owning
1464  * the key is about to be unloaded.
1465  */
1466 void lu_context_key_quiesce(struct lu_context_key *key)
1467 {
1468         struct lu_context *ctx;
1469         extern unsigned cl_env_cache_purge(unsigned nr);
1470
1471         if (!(key->lct_tags & LCT_QUIESCENT)) {
1472                 /*
1473                  * XXX layering violation.
1474                  */
1475                 cl_env_cache_purge(~0);
1476                 key->lct_tags |= LCT_QUIESCENT;
1477                 /*
1478                  * XXX memory barrier has to go here.
1479                  */
1480                 spin_lock(&lu_keys_guard);
1481                 cfs_list_for_each_entry(ctx, &lu_context_remembered,
1482                                         lc_remember)
1483                         key_fini(ctx, key->lct_index);
1484                 spin_unlock(&lu_keys_guard);
1485                 ++key_set_version;
1486         }
1487 }
1488 EXPORT_SYMBOL(lu_context_key_quiesce);
1489
1490 void lu_context_key_revive(struct lu_context_key *key)
1491 {
1492         key->lct_tags &= ~LCT_QUIESCENT;
1493         ++key_set_version;
1494 }
1495 EXPORT_SYMBOL(lu_context_key_revive);
1496
1497 static void keys_fini(struct lu_context *ctx)
1498 {
1499         int     i;
1500
1501         if (ctx->lc_value == NULL)
1502                 return;
1503
1504         for (i = 0; i < ARRAY_SIZE(lu_keys); ++i)
1505                 key_fini(ctx, i);
1506
1507         OBD_FREE(ctx->lc_value, ARRAY_SIZE(lu_keys) * sizeof ctx->lc_value[0]);
1508         ctx->lc_value = NULL;
1509 }
1510
1511 static int keys_fill(struct lu_context *ctx)
1512 {
1513         int i;
1514
1515         LINVRNT(ctx->lc_value != NULL);
1516         for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
1517                 struct lu_context_key *key;
1518
1519                 key = lu_keys[i];
1520                 if (ctx->lc_value[i] == NULL && key != NULL &&
1521                     (key->lct_tags & ctx->lc_tags) &&
1522                     /*
1523                      * Don't create values for a LCT_QUIESCENT key, as this
1524                      * will pin module owning a key.
1525                      */
1526                     !(key->lct_tags & LCT_QUIESCENT)) {
1527                         void *value;
1528
1529                         LINVRNT(key->lct_init != NULL);
1530                         LINVRNT(key->lct_index == i);
1531
1532                         value = key->lct_init(ctx, key);
1533                         if (unlikely(IS_ERR(value)))
1534                                 return PTR_ERR(value);
1535
1536                         LASSERT(key->lct_owner != NULL);
1537                         if (!(ctx->lc_tags & LCT_NOREF))
1538                                 cfs_try_module_get(key->lct_owner);
1539                         lu_ref_add_atomic(&key->lct_reference, "ctx", ctx);
1540                         cfs_atomic_inc(&key->lct_used);
1541                         /*
1542                          * This is the only place in the code, where an
1543                          * element of ctx->lc_value[] array is set to non-NULL
1544                          * value.
1545                          */
1546                         ctx->lc_value[i] = value;
1547                         if (key->lct_exit != NULL)
1548                                 ctx->lc_tags |= LCT_HAS_EXIT;
1549                 }
1550                 ctx->lc_version = key_set_version;
1551         }
1552         return 0;
1553 }
1554
1555 static int keys_init(struct lu_context *ctx)
1556 {
1557         OBD_ALLOC(ctx->lc_value, ARRAY_SIZE(lu_keys) * sizeof ctx->lc_value[0]);
1558         if (likely(ctx->lc_value != NULL))
1559                 return keys_fill(ctx);
1560
1561         return -ENOMEM;
1562 }
1563
1564 /**
1565  * Initialize context data-structure. Create values for all keys.
1566  */
1567 int lu_context_init(struct lu_context *ctx, __u32 tags)
1568 {
1569         int     rc;
1570
1571         memset(ctx, 0, sizeof *ctx);
1572         ctx->lc_state = LCS_INITIALIZED;
1573         ctx->lc_tags = tags;
1574         if (tags & LCT_REMEMBER) {
1575                 spin_lock(&lu_keys_guard);
1576                 cfs_list_add(&ctx->lc_remember, &lu_context_remembered);
1577                 spin_unlock(&lu_keys_guard);
1578         } else {
1579                 CFS_INIT_LIST_HEAD(&ctx->lc_remember);
1580         }
1581
1582         rc = keys_init(ctx);
1583         if (rc != 0)
1584                 lu_context_fini(ctx);
1585
1586         return rc;
1587 }
1588 EXPORT_SYMBOL(lu_context_init);
1589
1590 /**
1591  * Finalize context data-structure. Destroy key values.
1592  */
1593 void lu_context_fini(struct lu_context *ctx)
1594 {
1595         LINVRNT(ctx->lc_state == LCS_INITIALIZED || ctx->lc_state == LCS_LEFT);
1596         ctx->lc_state = LCS_FINALIZED;
1597
1598         if ((ctx->lc_tags & LCT_REMEMBER) == 0) {
1599                 LASSERT(cfs_list_empty(&ctx->lc_remember));
1600                 keys_fini(ctx);
1601
1602         } else { /* could race with key degister */
1603                 spin_lock(&lu_keys_guard);
1604                 keys_fini(ctx);
1605                 cfs_list_del_init(&ctx->lc_remember);
1606                 spin_unlock(&lu_keys_guard);
1607         }
1608 }
1609 EXPORT_SYMBOL(lu_context_fini);
1610
1611 /**
1612  * Called before entering context.
1613  */
1614 void lu_context_enter(struct lu_context *ctx)
1615 {
1616         LINVRNT(ctx->lc_state == LCS_INITIALIZED || ctx->lc_state == LCS_LEFT);
1617         ctx->lc_state = LCS_ENTERED;
1618 }
1619 EXPORT_SYMBOL(lu_context_enter);
1620
1621 /**
1622  * Called after exiting from \a ctx
1623  */
1624 void lu_context_exit(struct lu_context *ctx)
1625 {
1626         int i;
1627
1628         LINVRNT(ctx->lc_state == LCS_ENTERED);
1629         ctx->lc_state = LCS_LEFT;
1630         if (ctx->lc_tags & LCT_HAS_EXIT && ctx->lc_value != NULL) {
1631                 for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
1632                         if (ctx->lc_value[i] != NULL) {
1633                                 struct lu_context_key *key;
1634
1635                                 key = lu_keys[i];
1636                                 LASSERT(key != NULL);
1637                                 if (key->lct_exit != NULL)
1638                                         key->lct_exit(ctx,
1639                                                       key, ctx->lc_value[i]);
1640                         }
1641                 }
1642         }
1643 }
1644 EXPORT_SYMBOL(lu_context_exit);
1645
1646 /**
1647  * Allocate for context all missing keys that were registered after context
1648  * creation. key_set_version is only changed in rare cases when modules
1649  * are loaded and removed.
1650  */
1651 int lu_context_refill(struct lu_context *ctx)
1652 {
1653         return likely(ctx->lc_version == key_set_version) ? 0 : keys_fill(ctx);
1654 }
1655 EXPORT_SYMBOL(lu_context_refill);
1656
1657 /**
1658  * lu_ctx_tags/lu_ses_tags will be updated if there are new types of
1659  * obd being added. Currently, this is only used on client side, specifically
1660  * for echo device client, for other stack (like ptlrpc threads), context are
1661  * predefined when the lu_device type are registered, during the module probe
1662  * phase.
1663  */
1664 __u32 lu_context_tags_default = 0;
1665 __u32 lu_session_tags_default = 0;
1666
1667 void lu_context_tags_update(__u32 tags)
1668 {
1669         spin_lock(&lu_keys_guard);
1670         lu_context_tags_default |= tags;
1671         key_set_version++;
1672         spin_unlock(&lu_keys_guard);
1673 }
1674 EXPORT_SYMBOL(lu_context_tags_update);
1675
1676 void lu_context_tags_clear(__u32 tags)
1677 {
1678         spin_lock(&lu_keys_guard);
1679         lu_context_tags_default &= ~tags;
1680         key_set_version++;
1681         spin_unlock(&lu_keys_guard);
1682 }
1683 EXPORT_SYMBOL(lu_context_tags_clear);
1684
1685 void lu_session_tags_update(__u32 tags)
1686 {
1687         spin_lock(&lu_keys_guard);
1688         lu_session_tags_default |= tags;
1689         key_set_version++;
1690         spin_unlock(&lu_keys_guard);
1691 }
1692 EXPORT_SYMBOL(lu_session_tags_update);
1693
1694 void lu_session_tags_clear(__u32 tags)
1695 {
1696         spin_lock(&lu_keys_guard);
1697         lu_session_tags_default &= ~tags;
1698         key_set_version++;
1699         spin_unlock(&lu_keys_guard);
1700 }
1701 EXPORT_SYMBOL(lu_session_tags_clear);
1702
1703 int lu_env_init(struct lu_env *env, __u32 tags)
1704 {
1705         int result;
1706
1707         env->le_ses = NULL;
1708         result = lu_context_init(&env->le_ctx, tags);
1709         if (likely(result == 0))
1710                 lu_context_enter(&env->le_ctx);
1711         return result;
1712 }
1713 EXPORT_SYMBOL(lu_env_init);
1714
1715 void lu_env_fini(struct lu_env *env)
1716 {
1717         lu_context_exit(&env->le_ctx);
1718         lu_context_fini(&env->le_ctx);
1719         env->le_ses = NULL;
1720 }
1721 EXPORT_SYMBOL(lu_env_fini);
1722
1723 int lu_env_refill(struct lu_env *env)
1724 {
1725         int result;
1726
1727         result = lu_context_refill(&env->le_ctx);
1728         if (result == 0 && env->le_ses != NULL)
1729                 result = lu_context_refill(env->le_ses);
1730         return result;
1731 }
1732 EXPORT_SYMBOL(lu_env_refill);
1733
1734 /**
1735  * Currently, this API will only be used by echo client.
1736  * Because echo client and normal lustre client will share
1737  * same cl_env cache. So echo client needs to refresh
1738  * the env context after it get one from the cache, especially
1739  * when normal client and echo client co-exist in the same client.
1740  */
1741 int lu_env_refill_by_tags(struct lu_env *env, __u32 ctags,
1742                           __u32 stags)
1743 {
1744         int    result;
1745
1746         if ((env->le_ctx.lc_tags & ctags) != ctags) {
1747                 env->le_ctx.lc_version = 0;
1748                 env->le_ctx.lc_tags |= ctags;
1749         }
1750
1751         if (env->le_ses && (env->le_ses->lc_tags & stags) != stags) {
1752                 env->le_ses->lc_version = 0;
1753                 env->le_ses->lc_tags |= stags;
1754         }
1755
1756         result = lu_env_refill(env);
1757
1758         return result;
1759 }
1760 EXPORT_SYMBOL(lu_env_refill_by_tags);
1761
1762 static struct cfs_shrinker *lu_site_shrinker = NULL;
1763
1764 typedef struct lu_site_stats{
1765         unsigned        lss_populated;
1766         unsigned        lss_max_search;
1767         unsigned        lss_total;
1768         unsigned        lss_busy;
1769 } lu_site_stats_t;
1770
1771 static void lu_site_stats_get(cfs_hash_t *hs,
1772                               lu_site_stats_t *stats, int populated)
1773 {
1774         cfs_hash_bd_t bd;
1775         int           i;
1776
1777         cfs_hash_for_each_bucket(hs, &bd, i) {
1778                 struct lu_site_bkt_data *bkt = cfs_hash_bd_extra_get(hs, &bd);
1779                 cfs_hlist_head_t        *hhead;
1780
1781                 cfs_hash_bd_lock(hs, &bd, 1);
1782                 stats->lss_busy  += bkt->lsb_busy;
1783                 stats->lss_total += cfs_hash_bd_count_get(&bd);
1784                 stats->lss_max_search = max((int)stats->lss_max_search,
1785                                             cfs_hash_bd_depmax_get(&bd));
1786                 if (!populated) {
1787                         cfs_hash_bd_unlock(hs, &bd, 1);
1788                         continue;
1789                 }
1790
1791                 cfs_hash_bd_for_each_hlist(hs, &bd, hhead) {
1792                         if (!cfs_hlist_empty(hhead))
1793                                 stats->lss_populated++;
1794                 }
1795                 cfs_hash_bd_unlock(hs, &bd, 1);
1796         }
1797 }
1798
1799 #ifdef __KERNEL__
1800
1801 /*
1802  * There exists a potential lock inversion deadlock scenario when using
1803  * Lustre on top of ZFS. This occurs between one of ZFS's
1804  * buf_hash_table.ht_lock's, and Lustre's lu_sites_guard lock. Essentially,
1805  * thread A will take the lu_sites_guard lock and sleep on the ht_lock,
1806  * while thread B will take the ht_lock and sleep on the lu_sites_guard
1807  * lock. Obviously neither thread will wake and drop their respective hold
1808  * on their lock.
1809  *
1810  * To prevent this from happening we must ensure the lu_sites_guard lock is
1811  * not taken while down this code path. ZFS reliably does not set the
1812  * __GFP_FS bit in its code paths, so this can be used to determine if it
1813  * is safe to take the lu_sites_guard lock.
1814  *
1815  * Ideally we should accurately return the remaining number of cached
1816  * objects without taking the  lu_sites_guard lock, but this is not
1817  * possible in the current implementation.
1818  */
1819 static int lu_cache_shrink(SHRINKER_ARGS(sc, nr_to_scan, gfp_mask))
1820 {
1821         lu_site_stats_t stats;
1822         struct lu_site *s;
1823         struct lu_site *tmp;
1824         int cached = 0;
1825         int remain = shrink_param(sc, nr_to_scan);
1826         CFS_LIST_HEAD(splice);
1827
1828         if (!(shrink_param(sc, gfp_mask) & __GFP_FS)) {
1829                 if (remain != 0)
1830                         return -1;
1831                 else
1832                         /* We must not take the lu_sites_guard lock when
1833                          * __GFP_FS is *not* set because of the deadlock
1834                          * possibility detailed above. Additionally,
1835                          * since we cannot determine the number of
1836                          * objects in the cache without taking this
1837                          * lock, we're in a particularly tough spot. As
1838                          * a result, we'll just lie and say our cache is
1839                          * empty. This _should_ be ok, as we can't
1840                          * reclaim objects when __GFP_FS is *not* set
1841                          * anyways.
1842                          */
1843                         return 0;
1844         }
1845
1846         CDEBUG(D_INODE, "Shrink %d objects\n", remain);
1847
1848         mutex_lock(&lu_sites_guard);
1849         cfs_list_for_each_entry_safe(s, tmp, &lu_sites, ls_linkage) {
1850                 if (shrink_param(sc, nr_to_scan) != 0) {
1851                         remain = lu_site_purge(&lu_shrink_env, s, remain);
1852                         /*
1853                          * Move just shrunk site to the tail of site list to
1854                          * assure shrinking fairness.
1855                          */
1856                         cfs_list_move_tail(&s->ls_linkage, &splice);
1857                 }
1858
1859                 memset(&stats, 0, sizeof(stats));
1860                 lu_site_stats_get(s->ls_obj_hash, &stats, 0);
1861                 cached += stats.lss_total - stats.lss_busy;
1862                 if (shrink_param(sc, nr_to_scan) && remain <= 0)
1863                         break;
1864         }
1865         cfs_list_splice(&splice, lu_sites.prev);
1866         mutex_unlock(&lu_sites_guard);
1867
1868         cached = (cached / 100) * sysctl_vfs_cache_pressure;
1869         if (shrink_param(sc, nr_to_scan) == 0)
1870                 CDEBUG(D_INODE, "%d objects cached\n", cached);
1871         return cached;
1872 }
1873
1874 /*
1875  * Debugging stuff.
1876  */
1877
1878 /**
1879  * Environment to be used in debugger, contains all tags.
1880  */
1881 struct lu_env lu_debugging_env;
1882
1883 /**
1884  * Debugging printer function using printk().
1885  */
1886 int lu_printk_printer(const struct lu_env *env,
1887                       void *unused, const char *format, ...)
1888 {
1889         va_list args;
1890
1891         va_start(args, format);
1892         vprintk(format, args);
1893         va_end(args);
1894         return 0;
1895 }
1896
1897 void lu_debugging_setup(void)
1898 {
1899         lu_env_init(&lu_debugging_env, ~0);
1900 }
1901
1902 void lu_context_keys_dump(void)
1903 {
1904         int i;
1905
1906         for (i = 0; i < ARRAY_SIZE(lu_keys); ++i) {
1907                 struct lu_context_key *key;
1908
1909                 key = lu_keys[i];
1910                 if (key != NULL) {
1911                         CERROR("[%d]: %p %x (%p,%p,%p) %d %d \"%s\"@%p\n",
1912                                i, key, key->lct_tags,
1913                                key->lct_init, key->lct_fini, key->lct_exit,
1914                                key->lct_index, cfs_atomic_read(&key->lct_used),
1915                                key->lct_owner ? key->lct_owner->name : "",
1916                                key->lct_owner);
1917                         lu_ref_print(&key->lct_reference);
1918                 }
1919         }
1920 }
1921 EXPORT_SYMBOL(lu_context_keys_dump);
1922 #else  /* !__KERNEL__ */
1923 static int lu_cache_shrink(int nr, unsigned int gfp_mask)
1924 {
1925         return 0;
1926 }
1927 #endif /* __KERNEL__ */
1928
1929 int  cl_global_init(void);
1930 void cl_global_fini(void);
1931 int  lu_ref_global_init(void);
1932 void lu_ref_global_fini(void);
1933
1934 int dt_global_init(void);
1935 void dt_global_fini(void);
1936
1937 int llo_global_init(void);
1938 void llo_global_fini(void);
1939
1940 /* context key constructor/destructor: lu_ucred_key_init, lu_ucred_key_fini */
1941 LU_KEY_INIT_FINI(lu_ucred, struct lu_ucred);
1942
1943 static struct lu_context_key lu_ucred_key = {
1944         .lct_tags = LCT_SESSION,
1945         .lct_init = lu_ucred_key_init,
1946         .lct_fini = lu_ucred_key_fini
1947 };
1948
1949 /**
1950  * Get ucred key if session exists and ucred key is allocated on it.
1951  * Return NULL otherwise.
1952  */
1953 struct lu_ucred *lu_ucred(const struct lu_env *env)
1954 {
1955         if (!env->le_ses)
1956                 return NULL;
1957         return lu_context_key_get(env->le_ses, &lu_ucred_key);
1958 }
1959 EXPORT_SYMBOL(lu_ucred);
1960
1961 /**
1962  * Get ucred key and check if it is properly initialized.
1963  * Return NULL otherwise.
1964  */
1965 struct lu_ucred *lu_ucred_check(const struct lu_env *env)
1966 {
1967         struct lu_ucred *uc = lu_ucred(env);
1968         if (uc && uc->uc_valid != UCRED_OLD && uc->uc_valid != UCRED_NEW)
1969                 return NULL;
1970         return uc;
1971 }
1972 EXPORT_SYMBOL(lu_ucred_check);
1973
1974 /**
1975  * Get ucred key, which must exist and must be properly initialized.
1976  * Assert otherwise.
1977  */
1978 struct lu_ucred *lu_ucred_assert(const struct lu_env *env)
1979 {
1980         struct lu_ucred *uc = lu_ucred_check(env);
1981         LASSERT(uc != NULL);
1982         return uc;
1983 }
1984 EXPORT_SYMBOL(lu_ucred_assert);
1985
1986 /**
1987  * Initialization of global lu_* data.
1988  */
1989 int lu_global_init(void)
1990 {
1991         int result;
1992
1993         CDEBUG(D_INFO, "Lustre LU module (%p).\n", &lu_keys);
1994
1995         result = lu_ref_global_init();
1996         if (result != 0)
1997                 return result;
1998
1999         LU_CONTEXT_KEY_INIT(&lu_global_key);
2000         result = lu_context_key_register(&lu_global_key);
2001         if (result != 0)
2002                 return result;
2003
2004         LU_CONTEXT_KEY_INIT(&lu_ucred_key);
2005         result = lu_context_key_register(&lu_ucred_key);
2006         if (result != 0)
2007                 return result;
2008
2009         /*
2010          * At this level, we don't know what tags are needed, so allocate them
2011          * conservatively. This should not be too bad, because this
2012          * environment is global.
2013          */
2014         mutex_lock(&lu_sites_guard);
2015         result = lu_env_init(&lu_shrink_env, LCT_SHRINKER);
2016         mutex_unlock(&lu_sites_guard);
2017         if (result != 0)
2018                 return result;
2019
2020         /*
2021          * seeks estimation: 3 seeks to read a record from oi, one to read
2022          * inode, one for ea. Unfortunately setting this high value results in
2023          * lu_object/inode cache consuming all the memory.
2024          */
2025         lu_site_shrinker = cfs_set_shrinker(CFS_DEFAULT_SEEKS, lu_cache_shrink);
2026         if (lu_site_shrinker == NULL)
2027                 return -ENOMEM;
2028
2029 #ifdef __KERNEL__
2030         result = dt_global_init();
2031         if (result != 0)
2032                 return result;
2033
2034         result = llo_global_init();
2035         if (result != 0)
2036                 return result;
2037 #endif
2038         result = cl_global_init();
2039
2040         return result;
2041 }
2042
2043 /**
2044  * Dual to lu_global_init().
2045  */
2046 void lu_global_fini(void)
2047 {
2048         cl_global_fini();
2049 #ifdef __KERNEL__
2050         llo_global_fini();
2051         dt_global_fini();
2052 #endif
2053         if (lu_site_shrinker != NULL) {
2054                 cfs_remove_shrinker(lu_site_shrinker);
2055                 lu_site_shrinker = NULL;
2056         }
2057
2058         lu_context_key_degister(&lu_global_key);
2059         lu_context_key_degister(&lu_ucred_key);
2060
2061         /*
2062          * Tear shrinker environment down _after_ de-registering
2063          * lu_global_key, because the latter has a value in the former.
2064          */
2065         mutex_lock(&lu_sites_guard);
2066         lu_env_fini(&lu_shrink_env);
2067         mutex_unlock(&lu_sites_guard);
2068
2069         lu_ref_global_fini();
2070 }
2071
2072 struct lu_buf LU_BUF_NULL = {
2073         .lb_buf = NULL,
2074         .lb_len = 0
2075 };
2076 EXPORT_SYMBOL(LU_BUF_NULL);
2077
2078 static __u32 ls_stats_read(struct lprocfs_stats *stats, int idx)
2079 {
2080 #ifdef LPROCFS
2081         struct lprocfs_counter ret;
2082
2083         lprocfs_stats_collect(stats, idx, &ret);
2084         return (__u32)ret.lc_count;
2085 #else
2086         return 0;
2087 #endif
2088 }
2089
2090 /**
2091  * Output site statistical counters into a buffer. Suitable for
2092  * lprocfs_rd_*()-style functions.
2093  */
2094 int lu_site_stats_print(const struct lu_site *s, char *page, int count)
2095 {
2096         lu_site_stats_t stats;
2097
2098         memset(&stats, 0, sizeof(stats));
2099         lu_site_stats_get(s->ls_obj_hash, &stats, 1);
2100
2101         return snprintf(page, count, "%d/%d %d/%d %d %d %d %d %d %d %d\n",
2102                         stats.lss_busy,
2103                         stats.lss_total,
2104                         stats.lss_populated,
2105                         CFS_HASH_NHLIST(s->ls_obj_hash),
2106                         stats.lss_max_search,
2107                         ls_stats_read(s->ls_stats, LU_SS_CREATED),
2108                         ls_stats_read(s->ls_stats, LU_SS_CACHE_HIT),
2109                         ls_stats_read(s->ls_stats, LU_SS_CACHE_MISS),
2110                         ls_stats_read(s->ls_stats, LU_SS_CACHE_RACE),
2111                         ls_stats_read(s->ls_stats, LU_SS_CACHE_DEATH_RACE),
2112                         ls_stats_read(s->ls_stats, LU_SS_LRU_PURGED));
2113 }
2114 EXPORT_SYMBOL(lu_site_stats_print);
2115
2116 /**
2117  * Helper function to initialize a number of kmem slab caches at once.
2118  */
2119 int lu_kmem_init(struct lu_kmem_descr *caches)
2120 {
2121         int result;
2122         struct lu_kmem_descr *iter = caches;
2123
2124         for (result = 0; iter->ckd_cache != NULL; ++iter) {
2125                 *iter->ckd_cache = cfs_mem_cache_create(iter->ckd_name,
2126                                                         iter->ckd_size,
2127                                                         0, 0);
2128                 if (*iter->ckd_cache == NULL) {
2129                         result = -ENOMEM;
2130                         /* free all previously allocated caches */
2131                         lu_kmem_fini(caches);
2132                         break;
2133                 }
2134         }
2135         return result;
2136 }
2137 EXPORT_SYMBOL(lu_kmem_init);
2138
2139 /**
2140  * Helper function to finalize a number of kmem slab cached at once. Dual to
2141  * lu_kmem_init().
2142  */
2143 void lu_kmem_fini(struct lu_kmem_descr *caches)
2144 {
2145         int rc;
2146
2147         for (; caches->ckd_cache != NULL; ++caches) {
2148                 if (*caches->ckd_cache != NULL) {
2149                         rc = cfs_mem_cache_destroy(*caches->ckd_cache);
2150                         LASSERTF(rc == 0, "couldn't destroy %s slab\n",
2151                                  caches->ckd_name);
2152                         *caches->ckd_cache = NULL;
2153                 }
2154         }
2155 }
2156 EXPORT_SYMBOL(lu_kmem_fini);
2157
2158 /**
2159  * Temporary solution to be able to assign fid in ->do_create()
2160  * till we have fully-functional OST fids
2161  */
2162 void lu_object_assign_fid(const struct lu_env *env, struct lu_object *o,
2163                           const struct lu_fid *fid)
2164 {
2165         struct lu_site          *s = o->lo_dev->ld_site;
2166         struct lu_fid           *old = &o->lo_header->loh_fid;
2167         struct lu_site_bkt_data *bkt;
2168         struct lu_object        *shadow;
2169         cfs_waitlink_t           waiter;
2170         cfs_hash_t              *hs;
2171         cfs_hash_bd_t            bd;
2172         __u64                    version = 0;
2173
2174         LASSERT(fid_is_zero(old));
2175
2176         hs = s->ls_obj_hash;
2177         cfs_hash_bd_get_and_lock(hs, (void *)fid, &bd, 1);
2178         shadow = htable_lookup(s, &bd, fid, &waiter, &version);
2179         /* supposed to be unique */
2180         LASSERT(shadow == NULL);
2181         *old = *fid;
2182         bkt = cfs_hash_bd_extra_get(hs, &bd);
2183         cfs_hash_bd_add_locked(hs, &bd, &o->lo_header->loh_hash);
2184         bkt->lsb_busy++;
2185         cfs_hash_bd_unlock(hs, &bd, 1);
2186 }
2187 EXPORT_SYMBOL(lu_object_assign_fid);
2188
2189 /**
2190  * allocates object with 0 (non-assiged) fid
2191  * XXX: temporary solution to be able to assign fid in ->do_create()
2192  *      till we have fully-functional OST fids
2193  */
2194 struct lu_object *lu_object_anon(const struct lu_env *env,
2195                                  struct lu_device *dev,
2196                                  const struct lu_object_conf *conf)
2197 {
2198         struct lu_fid     fid;
2199         struct lu_object *o;
2200
2201         fid_zero(&fid);
2202         o = lu_object_alloc(env, dev, &fid, conf);
2203
2204         return o;
2205 }
2206 EXPORT_SYMBOL(lu_object_anon);